Feature/6490 display the number of assigned tracks per flow in the flow table

OTAnalytics/adapter_ui/abstract_frame_track_statistics.py

@@ -0,0 +1,13 @@
+from abc import ABC, abstractmethod
+
+from OTAnalytics.application.use_cases.track_statistics import TrackStatistics
+
+
+class AbstractFrameTrackStatistics(ABC):
+    @abstractmethod
+    def introduce_to_viewmodel(self) -> None:
+        raise NotImplementedError
+
+    @abstractmethod
+    def update_track_statistics(self, track_statistics: TrackStatistics) -> None:
+        raise NotImplementedError

OTAnalytics/adapter_ui/view_model.py

@@ -17,12 +17,15 @@
 from OTAnalytics.adapter_ui.abstract_frame_track_plotting import (
     AbstractFrameTrackPlotting,
 )
+from OTAnalytics.adapter_ui.abstract_frame_track_statistics import (
+    AbstractFrameTrackStatistics,
+)
 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.flow import Flow, FlowId
 from OTAnalytics.domain.section import Section
 from OTAnalytics.domain.video import Video
 
@@ -85,6 +88,10 @@ def set_filter_frame(self, filter_frame: AbstractFrameFilter) -> None:
     def set_frame_project(self, project_frame: AbstractFrameProject) -> None:
         pass
 
+    @abstractmethod
+    def set_frame_track_statistics(self, frame: AbstractFrameTrackStatistics) -> None:
+        pass
+
     @abstractmethod
     def set_button_quick_save_config(
         self, button_quick_save_config: AbstractButtonQuickSaveConfig
@@ -426,3 +433,7 @@ def set_svz_metadata_frame(self, frame: AbstractFrameSvzMetadata) -> None:
     @abstractmethod
     def get_save_path_suggestion(self, file_type: str, context_file_type: str) -> Path:
         raise NotImplementedError
+
+    @abstractmethod
+    def get_tracks_assigned_to_each_flow(self) -> dict[FlowId, int]:
+        raise NotImplementedError

OTAnalytics/application/application.py

@@ -36,6 +36,9 @@
     EnableFilterTrackByDate,
 )
 from OTAnalytics.application.use_cases.flow_repository import AddFlow
+from OTAnalytics.application.use_cases.flow_statistics import (
+    NumberOfTracksAssignedToEachFlow,
+)
 from OTAnalytics.application.use_cases.generate_flows import GenerateFlows
 from OTAnalytics.application.use_cases.load_otconfig import LoadOtconfig
 from OTAnalytics.application.use_cases.load_otflow import LoadOtflow
@@ -59,6 +62,10 @@
     GetAllTrackFiles,
     TrackRepositorySize,
 )
+from OTAnalytics.application.use_cases.track_statistics import (
+    CalculateTrackStatistics,
+    TrackStatistics,
+)
 from OTAnalytics.application.use_cases.update_project import ProjectUpdater
 from OTAnalytics.domain.date import DateRange
 from OTAnalytics.domain.filter import FilterElement, FilterElementSettingRestorer
@@ -131,6 +138,8 @@ def __init__(
         config_has_changed: ConfigHasChanged,
         export_road_user_assignments: ExportRoadUserAssignments,
         file_name_suggester: SavePathSuggester,
+        calculate_track_statistics: CalculateTrackStatistics,
+        number_of_tracks_assigned_to_each_flow: NumberOfTracksAssignedToEachFlow,
     ) -> None:
         self._datastore: Datastore = datastore
         self.track_state: TrackState = track_state
