OP#10282: Parse track files off the event loop

Author: randy-seng

OTAnalytics/adapter_ui/dummy_viewmodel.py

@@ -686,9 +686,9 @@ async def load_otconfig(self) -> None:
         )
         if not otconfig_file:
             return
-        self._load_otconfig(otconfig_file)
+        await self._load_otconfig(otconfig_file)
 
-    def _load_otconfig(self, otconfig_file: Path) -> None:
+    async def _load_otconfig(self, otconfig_file: Path) -> None:
         proceed = self._ui_factory.info_box(
             message=(
                 "This will load a stored configuration from file. \n"
@@ -701,7 +701,7 @@ def _load_otconfig(self, otconfig_file: Path) -> None:
         if proceed.canceled:
             return
         logger().info(f"{OTCONFIG_FILE_TYPE} file to load: {otconfig_file}")
-        self._application.load_otconfig(file=Path(otconfig_file))
+        await self._application.load_otconfig_async(file=Path(otconfig_file))
         self.show_current_project()
         self.update_svz_metadata_view()
 
@@ -813,7 +813,7 @@ async def load_tracks(self) -> None:
         if not track_files:
             return
         logger().info(f"Tracks files to load: {track_files}")
-        self._application.add_tracks_of_files(track_files=track_files)
+        await self._application.add_tracks_of_files_async(track_files=track_files)
 
     async def load_configuration(self) -> None:  # sourcery skip: avoid-builtin-shadow
         # INFO: Current behavior: Overwrites existing sections
@@ -832,7 +832,7 @@ async def load_configuration(self) -> None:  # sourcery skip: avoid-builtin-shad
         elif configuration_file.suffix == f".{OTFLOW_FILE_TYPE}":
             self._load_otflow(configuration_file)
         elif configuration_file.suffix == f".{OTCONFIG_FILE_TYPE}":
-            self._load_otconfig(configuration_file)
+            await self._load_otconfig(configuration_file)
         else:
             raise ValueError("Configuration file to load has unknown file extension")
 

OTAnalytics/application/application.py

@@ -309,17 +309,36 @@ def save_otconfig(self, file: Path) -> None:
         self._save_otconfig(file)
 
     def load_otconfig(self, file: Path) -> None:
+        """Load an otconfig, blocking. Outside an event loop only."""
         self._load_otconfig.load(file)
 
+    async def load_otconfig_async(self, file: Path) -> None:
+        """Load an otconfig without blocking the event loop.
+
+        Args:
+            file (Path): the otconfig file.
+        """
+        await self._load_otconfig.load_async(file)
+
     def add_tracks_of_files(self, track_files: list[Path]) -> None:
         """
         Load a multiple track files.
 
+        Outside an event loop only. Inside one use `add_tracks_of_files_async`.
+
         Args:
             track_files (list[Path]): files in ottrk format
         """
         self._load_track_files(track_files)
 
+    async def add_tracks_of_files_async(self, track_files: list[Path]) -> None:
+        """Load multiple track files without blocking the event loop.
+
+        Args:
+            track_files (list[Path]): files in ottrk format
+        """
+        await self._load_track_files.load(track_files)
+
     def delete_all_tracks(self) -> None:
         """Delete all tracks."""
         self._datastore.delete_all_tracks()

OTAnalytics/application/use_cases/load_otconfig.py

@@ -1,6 +1,7 @@
 from pathlib import Path
+from typing import Awaitable, Callable
 
-from OTAnalytics.application.parser.config_parser import ConfigParser
+from OTAnalytics.application.parser.config_parser import ConfigParser, OtConfig
 from OTAnalytics.application.parser.deserializer import Deserializer
 from OTAnalytics.application.state import ConfigurationFile
 from OTAnalytics.application.use_cases.add_new_remark import AddNewRemark
@@ -44,29 +45,66 @@ def __init__(
         self._subject = Subject[ConfigurationFile]()
 
     def load(self, file: Path) -> None:
-        self._reset_application.reset()
-        config = self._config_parser.parse(file)
+        """Load an otconfig and its track files, blocking.
+
+        For use outside an event loop only, such as preloading a config file given
+        on the command line at startup. Inside an event loop use `load_async`.
+
+        Args:
+            file (Path): the otconfig file.
+        """
+        self._apply(file, self._load_track_files)
+
+    async def load_async(self, file: Path) -> None:
+        """Load an otconfig and its track files, keeping the ui responsive.
+
+        Args:
+            file (Path): the otconfig file.
+        """
+        await self._apply_async(file, self._load_track_files.load)
+
+    def _apply(self, file: Path, load_track_files: Callable[[list], None]) -> None:
+        config = self._begin(file)
+        try:
+            self._publish_before_tracks(config)
+            load_track_files(list(config.analysis.track_files))
+            self._publish_after_tracks(file, config)
+        except (SectionAlreadyExists, FlowAlreadyExists) as cause:
+            self._abort(cause)
+
+    async def _apply_async(
+        self, file: Path, load_track_files: Callable[[list], Awaitable[None]]
+    ) -> None:
+        config = self._begin(file)
         try:
-            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)
-            self._load_track_files(list(config.analysis.track_files))
-            if config.remark:
-                self._add_new_remark.add(config.remark)
-            self._subject.notify(
-                ConfigurationFile(
-                    file,
-                    self._deserialize(file),
-                )
-            )
+            self._publish_before_tracks(config)
+            await load_track_files(list(config.analysis.track_files))
+            self._publish_after_tracks(file, config)
         except (SectionAlreadyExists, FlowAlreadyExists) as cause:
-            self._reset_application.reset()
-            raise UnableToLoadOtconfigFile(
-                "Error while loading otconfig file. Abort loading!"
-            ) from cause
+            self._abort(cause)
+
+    def _begin(self, file: Path) -> OtConfig:
+        self._reset_application.reset()
+        return self._config_parser.parse(file)
+
+    def _publish_before_tracks(self, config: OtConfig) -> None:
+        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)
+
+    def _publish_after_tracks(self, file: Path, config: OtConfig) -> None:
+        if config.remark:
+            self._add_new_remark.add(config.remark)
+        self._subject.notify(ConfigurationFile(file, self._deserialize(file)))
+
+    def _abort(self, cause: Exception) -> None:
+        self._reset_application.reset()
+        raise UnableToLoadOtconfigFile(
+            "Error while loading otconfig file. Abort loading!"
+        ) from cause
 
     def register(self, observer: OBSERVER[ConfigurationFile]) -> None:
         self._subject.register(observer)

OTAnalytics/application/use_cases/load_track_files.py

@@ -1,8 +1,9 @@
+import asyncio
 from pathlib import Path
 
 from OTAnalytics.application.datastore import VideoParser
 from OTAnalytics.application.logger import logger
-from OTAnalytics.application.parser.track_parser import TrackParser
+from OTAnalytics.application.parser.track_parser import TrackParser, TracksParseResult
 from OTAnalytics.application.state import TracksMetadata, VideosMetadata
 from OTAnalytics.domain.progress import ProgressbarBuilder
 from OTAnalytics.domain.track_repository import TrackFileRepository, TrackRepository
@@ -31,28 +32,95 @@ def __init__(
         self._videos_metadata = videos_metadata
 
     def __call__(self, files: list[Path]) -> None:
+        """Load and parse track files together with their videos, blocking.
+
+        For use outside an event loop only, such as preloading files given on the
+        command line at startup. Nothing is connected then, so there is nothing to
+        yield to. Inside an event loop use `load` instead, which keeps the ui
+        responsive.
+
+        Args:
+            files (list[Path]): files in ottrk format.
+
+        Raises:
+            RuntimeError: if called while an event loop is running.
         """
-        Load and parse the given track file together with the corresponding video file.
+        if self._event_loop_is_running():
+            raise RuntimeError(
+                "Parsing track files blocks the event loop and freezes the ui."
+                " Use load() instead."
+            )
+        if files_to_load := self._files_to_load(files):
+            self._publish(self._parse(files_to_load), files_to_load)
 
-        Each video is resolved relative to the parent folder of the track file it was
-        parsed from, so track files from different folders each find their own video.
-        This relies on `TrackParser.parse_files` returning exactly one `VideoMetadata`
-        per input file, in input order; `strict=True` turns a violation of that into
-        an error rather than silently dropping videos.
+    async def load(self, files: list[Path]) -> None:
+        """Load and parse track files together with their videos.
+
+        Parsing runs on a worker thread, so the ui keeps repainting while it runs.
+        Everything that touches a repository stays on the event loop.
+
+        Args:
+            files (list[Path]): files in ottrk format.
+        """
+        if files_to_load := self._files_to_load(files):
+            # parse_files must stay pure: no repository, no observer, no ui. It runs
+            # on a worker thread here, and repositories notify observers that mutate
+            # widgets, which is only safe on the event loop.
+            parse_result = await asyncio.to_thread(self._parse, files_to_load)
+            self._publish(parse_result, files_to_load)
+
+    @staticmethod
+    def _event_loop_is_running() -> bool:
+        try:
+            asyncio.get_running_loop()
+        except RuntimeError:
+            return False
+        return True
+
+    def _files_to_load(self, files: list[Path]) -> list[Path]:
+        """Select the files that are not loaded already.
 
         Args:
-            files (Path): files in ottrk format.
+            files (list[Path]): files the caller asked to load.
+
+        Returns:
+            list[Path]: the files still to be parsed.
         """
         if not files:
-            return
+            return []
         files_to_load = [
             file for file in files if not self._is_file_already_loaded(file)
         ]
         self._log_already_loaded_files(files, files_to_load)
-        if not files_to_load:
-            return
+        return files_to_load
+
+    def _parse(self, files_to_load: list[Path]) -> TracksParseResult:
+        """Parse the given track files. Pure: safe to run off the event loop.
+
+        Args:
+            files_to_load (list[Path]): files in ottrk format.
+
+        Returns:
+            TracksParseResult: the parsed tracks and their metadata.
+        """
         logger().info(f"Loading {len(files_to_load)} track files and videos...")
-        parse_result = self._track_parser.parse_files(files_to_load)
+        return self._track_parser.parse_files(files_to_load)
+
+    def _publish(
+        self, parse_result: TracksParseResult, files_to_load: list[Path]
+    ) -> None:
+        """Publish the parse result to the repositories. Event loop only.
+
+        Each video is resolved relative to the parent folder of the track file it was
+        parsed from, so track files from different folders each find their own video.
+        This relies on `TrackParser.parse_files` returning exactly one `VideoMetadata`
+        per input file, in input order; `strict=True` turns a violation of that into
+        an error rather than silently dropping videos.
+
+        Args:
+            parse_result (TracksParseResult): what the parser produced.
+            files_to_load (list[Path]): the files handed to the parser.
+        """
         for video_metadata in parse_result.videos_metadata:
             self._videos_metadata.update(video_metadata)
 

tests/unit/OTAnalytics/adapter_ui/test_dummy_viewmodel_loading.py

@@ -33,8 +33,11 @@ def create_given(
     provide_track_files.provide = AsyncMock(return_value=tracks or [])
     provide_video_files = Mock(spec=ProvideVideoFiles)
     provide_video_files.provide = AsyncMock(return_value=videos or [])
+    application = Mock()
+    application.add_tracks_of_files_async = AsyncMock()
+    application.load_otconfig_async = AsyncMock()
     return Given(
-        application=Mock(),
+        application=application,
         provide_track_files=provide_track_files,
         provide_video_files=provide_video_files,
     )
@@ -70,15 +73,20 @@ def _build(given: Given) -> DummyViewModel:
 
 class TestLoadTracks:
     async def test_forwards_the_provided_files(self) -> None:
-        """#Requirement https://openproject.platomo.de/wp/10280"""
+        """Loading is awaited so the ui keeps repainting while parsing runs.
+
+        #Requirement https://openproject.platomo.de/wp/10280
+        #Requirement https://openproject.platomo.de/wp/10282
+        """
         given = create_given(tracks=TRACK_FILES)
         target = create_target(given)
 
         await target.load_tracks()
 
-        given.application.add_tracks_of_files.assert_called_once_with(
+        given.application.add_tracks_of_files_async.assert_awaited_once_with(
             track_files=TRACK_FILES
         )
+        given.application.add_tracks_of_files.assert_not_called()
 
     async def test_loads_nothing_when_the_provider_returns_nothing(self) -> None:
         """#Requirement https://openproject.platomo.de/wp/10280"""
@@ -87,7 +95,7 @@ async def test_loads_nothing_when_the_provider_returns_nothing(self) -> None:
 
         await target.load_tracks()
 
-        given.application.add_tracks_of_files.assert_not_called()
+        given.application.add_tracks_of_files_async.assert_not_awaited()
 
 
 class TestAddVideo: