feature/4837-add-additional-metadata-to-otanalytics-otconfig

OTAnalytics/adapter_ui/abstract_frame_project.py

@@ -15,3 +15,13 @@ def update(self, name: str, start_date: Optional[datetime]) -> None:
     @abstractmethod
     def set_enabled_general_buttons(self, enabled: bool) -> None:
         raise NotImplementedError
+
+
+class AbstractFrameSvzMetadata:
+    @abstractmethod
+    def introduce_to_viewmodel(self) -> None:
+        pass
+
+    @abstractmethod
+    def update(self, metadata: dict) -> None:
+        pass

OTAnalytics/adapter_ui/text_resources.py

@@ -0,0 +1,51 @@
+from dataclasses import dataclass
+from typing import Iterator
+
+COLUMN_NAME = "column_name"
+
+
+@dataclass(frozen=True, order=True)
+class ColumnResource:
+    """
+    Represents a row in a treeview with an id and a dict of values to be shown.
+    The dicts keys represent the columns and the values represent the cell values.
+    """
+
+    id: str
+    values: dict[str, str]
+
+
+class ColumnResources:
+    def __init__(
+        self, resources: list[ColumnResource], lookup_column: str = COLUMN_NAME
+    ) -> None:
+        self._resources = resources
+        self._lookup_column = lookup_column
+        self._to_id = self._create_to_id(resources)
+        self._to_name = self._create_to_name(resources)
+
+    def _create_to_id(self, resources: list[ColumnResource]) -> dict[str, str]:
+        return {
+            resource.values[self._lookup_column]: resource.id for resource in resources
+        }
+
+    def _create_to_name(self, resources: list[ColumnResource]) -> dict[str, str]:
+        return {
+            resource.id: resource.values[self._lookup_column] for resource in resources
+        }
+
+    @property
+    def names(self) -> list[str]:
+        return [resource.values[self._lookup_column] for resource in self._resources]
+
+    def get_name_for(self, resource_id: str) -> str:
+        return self._to_name.get(resource_id, "")
+
+    def get_id_for(self, name: str) -> str:
+        return self._to_id.get(name, "")
+
+    def has(self, resource_id: str) -> bool:
+        return resource_id in [resource.id for resource in self._resources]
+
+    def __iter__(self) -> Iterator[ColumnResource]:
+        return self._resources.__iter__()

OTAnalytics/adapter_ui/ui_texts.py

@@ -0,0 +1,14 @@
+from OTAnalytics.application.project import DirectionOfStationing, WeatherType
+
+DIRECTIONS_OF_STATIONING = {
+    DirectionOfStationing.IN_DIRECTION: "In Stationierungsrichtung",
+    DirectionOfStationing.OPPOSITE_DIRECTION: "Gegen Stationierungsrichtung",
+}
+
+WEATHER_TYPES = {
+    WeatherType.SUN: "sonnig",
+    WeatherType.CLOUD: "bewölkt",
+    WeatherType.RAIN: "Regen",
+    WeatherType.SNOW: "Schnee",
+    WeatherType.FOG: "Nebel",
+}

OTAnalytics/adapter_ui/view_model.py

@@ -10,13 +10,17 @@
 from OTAnalytics.adapter_ui.abstract_frame import AbstractFrame
 from OTAnalytics.adapter_ui.abstract_frame_canvas import AbstractFrameCanvas
 from OTAnalytics.adapter_ui.abstract_frame_filter import AbstractFrameFilter
-from OTAnalytics.adapter_ui.abstract_frame_project import AbstractFrameProject
+from OTAnalytics.adapter_ui.abstract_frame_project import (
+    AbstractFrameProject,
+    AbstractFrameSvzMetadata,
+)
 from OTAnalytics.adapter_ui.abstract_frame_track_plotting import (
     AbstractFrameTrackPlotting,
 )
 from OTAnalytics.adapter_ui.abstract_frame_tracks import AbstractFrameTracks
 from OTAnalytics.adapter_ui.abstract_main_window import AbstractMainWindow
 from OTAnalytics.adapter_ui.abstract_treeview_interface import AbstractTreeviewInterface
+from OTAnalytics.adapter_ui.text_resources import ColumnResources
 from OTAnalytics.domain.date import DateRange
 from OTAnalytics.domain.flow import Flow
 from OTAnalytics.domain.section import Section
@@ -393,3 +397,19 @@ def set_video_control_frame(self, frame: AbstractFrame) -> None:
     @abstractmethod
     def export_road_user_assignments(self) -> None:
         raise NotImplementedError
+
+    @abstractmethod
+    def update_svz_metadata(self, metadata: dict) -> None:
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_directions_of_stationing(self) -> ColumnResources:
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_weather_types(self) -> ColumnResources:
+        raise NotImplementedError
+
+    @abstractmethod
+    def set_svz_metadata_frame(self, frame: AbstractFrameSvzMetadata) -> None:
+        raise NotImplementedError

