Merge pull request #496 from OpenTrafficCam/feature/8704-download-files-from-file-server-via-ftp-in-otvision-remote-worker

Feature/8704 download files from file server via ftp in otvision remote worker

Author: briemla

OTVision/application/config_parser.py

@@ -73,6 +73,11 @@
     _UndistortConfig,
 )
 from OTVision.domain.serialization import Deserializer
+from OTVision.plugin.ffmpeg_video_writer import (
+    ConstantRateFactor,
+    EncodingSpeed,
+    VideoCodec,
+)
 
 
 class InvalidOtvisionConfigError(Exception):
@@ -190,6 +195,21 @@ def parse_detect_config(self, data: dict) -> DetectConfig:
         if expected_duration is not None:
             expected_duration = timedelta(seconds=int(expected_duration))
 
+        if video_codec := data.get(VIDEO_CODEC, None):
+            video_codec = VideoCodec(video_codec)
+        else:
+            video_codec = DetectConfig.video_codec
+
+        if encoding_speed := data.get(ENCODING_SPEED, None):
+            encoding_speed = EncodingSpeed(encoding_speed)
+        else:
+            encoding_speed = DetectConfig.encoding_speed
+
+        if data.get(CRF, None) is not None:
+            crf = ConstantRateFactor[data[CRF]]
+        else:
+            crf = DetectConfig.crf
+
         start_time = self._parse_start_time(data)
         return DetectConfig(
             paths=sources,
@@ -202,9 +222,9 @@ def parse_detect_config(self, data: dict) -> DetectConfig:
             detect_start=data.get(DETECT_START, DetectConfig.detect_start),
             detect_end=data.get(DETECT_END, DetectConfig.detect_end),
             write_video=data.get(WRITE_VIDEO, DetectConfig.write_video),
-            video_codec=data.get(VIDEO_CODEC, DetectConfig.video_codec),
-            encoding_speed=data.get(ENCODING_SPEED, DetectConfig.encoding_speed),
-            crf=data.get(CRF, DetectConfig.crf),
+            video_codec=video_codec,
+            encoding_speed=encoding_speed,
+            crf=crf,
         )
 
     def parse_yolo_config(self, data: dict) -> YoloConfig:

tests/application/test_config_parser.py

@@ -1,15 +1,97 @@
+from datetime import datetime, timedelta
 from pathlib import Path
 from unittest.mock import Mock
 
 import pytest
 
 from OTVision.application.config import (
     Config,
+    DetectConfig,
     StreamConfig,
     TrackConfig,
+    YoloConfig,
     _TrackIouConfig,
 )
 from OTVision.application.config_parser import ConfigParser, InvalidOtvisionConfigError
+from OTVision.plugin.ffmpeg_video_writer import (
+    ConstantRateFactor,
+    EncodingSpeed,
+    VideoCodec,
+)
+
+# Ensure paths are translated to the respective platform (unix, windows)
+VIDEO_1 = str(Path("tests/data/video1.mp4"))
+VIDEO_2 = str(Path("tests/data/video2.mp4"))
+
+
+class TestConfigParser:
+    @pytest.fixture
+    def given_deserializer(self) -> Mock:
+        return Mock()
+
+    @pytest.fixture
+    def given_config_parser(self, given_deserializer: Mock) -> ConfigParser:
+        return ConfigParser(given_deserializer)
+
+    def test_parse_detect_config_with_fully_specified_config(
+        self, given_config_parser: ConfigParser
+    ) -> None:
+        detect_dict = {
+            "PATHS": [VIDEO_1, VIDEO_2],
+            "RUN_CHAINED": False,
+            "YOLO": {
+                "WEIGHTS": "yolov8m.pt",
+                "CONF": 0.35,
+                "IOU": 0.5,
+                "IMGSIZE": 1280,
+                "NORMALIZED": False,
+            },
+            "EXPECTED_DURATION": 3600,
+            "OVERWRITE": False,
+            "HALF_PRECISION": True,
+            "START_TIME": "2025-10-15_14-30-00",
+            "DETECT_START": 10,
+            "DETECT_END": 100,
+            "WRITE_VIDEO": True,
+            "VIDEO_CODEC": "h264_nvenc",
+            "ENCODING_SPEED": "medium",
+            "CRF": "HIGH_QUALITY",
+        }
+
+        result = given_config_parser.parse_detect_config(detect_dict)
+
+        expected = DetectConfig(
+            paths=[VIDEO_1, VIDEO_2],
+            run_chained=False,
+            yolo_config=YoloConfig(
+                weights="yolov8m.pt",
+                conf=0.35,
+                iou=0.5,
+                img_size=1280,
+                normalized=False,
+            ),
+            expected_duration=timedelta(seconds=3600),
+            overwrite=False,
+            half_precision=True,
+            start_time=datetime(2025, 10, 15, 14, 30, 0),
+            detect_start=10,
+            detect_end=100,
+            write_video=True,
+            video_codec=VideoCodec.H264_NVENC,
+            encoding_speed=EncodingSpeed.MEDIUM,
+            crf=ConstantRateFactor.HIGH_QUALITY,
+        )
+        assert result == expected
+
+    def test_parse_detect_config_with_empty_config(
+        self, given_config_parser: ConfigParser
+    ) -> None:
+        detect_dict: dict = {}
+
+        result = given_config_parser.parse_detect_config(detect_dict)
+
+        expected = DetectConfig()
+        assert result == expected
 
 
 class TestConfigParserValidateFlushBufferSupportTrackLifecycle: