import subprocess import os from dataclasses import dataclass from config import settings from services.hardware import detect_encoder def _rendition_slug(quality: dict) -> str: """Path slug for a rendition, falling back to its label.""" return quality.get("path") or quality["label"] @dataclass class _Rendition: """One encoder process, remembered per rendition rather than per stream.""" slug: str process: subprocess.Popen class TranscodeManager: # Keyed by stream, then by rendition slug. Per-rendition tracking is what lets # a single dead encoder be noticed: a stream-level list made a dead 原画 # invisible as long as any sibling encoder was still alive. _renditions: dict[str, dict[str, _Rendition]] = {} # Signature of the last successful start, so a supervisor can tell whether a # running set of processes still matches the stream's current configuration. _signatures: dict[str, str] = {} # Number of renditions that failed to launch on the last attempt. _failed_starts: dict[str, int] = {} @staticmethod def _signature(qualities: list[dict]) -> str: # The hardware encoder is part of the signature: if detection ever changes # (a driver appears, the GPU is removed), running processes are stale. encoder = detect_encoder().name return "|".join( f"{q.get('label')}:{q.get('width')}x{q.get('height')}:{q.get('bitrate')}:" f"{'copy' if q.get('copy') else encoder}" for q in qualities ) def is_running(self, stream_key: str) -> bool: """True when every rendition this stream should have is still alive.""" return not self.dead_renditions(stream_key) def dead_renditions(self, stream_key: str) -> list[str]: """Slugs of renditions that have exited. Empty when all are healthy. Missing sessions count as dead, so a stream that was never started reports every rendition as needing one. """ renditions = self._renditions.get(stream_key) if not renditions: return ["*"] return [slug for slug, r in renditions.items() if r.process.poll() is not None] def needs_restart(self, stream_key: str, qualities: list[dict]) -> bool: if not qualities: return False if self._signatures.get(stream_key) != self._signature(qualities): return True return bool(self.dead_renditions(stream_key)) def _clear_stale_output(self, hls_dir: str) -> None: """Drop the previous run's playlist and segments. ffmpeg writes a live playlist with no #EXT-X-ENDLIST, so a playlist left behind by an encoder that died looks to hls.js exactly like a stream that simply stopped: it keeps reloading a file that will never change and buffers forever. Removing it makes the player see a missing manifest instead, which it retries until the supervisor has the encoder back up. """ if not os.path.isdir(hls_dir): return for name in os.listdir(hls_dir): if name.endswith(".ts") or name.endswith(".m3u8") or name.endswith(".m3u8.tmp"): try: os.remove(os.path.join(hls_dir, name)) except OSError: # A player holding the file open (Windows) is fine: it will be # overwritten by the restarting encoder. pass def start_transcode(self, stream_key: str, input_url: str, qualities: list[dict]): """Start whichever renditions are not already alive. Restarting only the dead ones keeps the survivors streaming: tearing down every encoder because one of them exited would black out the whole stream, and a viewer on a healthy rendition has no reason to be interrupted. """ signature = self._signature(qualities) if self._signatures.get(stream_key) != signature: # The configuration changed, so nothing on disk is known-good. self.stop_transcode(stream_key) else: self._reap_dead(stream_key) live = self._renditions.setdefault(stream_key, {}) started = 0 segment_seconds = max(1, settings.hls_segment_seconds) for quality in qualities: slug = _rendition_slug(quality) if slug in live: continue # still running; leave it alone cmd = self._build_command(stream_key, quality, input_url, segment_seconds) try: proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) except FileNotFoundError: continue live[slug] = _Rendition(slug=slug, process=proc) started += 1 self._signatures[stream_key] = signature self._failed_starts[stream_key] = max(0, len(qualities) - len(live)) return started def _reap_dead(self, stream_key: str) -> None: """Forget exited renditions so the next start can replace them.""" live = self._renditions.get(stream_key) if not live: return for slug, rendition in list(live.items()): if rendition.process.poll() is not None: del live[slug] def _build_command(self, stream_key: str, quality: dict, input_url: str, segment_seconds: int) -> list[str]: label = quality["label"] # The stream-copy ("原画") rendition carries the top rendition's numbers # only so the signature stays complete; ffmpeg copies the input rather # than encoding to them. See StreamService.transcode_renditions. is_copy = bool(quality.get("copy")) # Files land under the rendition's path slug, never its label: MediaMTX # refuses non-alphanumeric HLS paths and the label may be Chinese. path_slug = _rendition_slug(quality) width = quality["width"] height = quality["height"] fps = quality.get("fps", 30) codec = quality.get("codec", "libx264") bitrate = quality["bitrate"] crf = quality.get("crf", 23) preset = quality.get("preset", "ultrafast") audio_bitrate = quality.get("audio_bitrate", "128k") keyframe_seconds = quality.get("keyframe_seconds", segment_seconds) keyframe_frames = keyframe_seconds * fps hls_dir = os.path.join(settings.hls_runtime_dir, stream_key, path_slug) os.makedirs(hls_dir, exist_ok=True) self._clear_stale_output(hls_dir) hls_output = [ "-f", "hls", "-hls_time", str(segment_seconds), "-hls_list_size", "6", "-hls_flags", "delete_segments+independent_segments+program_date_time", "-hls_segment_filename", os.path.join(hls_dir, "segment_%03d.ts"), os.path.join(hls_dir, "index.m3u8"), ] # NOTE: ffmpeg's -reconnect* flags are HTTP/HTTPS-only protocol # options and make RTMP inputs fail with "Option not found". # RTMP drop recovery is handled by the supervisor in main.py, # which restarts exited encoders on its next tick. hls_input = ["-rw_timeout", "10000000", "-i", input_url] if is_copy: # Source ("原画") rendition: keep the publisher's video as-is and only # re-wrap it, so the output keeps the exact incoming resolution, # frame rate and bitrate. Video copy costs no CPU; audio is re-encoded # because MPEG-TS cannot carry every codec a publisher might send, and # one AAC track is a rounding error next to the video encoders. return ( ["ffmpeg", "-y"] + hls_input + [ "-c:v", "copy", "-c:a", "aac", "-b:a", "192k", "-copyts", "-start_at_zero", "-map", "0:v:0", "-map", "0:a:0?", ] + hls_output ) # Encode on the GPU when the host has one. The configured codec/preset # describe the *intended* quality, but the values themselves are # encoder-specific: libx264's -crf and "ultrafast" mean nothing to # h264_nvenc, which wants -cq and p1..p7. So the target is re-expressed # in the chosen encoder's own terms rather than passed through. encoder = detect_encoder() encode_args = ["-c:v", encoder.name] if encoder.name == codec: # The configured encoder is the one we picked: honour its settings. encode_args += ["-preset", preset, encoder.quality_flag, str(crf)] else: encode_args += ["-preset", encoder.preset, encoder.quality_flag, str(encoder.quality_value)] if encoder.supports_sc_threshold: # Only the software encoders take this; nvenc rejects it. encode_args += ["-sc_threshold", "0"] return ( ["ffmpeg", "-y"] + hls_input + encode_args + [ "-r", str(fps), "-b:v", bitrate, "-s", f"{width}x{height}", "-g", str(keyframe_frames), "-keyint_min", str(keyframe_frames), "-force_key_frames", f"expr:gte(t,n_forced*{keyframe_seconds})", "-copyts", "-start_at_zero", "-c:a", "aac", "-b:a", audio_bitrate, ] + hls_output ) def stop_transcode(self, stream_key: str): live = self._renditions.pop(stream_key, {}) for rendition in live.values(): proc = rendition.process try: proc.terminate() proc.wait(timeout=5) except Exception: try: proc.kill() except Exception: pass def stop_all(self): for key in list(self._renditions.keys()): self.stop_transcode(key) transcode_mgr = TranscodeManager()