refactor

data/rules/occupancy/building_and_PoIs_tags.csv → rabotnikobm/rules/gem_occupancy/data/building_and_PoIs_tags.csv

-key,value,taginfo_description,GEM_occupancy,comments
+key,value,taginfo_description,occupancy,comments
 aerialway,station,"A station where passengers and/or goods can enter and/or leave the aerialway (forms of transport that use wires, including cable-cars, chair-lifts and drag-lifts)",COM,COM contains other kinds of transport stations but not one specific for aerialways
 aeroway,hangar,A large airport building with extensive floor areas for housing aircraft or spacecraft,COM10,Unclear if hangar is treated as standard part of an airport or not
 aeroway,terminal,An airport passenger building,COM10,

data/rules/occupancy/landuse_tags.csv → rabotnikobm/rules/gem_occupancy/data/landuse_tags.csv

-key,value,taginfo_description,GEM_occupancy,comments
+key,value,taginfo_description,occupancy,comments
 amenity,university,"An educational institution designed for instruction, examination, or both, of students in many branches of advanced learning.",EDU3,
 amenity,school,A primary or secondary school (pupils typically aged 6 to 18).,EDU2,
 amenity,college,"A place for further education, a post-secondary education institution which is not a University",EDU3,

rabotnikobm/rules/get_building_occupancy.py → rabotnikobm/rules/gem_occupancy/get_building_occupancy.py

@@ -16,15 +16,48 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program. If not, see http://www.gnu.org/licenses/.
 
+from __future__ import annotations
+
 import logging
+import csv
+from typing import Optional
 
 from rabotnik import Rule
 from rabotnik.storages.base import StorageBase
-from rabotnikobm.occupancy.mapping import OccupancyMapper
+from rabotnikobm.rules.gem_occupancy.mapping import OccupancyMapper
 
 logger = logging.getLogger()
 
 
+class OverridingOccupancy:
+
+    """Takes precedence over other mappings.
+
+    If `OverridingOccupancy.apply` returns a result this will be the designated occupancy.
+    Otherwise will return None.
+    """
+
+    def __init__(self, mapping):
+        self.mapping = mapping
+
+    def apply(self, occupancies: list[str]) -> Optional[str]:
+        """Apply the loaded mapping to a list of `occupancies`."""
+
+        occupancies = set(occupancies)
+        for candidate in self.mapping.keys():
+            if candidate in occupancies:
+                return candidate
+
+    @classmethod
+    def from_csv(cls, fn: str) -> OverridingOccupancy:
+        """Read a csv and initialize a `OverridingOccupancy` instance."""
+
+        with open(fn) as csvfile:
+            occupancy_mapping = {k.strip(): v.strip() for (k, v) in csv.reader(csvfile)}
+
+        return cls(mapping=occupancy_mapping)
+
+
 class GetBuildingOccupancy(Rule):
     """A rule to map OSM tags to building occupancies."""
 
@@ -32,7 +65,7 @@ class GetBuildingOccupancy(Rule):
         self.storage = storage
         self.occupancy_mapper = occupancy_mapper
 
