Add tool for finding removed buildings

Adds a tool for finding OSM buildings that have been removed from the
cadastral registry. From testing in Lillestrøm this is unfortunately not
a reliable indicator of the building actually having been removed, so
the output format is intentionally not suitable for automatic uploading.

Makefile

@@ -4,9 +4,11 @@
 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 \
 		;
 
 

find_removed.py

@@ -0,0 +1,90 @@
+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_raw, osm_raw):
+    cadastral_buildings = json.loads(cadastral_raw)['features']
+    osm_buildings = json.loads(osm_raw)['elements']
+
+    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 json.dumps({
+            'type': 'FeatureCollection',
+            'generator': 'find_removed.py',
+            'features': 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()
+
+    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,
+                                        with_position=True)
+
+    print('Comparing data')
+    output = find_removed(cadastral_raw, osm_raw)
+    with open(args.output, 'w', encoding='utf-8') as file:
+        file.write(output)
+
+
+if __name__ == '__main__':
+    main()

shared.py

@@ -17,13 +17,14 @@ def run_overpass_query(query):
     return request.text
 
 
-def load_building_tags(municipality_id):
+def load_building_tags(municipality_id, with_position=False):
+    center = 'center' if with_position else ''
     query = f'''[out:json][timeout:60];
                 (area[ref={municipality_id}]
                      [admin_level=7]
                      [place=municipality];
                 ) -> .county;
                 nwr["ref:bygningsnr"](area.county);
-                out tags noids;
+                out tags noids {center};
              '''
     return run_overpass_query(query)

tests/test_find_removed.py

@@ -0,0 +1,79 @@
+import json
+import unittest
+
+import find_removed
+
+
+expected_output_point = {
+        'type': 'Feature',
+        'geometry': {
+            'type': 'Point',
+            'coordinates': [
+                11.0,
+                59.0,
+                ]
+            },
+        'properties': {
+            'ref:bygningsnr': '1',
+            'building': 'yes',
+            }
+        }
+
+
+def cadastral(ref):
+    return {'properties': {'ref:bygningsnr': str(ref)}}
+
+
+def osm_node(ref):
+    return {
+            'type': 'node',
+            'lat': 59.0,
+            'lon': 11.0,
+            'tags': {
+                'building': 'yes',
+                'ref:bygningsnr': str(ref),
+                }
+            }
+
+
+def osm_way(ref):
+    return {
+            'type': 'way',
+            'center': {
+                'lat': 59.0,
+                'lon': 11.0,
+                },
+            'tags': {
+                'building': 'yes',
+                'ref:bygningsnr': str(ref),
+                }
+            }
+
+
+class TestFindRemoved(unittest.TestCase):
+    def _find_removed(self, cadastral_buildings, osm_buildings):
+        cadastral_raw = json.dumps({'features': cadastral_buildings})
+        osm_raw = json.dumps({'elements': osm_buildings})
+
+        raw_output = find_removed.find_removed(cadastral_raw, osm_raw)
+        output = json.loads(raw_output)
+        self.assertEqual('FeatureCollection', output['type'])
+        self.assertEqual('find_removed.py', output['generator'])
+        self.assertEqual(list, type(output['features']))
+        return output['features']
+
+    def test_ignore_building_still_in_cadastral_data(self):
+        removed = self._find_removed([cadastral(1)], [osm_node(1)])
+        self.assertEqual([], removed)
+
+    def test_ignore_building_missing_from_osm(self):
+        removed = self._find_removed([cadastral(1)], [])
+        self.assertEqual([], removed)
+
+    def test_output_removed_building_node(self):
+        removed = self._find_removed([], [osm_node(1)])
+        self.assertEqual([expected_output_point], removed)
+
+    def test_output_removed_building_way(self):
+        removed = self._find_removed([], [osm_way(1)])
+        self.assertEqual([expected_output_point], removed)