Commit 866e09cc authored by Marius Kriegerowski's avatar Marius Kriegerowski
Browse files

Added building occupancy rule

parent d2db3301
......@@ -15,6 +15,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/.
from __future__ import annotations
import csv
from collections import defaultdict
......@@ -32,29 +33,21 @@ def group_tags(occupancy_mapping: list[dict[str, str]]) -> dict[str, dict[str, s
return grouped_tag_mapping
def read_csv(fn: str) -> list[dict[str, str]]:
"""Read content from csv into list of dicts."""
with open(fn) as csvfile:
mapping = list(csv.DictReader(csvfile))
return mapping
class OccupancyMapper:
"""Map building tags to `GEM_taxonomy_occupancy`"""
"""Map building tags to `GEM_occupancy`"""
def __init__(self, mapping):
self.mapping = mapping
def tags_to_occupancy(self, osm_tags: dict[dict[str, str]]) -> list[str]:
"""Map `osm_tags` to lists of GEM_taxonomy_occupancy as defined in
def apply(self, osm_tags: dict[dict[str, str]]) -> list[str]:
"""Map `osm_tags` to lists of GEM_occupancy 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_taxonomy_occupancy`s
list of `GEM_occupancy`s
"""
occupancies = []
......@@ -67,9 +60,45 @@ class OccupancyMapper:
return occupancies
@classmethod
def from_csv(cls, fn: str):
def read_csv(cls, fn: str) -> list[dict[str, str]]:
"""Read content from csv into list of dicts."""
with open(fn) as csvfile:
mapping = list(csv.DictReader(csvfile))
return mapping
@classmethod
def from_csv(cls, fn: str) -> OccupancyMapper:
"""Read a csv and initialize a `OccupancyMapper`."""
occupancy_mapping = read_csv(fn)
occupancy_mapping = cls.read_csv(fn)
occupancy_mapping_grouped = group_tags(occupancy_mapping)
return cls(mapping=occupancy_mapping_grouped)
class OverridingOccupancy:
def __init__(self, mapping):
self.mapping = mapping
def apply(self, occupancies: list[str]):
for candidate, occupancy in self.mapping.items():
if candidate in occupancies:
return occupancy
else:
raise KeyError("No overriding occupancy matched")
@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)
#!/usr/bin/env python3
# Copyright (C) 2021:
# Helmholtz-Zentrum Potsdam Deutsches GeoForschungsZentrum GFZ
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or (at
# your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero
# General Public License for more details.
#
# 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 logging
from rabotnik import Rule
from rabotnik.storages.base import StorageBase
from rabotnikobm.occupancy.mapping import OccupancyMapper
logger = logging.getLogger()
class GetBuildingTaxonomy(Rule):
"""A rule to map OSM tags to building occupancies."""
def __init__(self, storage: StorageBase, occupancy_mapper: OccupancyMapper):
self.storage = storage
self.occupancy_mapper = occupancy_mapper
async def evaluate(self, payload: dict):
building_id = payload["building_id"]
logger.debug("Processing building: %s", building_id)
tags = await self.storage.expect_one(
f"SELECT tags FROM osm_building_relations WHERE osm_id={building_id} AND index=0"
)
occupancies = self.occupancy_mapper.apply(tags)
logger.debug("occupancies %s: %s", building_id, occupancies)
return occupancies
......@@ -3,6 +3,8 @@ import logging
import pytest
import rabotnik
from rabotnikobm.occupancy.mapping import OccupancyMapper
logger = logging.getLogger(__name__)
......@@ -48,6 +50,20 @@ def storage_consumer(pytestconfig):
storage.disconnect()
@pytest.fixture
def building_poi_mapper(pytestconfig):
fn_mapping = pytestconfig.rootpath / "data/rules/occupancy/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"
mapper = OccupancyMapper.from_csv(fn_mapping)
yield mapper
def pytest_collection_modifyitems(config, items):
storage_configuration = config.getoption("storage_contributor")
if storage_configuration:
......
#!/usr/bin/env python3
# Copyright (C) 2021:
# Helmholtz-Zentrum Potsdam Deutsches GeoForschungsZentrum GFZ
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or (at
# your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero
# General Public License for more details.
#
# 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.rules.get_building_occupancy import GetBuildingTaxonomy
@pytest.mark.requires_storage
@pytest.mark.asyncio
async def test_get_building_taxonomy(connected_storage, building_poi_mapper):
rule = GetBuildingTaxonomy(storage=connected_storage, occupancy_mapper=building_poi_mapper)
payload = {"building_id": -6744517}
result = await rule.evaluate(payload=payload)
assert result == ["ASS4", "UNDECIDABLE"]
......@@ -16,7 +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/.
from rabotnikobm.occupancy.mapping import group_tags, OccupancyMapper
from rabotnikobm.occupancy.mapping import group_tags, OverridingOccupancy
def test_group_tags():
......@@ -30,25 +30,20 @@ def test_group_tags():
assert grouped == {"a": {0: {}, 1: {}}, "b": {2: {}}}
def test_mapper_from_csv(pytestconfig):
demo_file = pytestconfig.rootpath / "data/rules/occupancy/building_and_PoIs_tags.csv"
mapper = OccupancyMapper.from_csv(demo_file)
assert mapper
def test_occupancy_mapper_building_pois(pytestconfig):
demo_file = pytestconfig.rootpath / "data/rules/occupancy/building_and_PoIs_tags.csv"
mapper = OccupancyMapper.from_csv(demo_file)
def test_mapper_building_pois(building_poi_mapper):
sample_tags = [{"amenity": "community_centre"}, {"amenity": "cafe"}, {"x": "y"}]
assert building_poi_mapper.apply(sample_tags) == ["ASS4", "COM5"]
assert mapper.tags_to_occupancy(sample_tags) == ["ASS4", "COM5"]
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_occupancy_mapper_landuse(pytestconfig):
demo_file = pytestconfig.rootpath / "data/rules/occupancy/landuse_tags.csv"
mapper = OccupancyMapper.from_csv(demo_file)
sample_tags = [{"amenity": "university", "a": "b"}, {"landuse": "brownfield"}, {"x": "y"}]
def test_occupancy_mapper(pytestconfig):
fn_mapping = pytestconfig.rootpath / "data/rules/occupancy/overriding_occupancies.csv"
overriding_mapping = OverridingOccupancy.from_csv(fn_mapping)
assert mapper.tags_to_occupancy(sample_tags) == ["EDU3", "UNDECIDABLE"]
demo_tags = ["ASS1", "COM10"]
occupancy = overriding_mapping.apply(demo_tags)
assert occupancy == "airport"
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment