Merge pull request #849 from OpenTrafficCam/bug/10484-missing-assignment-of-tracks-to-flows

Bug/10484 missing assignment of tracks to flows

Author: schubsen

OTAnalytics/adapter_ui/dummy_viewmodel.py

@@ -476,9 +476,6 @@ def _update_enabled_video_control_buttons(self) -> None:
     def _on_section_changed(self, section: SectionId) -> None:
         self._refresh_sections_in_ui()
 
-    def on_flow_changed(self, flow_id: FlowId) -> None:
-        self.notify_flows([flow_id])
-
     def on_background_updated(self, image: Optional[TrackImage]) -> None:
         if image:
             self.frame_canvas.update_background(image)

OTAnalytics/adapter_ui/view_model.py

@@ -461,10 +461,6 @@ def get_tracks_assigned_to_each_flow(self) -> dict[FlowId, int]:
     async def export_track_statistics(self) -> None:
         raise NotImplementedError
 
-    @abstractmethod
-    def on_flow_changed(self, flow_id: FlowId) -> None:
-        raise NotImplementedError
-
     @abstractmethod
     def on_background_updated(self, image: Optional[TrackImage]) -> None:
         raise NotImplementedError

OTAnalytics/application/application.py

@@ -86,7 +86,7 @@
 from OTAnalytics.application.use_cases.update_project import ProjectUpdater
 from OTAnalytics.domain.date import DateRange
 from OTAnalytics.domain.filter import FilterElement, FilterElementSettingRestorer
