Merge pull request #515 from OpenTrafficCam/feature/4995-fallback-to-…

…video-file-name-if-relative-path-can-not-be-found

Feature/4995 fallback to video file name if relative path can not be found

OTAnalytics/plugin_parser/otconfig_parser.py

@@ -1,7 +1,7 @@
 from abc import ABC, abstractmethod
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Iterable
+from typing import Iterable, Sequence
 
 from OTAnalytics.application import project
 from OTAnalytics.application.config import (
@@ -15,6 +15,7 @@
 )
 from OTAnalytics.application.config_specification import OtConfigDefaultValueProvider
 from OTAnalytics.application.datastore import VideoParser
+from OTAnalytics.application.logger import logger
 from OTAnalytics.application.parser.config_parser import (
     AnalysisConfig,
     ConfigParser,
@@ -130,7 +131,7 @@ def parse_from_dict(self, data: dict, base_folder: Path) -> OtConfig:
         fixed_content = self._format_fixer.fix(data)
         _project = self._parse_project(fixed_content[PROJECT])
         analysis_config = self._parse_analysis(fixed_content[ANALYSIS], base_folder)
-        videos = self._video_parser.parse_list(fixed_content[video.VIDEOS], base_folder)
+        videos = self._parse_videos(fixed_content[video.VIDEOS], base_folder)
         sections, flows = self._flow_parser.parse_content(
             fixed_content[section.SECTIONS], fixed_content[flow.FLOWS]
         )
@@ -142,6 +143,30 @@ def parse_from_dict(self, data: dict, base_folder: Path) -> OtConfig:
             flows=flows,
         )
 
+    def _parse_videos(
+        self, video_entries: list[dict], base_folder: Path
+    ) -> Sequence[Video]:
+        existing_entries = []
+        for video_entry in video_entries:
+            video_file = base_folder / video_entry[PATH]
+            if video_file.exists():
+                existing_entries.append(video_entry)
+            else:
+                alternative_file = base_folder / video_file.name
+                logger().warning(
+                    f"Unable to find video file '{video_file}'. "
+                    "Try searching for video file with same name in "
+                    f"base_folder '{base_folder}'."
+                )
+                if alternative_file.exists():
+                    existing_entries.append({PATH: alternative_file.name})
+                else:
+                    raise FileNotFoundError(
+                        f"Searching for alternative video file '{alternative_file}'"
+                        "unsuccessful. Can not parse OTConfig."
+                    )
+        return self._video_parser.parse_list(existing_entries, base_folder)
+
     def _parse_project(self, data: dict) -> Project:
         _validate_data(data, [project.NAME, project.START_DATE])
         name = data[project.NAME]
@@ -217,7 +242,26 @@ def _parse_export(self, data: dict) -> ExportConfig:
     def _parse_track_files(
         self, track_files: list[str], base_folder: Path
     ) -> set[Path]:
-        return {base_folder / _file for _file in track_files}
+        existing_track_files: set[Path] = set()
+        for _file in track_files:
+            file_in_config = base_folder / _file
+            if file_in_config.exists():
+                existing_track_files.add(file_in_config)
+            else:
+                alternative_file = base_folder / file_in_config.name
+                logger().warning(
+                    f"Unable to find track file '{file_in_config}'. "
+                    "Try searching for track file with same name in "
+                    f"base_folder '{base_folder}'."
+                )
+                if alternative_file.exists():
+                    existing_track_files.add(alternative_file)
+                else:
+                    raise FileNotFoundError(
+                        f"Searching for alternative track file '{alternative_file}'"
+                        "unsuccessful. Can not parse OTConfig."
+                    )
+        return existing_track_files
 
     def serialize(
         self,

tests/OTAnalytics/plugin_parser/test_otconfig_parser.py

@@ -1,3 +1,4 @@
+import shutil
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Sequence
@@ -39,6 +40,7 @@
     EXPORT,
     LOGFILE,
     NUM_PROCESSES,
+    PATH,
     PROJECT,
     SAVE_NAME,
     SAVE_SUFFIX,
@@ -48,7 +50,7 @@
     OtConfigFormatFixer,
     OtConfigParser,
 )
-from tests.conftest import do_nothing
+from tests.conftest import YieldFixture, do_nothing
 
 
 @pytest.fixture
@@ -65,6 +67,16 @@ def mock_otconfig() -> OtConfig:
     return OtConfig(project, analysis, videos, sections, flows)
 
 
+@pytest.fixture
+def base_dir(test_data_tmp_dir: Path) -> YieldFixture[Path]:
+    base_folder = test_data_tmp_dir / "my_base_folder"
+    if base_folder.exists():
+        shutil.rmtree(base_folder)
+    base_folder.mkdir(exist_ok=False)
+    yield base_folder
+    shutil.rmtree(base_folder)
+
+
 class TestOtConfigParser:
     def test_serialize_config(
         self, test_data_tmp_dir: Path, do_nothing_fixer: OtConfigFormatFixer
@@ -233,6 +245,33 @@ def test_validate(self) -> None:
         with pytest.raises(StartDateMissing):
             OtConfigParser._validate_data(project)
 
+    def test_parse_videos_alternative_files(self, base_dir: Path) -> None:
+        expected_in_config_first_video = Path("relpath/to/video/first.mp4")
+        expected_in_config_second_video = Path("relpath/to/video/second.mp4")
+
+        actual_first_video = base_dir / expected_in_config_first_video.name
+        actual_second_video = base_dir / expected_in_config_second_video.name
+        actual_first_video.touch()
+        actual_second_video.touch()
+
+        video_entries = [
+            {PATH: expected_in_config_first_video},
+            {PATH: expected_in_config_second_video},
+        ]
+        video_parser = Mock()
+        parser = OtConfigParser(Mock(), video_parser, Mock())
+        parser._parse_videos(video_entries, base_dir)
+
+        video_parser.parse_list.assert_called_once_with(
+            [
+                {PATH: expected_in_config_first_video.name},
+                {PATH: expected_in_config_second_video.name},
+            ],
+            base_dir,
+        )
+        actual_first_video.unlink()
+        actual_second_video.unlink()
+
 
 class TestMultiFixer:
     def test_fix_all(self) -> None: