239 lines
8.1 KiB
Python
239 lines
8.1 KiB
Python
"""SRT streaming for Deltacast SDI input."""
|
|
|
|
import asyncio
|
|
import logging
|
|
from urllib.parse import urlparse, parse_qs, urlencode, urlunparse
|
|
from backend.app.config import Settings
|
|
from backend.app.models import RecorderConfig, CodecType
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class SRTStreamer:
|
|
"""
|
|
Independent SRT output manager for streaming without recording.
|
|
Manages a separate FFmpeg process for SRT-only output.
|
|
"""
|
|
|
|
def __init__(self, port_index: int, settings: Settings):
|
|
"""Initialize SRT streamer.
|
|
|
|
Args:
|
|
port_index: 0-based Deltacast port index
|
|
settings: Application settings
|
|
"""
|
|
self.port_index = port_index
|
|
self.settings = settings
|
|
self._process: asyncio.subprocess.Process | None = None
|
|
self._config: RecorderConfig | None = None
|
|
self._is_streaming: bool = False
|
|
|
|
async def start_stream(self, config: RecorderConfig) -> None:
|
|
"""Start SRT-only FFmpeg stream.
|
|
|
|
Args:
|
|
config: RecorderConfig with codec and SRT settings
|
|
|
|
Raises:
|
|
ValueError: If SRT not enabled in config or destination not set
|
|
RuntimeError: If already streaming
|
|
"""
|
|
if not config.srt_enabled:
|
|
raise ValueError("SRT is not enabled in config")
|
|
|
|
if not config.srt_destination:
|
|
raise ValueError("SRT destination is not set in config")
|
|
|
|
if self._is_streaming:
|
|
raise RuntimeError(f"Port {self.port_index} is already streaming SRT")
|
|
|
|
self._config = config
|
|
|
|
# Build FFmpeg command
|
|
command = self._build_srt_command(config)
|
|
logger.info(f"Port {self.port_index}: Starting SRT stream: {' '.join(command)}")
|
|
|
|
# Spawn subprocess
|
|
self._process = await asyncio.create_subprocess_exec(
|
|
*command,
|
|
stdin=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
)
|
|
|
|
self._is_streaming = True
|
|
logger.info(f"Port {self.port_index}: SRT stream started (PID: {self._process.pid})")
|
|
|
|
async def stop_stream(self) -> None:
|
|
"""Stop SRT stream gracefully: stdin q -> SIGTERM -> SIGKILL.
|
|
|
|
The subprocess is terminated using escalating methods:
|
|
1. Send 'q\n' to stdin (graceful quit)
|
|
2. Wait up to 3 seconds for process to exit
|
|
3. Send SIGTERM
|
|
4. Wait up to 2 seconds for process to exit
|
|
5. Send SIGKILL
|
|
|
|
If already stopped, this is a no-op.
|
|
"""
|
|
if not self._is_streaming:
|
|
logger.info(f"Port {self.port_index}: Not currently streaming")
|
|
return
|
|
|
|
logger.info(f"Port {self.port_index}: Stopping SRT stream (PID: {self._process.pid})")
|
|
|
|
try:
|
|
# Step 1: Try graceful quit via stdin
|
|
if self._process and self._process.stdin and not self._process.stdin.is_closing():
|
|
self._process.stdin.write(b"q\n")
|
|
await self._process.stdin.drain()
|
|
logger.debug(f"Port {self.port_index}: Sent quit command to stdin")
|
|
|
|
# Step 2: Wait up to 3 seconds for graceful shutdown
|
|
try:
|
|
await asyncio.wait_for(self._process.wait(), timeout=3.0)
|
|
logger.info(f"Port {self.port_index}: SRT stream exited gracefully")
|
|
self._process = None
|
|
self._is_streaming = False
|
|
return
|
|
except asyncio.TimeoutError:
|
|
logger.warning(f"Port {self.port_index}: Graceful quit timeout, sending SIGTERM")
|
|
|
|
# Step 3: Send SIGTERM
|
|
if self._process:
|
|
try:
|
|
self._process.terminate()
|
|
except ProcessLookupError:
|
|
pass # Process already terminated
|
|
except Exception as e:
|
|
logger.error(f"Port {self.port_index}: Error sending SIGTERM: {e}")
|
|
|
|
# Step 4: Wait up to 2 seconds for SIGTERM to work
|
|
try:
|
|
await asyncio.wait_for(self._process.wait(), timeout=2.0)
|
|
logger.info(f"Port {self.port_index}: SRT stream exited after SIGTERM")
|
|
self._process = None
|
|
self._is_streaming = False
|
|
return
|
|
except asyncio.TimeoutError:
|
|
logger.warning(f"Port {self.port_index}: SIGTERM timeout, sending SIGKILL")
|
|
|
|
# Step 5: Send SIGKILL
|
|
if self._process:
|
|
try:
|
|
self._process.kill()
|
|
await self._process.wait()
|
|
logger.info(f"Port {self.port_index}: SRT stream killed")
|
|
except ProcessLookupError:
|
|
pass # Process already terminated
|
|
except Exception as e:
|
|
logger.error(f"Port {self.port_index}: Error sending SIGKILL: {e}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Port {self.port_index}: Error stopping SRT stream: {e}")
|
|
finally:
|
|
self._process = None
|
|
self._is_streaming = False
|
|
|
|
@property
|
|
def is_streaming(self) -> bool:
|
|
"""Return current streaming state."""
|
|
return self._is_streaming
|
|
|
|
def _build_srt_command(self, config: RecorderConfig) -> list[str]:
|
|
"""Build FFmpeg command for SRT-only output (no file recording).
|
|
|
|
Command structure:
|
|
ffmpeg -f deltacast -i deltacast://{port_index} [codec_args] -f mpegts {srt_destination}
|
|
|
|
Args:
|
|
config: RecorderConfig with port, codec, and SRT settings
|
|
|
|
Returns:
|
|
List of command arguments ready for subprocess execution
|
|
"""
|
|
command = [self.settings.ffmpeg_path]
|
|
|
|
# Add input device arguments
|
|
command.extend([
|
|
"-f", "deltacast",
|
|
"-i", f"deltacast://{self.port_index}"
|
|
])
|
|
|
|
# Add codec arguments
|
|
command.extend(self._get_codec_args(config))
|
|
|
|
# Add SRT output with mpegts format
|
|
srt_destination = self._add_latency_to_srt_url(config.srt_destination)
|
|
command.extend([
|
|
"-f", "mpegts",
|
|
srt_destination
|
|
])
|
|
|
|
return command
|
|
|
|
def _get_codec_args(self, config: RecorderConfig) -> list[str]:
|
|
"""Map CodecType to FFmpeg codec arguments for SRT output.
|
|
|
|
Args:
|
|
config: RecorderConfig with codec selection
|
|
|
|
Returns:
|
|
Codec-specific FFmpeg arguments
|
|
"""
|
|
args = []
|
|
|
|
if config.codec == CodecType.PRORES:
|
|
args.extend(["-c:v", "prores_ks"])
|
|
profile_map = {
|
|
"hq": "3",
|
|
"mq": "2",
|
|
"lq": "0",
|
|
}
|
|
profile = profile_map.get(config.quality_profile, "3")
|
|
args.extend(["-profile:v", profile])
|
|
|
|
elif config.codec == CodecType.DNXHD:
|
|
args.extend(["-c:v", "dnxhd"])
|
|
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"])
|
|
bitrate = f"{config.bitrate}M" if config.bitrate else "8M"
|
|
args.extend(["-b:v", bitrate])
|
|
|
|
return args
|
|
|
|
def _add_latency_to_srt_url(self, srt_url: str) -> str:
|
|
"""Add or override latency parameter in SRT URL.
|
|
|
|
Args:
|
|
srt_url: SRT destination URL (e.g., srt://host:port or srt://host:port?latency=...)
|
|
|
|
Returns:
|
|
SRT URL with latency parameter set from settings
|
|
"""
|
|
parsed = urlparse(srt_url)
|
|
params = parse_qs(parsed.query, keep_blank_values=True)
|
|
|
|
# Set latency from settings
|
|
params["latency"] = [str(self.settings.srt_latency)]
|
|
|
|
# Reconstruct query string
|
|
new_query = urlencode(params, doseq=True)
|
|
|
|
# Reconstruct URL
|
|
new_url = urlunparse((
|
|
parsed.scheme,
|
|
parsed.netloc,
|
|
parsed.path,
|
|
parsed.params,
|
|
new_query,
|
|
parsed.fragment
|
|
))
|
|
|
|
return new_url
|