OTAnalytics/application/application.py

@@ -8,6 +8,7 @@
     ExportFormat,
 )
 from OTAnalytics.application.datastore import Datastore
+from OTAnalytics.application.project import SvzMetadata
 from OTAnalytics.application.state import (
     ActionState,
     FileState,
@@ -619,6 +620,9 @@ def update_project_name(self, name: str) -> None:
     def update_project_start_date(self, start_date: datetime | None) -> None:
         self._project_updater.update_start_date(start_date)
 
+    def update_svz_metadata(self, metadata: SvzMetadata) -> None:
+        self._project_updater.update_svz_metadata(metadata)
+
     def get_track_repository_size(self) -> int:
         return self._track_repository_size.get()
 

OTAnalytics/application/datastore.py

@@ -226,7 +226,7 @@ def __init__(
         self._video_repository = video_repository
         self._track_to_video_repository = track_to_video_repository
         self._progressbar = progressbar
-        self.project = Project(name="", start_date=None)
+        self.project = Project(name="", start_date=None, metadata=None)
 
     def register_video_observer(self, observer: VideoListObserver) -> None:
         self._video_repository.register_videos_observer(observer)

OTAnalytics/application/project.py

@@ -1,18 +1,107 @@
 from dataclasses import dataclass
 from datetime import datetime
+from enum import Enum
 from typing import Optional
 
 NAME: str = "name"
 START_DATE: str = "start_date"
+METADATA: str = "metadata"
+TK_NUMBER: str = "tk_number"
+COUNTING_LOCATION_NUMBER: str = "counting_location_number"
+DIRECTION: str = "direction"
+WEATHER: str = "weather"
+REMARK: str = "remark"
+COORDINATE_X: str = "coordinate_x"
+COORDINATE_Y: str = "coordinate_y"
+
+
+class DirectionOfStationingParseError(Exception):
+    pass
+
+
+class DirectionOfStationing(Enum):
+    IN_DIRECTION = "1"
+    OPPOSITE_DIRECTION = "2"
+
+    def serialize(self) -> str:
+        return self.value
+
+    @staticmethod
+    def parse(direction: str) -> "DirectionOfStationing":
+        match direction:
+            case DirectionOfStationing.IN_DIRECTION.value:
+                return DirectionOfStationing.IN_DIRECTION
+            case DirectionOfStationing.OPPOSITE_DIRECTION.value:
+                return DirectionOfStationing.OPPOSITE_DIRECTION
+            case _:
+                raise DirectionOfStationingParseError(
+                    f"Unable to parse not existing direction '{direction}'"
+                )
+
+
+class WeatherTypeParseError(Exception):
+    pass
+
+
+class WeatherType(Enum):
+    SUN = "1"
+    CLOUD = "2"
+    RAIN = "3"
+    SNOW = "4"
+    FOG = "5"
+
+    def serialize(self) -> str:
+        return self.value
+
+    @staticmethod
+    def parse(weather_type: str) -> "WeatherType":
+        for type in [
+            WeatherType.SUN,
+            WeatherType.CLOUD,
+            WeatherType.RAIN,
+            WeatherType.SNOW,
+            WeatherType.FOG,
+        ]:
+            if type.value == weather_type:
+                return type
+        raise WeatherTypeParseError(
+            f"Unable to parse not existing weather type '{weather_type}'"
+        )
+
+
+@dataclass
+class SvzMetadata:
+    tk_number: str | None
+    counting_location_number: str | None
+    direction: DirectionOfStationing | None
+    weather: WeatherType | None
+    remark: str | None
+    coordinate_x: str | None
+    coordinate_y: str | None
+
+    def to_dict(self) -> dict:
+        return {
+            TK_NUMBER: self.tk_number if self.tk_number else None,
+            COUNTING_LOCATION_NUMBER: (
+                self.counting_location_number if self.counting_location_number else None
+            ),
+            DIRECTION: self.direction.serialize() if self.direction else None,
+            WEATHER: self.weather.serialize() if self.weather else None,
+            REMARK: self.remark if self.remark else None,
+            COORDINATE_X: self.coordinate_x if self.coordinate_x else None,
+            COORDINATE_Y: self.coordinate_y if self.coordinate_y else None,
+        }
 
 
 @dataclass
 class Project:
     name: str
-    start_date: Optional[datetime]
+    start_date: Optional[datetime] = None
+    metadata: Optional[SvzMetadata] = None
 
     def to_dict(self) -> dict:
         return {
             NAME: self.name,
             START_DATE: self.start_date.timestamp() if self.start_date else None,
+            METADATA: self.metadata.to_dict() if self.metadata else None,
         }

OTAnalytics/application/use_cases/load_otconfig.py

@@ -42,7 +42,9 @@ def load(self, file: Path) -> None:
         self._clear_repositories()
         config = self._config_parser.parse(file)
         try:
-            self._update_project(config.project.name, config.project.start_date)
+            self._update_project(
+                config.project.name, config.project.start_date, config.project.metadata
+            )
             self._add_videos.add(config.videos)
             self._add_sections.add(config.sections)
             self._add_flows.add(config.flows)

OTAnalytics/application/use_cases/reset_project_config.py

@@ -13,4 +13,4 @@ def __init__(self, update_project: ProjectUpdater):
 
     def __call__(self) -> None:
         """Reset the project configuration."""
-        self._update_project("", None)
+        self._update_project("", None, None)

OTAnalytics/application/use_cases/update_project.py

@@ -2,7 +2,7 @@
 from typing import Optional
 
 from OTAnalytics.application.datastore import Datastore
-from OTAnalytics.application.project import Project
+from OTAnalytics.application.project import Project, SvzMetadata
 from OTAnalytics.domain.observer import OBSERVER, Subject
 
 
@@ -13,22 +13,30 @@ def __init__(self, datastore: Datastore) -> None:
         self._datastore = datastore
         self._subject = Subject[Project]()
 
-    def __call__(self, name: str, start_date: Optional[datetime]) -> None:
-        project = Project(name=name, start_date=start_date)
+    def __call__(
+        self, name: str, start_date: Optional[datetime], metadata: Optional[SvzMetadata]
+    ) -> None:
+        project = Project(name=name, start_date=start_date, metadata=metadata)
         self._datastore.project = project
         self._subject.notify(project)
 
     def update_name(self, name: str) -> None:
         old_project = self._datastore.project
-        new_project = Project(name, old_project.start_date)
+        new_project = Project(name, old_project.start_date, old_project.metadata)
         self._datastore.project = new_project
         self._subject.notify(new_project)
 
     def update_start_date(self, start_date: Optional[datetime]) -> None:
         old_project = self._datastore.project
-        new_project = Project(old_project.name, start_date)
+        new_project = Project(old_project.name, start_date, old_project.metadata)
         self._datastore.project = new_project
         self._subject.notify(new_project)
 
     def register(self, observer: OBSERVER[Project]) -> None:
         self._subject.register(observer)
+
+    def update_svz_metadata(self, metadata: SvzMetadata) -> None:
+        old_project = self._datastore.project
+        new_project = Project(old_project.name, old_project.start_date, metadata)
+        self._datastore.project = new_project
+        self._subject.notify(new_project)

OTAnalytics/plugin_parser/otconfig_parser.py

@@ -14,7 +14,19 @@
     StartDateMissing,
 )
 from OTAnalytics.application.parser.flow_parser import FlowParser
-from OTAnalytics.application.project import Project
+from OTAnalytics.application.project import (
+    COORDINATE_X,
+    COORDINATE_Y,
+    COUNTING_LOCATION_NUMBER,
+    DIRECTION,
+    REMARK,
+    TK_NUMBER,
+    WEATHER,
+    DirectionOfStationing,
+    Project,
+    SvzMetadata,
+    WeatherType,
+)
 from OTAnalytics.domain import flow, section, video
 from OTAnalytics.domain.flow import Flow
 from OTAnalytics.domain.section import Section
@@ -119,7 +131,28 @@ def _parse_project(self, data: dict) -> Project:
         _validate_data(data, [project.NAME, project.START_DATE])
         name = data[project.NAME]
         start_date = datetime.fromtimestamp(data[project.START_DATE], timezone.utc)
-        return Project(name=name, start_date=start_date)
+        svz_metadata = None
+        if svz_data := data.get(project.METADATA):
+            svz_metadata = self._parse_svz_metadata(svz_data)
+        return Project(name=name, start_date=start_date, metadata=svz_metadata)
+
+    def _parse_svz_metadata(self, data: dict) -> SvzMetadata:
+        tk_number = data[TK_NUMBER]
+        counting_location_number = data[COUNTING_LOCATION_NUMBER]
+        direction = DirectionOfStationing.parse(data[DIRECTION])
+        weather = WeatherType.parse(data[WEATHER])
+        remark = data[REMARK]
+        coordinate_x = data[COORDINATE_X]
+        coordinate_y = data[COORDINATE_Y]
+        return SvzMetadata(
+            tk_number=tk_number,
+            counting_location_number=counting_location_number,
+            direction=direction,
+            weather=weather,
+            remark=remark,
+            coordinate_x=coordinate_x,
+            coordinate_y=coordinate_y,
+        )
 
     def _parse_analysis(self, data: dict, base_folder: Path) -> AnalysisConfig:
         _validate_data(