Source code for mkv2cast.json_progress

"""
JSON progress output for mkv2cast.

This module provides structured JSON output for integration
with other applications (web UIs, monitoring tools, etc.).
"""

import json
import sys
import time
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Optional


[docs] @dataclass class FileProgress: """Progress information for a single file.""" filename: str filepath: str status: str # "queued", "checking", "encoding", "done", "skipped", "failed" progress_percent: float = 0.0 current_frame: int = 0 total_frames: int = 0 current_time_ms: int = 0 duration_ms: int = 0 fps: float = 0.0 speed: str = "" bitrate: str = "" size_bytes: int = 0 eta_seconds: float = 0.0 error: Optional[str] = None started_at: Optional[float] = None finished_at: Optional[float] = None output_path: Optional[str] = None
[docs] @dataclass class OverallProgress: """Overall progress for the entire batch.""" total_files: int = 0 processed_files: int = 0 converted_files: int = 0 skipped_files: int = 0 failed_files: int = 0 current_file: Optional[str] = None overall_percent: float = 0.0 eta_seconds: float = 0.0 started_at: Optional[float] = None backend: str = "" encode_workers: int = 1 integrity_workers: int = 1
[docs] @dataclass class JSONProgressState: """Complete state for JSON progress output.""" version: str = "1.0" timestamp: float = field(default_factory=time.time) event: str = "progress" # "start", "progress", "file_start", "file_done", "complete" overall: OverallProgress = field(default_factory=OverallProgress) files: Dict[str, FileProgress] = field(default_factory=dict) current_encoding: List[str] = field(default_factory=list) current_checking: List[str] = field(default_factory=list)
[docs] class JSONProgressOutput: """Manages JSON progress output to stdout.""" def __init__(self, stream=sys.stdout): self.stream = stream self.state = JSONProgressState() self._file_durations: Dict[str, int] = {} def _emit(self, event: str, extra: Optional[Dict[str, Any]] = None) -> None: """Emit a JSON progress event.""" self.state.timestamp = time.time() self.state.event = event output = asdict(self.state) if extra: output.update(extra) print(json.dumps(output), file=self.stream, flush=True)
[docs] def start( self, total_files: int, backend: str, encode_workers: int, integrity_workers: int, ) -> None: """Signal start of processing.""" self.state.overall = OverallProgress( total_files=total_files, backend=backend, encode_workers=encode_workers, integrity_workers=integrity_workers, started_at=time.time(), ) self._emit("start")
[docs] def set_file_duration(self, filepath: str, duration_ms: int) -> None: """Set the duration for a file (from probe).""" self._file_durations[filepath] = duration_ms
[docs] def file_queued(self, filepath: Path, duration_ms: int = 0) -> None: """Signal a file has been queued.""" key = str(filepath) self.state.files[key] = FileProgress( filename=filepath.name, filepath=key, status="queued", duration_ms=duration_ms or self._file_durations.get(key, 0), ) self._file_durations[key] = duration_ms
[docs] def file_checking(self, filepath: Path) -> None: """Signal a file integrity check has started.""" key = str(filepath) if key in self.state.files: self.state.files[key].status = "checking" self.state.files[key].started_at = time.time() self.state.current_checking.append(filepath.name) self._emit("file_checking", {"file": filepath.name})
[docs] def file_check_done(self, filepath: Path) -> None: """Signal a file integrity check has finished.""" if filepath.name in self.state.current_checking: self.state.current_checking.remove(filepath.name)
[docs] def file_encoding_start(self, filepath: Path, duration_ms: int = 0) -> None: """Signal encoding has started for a file.""" key = str(filepath) if key in self.state.files: self.state.files[key].status = "encoding" self.state.files[key].started_at = time.time() if duration_ms: self.state.files[key].duration_ms = duration_ms else: self.state.files[key] = FileProgress( filename=filepath.name, filepath=key, status="encoding", duration_ms=duration_ms or self._file_durations.get(key, 0), started_at=time.time(), ) self.state.current_encoding.append(filepath.name) self.state.overall.current_file = filepath.name self._emit("file_start", {"file": filepath.name})
[docs] def file_progress( self, filepath: Path, frame: int = 0, fps: float = 0.0, time_ms: int = 0, bitrate: str = "", speed: str = "", size_bytes: int = 0, ) -> None: """Update encoding progress for a file.""" key = str(filepath) if key not in self.state.files: return fp = self.state.files[key] fp.current_frame = frame fp.current_time_ms = time_ms fp.fps = fps fp.bitrate = bitrate fp.speed = speed fp.size_bytes = size_bytes # Calculate progress percentage if fp.duration_ms > 0: fp.progress_percent = min(100.0, (time_ms / fp.duration_ms) * 100) # Calculate ETA if time_ms > 0 and fp.started_at: elapsed = time.time() - fp.started_at remaining_ms = fp.duration_ms - time_ms if time_ms > 0: rate = elapsed / time_ms fp.eta_seconds = (remaining_ms * rate) / 1000 self._update_overall() self._emit("progress")
[docs] def file_done( self, filepath: Path, output_path: Optional[Path] = None, skipped: bool = False, error: Optional[str] = None, ) -> None: """Signal a file has finished processing.""" key = str(filepath) if key in self.state.files: fp = self.state.files[key] fp.finished_at = time.time() if error: fp.status = "failed" fp.error = error self.state.overall.failed_files += 1 elif skipped: fp.status = "skipped" self.state.overall.skipped_files += 1 else: fp.status = "done" fp.progress_percent = 100.0 self.state.overall.converted_files += 1 if output_path: fp.output_path = str(output_path) if filepath.name in self.state.current_encoding: self.state.current_encoding.remove(filepath.name) self.state.overall.processed_files += 1 self._update_overall() self._emit("file_done", {"file": filepath.name, "status": fp.status if key in self.state.files else "unknown"})
[docs] def complete(self) -> None: """Signal all processing is complete.""" self.state.overall.overall_percent = 100.0 self._emit("complete")
def _update_overall(self) -> None: """Update overall progress statistics.""" total = self.state.overall.total_files if total > 0: # Weight: completed files + partial progress of encoding files completed_weight = self.state.overall.processed_files encoding_weight = sum( fp.progress_percent / 100.0 for fp in self.state.files.values() if fp.status == "encoding" ) self.state.overall.overall_percent = ((completed_weight + encoding_weight) / total) * 100 # Estimate overall ETA based on average processing time if self.state.overall.started_at and self.state.overall.processed_files > 0: elapsed = time.time() - self.state.overall.started_at avg_time = elapsed / (self.state.overall.processed_files + encoding_weight) remaining = total - self.state.overall.processed_files - encoding_weight self.state.overall.eta_seconds = avg_time * remaining
[docs] def parse_ffmpeg_progress_for_json(line: str) -> Dict[str, Any]: """Parse FFmpeg progress line for JSON output. Returns a dict with parsed values. """ result: Dict[str, Any] = {} # frame=12345 if "frame=" in line: try: frame_part = line.split("frame=")[1].split()[0] result["frame"] = int(frame_part) except (IndexError, ValueError): pass # fps=123.45 if "fps=" in line: try: fps_part = line.split("fps=")[1].split()[0] result["fps"] = float(fps_part) except (IndexError, ValueError): pass # time=00:01:23.45 if "time=" in line: try: time_part = line.split("time=")[1].split()[0] if time_part and time_part != "N/A": parts = time_part.split(":") if len(parts) == 3: h, m, s = parts time_ms = int((int(h) * 3600 + int(m) * 60 + float(s)) * 1000) result["time_ms"] = time_ms except (IndexError, ValueError): pass # bitrate=1234.5kbits/s if "bitrate=" in line: try: bitrate_part = line.split("bitrate=")[1].split()[0] result["bitrate"] = bitrate_part except (IndexError, ValueError): pass # speed=1.23x if "speed=" in line: try: speed_part = line.split("speed=")[1].split()[0] result["speed"] = speed_part except (IndexError, ValueError): pass # size=12345kB if "size=" in line: try: size_part = line.split("size=")[1].split()[0] if size_part.endswith("kB"): result["size_bytes"] = int(float(size_part[:-2]) * 1024) elif size_part.endswith("mB"): result["size_bytes"] = int(float(size_part[:-2]) * 1024 * 1024) except (IndexError, ValueError): pass return result