Tools for lifecycle updates and removed buildings

Makefile

@@ -0,0 +1,22 @@
+.PHONY : all flake8 test
+
+
+FLAKE8_FILES := \
+		filter_buildings.py \
+		find_lifecycle_updates.py \
+		find_removed.py \
+		shared.py \
+		tests/test_filter.py \
+		tests/test_find_lifecycle_updates.py \
+		tests/test_find_removed.py \
+		tests/test_shared.py \
+		;
+
+
+all : flake8 test
+
+flake8 : $(FLAKE8_FILES)
+	flake8 $?
+
+test :
+	python3 -m unittest discover -s tests

filter_buildings.py

@@ -2,71 +2,51 @@
 import json
 import sys
 
-import requests
+import shared
 
 
-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.json()['elements']
-
-
-def load_osm_refs(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)
-    elements = run_overpass_query(query)
+def load_osm_refs(osm_raw):
+    elements = json.loads(osm_raw)['elements']
 
     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
 
 
+def filter_buildings(cadastral_buildings, osm_refs):
+    def in_osm(building):
+        raw_ref = building['properties']['ref:bygningsnr']
+        building_refs = shared.parse_ref(raw_ref)
+        return bool(building_refs & osm_refs)
+
+    return [b for b in cadastral_buildings if not in_osm(b)]
+
+
 def main():
     parser = argparse.ArgumentParser()
     parser.add_argument('--input', required=True)
     parser.add_argument('--output', required=True)
-    parser.add_argument('--municipality', required=True, type=int)
+    parser.add_argument('--municipality', required=True)
     args = parser.parse_args()
 
-    with open(args.input, 'r', encoding='utf-8') as file:
-        data = json.load(file)
-        import_buildings = data['features']
-    print('Loaded {} buildings'.format(len(import_buildings)))
+    muni_id = shared.handle_municipality_argument(args.municipality)
 
-    osm_refs = load_osm_refs(args.municipality)
-    print('Loaded {} unique references from OSM'.format(len(osm_refs)))
+    with open(args.input, 'r', encoding='utf-8') as file:
+        cadastral = shared.parse_cadastral_data(file.read())
+    print(f'Loaded {len(cadastral)} buildings')
 
-    def in_osm(building):
-        raw_ref = building['properties']['ref:bygningsnr']
-        building_refs = parse_ref(raw_ref)
-        return bool(building_refs & osm_refs)
+    osm_raw = shared.load_building_tags(muni_id)
+    osm_refs = load_osm_refs(osm_raw)
+    print(f'Loaded {len(osm_refs)} unique references from OSM')
 
-    missing_in_osm = [b for b in import_buildings if not in_osm(b)]
-    print('Writing {} buildings missing from OSM'.format(len(missing_in_osm)))
+    output = filter_buildings(cadastral, osm_refs)
+    print(f'Writing {len(output)} buildings missing from OSM')
 
     with open(args.output, 'w', encoding='utf-8') as file:
-        geojson = {
-                'type': 'FeatureCollection',
-                'generator': 'filter_buildings.py',
-                'features': missing_in_osm,
-                }
-        json.dump(geojson, file)
+        file.write(shared.format_geojson(output))
 
     return 0
 

find_lifecycle_updates.py

@@ -0,0 +1,99 @@
+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_buildings, osm_by_ref):
+    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
+
+    return updated
+
+
+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()
+
+    muni_id = shared.handle_municipality_argument(args.municipality)
+
+    with open(args.input, 'r', encoding='utf-8') as file:
+        cadastral = shared.parse_cadastral_data(file.read())
+    print(f'Loaded {len(cadastral)} buildings')
+
+    osm_raw = shared.load_building_tags(muni_id)
+    osm_buildings = json.loads(osm_raw)['elements']
+    osm_by_ref = osm_buildings_by_ref(osm_buildings)
+    print(f'Loaded {len(osm_buildings)} buildings from OSM')
+
+    output = find_lifecycle_updates(cadastral, osm_by_ref)
+    print(f'Writing {len(output)} updated buildings')
+    with open(args.output, 'w', encoding='utf-8') as file:
+        file.write(shared.format_geojson(output))
+
+    return 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())

find_removed.py

@@ -0,0 +1,87 @@
+import argparse
+import json
+
+import shared
+
+
+def collect_refs(buildings):
+    refs = set()
+
+    for building in buildings:
+        try:
+            tags = building['tags']
+        except KeyError:
+            tags = building['properties']
+
+        raw_ref = tags['ref:bygningsnr']
+        for ref in shared.parse_ref(raw_ref):
+            refs.add(ref)
+
+    return refs
+
+
+def to_output(building):
+    if building['type'] == 'node':
+        lon = building['lon']
+        lat = building['lat']
+    else:
+        lon = building['center']['lon']
+        lat = building['center']['lat']
+
+    return {
+            'type': 'Feature',
+            'geometry': {
+                'type': 'Point',
+                'coordinates': [
+                    lon,
+                    lat,
+                    ]
+                },
+            'properties': building['tags'],
+            }
+
+
+def find_removed(cadastral_buildings, osm_buildings):
+    cadastral_refs = collect_refs(cadastral_buildings)
+    osm_refs = collect_refs(osm_buildings)
+
+    removed_buildings = []
+    for ref in osm_refs - cadastral_refs:
+        for osm_building in osm_buildings:
+            if ref in collect_refs([osm_building]):
+                try:
+                    removed_buildings.append(to_output(osm_building))
+                except Exception:
+                    print(osm_building)
+                    raise
+
+    return removed_buildings
+
+
+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()
+
+    muni_id = shared.handle_municipality_argument(args.municipality)
+
+    with open(args.input, 'r', encoding='utf-8') as file:
+        cadastral = shared.parse_cadastral_data(file.read())
+    print(f'Loaded {len(cadastral)} buildings')
+
+    osm_raw = shared.load_building_tags(muni_id,
+                                        with_position=True)
+    osm_buildings = json.loads(osm_raw)['elements']
+    print(f'Loaded {len(osm_buildings)} buildings from OSM')
+
+    output = find_removed(cadastral, osm_buildings)
+    print(f'Writing {len(output)} buildings that have been removed')
+
+    with open(args.output, 'w', encoding='utf-8') as file:
+        file.write(shared.format_geojson(output))
+
+
+if __name__ == '__main__':
+    main()