Add backend/app/utils/ffmpeg.py

backend/app/utils/ffmpeg.py

@@ -0,0 +1,122 @@
+"""FFmpeg command builder for Deltacast SDI recording."""
+
+from datetime import datetime
+from backend.app.config import Settings
+from backend.app.models import RecorderConfig, CodecType
+
+
+class FFmpegCommandBuilder:
+    """Builds FFmpeg commands for Deltacast SDI recording."""
+
+    def __init__(self, settings: Settings):
+        self.settings = settings
+
+    def build_command(self, config: RecorderConfig) -> list[str]:
+        """Build complete FFmpeg command for recording a port.
+
+        Args:
+            config: RecorderConfig with port, codec, and output settings
+
+        Returns:
+            List of command arguments ready for subprocess execution
+        """
+        command = [self.settings.ffmpeg_path]
+
+        # Add input device arguments
+        command.extend(self._get_input_args(config.port_index))
+
+        # Add codec arguments
+        command.extend(self._get_codec_args(config))
+
+        # Add output file arguments
+        command.extend(self._get_output_args(config))
+
+        # Add SRT streaming if enabled
+        if config.srt_enabled:
+            command.extend(self._get_srt_output_args(config))
+
+        return command
+
+    def _get_input_args(self, port_index: int) -> list[str]:
+        """Get Deltacast input device args for the given port.
+
+        Args:
+            port_index: Port index (0-based)
+
+        Returns:
+            FFmpeg input arguments: ["-f", "deltacast", "-i", "deltacast://N"]
+        """
+        return [
+            "-f", "deltacast",
+            "-i", f"deltacast://{port_index}"
+        ]
+
+    def _get_codec_args(self, config: RecorderConfig) -> list[str]:
+        """Map CodecType to FFmpeg codec arguments.
+
+        Args:
+            config: RecorderConfig with codec selection
+
+        Returns:
+            Codec-specific FFmpeg arguments
+        """
+        args = []
+
+        if config.codec == CodecType.PRORES:
+            args.extend(["-c:v", "prores_ks"])
+            # Map quality profiles to prores_ks profile levels
+            profile_map = {
+                "hq": "3",      # High quality
+                "mq": "2",      # Medium quality
+                "lq": "0",      # Low quality
+            }
+            profile = profile_map.get(config.quality_profile, "3")
+            args.extend(["-profile:v", profile])
+
+        elif config.codec == CodecType.DNXHD:
+            args.extend(["-c:v", "dnxhd"])
+            # Use bitrate from config, default to 185M if not set
+            bitrate = f"{config.bitrate}M" if config.bitrate else "185M"
+            args.extend(["-b:v", bitrate])
+
+        elif config.codec == CodecType.UNCOMPRESSED:
+            args.extend(["-c:v", "rawvideo", "-pix_fmt", "uyvy422"])
+
+        elif config.codec == CodecType.H264:
+            args.extend(["-c:v", "libx264"])
+            # Use bitrate from config, default to 50M if not set
+            bitrate = f"{config.bitrate}M" if config.bitrate else "50M"
+            args.extend(["-b:v", bitrate])
+
+        return args
+
+    def _get_output_args(self, config: RecorderConfig) -> list[str]:
+        """Get output file arguments including format and path.
+
+        Args:
+            config: RecorderConfig with output path
+
+        Returns:
+            Output format and file path arguments
+        """
+        # Substitute {timestamp} with current datetime in format YYYYMMDD_HHMMSS
+        output_path = config.recording_path
+        if "{timestamp}" in output_path:
+            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+            output_path = output_path.replace("{timestamp}", timestamp)
+
+        return ["-f", "mxf", output_path]
+
+    def _get_srt_output_args(self, config: RecorderConfig) -> list[str]:
+        """Get SRT streaming output arguments if enabled.
+
+        Args:
+            config: RecorderConfig with SRT settings
+
+        Returns:
+            SRT output arguments or empty list if SRT disabled
+        """
+        if not config.srt_enabled or not config.srt_destination:
+            return []
+
+        return ["-f", "mpegts", config.srt_destination]