File size: 3,080 Bytes
345855e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 | from __future__ import annotations
import subprocess
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable
try:
import psutil
except Exception: # pragma: no cover - optional runtime dependency fallback
psutil = None
class FFmpegError(RuntimeError):
def __init__(self, message: str, command: list[str], stderr: str = "") -> None:
super().__init__(message)
self.command = command
self.stderr = stderr
@dataclass
class CommandResult:
command: list[str]
returncode: int
duration_seconds: float
stdout: str = ""
stderr: str = ""
metrics: dict[str, float | int] = field(default_factory=dict)
class FFmpegRunner:
def __init__(
self,
timeout_seconds: int = 900,
log: Callable[[str], None] | None = None,
on_command: Callable[[list[str]], None] | None = None,
) -> None:
self.timeout_seconds = timeout_seconds
self.log = log or (lambda _: None)
self.on_command = on_command or (lambda _: None)
def run(self, command: list[str], cwd: Path | None = None) -> CommandResult:
started = time.time()
self.on_command(command)
self.log("$ " + " ".join(command))
process = subprocess.Popen(
command,
cwd=str(cwd) if cwd else None,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
encoding="utf-8",
errors="replace",
)
peak_rss = 0
cpu_percent = 0.0
proc = psutil.Process(process.pid) if psutil else None
try:
stdout, stderr = process.communicate(timeout=self.timeout_seconds)
if proc:
try:
peak_rss = max(peak_rss, proc.memory_info().rss)
cpu_percent = proc.cpu_percent(interval=None)
except Exception:
pass
except subprocess.TimeoutExpired as exc:
self._kill_process(process)
stdout, stderr = process.communicate()
raise FFmpegError(f"FFmpeg timed out after {self.timeout_seconds}s", command, stderr) from exc
duration = time.time() - started
result = CommandResult(
command=command,
returncode=process.returncode,
duration_seconds=duration,
stdout=stdout,
stderr=stderr,
metrics={"duration_seconds": duration, "peak_rss_bytes": peak_rss, "cpu_percent": cpu_percent},
)
if process.returncode != 0:
raise FFmpegError("FFmpeg failed", command, stderr)
return result
@staticmethod
def _kill_process(process: subprocess.Popen[str]) -> None:
if psutil:
try:
parent = psutil.Process(process.pid)
for child in parent.children(recursive=True):
child.kill()
parent.kill()
return
except Exception:
pass
process.kill()
|