Merge pull request #619 from OpenTrafficCam/feature/7185-refactor-use…

…-case-to-add-section-with-its-geometry

feature/7185-refactor-use-case-to-add-section-with-its-geometry

OTAnalytics/adapter_ui/view_model.py

@@ -1,7 +1,7 @@
 from abc import ABC, abstractmethod
 from datetime import datetime
 from pathlib import Path
-from typing import Callable, Iterable, Optional
+from typing import Iterable, Optional
 
 from OTAnalytics.adapter_ui.abstract_button_quick_save_config import (
     AbstractButtonQuickSaveConfig,
@@ -24,6 +24,7 @@
 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.application.use_cases.editor.section_editor import MetadataProvider
 from OTAnalytics.domain.date import DateRange
 from OTAnalytics.domain.flow import Flow, FlowId
 from OTAnalytics.domain.section import Section
@@ -32,12 +33,6 @@
 
 DISTANCES: str = "distances"
 
-MetadataProvider = Callable[[], dict]
-
-
-class MissingCoordinate(Exception):
-    pass
-
 
 class ViewModel(ABC):
 

OTAnalytics/application/use_cases/editor/section_editor.py

@@ -0,0 +1,138 @@
+import contextlib
+from typing import Callable
+
+from OTAnalytics.application.application import CancelAddSection
+from OTAnalytics.application.use_cases.section_repository import AddSection
+from OTAnalytics.domain import geometry
+from OTAnalytics.domain.geometry import coordinate_from_tuple
+from OTAnalytics.domain.section import (
+    ID,
+    NAME,
+    RELATIVE_OFFSET_COORDINATES,
+    Area,
+    LineSection,
+    MissingSection,
+    Section,
+    SectionId,
+    SectionRepository,
+)
+from OTAnalytics.domain.types import EventType
+
+MetadataProvider = Callable[[], dict]
+
+
+class MissingCoordinate(Exception):
+    pass
+
+
+def validate_coordinates(coordinates: list[tuple[int, int]]) -> None:
+    if not coordinates:
+        raise MissingCoordinate("First coordinate is missing")
+    elif len(coordinates) == 1:
+        raise MissingCoordinate("Second coordinate is missing")
+
+
+def validate_section_information(
+    meta_data: dict, coordinates: list[tuple[int, int]]
+) -> None:
+    validate_coordinates(coordinates)
+    if not meta_data:
+        raise ValueError("Metadata of line_section are not defined")
+
+
+class CreateSectionId:
+    def __init__(self, section_repository: SectionRepository) -> None:
+        self._section_repository = section_repository
+
+    def create_id(self) -> SectionId:
+        return self._section_repository.get_id()
+
+
+class AddNewSection:
+    def __init__(self, create_section_id: CreateSectionId, add_section: AddSection):
+        self._create_section_id = create_section_id
+        self._add_section = add_section
+
+    def add_new_section(
+        self,
+        coordinates: list[tuple[int, int]],
+        is_area_section: bool,
+        get_metadata: MetadataProvider,
+    ) -> Section | None:
+        validate_coordinates(coordinates)
+        with contextlib.suppress(CancelAddSection):
+            return self.__create_section(coordinates, is_area_section, get_metadata)
+        return None
+
+    def __create_section(
+        self,
+        coordinates: list[tuple[int, int]],
+        is_area_section: bool,
+        get_metadata: MetadataProvider,
+    ) -> Section:
+        metadata = self.__get_metadata(get_metadata)
+        relative_offset_coordinates_enter = metadata[RELATIVE_OFFSET_COORDINATES][
+            EventType.SECTION_ENTER.serialize()
+        ]
+        section: Section | None = None
+        if is_area_section:
+            section = Area(
+                id=self._create_section_id.create_id(),
+                name=metadata[NAME],
+                relative_offset_coordinates={
+                    EventType.SECTION_ENTER: geometry.RelativeOffsetCoordinate(
+                        **relative_offset_coordinates_enter
+                    )
+                },
+                plugin_data={},
+                coordinates=[
+                    coordinate_from_tuple(coordinate) for coordinate in coordinates
+                ],
+            )
+        else:
+            section = LineSection(
+                id=self._create_section_id.create_id(),
+                name=metadata[NAME],
+                relative_offset_coordinates={
+                    EventType.SECTION_ENTER: geometry.RelativeOffsetCoordinate(
+                        **relative_offset_coordinates_enter
+                    )
+                },
+                plugin_data={},
+                coordinates=[
+                    coordinate_from_tuple(coordinate) for coordinate in coordinates
+                ],
+            )
+        if section is None:
+            raise TypeError("section has to be LineSection or Area, but is None")
+        self._add_section(section)
+        return section
+
+    def __get_metadata(self, get_metadata: MetadataProvider) -> dict:
+        metadata = get_metadata()
+        while (
+            (not metadata)
+            or (NAME not in metadata)
+            or (not self._add_section.is_section_name_valid(metadata[NAME]))
+            or (RELATIVE_OFFSET_COORDINATES not in metadata)
+        ):
+            metadata = get_metadata()
+        return metadata
+
+
+class UpdateSectionCoordinates:
+    def __init__(self, section_repository: SectionRepository) -> None:
+        self._section_repository = section_repository
+
+    def update(self, meta_data: dict, coordinates: list[tuple[int, int]]) -> SectionId:
+        validate_section_information(meta_data, coordinates)
+        section_id = SectionId(meta_data[ID])
+        if not (section := self._section_repository.get(section_id)):
+            raise MissingSection(
+                f"Could not update section '{section_id.serialize()}' after editing"
+            )
+        section.update_coordinates(
+            [coordinate_from_tuple(coordinate) for coordinate in coordinates]
+        )
+        self._section_repository.update(section)
+        return section_id

OTAnalytics/domain/geometry.py

@@ -23,6 +23,10 @@ def to_list(self) -> list[float]:
         return [self.x, self.y]
 
 
+def coordinate_from_tuple(coordinate: tuple[float, float]) -> Coordinate:
+    return Coordinate(coordinate[0], coordinate[1])
+
+
 @dataclass(frozen=True)
 class Line(DataclassValidation):
     """A `Line` is a geometry that can consist of multiple line segments.

OTAnalytics/plugin_ui/customtkinter_gui/dummy_viewmodel.py

@@ -47,11 +47,7 @@
     DIRECTIONS_OF_STATIONING,
     WEATHER_TYPES,
 )
-from OTAnalytics.adapter_ui.view_model import (
-    MetadataProvider,
-    MissingCoordinate,
-    ViewModel,
-)
+from OTAnalytics.adapter_ui.view_model import ViewModel
 from OTAnalytics.application.analysis.traffic_counting_specification import (
     CountingSpecificationDto,
 )
@@ -93,6 +89,11 @@
 from OTAnalytics.application.use_cases.config import MissingDate
 from OTAnalytics.application.use_cases.config_has_changed import NoExistingConfigFound
 from OTAnalytics.application.use_cases.cut_tracks_with_sections import CutTracksDto
+from OTAnalytics.application.use_cases.editor.section_editor import (
+    AddNewSection,
+    MetadataProvider,
+    UpdateSectionCoordinates,
+)
 from OTAnalytics.application.use_cases.export_events import (
     EventListExporter,
     ExporterNotFoundError,
@@ -124,11 +125,8 @@
 from OTAnalytics.domain.section import (
     COORDINATES,
     ID,
-    NAME,
-    RELATIVE_OFFSET_COORDINATES,
     Area,
     LineSection,
-    MissingSection,
     Section,
     SectionId,
     SectionListObserver,
@@ -349,12 +347,16 @@ def __init__(
         name_generator: FlowNameGenerator,
         event_list_export_formats: dict,
         show_svz: bool,
+        add_new_section: AddNewSection,
+        update_section_coordinates: UpdateSectionCoordinates,
     ) -> None:
         self._application = application
         self._flow_parser: FlowParser = flow_parser
         self._name_generator = name_generator
         self._event_list_export_formats = event_list_export_formats
         self._show_svz = show_svz
+        self._add_new_section = add_new_section
+        self._update_section_coordinates = update_section_coordinates
         self._window: Optional[AbstractMainWindow] = None
         self._frame_project: Optional[AbstractFrameProject] = None
         self._frame_tracks: Optional[AbstractFrameTracks] = None
@@ -949,102 +951,25 @@ def add_new_section(
         is_area_section: bool,
         get_metadata: MetadataProvider,
     ) -> None:
-        if not coordinates:
-            raise MissingCoordinate("First coordinate is missing")
-        elif len(coordinates) == 1:
-            raise MissingCoordinate("Second coordinate is missing")
-        with contextlib.suppress(CancelAddSection):
-            section = self.__create_section(coordinates, is_area_section, get_metadata)
+        section = self._add_new_section.add_new_section(
+            coordinates=coordinates,
+            is_area_section=is_area_section,
+            get_metadata=get_metadata,
+        )
+        if section:
             if not section.name.startswith(CUTTING_SECTION_MARKER):
                 logger().info(f"New section created: {section.id}")
                 self._update_selected_sections([section.id])
         self._finish_action()
 
-    def __create_section(
-        self,
-        coordinates: list[tuple[int, int]],
-        is_area_section: bool,
-        get_metadata: MetadataProvider,
-    ) -> Section:
-        metadata = self.__get_metadata(get_metadata)
-        relative_offset_coordinates_enter = metadata[RELATIVE_OFFSET_COORDINATES][
-            EventType.SECTION_ENTER.serialize()
-        ]
-        section: Section | None = None
-        if is_area_section:
-            section = Area(
-                id=self._application.get_section_id(),
-                name=metadata[NAME],
-                relative_offset_coordinates={
-                    EventType.SECTION_ENTER: geometry.RelativeOffsetCoordinate(
-                        **relative_offset_coordinates_enter
-                    )
-                },
-                plugin_data={},
-                coordinates=[
-                    self._to_coordinate(coordinate) for coordinate in coordinates
-                ],
-            )
-        else:
-            section = LineSection(
-                id=self._application.get_section_id(),
-                name=metadata[NAME],
-                relative_offset_coordinates={
-                    EventType.SECTION_ENTER: geometry.RelativeOffsetCoordinate(
-                        **relative_offset_coordinates_enter
-                    )
-                },
-                plugin_data={},
-                coordinates=[
-                    self._to_coordinate(coordinate) for coordinate in coordinates
-                ],
-            )
-        if section is None:
-            raise TypeError("section has to be LineSection or Area, but is None")
-        self._application.add_section(section)
-        return section
-
-    def __get_metadata(self, get_metadata: MetadataProvider) -> dict:
-        metadata = get_metadata()
-        while (
-            (not metadata)
-            or (NAME not in metadata)
-            or (not self.is_section_name_valid(metadata[NAME]))
-            or (RELATIVE_OFFSET_COORDINATES not in metadata)
-        ):
-            metadata = get_metadata()
-        return metadata
-
-    def __validate_section_information(
-        self, meta_data: dict, coordinates: list[tuple[int, int]]
-    ) -> None:
-        if not coordinates:
-            raise MissingCoordinate("First coordinate is missing")
-        elif len(coordinates) == 1:
-            raise MissingCoordinate("Second coordinate is missing")
-        if not meta_data:
-            raise ValueError("Metadata of line_section are not defined")
-
     def update_section_coordinates(
         self, meta_data: dict, coordinates: list[tuple[int, int]]
     ) -> None:
-        self.__validate_section_information(meta_data, coordinates)
-        section_id = SectionId(meta_data[ID])
-        if not (section := self._application.get_section_for(section_id)):
-            raise MissingSection(
-                f"Could not update section '{section_id.serialize()}' after editing"
-            )
-        section.update_coordinates(
-            [self._to_coordinate(coordinate) for coordinate in coordinates]
-        )
-        self._application.update_section(section)
-        logger().info(f"Update section: {section.id}")
-        self._update_selected_sections([section.id])
+        section_id = self._update_section_coordinates.update(meta_data, coordinates)
+        logger().info(f"Update section: {section_id}")
+        self._update_selected_sections([section_id])
         self._finish_action()
 
-    def _to_coordinate(self, coordinate: tuple[int, int]) -> geometry.Coordinate:
-        return geometry.Coordinate(coordinate[0], coordinate[1])
-
     def _is_area_section(self, section: Section | None) -> bool:
         return isinstance(section, Area)
 

OTAnalytics/plugin_ui/main_application.py

@@ -80,6 +80,11 @@
 from OTAnalytics.application.use_cases.cut_tracks_with_sections import (
     CutTracksIntersectingSection,
 )
+from OTAnalytics.application.use_cases.editor.section_editor import (
+    AddNewSection,
+    CreateSectionId,
+    UpdateSectionCoordinates,
+)
 from OTAnalytics.application.use_cases.event_repository import (
     AddEvents,
     ClearAllEvents,
@@ -652,12 +657,20 @@ def start_gui(self, run_config: RunConfiguration) -> None:
         cut_tracks_intersecting_section.register(clear_all_events.on_tracks_cut)
         application.connect_clear_event_repository_observer()
         name_generator = ArrowFlowNameGenerator()
+        create_section_id = CreateSectionId(section_repository)
+        add_new_section = AddNewSection(
+            create_section_id=create_section_id,
+            add_section=add_section,
+        )
+        update_section_coordinates = UpdateSectionCoordinates(section_repository)
         dummy_viewmodel = DummyViewModel(
             application,
             flow_parser,
             name_generator,
             event_list_export_formats=AVAILABLE_EVENTLIST_EXPORTERS,
             show_svz=run_config.show_svz,
+            add_new_section=add_new_section,
+            update_section_coordinates=update_section_coordinates,
         )
         application.register_video_observer(dummy_viewmodel)
         application.register_sections_observer(dummy_viewmodel)