Add tool for finding lifecycle updates

Adds a tool for finding buildings that are planned or under construction
in OSM, but are finished in the cadastral data. Since IG doesn't mean
that construction has actually started, planned and construction are
treated as the same thing.

Author: zidel

Makefile

@@ -3,7 +3,10 @@
 
 FLAKE8_FILES := \
 		filter_buildings.py \
+		find_lifecycle_updates.py \
+		shared.py \
 		tests/test_filter.py \
+		tests/test_find_lifecycle_updates.py \
 		;
 
 

filter_buildings.py

@@ -2,32 +2,7 @@
 import json
 import sys
 
-import requests
-
-
-def parse_ref(raw_ref):
-    return {int(ref) for ref in raw_ref.split(';') if ref}
-
-
-def run_overpass_query(query):
-    overpass_url = "https://overpass-api.de/api/interpreter"
-    params = {'data': query}
-    version = '0.8.0'
-    headers = {'User-Agent': 'building2osm/' + version}
-    request = requests.get(overpass_url,
-                           params=params,
-                           headers=headers)
-    return request.text
-
-
-def load_osm_data(municipality_id):
-    query_fmt = '''[out:json][timeout:60];
-                   (area[ref={}][admin_level=7][place=municipality];)->.county;
-                   nwr["ref:bygningsnr"](area.county);
-                   out tags noids;
-                '''
-    query = query_fmt.format(municipality_id)
-    return run_overpass_query(query)
+import shared
 
 
 def load_osm_refs(osm_raw):
@@ -36,7 +11,7 @@ def load_osm_refs(osm_raw):
     osm_refs = set()
     for element in elements:
         raw_ref = element['tags']['ref:bygningsnr']
-        osm_refs |= parse_ref(raw_ref)
+        osm_refs |= shared.parse_ref(raw_ref)
 
     return osm_refs
 
@@ -47,7 +22,7 @@ def filter_buildings(cadastral_raw, osm_raw):
 
     def in_osm(building):
         raw_ref = building['properties']['ref:bygningsnr']
-        building_refs = parse_ref(raw_ref)
+        building_refs = shared.parse_ref(raw_ref)
         return bool(building_refs & osm_refs)
 
     missing_in_osm = [b for b in cadastral_buildings if not in_osm(b)]
@@ -72,7 +47,7 @@ def main():
         cadastral_raw = file.read()
 
     print('Running Overpass query')
-    osm_raw = load_osm_data(args.municipality)
+    osm_raw = shared.load_building_tags(args.municipality)
 
     print('Filtering buildings')
     output = filter_buildings(cadastral_raw, osm_raw)

find_lifecycle_updates.py

@@ -0,0 +1,105 @@
+import argparse
+import json
+import re
+import sys
+
+import shared
+
+
+def osm_buildings_by_ref(osm_buildings):
+    by_ref = {}
+    for osm_building in osm_buildings:
+        tags = osm_building['tags']
+        raw_ref = tags['ref:bygningsnr']
+        for osm_ref in shared.parse_ref(raw_ref):
+            try:
+                by_ref[osm_ref].append(osm_building)
+            except KeyError:
+                by_ref[osm_ref] = [osm_building]
+
+    return by_ref
+
+
+def cadastral_construction_finished(building):
+    tags = building['properties']
+    if 'STATUS' not in tags:
+        raise RuntimeError
+
+    if re.match('#(RA|IG) .*', tags['STATUS']):
+        return False
+
+    return True
+
+
+def osm_construction_finished(building):
+    tags = building['tags']
+    if 'planned:building' in tags:
+        return False
+    elif 'building' in tags and tags['building'] == 'construction':
+        return False
+    else:
+        return True
+
+
+def has_lifecycle_update(cadastral_building, osm_buildings):
+    for osm_building in osm_buildings:
+        cadastral_done = cadastral_construction_finished(cadastral_building)
+        osm_done = osm_construction_finished(osm_building)
+
+        if cadastral_done and not osm_done:
+            return True
+
+    return False
+
+
+def find_lifecycle_updates(cadastral_raw, osm_raw):
+    cadastral_buildings = json.loads(cadastral_raw)['features']
+
+    osm_buildings = json.loads(osm_raw)['elements']
+    osm_by_ref = osm_buildings_by_ref(osm_buildings)
+
+    updated = []
+    for cadastral_building in cadastral_buildings:
+        cadastral_ref = int(cadastral_building['properties']['ref:bygningsnr'])
+        try:
+            osm_buildings = osm_by_ref[cadastral_ref]
+        except KeyError:
+            # Building is missing from OSM
+            continue
+
+        if has_lifecycle_update(cadastral_building, osm_buildings):
+            updated.append(cadastral_building)
+            continue
+
+    geojson = {
+            'type': 'FeatureCollection',
+            'generator': 'find_lifecycle_updates.py',
+            'features': updated,
+            }
+    return json.dumps(geojson)
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--input', required=True)
+    parser.add_argument('--output', required=True)
+    parser.add_argument('--municipality', required=True)
+    args = parser.parse_args()
+
+    print('Reading cadastral file')
+    with open(args.input, 'r', encoding='utf-8') as file:
+        cadastral_raw = file.read()
+
+    print('Running Overpass query')
+    osm_raw = shared.load_building_tags(args.municipality)
+
+    print('Filtering buildings')
+    output = find_lifecycle_updates(cadastral_raw, osm_raw)
+    with open(args.output, 'w', encoding='utf-8') as file:
+        file.write(output)
+
+    return 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())

shared.py

@@ -0,0 +1,29 @@
+import requests
+
+
+def parse_ref(raw_ref):
+    return {int(ref) for ref in raw_ref.split(';') if ref}
+
+
+def run_overpass_query(query):
+    overpass_url = "https://overpass-api.de/api/interpreter"
+    params = {'data': query}
+    version = '0.8.0'
+    headers = {'User-Agent': 'building2osm/' + version}
+    request = requests.get(overpass_url,
+                           params=params,
+                           headers=headers)
+    request.raise_for_status()
+    return request.text
+
+
+def load_building_tags(municipality_id):
+    query = f'''[out:json][timeout:60];
+                (area[ref={municipality_id}]
+                     [admin_level=7]
+                     [place=municipality];
+                ) -> .county;
+                nwr["ref:bygningsnr"](area.county);
+                out tags noids;
+             '''
+    return run_overpass_query(query)

tests/test_find_lifecycle_updates.py

@@ -0,0 +1,71 @@
+import json
+import unittest
+
+import find_lifecycle_updates
+
+
+def cadastral(ref, status):
+    if status == 'MB':
+        status = '#MB Midlertidig brukstillatelse'
+    elif status == 'IG':
+        status = '#IG Igangsettingstillatelse'
+    else:
+        raise RuntimeError
+
+    return {
+        'properties': {
+            'ref:bygningsnr': str(ref),
+            'STATUS': status,
+            },
+        }
+
+
+def osm(ref, planned=False, construction=False):
+    tags = {
+            'ref:bygningsnr': str(ref),
+            }
+
+    if planned:
+        tags['planned:building'] = 'yes'
+    elif construction:
+        tags['building'] = 'construction'
+
+    return {'tags': tags}
+
+
+class TestFindLifecycleUpdate(unittest.TestCase):
+    def _run_filter(self, cadastral_buildings, osm_buildings):
+        cadastral_raw = json.dumps({'features': cadastral_buildings})
+        osm_raw = json.dumps({'elements': osm_buildings})
+        raw_output = find_lifecycle_updates.find_lifecycle_updates(
+                cadastral_raw,
+                osm_raw)
+        output = json.loads(raw_output)
+        self.assertEqual('FeatureCollection', output['type'])
+        self.assertEqual('find_lifecycle_updates.py', output['generator'])
+        return output['features']
+
+    def test_provisional_use_permit_is_update_from_planned(self):
+        cadastral_buildings = [cadastral(1, status='MB')]
+        osm_buildings = [osm(1, planned=True)]
+        output = self._run_filter(cadastral_buildings, osm_buildings)
+        self.assertEqual(cadastral_buildings, output)
+
+    def test_provisional_use_permit_is_update_from_construction(self):
+        cadastral_buildings = [cadastral(1, status='MB')]
+        osm_buildings = [osm(1, construction=True)]
+        output = self._run_filter(cadastral_buildings, osm_buildings)
+        self.assertEqual(cadastral_buildings, output)
+
+    def test_dont_include_construction_permit_when_osm_has_planned(self):
+        # IG doesn't imply that construction has actually started, so planned
+        # might still be the correct OSM tagging
+        cadastral_buildings = [cadastral(1, status='IG')]
+        osm_buildings = [osm(1, planned=True)]
+        output = self._run_filter(cadastral_buildings, osm_buildings)
+        self.assertEqual([], output)
+
+    def test_ignore_building_missing_from_osm(self):
+        cadastral_buildings = [cadastral(1, status='MB')]
+        output = self._run_filter(cadastral_buildings, [])
+        self.assertEqual([], output)