-    async def evaluate(self, payload: dict):
+    async def evaluate(self, payload: dict) -> list[str]:
         building_id = payload["building_id"]
         logger.debug("Processing building: %s", building_id)
         tags = await self.storage.expect_one(

rabotnikobm/occupancy/mapping.py → rabotnikobm/rules/gem_occupancy/mapping.py

@@ -36,19 +36,19 @@ def group_tags(occupancy_mapping: list[dict[str, str]]) -> dict[str, dict[str, s
 
 class OccupancyMapper:
 
-    """Map osm tags to `GEM_occupancy` categories."""
+    """Map osm tags to `occupancy` categories."""
 
     def __init__(self, mapping):
         self.mapping = mapping
 
     def apply(self, osm_tags: dict[dict[str, str]]) -> list[str]:
-        """Map `osm_tags` to lists of GEM_occupancy as defined in
+        """Map `osm_tags` to lists of occupancies as defined in
         building_and_POIs_tags.csv and landuse_tags.csv. Tags may be duplicated.
 
         Args:
             osm_tags: list of OSM building tag strings associated with a specific building
         Returns:
-            list of `GEM_occupancy`s
+            list of `occupancy`s
         """
 
         occupancies = []
@@ -56,7 +56,7 @@ class OccupancyMapper:
             for key, value in osm_tag.items():
                 occupancy = self.mapping.get(key, {}).get(value, None)
                 if occupancy is not None:
-                    occupancies.append(occupancy["GEM_occupancy"])
+                    occupancies.append(occupancy["occupancy"])
 
         return occupancies
 
@@ -77,41 +77,3 @@ class OccupancyMapper:
         occupancy_mapping_grouped = group_tags(occupancy_mapping)
 
         return cls(mapping=occupancy_mapping_grouped)
-
-
-class OverridingOccupancy:
-
-    """Takes precedence over other mappings.
-
-    If `OverridingOccupancy.apply` returns a result this will be the designated occupancy.
-    Otherwise will return None.
-    """
-
-    def __init__(self, mapping):
-        self.mapping = mapping
-
-    def apply(self, occupancies: list[str]):
-        """Apply the loaded mapping to a list of `occupancies`."""
-
-        for candidate, occupancy in self.mapping.items():
-            if candidate in occupancies:
-                return occupancy
-
-        else:
-            return None
-
-    @classmethod
-    def read_csv(cls, fn: str) -> dict[str, str]:
-        """Read content from csv into list of dicts."""
-
-        with open(fn) as csvfile:
-            mapping = {k.strip(): v.strip() for (k, v) in csv.reader(csvfile)}
-
-        return mapping
-
-    @classmethod
-    def from_csv(cls, fn: str) -> OverridingOccupancy:
-        """Read a csv and initialize a `OverridingOccupancy` instance."""
-
-        occupancy_mapping = cls.read_csv(fn)
-        return cls(mapping=occupancy_mapping)

tests/conftest.py

@@ -3,7 +3,7 @@ import logging
 import pytest
 import rabotnik
 
-from rabotnikobm.occupancy.mapping import OccupancyMapper
+from rabotnikobm.rules.gem_occupancy.mapping import OccupancyMapper
 
 logger = logging.getLogger(__name__)
 
@@ -51,15 +51,20 @@ def storage_consumer(pytestconfig):
 
 
 @pytest.fixture
-def building_poi_mapper(pytestconfig):
-    fn_mapping = pytestconfig.rootpath / "data/rules/occupancy/building_and_PoIs_tags.csv"
+def gem_data_path(pytestconfig):
+    yield pytestconfig.rootpath / "rabotnikobm/rules/gem_occupancy/data"
+
+
+@pytest.fixture
+def building_poi_mapper(gem_data_path):
+    fn_mapping = gem_data_path / "building_and_PoIs_tags.csv"
     mapper = OccupancyMapper.from_csv(fn_mapping)
     yield mapper
 
 
 @pytest.fixture
-def landuse_mapper(pytestconfig):
-    fn_mapping = pytestconfig.rootpath / "data/rules/occupancy/landuse_tags.csv"
+def landuse_mapper(gem_data_path):
+    fn_mapping = gem_data_path / "landuse_tags.csv"
     mapper = OccupancyMapper.from_csv(fn_mapping)
     yield mapper
 

tests/test_get_building_occupancy.py → tests/test_get_gem_occupancy.py

@@ -17,15 +17,37 @@
 # along with this program. If not, see http://www.gnu.org/licenses/.
 
 import pytest
+from rabotnikobm.rules.gem_occupancy.get_building_occupancy import (
+    OverridingOccupancy,
+    GetBuildingOccupancy,
+)
 
-from rabotnikobm.rules.get_building_occupancy import GetBuildingOccupancy
+
+@pytest.fixture()
+def overriding_occupancies(gem_data_path):
+    fn_mapping = gem_data_path / "overriding_occupancies.csv"
+    overriding_occupancies = OverridingOccupancy.from_csv(fn_mapping)
+
+    yield overriding_occupancies
 
 
 @pytest.mark.requires_storage
 @pytest.mark.asyncio
-async def test_rule_get_building_occupancy(connected_storage, building_poi_mapper):
-    rule = GetBuildingOccupancy(storage=connected_storage, occupancy_mapper=building_poi_mapper)
+async def test_rule_get_building_occupancy(storage_consumer, building_poi_mapper):
+    rule = GetBuildingOccupancy(storage=storage_consumer, occupancy_mapper=building_poi_mapper)
 
     payload = {"building_id": -6744517}
     result = await rule.evaluate(payload=payload)
     assert result == ["ASS4", "UNDECIDABLE"]
+
+
+def test_overriding_occupancy(overriding_occupancies):
+    demo_tags = ["ASS1", "COM10"]
+    occupancy = overriding_occupancies.apply(demo_tags)
+    assert occupancy == "COM10"
+
+
+def test_overriding_occupancy_unknown(overriding_occupancies):
+    demo_tags = ["unknown tag"]
+    occupancy = overriding_occupancies.apply(demo_tags)
+    assert occupancy is None

tests/test_occupancy_mapping.py

@@ -16,16 +16,7 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program. If not, see http://www.gnu.org/licenses/.
 
-import pytest
-from rabotnikobm.occupancy.mapping import group_tags, OverridingOccupancy
-
-
-@pytest.fixture()
-def overriding_occupancies(pytestconfig):
-    fn_mapping = pytestconfig.rootpath / "data/rules/occupancy/overriding_occupancies.csv"
-    overriding_occupancies = OverridingOccupancy.from_csv(fn_mapping)
-
-    yield overriding_occupancies
+from rabotnikobm.rules.gem_occupancy.mapping import group_tags
 
 
 def test_group_tags():
@@ -47,15 +38,3 @@ def test_mapper_building_pois(building_poi_mapper):
 def test_mapper_landuse(landuse_mapper):
     sample_tags = [{"amenity": "university", "a": "b"}, {"landuse": "brownfield"}, {"x": "y"}]
     assert landuse_mapper.apply(sample_tags) == ["EDU3", "UNDECIDABLE"]
-
-
-def test_overriding_occupancy(overriding_occupancies):
-    demo_tags = ["ASS1", "COM10"]
-    occupancy = overriding_occupancies.apply(demo_tags)
-    assert occupancy == "airport"
-
-
-def test_overriding_occupancy_unknown(overriding_occupancies):
-    demo_tags = ["unknown tag"]
-    occupancy = overriding_occupancies.apply(demo_tags)
-    assert occupancy is None