Commit 3dce5c7b authored by Marius Kriegerowski's avatar Marius Kriegerowski
Browse files

fix rules 1 and 2

parent 4edb8fde
......@@ -61,12 +61,30 @@ async def start_rabotnik_obm(message_bus):
await message_bus.subscribe("building",
async def start_rabotnik_gem_occupancy(
message_bus, storage_consumer, storage_contributor, n_processes_max
rules = [
GetBuilding(storage_consumer, storage_contributor),
GetFloorspace(storage_consumer, storage_contributor),
rules = Assembly(rules=rules, n_processes_max=n_processes_max)
await message_bus.subscribe("building",
async def start_rabotnik(args):"start rabotnik")
message_bus = await connected_message_bus(args.config_message_bus)
await start_rabotnik_obm(message_bus)
# await start_rabotnik_obm(
# message_bus, storage_consumer, storage_contributor, args.n_processes_max
# )
def main():
......@@ -17,18 +17,19 @@
# along with this program. If not, see
from __future__ import annotations
import logging
import csv
import logging
from rabotnik import Rule
from rabotnik.storages.base import StorageBase
from rabotnikobm.rules.gem_occupancy.mapping import OccupancyMapper, TagStatistics, GEMTag
logger = logging.getLogger()
class TagResult(Exception):
"""Raise this exception with the identified occupancy"""
def __init__(self, tag: GEMTag):
......@@ -36,7 +37,6 @@ class TagResult(Exception):
class OverridingOccupancy:
"""Takes precedence over other mappings.
If `OverridingOccupancy.apply` returns a result this will be the designated occupancy.
......@@ -66,15 +66,23 @@ class OverridingOccupancy:
def check_exactly_one_unique_tag(occupancies: TagStatistics):
"""raises `TagResult` if there is exactly one unique type"""
"""raises `TagResult` if there is exactly one unique group"""
if occupancies.exactly_one_unique_tag():
raise TagResult(occupancies.tags[0])
def check_one_unique_sub_type(occupancies: TagStatistics):
if occupancies.exactly_one_unique_type() and occupancies.number_of_sub_groups == 1:
# TODO: find raise TagResult(occupancies....)
raise TagResult("ASDASDF")
def check_one_unique_sub_group(occupancies: TagStatistics):
if occupancies.number_of_unique_groups == 1 and occupancies.number_of_unique_subgroups <= 2:
raise TagResult(
class GetBuildingOccupancy(Rule):
......@@ -29,6 +29,10 @@ from typing import Optional
REGEX_GEM_GROUP = r"(?P<group>^[A-Z]*)(?P<sub_group>[0-9]*)(?P<sub_sub_group>[A-Z]*$)"
class GEMTagException(Exception):
class GEMTag:
"""Represents a single GEM classification
......@@ -70,25 +74,69 @@ class TagStatistics(
def exactly_one_unique_tag(self) -> bool:
return self.number_of_unique_tags == 1
def subsubgroups(self):
"""list of available sub_sub_groups without None"""
return list(filter(None, self.subsubgroups_counter.keys()))
def subgroups(self):
"""list of available sub_groups without None"""
return list(filter(None, self.subgroups_counter.keys()))
def number_of_unique_tags(self) -> int:
return len(self.tags_counter)
return len(set(self.tags_counter))
def number_of_unique_types(self) -> int:
return len(self.tags_counter)
def number_of_unique_groups(self) -> int:
return len(list(self.groups_counter))
def number_of_unique_subtypes(self) -> int:
return len(self.subtypes_counter)
def number_of_unique_subgroups(self) -> int:
return len(self.subgroups_counter)
def unique_group(self) -> str:
"""Get the unique group ID. In case of multiple available groups, raise GEMTagException."""
if self.number_of_unique_groups > 1:
raise GEMTagException(f"{self} has more than unique group.")
return list(self.groups_counter.keys())[0]
def unique_sub_group(self) -> Optional[int]:
"""Get the unique group ID. In case of multiple available groups, raise GEMTagException."""
if len(self.subgroups) > 1:
raise GEMTagException(f"{self} has more than unique sub_group.")
if len(self.subgroups) == 0:
return None
return self.subgroups[0]
def unique_sub_sub_group(self) -> Optional[str]:
"""Get the unique group ID. In case of multiple available groups, raise GEMTagException."""
if len(self.subsubgroups) > 1:
raise GEMTagException(f"{self} has more than unique sub_sub_group.")
if len(self.subsubgroups) == 0:
return None
return self.subsubgroups[0]
def from_strings(cls, tags: list[str]) -> TagStatistics:
......@@ -99,17 +147,21 @@ class TagStatistics(
def from_tags(cls, tags: list[GEMTag]) -> TagStatistics:
tags_counter = Counter(tags)
types = [ for tag in tags]
types_counter = Counter(types)
groups = [ for tag in tags]
groups_counter = Counter(groups)
subgroups = [tag.sub_group for tag in tags]
subgroups_counter = Counter(subgroups)
subtypes = [tag.sub_group for tag in tags]
subtypes_counter = Counter(subtypes)
subsubgroups = [tag.sub_sub_group for tag in tags]
subsubgroups_counter = Counter(subsubgroups)
return cls(
......@@ -23,6 +23,7 @@ from rabotnikobm.rules.gem_occupancy.get_building_occupancy import (
from rabotnikobm.rules.gem_occupancy.mapping import GEMTag
......@@ -57,15 +58,27 @@ def test_overriding_occupancy(overriding_occupancies):
def test_overriding_occupancy_unknown():
"""Test Rule #1 """
with pytest.raises(AssertionError):
TagStatistics.from_strings(["unknown tag"])
def test_unique_tags():
"""Rule #2"""
tags = TagStatistics.from_tags([GEMTag.from_string("COM"), GEMTag.from_string("COM")])
tags = TagStatistics.from_strings(["COM", "COM"])
with pytest.raises(TagResult) as e:
assert == "COM"
def test_rule2():
"""Rule #2"""
tags = TagStatistics.from_strings(["RES", "RES1"])
with pytest.raises(TagResult) as e:
assert == "RES"
assert e.value.tag.sub_group == 1
......@@ -16,18 +16,41 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see
from rabotnikobm.rules.gem_occupancy.mapping import group_tags, GEMTag, TagStatistics
from rabotnikobm.rules.gem_occupancy.mapping import (
import pytest
def test_tag_statistics():
stats = TagStatistics.from_tags([GEMTag.from_string("COM"), GEMTag.from_string("COM")])
stats = TagStatistics.from_strings(["COM", "COM"])
assert stats.exactly_one_unique_tag() is True
stats = TagStatistics.from_tags([GEMTag.from_string("COM"), GEMTag.from_string("COM1")])
stats = TagStatistics.from_strings(["COM", "COM1"])
assert stats.exactly_one_unique_tag() is False
def test_tag_statistics_unique_getters():
stats = TagStatistics.from_strings(["COM", "COM"])
assert stats.unique_group == "COM"
assert stats.unique_sub_group is None
stats = TagStatistics.from_strings(["COM", "COM1"])
assert stats.unique_group == "COM"
assert stats.unique_sub_group is 1
assert stats.unique_sub_sub_group is None
stats = TagStatistics.from_strings(["COM", "COM1", "RES2"])
with pytest.raises(GEMTagException):
assert stats.unique_group != "COM"
with pytest.raises(GEMTagException):
assert stats.unique_sub_group != 1
def test_tag_hash():
assert hash(GEMTag.from_string("COM")) == hash(GEMTag.from_string("COM"))
assert hash(GEMTag.from_string("COM1")) != hash(GEMTag.from_string("COM"))
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