Fix config key mapping for saved setups

Author: steveseguin

config_loader.py

@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+from argparse import ArgumentParser, Namespace
+from typing import Any, Dict
+
+
+IGNORED_CONFIG_KEYS = {"platform", "auto_start", "custom_video_pipeline"}
+CONFIG_ARG_ALIASES = {"stream_id": "streamid"}
+VIDEO_SOURCE_OVERRIDE_ATTRS = (
+    "test",
+    "hdmi",
+    "camlink",
+    "z1",
+    "z1passthru",
+    "apple",
+    "v4l2",
+    "libcamera",
+    "rpicam",
+    "nvidiacsi",
+    "pipeline",
+    "video_pipeline",
+    "filesrc",
+    "filesrc2",
+    "pipein",
+    "novideo",
+)
+
+
+def _arg_is_default(args: Namespace, parser: ArgumentParser, attr: str) -> bool:
+    if not hasattr(args, attr):
+        return False
+    return getattr(args, attr) == parser.get_default(attr)
+
+
+def _video_source_has_cli_override(args: Namespace, parser: ArgumentParser) -> bool:
+    for attr in VIDEO_SOURCE_OVERRIDE_ATTRS:
+        if hasattr(args, attr) and getattr(args, attr) != parser.get_default(attr):
+            return True
+    return False
+
+
+def _apply_video_source_override(
+    args: Namespace,
+    parser: ArgumentParser,
+    value: Any,
+    config: Dict[str, Any],
+) -> None:
+    if _video_source_has_cli_override(args, parser):
+        return
+
+    if value == "test" and _arg_is_default(args, parser, "test"):
+        args.test = True
+    elif value == "libcamera" and _arg_is_default(args, parser, "libcamera"):
+        args.libcamera = True
+    elif value == "v4l2" and _arg_is_default(args, parser, "v4l2"):
+        args.v4l2 = "/dev/video0"
+    elif value == "custom" and _arg_is_default(args, parser, "video_pipeline"):
+        custom_pipeline = config.get("custom_video_pipeline")
+        if custom_pipeline:
+            args.video_pipeline = custom_pipeline
+
+
+def apply_config_overrides(
+    args: Namespace,
+    parser: ArgumentParser,
+    config: Dict[str, Any],
+) -> Namespace:
+    for key, value in config.items():
+        if key in IGNORED_CONFIG_KEYS:
+            continue
+
+        if key == "audio_enabled":
+            if value is False and _arg_is_default(args, parser, "noaudio"):
+                args.noaudio = True
+            continue
+
+        if key == "video_source":
+            _apply_video_source_override(args, parser, value, config)
+            continue
+
+        target_key = CONFIG_ARG_ALIASES.get(key, key)
+        if _arg_is_default(args, parser, target_key):
+            setattr(args, target_key, value)
+
+    return args

publish.py

@@ -24,6 +24,7 @@
 from pathlib import Path
 from typing import Dict, Any, Optional, Tuple, Set, List
 from functools import lru_cache
+from config_loader import apply_config_overrides
 try:
     import hashlib
     from urllib.parse import urlparse, urlencode
@@ -12271,22 +12272,7 @@ async def main():
                     config = json.load(f)
 
                 # Override args with config values (but command line args take precedence)
-                for key, value in config.items():
-                    if hasattr(args, key) and getattr(args, key) == parser.get_default(key):
-                        # Only override if the current value is the default
-                        if key == 'audio_enabled' and value is False:
-                            setattr(args, 'noaudio', True)
-                        elif key == 'video_source':
-                            if value == 'test':
-                                setattr(args, 'test', True)
-                            elif value == 'libcamera':
-                                setattr(args, 'libcamera', True)
-                            elif value == 'v4l2':
-                                setattr(args, 'v4l2', '/dev/video0')
-                            elif value == 'custom' and 'custom_video_pipeline' in config:
-                                setattr(args, 'video_pipeline', config['custom_video_pipeline'])
-                        elif key != 'platform' and key != 'auto_start' and key != 'audio_enabled' and key != 'video_source' and key != 'custom_video_pipeline':
-                            setattr(args, key, value)
+                apply_config_overrides(args, parser, config)
 
                 print(f"Loaded configuration from: {config_path}")
             except Exception as e:

tests/test_config_loader.py

@@ -0,0 +1,99 @@
+import argparse
+import unittest
+
+from config_loader import apply_config_overrides
+
+
+def build_parser():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--streamid", default="cli-default")
+    parser.add_argument("--bitrate", type=int, default=2500)
+    parser.add_argument("--test", action="store_true")
+    parser.add_argument("--hdmi", action="store_true")
+    parser.add_argument("--camlink", action="store_true")
+    parser.add_argument("--z1", action="store_true")
+    parser.add_argument("--z1passthru", action="store_true")
+    parser.add_argument("--apple", default=None)
+    parser.add_argument("--v4l2", default=None)
+    parser.add_argument("--libcamera", action="store_true")
+    parser.add_argument("--rpicam", action="store_true")
+    parser.add_argument("--nvidiacsi", action="store_true")
+    parser.add_argument("--pipeline", default=None)
+    parser.add_argument("--video-pipeline", dest="video_pipeline", default=None)
+    parser.add_argument("--filesrc", default=None)
+    parser.add_argument("--filesrc2", default=None)
+    parser.add_argument("--pipein", default=None)
+    parser.add_argument("--novideo", action="store_true")
+    parser.add_argument("--noaudio", action="store_true")
+    return parser
+
+
+class TestConfigLoader(unittest.TestCase):
+    def test_installer_config_maps_stream_id_and_test_source(self):
+        parser = build_parser()
+        args = parser.parse_args([])
+
+        apply_config_overrides(
+            args,
+            parser,
+            {
+                "stream_id": "config-stream",
+                "bitrate": 1800,
+                "video_source": "test",
+                "audio_enabled": True,
+            },
+        )
+
+        self.assertEqual(args.streamid, "config-stream")
+        self.assertEqual(args.bitrate, 1800)
+        self.assertTrue(args.test)
+        self.assertIsNone(args.v4l2)
+        self.assertFalse(args.noaudio)
+
+    def test_custom_video_source_uses_custom_pipeline(self):
+        parser = build_parser()
+        args = parser.parse_args([])
+
+        apply_config_overrides(
+            args,
+            parser,
+            {
+                "video_source": "custom",
+                "custom_video_pipeline": "videotestsrc pattern=ball",
+            },
+        )
+
+        self.assertEqual(args.video_pipeline, "videotestsrc pattern=ball")
+
+    def test_config_video_source_does_not_override_explicit_cli_source(self):
+        parser = build_parser()
+        args = parser.parse_args(["--hdmi"])
+
+        apply_config_overrides(
+            args,
+            parser,
+            {
+                "video_source": "test",
+            },
+        )
+
+        self.assertTrue(args.hdmi)
+        self.assertFalse(args.test)
+
+    def test_audio_enabled_false_sets_noaudio_when_not_explicitly_set(self):
+        parser = build_parser()
+        args = parser.parse_args([])
+
+        apply_config_overrides(
+            args,
+            parser,
+            {
+                "audio_enabled": False,
+            },
+        )
+
+        self.assertTrue(args.noaudio)
+
+
+if __name__ == "__main__":
+    unittest.main()