-from OTAnalytics.domain.flow import Flow, FlowChangedObserver, FlowId, FlowListObserver
+from OTAnalytics.domain.flow import Flow, FlowId, FlowListObserver
 from OTAnalytics.domain.geometry import RelativeOffsetCoordinate
 from OTAnalytics.domain.section import (
     Section,
@@ -238,9 +238,6 @@ def register_section_changed_observer(
     def register_flows_observer(self, observer: FlowListObserver) -> None:
         self._datastore.register_flows_observer(observer)
 
-    def register_flow_changed_observer(self, observer: FlowChangedObserver) -> None:
-        self._datastore.register_flow_changed_observer(observer)
-
     def get_all_sections(self) -> list[Section]:
         return self._datastore.get_all_sections()
 

OTAnalytics/application/datastore.py

@@ -10,13 +10,7 @@
     EventListExporter,
 )
 from OTAnalytics.domain.event import Event, EventRepository
-from OTAnalytics.domain.flow import (
-    Flow,
-    FlowChangedObserver,
-    FlowId,
-    FlowListObserver,
-    FlowRepository,
-)
+from OTAnalytics.domain.flow import Flow, FlowId, FlowListObserver, FlowRepository
 from OTAnalytics.domain.progress import ProgressbarBuilder
 from OTAnalytics.domain.remark import RemarkRepository
 from OTAnalytics.domain.section import (
@@ -301,15 +295,6 @@ def register_section_changed_observer(
         """
         self._section_repository.register_section_changed_observer(observer)
 
-    def register_flow_changed_observer(self, observer: FlowChangedObserver) -> None:
-        """
-        Listen to changes of sections in the repository.
-
-        Args:
-            observer (FlowChangedObserver): observer to notify about changes
-        """
-        self._flow_repository.register_flow_changed_observer(observer)
-
     def update_section(self, section: Section) -> None:
         """
         Update the section in the repository.

OTAnalytics/application/use_cases/assignment_repository.py

@@ -6,7 +6,6 @@
 from OTAnalytics.application.use_cases.create_road_user_assignments import (
     CreateRoadUserAssignments,
 )
-from OTAnalytics.application.use_cases.flow_repository import GetAllFlows
 from OTAnalytics.domain.event import EventRepository, EventRepositoryEvent
 from OTAnalytics.domain.flow import FlowId, FlowListObserver
 
@@ -70,34 +69,22 @@ def __call__(self, repo_event: EventRepositoryEvent) -> None:
             self._assignment_repository.remove_assignments_of_event(event)
 
 
-class RemoveAssignmentsOfChangedFlow:  # is a FlowChangedObserver
-    def __init__(self, assignment_repository: RoadUserAssignmentRepository) -> None:
-        self._assignment_repository = assignment_repository
-
-    def __call__(self, flow: FlowId) -> None:
-        self._assignment_repository.remove_assignments_of_flow(flow)
+class ClearAssignmentsOnChangedFlows(FlowListObserver):
+    """Clear all assignments.
 
+    Args:
+        assignment_repository (RoadUserAssignmentRepository): the repository
+            holding the assignments to clear.
+    """
 
-class RemoveAssignmentsOfRemovedFlows(FlowListObserver):
-    def __init__(
-        self,
-        assignment_repository: RoadUserAssignmentRepository,
-        get_all_flows: GetAllFlows,
-    ) -> None:
+    def __init__(self, assignment_repository: RoadUserAssignmentRepository) -> None:
         self._assignment_repository = assignment_repository
-        self._get_all_flows = get_all_flows
 
     def notify_flows(self, flows: list[FlowId]) -> None:
-        # TODO notification [] means flow is removed or all are removed!
-        # FlowRepositoryEvent would be helpful here
-        if not flows:
-            observed_flows = self._assignment_repository.get_observed_flows()
-            current_flows = [flow.id for flow in self._get_all_flows.get()]
-
-            if len(current_flows) == 0:
-                self._assignment_repository.clear()
-                return
-
-            gen = (flow for flow in observed_flows if flow not in current_flows)
-            for flow in gen:
-                self._assignment_repository.remove_assignments_of_flow(flow)
+        """Clear all assignments after flows have been changed.
+
+        Args:
+            flows (list[FlowId]): the added or updated flows, empty if flows
+                have been removed.
+        """
+        self._assignment_repository.clear()

OTAnalytics/domain/flow.py

@@ -1,6 +1,6 @@
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from typing import Callable, Generic, Iterable, Optional, TypeVar
+from typing import Iterable, Optional
 
 from OTAnalytics.domain.section import SectionId
 
@@ -84,44 +84,15 @@ class FlowListObserver(ABC):
     @abstractmethod
     def notify_flows(self, flows: list[FlowId]) -> None:
         """
-        Notifies that the given flows have been added or removed.
+        Notifies that the given flows have been added or updated.
 
-        Args:
-            flows (list[FlowId]): list of added or removed flows
-        """
-        pass
-
-
-VALUE = TypeVar("VALUE")
-
-FlowChangedObserver = Callable[[FlowId], None]
-
-
-class FlowChangedSubject(Generic[VALUE]):
-    """
-    Helper class to handle and notify observers
-    """
-
-    def __init__(self) -> None:
-        self.observers: set[FlowChangedObserver] = set()
-
-    def register(self, observer: FlowChangedObserver) -> None:
-        """
-        Listen to events.
-
-        Args:
-            observer (FlowChangedObserver): listener to add
-        """
-        self.observers.add(observer)
-
-    def notify(self, value: FlowId) -> None:
-        """
-        Notifies observers about the changed value.
+        An empty list indicates that flows have been removed.
 
         Args:
-            value (FlowId): changed value
+            flows (list[FlowId]): list of added or updated flows, empty if flows
+                have been removed
         """
-        [observer(value) for observer in self.observers]
+        pass
 
 
 class FlowListSubject:
@@ -146,7 +117,8 @@ def notify(self, flows: list[FlowId]) -> None:
         Notifies observers about the list of flows.
 
         Args:
-            flows (list[FlowId]): list of added flows
+            flows (list[FlowId]): list of added or updated flows, empty if flows
+                have been removed
         """
         [observer.notify_flows(flows) for observer in self.observers]
 
@@ -156,14 +128,10 @@ def __init__(self) -> None:
         self._flows: dict[FlowId, Flow] = {}
         self._current_id = 0
         self._repository_content_observers: FlowListSubject = FlowListSubject()
-        self._flow_content_observers: FlowChangedSubject = FlowChangedSubject()
 
     def register_flows_observer(self, observer: FlowListObserver) -> None:
         self._repository_content_observers.register(observer)
 
-    def register_flow_changed_observer(self, observer: FlowChangedObserver) -> None:
-        self._flow_content_observers.register(observer)
-
     def get_id(self) -> FlowId:
         """
         Get an id for a new flow
@@ -219,7 +187,7 @@ def remove(self, flow_id: FlowId) -> None:
 
     def update(self, flow: Flow) -> None:
         self._flows[flow.id] = flow
-        self._flow_content_observers.notify(flow.id)
+        self._repository_content_observers.notify([flow.id])
 
     def get(self, flow_id: FlowId) -> Optional[Flow]:
         return self._flows.get(flow_id)

OTAnalytics/plugin_prototypes/track_visualization/track_viz.py

@@ -97,17 +97,24 @@ def __init__(
         super().__init__(plotter_factory, flow_state.selected_flows, self.get_flow_ids)
 
         flow_repository.register_flows_observer(self)
-        flow_repository.register_flow_changed_observer(self.notify_flow)
         track_repository.observers.register(self.notify_invalidate)
         event_repository.register_observer(self.notify_events)
 
         self._repository = flow_repository
         self._flow_id_resolver = flow_id_resolver
 
-    def notify_flow(self, flow: FlowId) -> None:
-        self.notify_layers_changed([flow])
-
     def notify_flows(self, flows: list[FlowId]) -> None:
+        """Handle added, updated or removed flows.
+
+        All layers are invalidated because road users are assigned to exactly
+        one flow. Changing a single flow can therefore change the tracks
+        plotted for every other flow as well.
+
+        Args:
+            flows (list[FlowId]): the added or updated flows, empty if flows
+                have been removed.
+        """
+        self.notify_invalidate(None)
         self.notify_layers_changed(flows)
 
     def notify_events(self, events: EventRepositoryEvent) -> None:

OTAnalytics/plugin_ui/base_application.py

@@ -62,10 +62,9 @@
 from OTAnalytics.application.use_cases.add_new_remark import AddNewRemark
 from OTAnalytics.application.use_cases.apply_cli_cuts import ApplyCliCuts
 from OTAnalytics.application.use_cases.assignment_repository import (
+    ClearAssignmentsOnChangedFlows,
     GetRoadUserAssignments,
-    RemoveAssignmentsOfChangedFlow,
     RemoveAssignmentsOfRemovedEvents,
-    RemoveAssignmentsOfRemovedFlows,
 )
 from OTAnalytics.application.use_cases.clear_repositories import ClearRepositories
 from OTAnalytics.application.use_cases.config import SaveOtconfig
@@ -971,14 +970,8 @@ def remove_assignments_of_removed_events(self) -> RemoveAssignmentsOfRemovedEven
         )
 
     @cached_property
-    def remove_assignments_of_removed_flows(self) -> RemoveAssignmentsOfRemovedFlows:
-        return RemoveAssignmentsOfRemovedFlows(
-            self.assignment_repository, self.get_all_flows
-        )
-
-    @cached_property
-    def remove_assignments_of_changed_flows(self) -> RemoveAssignmentsOfChangedFlow:
-        return RemoveAssignmentsOfChangedFlow(self.assignment_repository)
+    def clear_assignments_on_changed_flows(self) -> ClearAssignmentsOnChangedFlows:
+        return ClearAssignmentsOnChangedFlows(self.assignment_repository)
 
     @cached_property
     def get_all_assignments(self) -> GetRoadUserAssignments:

OTAnalytics/plugin_ui/gui_application.py

@@ -67,7 +67,6 @@ def register_observers(self) -> None:
         self.application.register_video_observer(self.view_model)
         self.application.register_sections_observer(self.view_model)
         self.application.register_flows_observer(self.view_model)
-        self.application.register_flow_changed_observer(self.view_model.on_flow_changed)
         self.track_view_state.selected_videos.register(
             self.view_model.update_selected_videos
         )
@@ -127,15 +126,12 @@ def register_observers(self) -> None:
         # configure observers for count plot saver
         self.track_view_state.count_plots.register(self.save_count_plots.save)
 
-        # configure assignment repository update on flow/event deletion
+        # configure assignment repository update on flow and event changes
         self.event_repository.register_observer(
             self.remove_assignments_of_removed_events
         )
-        self.flow_repository.register_flow_changed_observer(
-            self.remove_assignments_of_changed_flows
-        )
         self.flow_repository.register_flows_observer(
-            self.remove_assignments_of_removed_flows
+            self.clear_assignments_on_changed_flows
         )
 
     @cached_property

tests/unit/OTAnalytics/application/use_cases/test_clear_assignments_on_changed_flows.py

@@ -0,0 +1,35 @@
+from unittest.mock import Mock
+
+from OTAnalytics.application.analysis.road_user_assignment import (
+    RoadUserAssignmentRepository,
+)
+from OTAnalytics.application.use_cases.assignment_repository import (
+    ClearAssignmentsOnChangedFlows,
+)
+from OTAnalytics.domain.flow import FlowId
+
+
+class TestClearAssignmentsOnChangedFlows:
+    def test_clear_on_changed_flow(self) -> None:
+        assignment_repository = Mock(spec=RoadUserAssignmentRepository)
+        target = ClearAssignmentsOnChangedFlows(assignment_repository)
+
+        target.notify_flows([FlowId("1")])
+
+        assignment_repository.clear.assert_called_once()
+
+    def test_clear_on_added_flows(self) -> None:
+        assignment_repository = Mock(spec=RoadUserAssignmentRepository)
+        target = ClearAssignmentsOnChangedFlows(assignment_repository)
+
+        target.notify_flows([FlowId("1"), FlowId("2")])
+
+        assignment_repository.clear.assert_called_once()
+
+    def test_clear_on_removed_flows(self) -> None:
+        assignment_repository = Mock(spec=RoadUserAssignmentRepository)
+        target = ClearAssignmentsOnChangedFlows(assignment_repository)
+
+        target.notify_flows([])
+
+        assignment_repository.clear.assert_called_once()