@@ -171,6 +180,10 @@ def __init__(
         self._config_has_changed = config_has_changed
         self._export_road_user_assignments = export_road_user_assignments
         self._file_name_suggester = file_name_suggester
+        self._calculate_track_statistics = calculate_track_statistics
+        self._number_of_tracks_assigned_to_each_flow = (
+            number_of_tracks_assigned_to_each_flow
+        )
 
     def connect_observers(self) -> None:
         """
@@ -667,6 +680,12 @@ def suggest_save_path(self, file_type: str, context_file_type: str = "") -> Path
         """
         return self._file_name_suggester.suggest(file_type, context_file_type)
 
+    def calculate_track_statistics(self) -> TrackStatistics:
+        return self._calculate_track_statistics.get_statistics()
+
+    def number_of_tracks_assigned_to_each_flow(self) -> dict[FlowId, int]:
+        return self._number_of_tracks_assigned_to_each_flow.get()
+
 
 class MissingTracksError(Exception):
     pass

OTAnalytics/application/use_cases/flow_statistics.py

@@ -0,0 +1,25 @@
+from OTAnalytics.application.use_cases.get_road_user_assignments import (
+    GetRoadUserAssignments,
+)
+from OTAnalytics.domain.flow import FlowId, FlowRepository
+
+
+class NumberOfTracksAssignedToEachFlow:
+    def __init__(
+        self, get_assignments: GetRoadUserAssignments, flow_repository: FlowRepository
+    ) -> None:
+        self._get_assignments = get_assignments
+        self._flow_repository = flow_repository
+
+    def get(self) -> dict[FlowId, int]:
+        result = self._tracks_assigned_to_flows()
+        for road_user_assignment in self._get_assignments.get():
+            flow_id = road_user_assignment.assignment.id
+            result[flow_id] += 1
+        return result
+
+    def _tracks_assigned_to_flows(self) -> dict[FlowId, int]:
+        flows = {}
+        for flow in self._flow_repository.get_all():
+            flows[flow.id] = 0
+        return flows

OTAnalytics/application/use_cases/get_road_user_assignments.py

@@ -0,0 +1,23 @@
+from OTAnalytics.application.analysis.traffic_counting import (
+    RoadUserAssigner,
+    RoadUserAssignment,
+)
+from OTAnalytics.domain.event import EventRepository
+from OTAnalytics.domain.flow import FlowRepository
+
+
+class GetRoadUserAssignments:
+    def __init__(
+        self,
+        flow_repository: FlowRepository,
+        event_repository: EventRepository,
+        assigner: RoadUserAssigner,
+    ) -> None:
+        self._flow_repository = flow_repository
+        self._event_repository = event_repository
+        self._assigner = assigner
+
+    def get(self) -> list[RoadUserAssignment]:
+        return self._assigner.assign(
+            self._event_repository.get_all(), self._flow_repository.get_all()
+        ).as_list()

OTAnalytics/application/use_cases/highlight_intersections.py

@@ -5,7 +5,11 @@
 from OTAnalytics.application.analysis.intersect import TracksIntersectingSections
 from OTAnalytics.application.analysis.traffic_counting import RoadUserAssigner
 from OTAnalytics.application.state import FlowState, SectionState, TrackViewState
-from OTAnalytics.application.use_cases.section_repository import GetSectionsById
+from OTAnalytics.application.use_cases.section_repository import (
+    GetAllSections,
+    GetCuttingSections,
+    GetSectionsById,
+)
 from OTAnalytics.domain.event import EventRepository
 from OTAnalytics.domain.flow import FlowId, FlowRepository
 from OTAnalytics.domain.section import SectionId
@@ -59,6 +63,76 @@ def get_ids(self) -> set[TrackId]:
         ).get_ids()
 
 
+class TracksIntersectingAllNonCuttingSections(TrackIdProvider):
+    """Returns track ids intersecting all sections which are not a cutting section.
+
+    Args:
+        get_all_sections (GetAllSections): the use case to get all sections.
+        tracks_intersecting_sections (TracksIntersectingSections): get track ids
+            intersecting sections.
+        get_section_by_id (GetSectionsById): use case to get sections by id.
+    """
+
+    def __init__(
+        self,
+        get_cutting_sections: GetCuttingSections,
+        get_all_sections: GetAllSections,
+        tracks_intersecting_sections: TracksIntersectingSections,
+        get_section_by_id: GetSectionsById,
+        intersection_repository: IntersectionRepository,
+    ) -> None:
+        self._get_cutting_sections = get_cutting_sections
+        self._get_all_sections = get_all_sections
+        self._tracks_intersecting_sections = tracks_intersecting_sections
+        self._get_section_by_id = get_section_by_id
+        self._intersection_repository = intersection_repository
+
+    def get_ids(self) -> set[TrackId]:
+        ids_non_cutting_sections = {
+            section.id
+            for section in self._get_all_sections()
+            if section not in self._get_cutting_sections()
+        }
+        return TracksIntersectingGivenSections(
+            ids_non_cutting_sections,
+            self._tracks_intersecting_sections,
+            self._get_section_by_id,
+            self._intersection_repository,
+        ).get_ids()
+
+
+class TracksIntersectingAllSections(TrackIdProvider):
+    """Returns track ids intersecting all sections.
+
+    Args:
+        get_all_sections (GetAllSections): the use case to get all sections.
+        tracks_intersecting_sections (TracksIntersectingSections): get track ids
+            intersecting sections.
+        get_section_by_id (GetSectionsById): use case to get sections by id.
+    """
+
+    def __init__(
+        self,
+        get_all_sections: GetAllSections,
+        tracks_intersecting_sections: TracksIntersectingSections,
+        get_section_by_id: GetSectionsById,
+        intersection_repository: IntersectionRepository,
+    ) -> None:
+        self._get_all_sections = get_all_sections
+        self._tracks_intersecting_sections = tracks_intersecting_sections
+        self._get_section_by_id = get_section_by_id
+        self._intersection_repository = intersection_repository
+
+    def get_ids(self) -> set[TrackId]:
+        ids_all_sections = {section.id for section in self._get_all_sections()}
+        return TracksIntersectingGivenSections(
+            ids_all_sections,
+            self._tracks_intersecting_sections,
+            self._get_section_by_id,
+            self._intersection_repository,
+        ).get_ids()
+
+
 class TracksIntersectingGivenSections(TrackIdProvider):
     """Returns track ids intersecting given sections.
 
@@ -158,14 +232,40 @@ def get_ids(self) -> Iterable[TrackId]:
         return ids
 
 
+class TracksAssignedToAllFlows(TrackIdProvider):
+    """Returns track ids that are assigned to all flows.
+
+    Args:
+        assigner (RoadUserAssigner): to assign tracks to flows.
+        event_repository (EventRepository): the event repository.
+        flow_repository (FlowRepository): the track repository.
+    """
+
+    def __init__(
+        self,
+        assigner: RoadUserAssigner,
+        event_repository: EventRepository,
+        flow_repository: FlowRepository,
+    ) -> None:
+        self._assigner = assigner
+        self._event_repository = event_repository
+        self._flow_repository = flow_repository
+
+    def get_ids(self) -> Iterable[TrackId]:
+        all_flow_ids = [flow.id for flow in self._flow_repository.get_all()]
+        return TracksAssignedToGivenFlows(
+            self._assigner, self._event_repository, self._flow_repository, all_flow_ids
+        ).get_ids()
+
+
 class TracksAssignedToGivenFlows(TrackIdProvider):
     """Returns track ids that are assigned to the given flows.
 
     Args:
         assigner (RoadUserAssigner): to assign tracks to flows.
         event_repository (EventRepository): the event repository.
         flow_repository (FlowRepository): the track repository.
