fix: replace absolute backend.app imports with relative imports

This commit is contained in:
Zac Gaetano 2026-04-14 09:55:39 -04:00
parent 979616e43b
commit 34a679d365

View file

@ -1,8 +1,8 @@
"""FFmpeg command builder for Deltacast SDI recording.""" """FFmpeg command builder for Deltacast SDI recording."""
from datetime import datetime from datetime import datetime
from backend.app.config import Settings from ..config import Settings
from backend.app.models import RecorderConfig, CodecType from ..models import RecorderConfig, CodecType
class FFmpegCommandBuilder: class FFmpegCommandBuilder:
@ -12,111 +12,40 @@ class FFmpegCommandBuilder:
self.settings = settings self.settings = settings
def build_command(self, config: RecorderConfig) -> list[str]: 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] command = [self.settings.ffmpeg_path]
# Add input device arguments
command.extend(self._get_input_args(config.port_index)) command.extend(self._get_input_args(config.port_index))
# Add codec arguments
command.extend(self._get_codec_args(config)) command.extend(self._get_codec_args(config))
# Add output file arguments
command.extend(self._get_output_args(config)) command.extend(self._get_output_args(config))
# Add SRT streaming if enabled
if config.srt_enabled: if config.srt_enabled:
command.extend(self._get_srt_output_args(config)) command.extend(self._get_srt_output_args(config))
return command return command
def _get_input_args(self, port_index: int) -> list[str]: def _get_input_args(self, port_index: int) -> list[str]:
"""Get Deltacast input device args for the given port. return ["-f", "deltacast", "-i", f"deltacast://{port_index}"]
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]: 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 = [] args = []
if config.codec == CodecType.PRORES: if config.codec == CodecType.PRORES:
args.extend(["-c:v", "prores_ks"]) args.extend(["-c:v", "prores_ks"])
# Map quality profiles to prores_ks profile levels profile_map = {"hq": "3", "mq": "2", "lq": "0"}
profile_map = { args.extend(["-profile:v", profile_map.get(config.quality_profile, "3")])
"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: elif config.codec == CodecType.DNXHD:
args.extend(["-c:v", "dnxhd"]) args.extend(["-c:v", "dnxhd", "-b:v", f"{config.bitrate}M" if config.bitrate else "185M"])
# 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: elif config.codec == CodecType.UNCOMPRESSED:
args.extend(["-c:v", "rawvideo", "-pix_fmt", "uyvy422"]) args.extend(["-c:v", "rawvideo", "-pix_fmt", "uyvy422"])
elif config.codec == CodecType.H264: elif config.codec == CodecType.H264:
args.extend(["-c:v", "libx264"]) args.extend(["-c:v", "libx264", "-b:v", f"{config.bitrate}M" if config.bitrate else "50M"])
# 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 return args
def _get_output_args(self, config: RecorderConfig) -> list[str]: 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 output_path = config.recording_path
if "{timestamp}" in output_path: if "{timestamp}" in output_path:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") output_path = output_path.replace("{timestamp}", datetime.now().strftime("%Y%m%d_%H%M%S"))
output_path = output_path.replace("{timestamp}", timestamp) if "{port_index}" in output_path:
output_path = output_path.replace("{port_index}", str(config.port_index))
return ["-f", "mxf", output_path] return ["-f", "mxf", output_path]
def _get_srt_output_args(self, config: RecorderConfig) -> list[str]: 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: if not config.srt_enabled or not config.srt_destination:
return [] return []
return ["-f", "mpegts", config.srt_destination] return ["-f", "mpegts", config.srt_destination]