-        flow_ids (list[FlowId]): the flows fo identify assigned tracks for.
+        flow_ids (list[FlowId]): the flows to identify assigned tracks for.
     """
 
     def __init__(

OTAnalytics/application/use_cases/inside_cutting_section.py

@@ -0,0 +1,44 @@
+from OTAnalytics.application.use_cases.section_repository import GetCuttingSections
+from OTAnalytics.application.use_cases.track_repository import GetAllTracks
+from OTAnalytics.domain.section import SectionId
+from OTAnalytics.domain.track import TrackId
+from OTAnalytics.domain.types import EventType
+
+
+class TrackIdsInsideCuttingSections:
+    def __init__(
+        self, get_tracks: GetAllTracks, get_cutting_sections: GetCuttingSections
+    ):
+        self._get_tracks = get_tracks
+        self._get_cutting_sections = get_cutting_sections
+
+    def __call__(self) -> set[TrackId]:
+        track_dataset = self._get_tracks.as_dataset()
+        cutting_sections = self._get_cutting_sections()
+        if not cutting_sections:
+            return set()
+
+        results: set[TrackId] = set()
+        for cutting_section in cutting_sections:
+            offset = cutting_section.get_offset(EventType.SECTION_ENTER)
+            # set of all tracks where at least one coordinate is contained
+            # by at least one cutting section
+            results.update(
+                set(
+                    track_id
+                    for track_id, section_data in (
+                        track_dataset.contained_by_sections(
+                            [cutting_section], offset
+                        ).items()
+                    )
+                    if contains_true(section_data)
+                )
+            )
+        return results
+
+
+def contains_true(section_data: list[tuple[SectionId, list[bool]]]) -> bool:
+    for _, bool_list in section_data:
+        if any(bool_list):
+            return True
+    return False

OTAnalytics/application/use_cases/section_repository.py

@@ -1,7 +1,13 @@
 from typing import Iterable
 
+from OTAnalytics.application.config import CLI_CUTTING_SECTION_MARKER
 from OTAnalytics.domain.geometry import RelativeOffsetCoordinate
-from OTAnalytics.domain.section import Section, SectionId, SectionRepository
+from OTAnalytics.domain.section import (
+    Section,
+    SectionId,
+    SectionRepository,
+    SectionType,
+)
 from OTAnalytics.domain.types import EventType
 
 
@@ -23,6 +29,25 @@ def __call__(self) -> list[Section]:
         return self._section_repository.get_all()
 
 
+class GetCuttingSections:
+    """Get all cutting sections from the repository."""
+
+    def __init__(self, section_repository: SectionRepository) -> None:
+        self._section_repository = section_repository
+
+    def __call__(self) -> list[Section]:
+        cutting_sections = sorted(
+            [
+                section
+                for section in self._section_repository.get_all()
+                if section.get_type() == SectionType.CUTTING
+                or section.name.startswith(CLI_CUTTING_SECTION_MARKER)
+            ],
+            key=lambda section: section.id.id,
+        )
+        return cutting_sections
+
+
 class GetSectionsById:
     """Get sections by their id.