diff --git a/docs/architecture.md b/docs/architecture.md index 6d8259c..0c06016 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -31,3 +31,16 @@ worker manager; import itself does not start work, and test configurations can disable autostart. Processing and muxing poll persisted cancellation state. The current Flask process serves the interface and API together on port 5000. A future 5175/4200 frontend/API split is a plan, not current behavior. + +Runtime render telemetry belongs to the worker manager, not the repository. A +bounded store holds caller-independent snapshots under a short lock, retains +only recent speed samples, derives elapsed time from an injectable monotonic +clock, and publishes UTC timestamps. Frame callbacks may be frequent, but the +runtime view is limited to about four updates per second and manifest progress +remains coarse. + +Durable job milestones record only meaningful phase transitions. Terminal +timing summaries survive restart; rolling speed and ETA restart as unknown. +FFmpeg stdout carries structured program progress while a separate reader drains +bounded stderr. The polling UI owns one timer and one request, rejects obsolete +job revisions, and stops on terminal or reset states. diff --git a/docs/product-direction.md b/docs/product-direction.md index 65b9f10..2a9d96a 100644 --- a/docs/product-direction.md +++ b/docs/product-direction.md @@ -22,6 +22,12 @@ readiness, and testable boundaries needed by that future workspace. GlitchCraft owns creative treatment; Web Video Optimizer remains the detailed delivery-optimization and packaging tool. +The transitional interface explains long video work with concise, progressively +disclosed telemetry. It truthfully describes applying the enabled stack to each +frame rather than pretending effects run as separate passes. Advanced temporal +feedback, optical flow, frame reordering, codec datamosh, signal modeling, and +new effect algorithms remain intentionally deferred. + A future orchestration dashboard may discover and check GlitchCraft, ColorCraft, and Web Video Optimizer through related contracts. It is not implemented here and cannot launch or remotely control this service. The family-aligned React diff --git a/docs/service-contract.md b/docs/service-contract.md index 4793e36..eb19d5f 100644 --- a/docs/service-contract.md +++ b/docs/service-contract.md @@ -39,3 +39,11 @@ Persistent storage covers image/video sources, explicit outputs, and video job state. Readiness reports queue capacity, worker concurrency, manifest migration or reconciliation state, and both required video tools. Saved recipe management, WVO handoff, authentication, and a visible library interface are deferred. + +Application version 0.2.1 advertises video-job contract v2 and the +`video-render-telemetry` capability. `/metadata` publishes supported phase +codes, the client-polling model, a four-Hz runtime update target, FFmpeg +program-progress support, and redacted queue totals. The capability is available +only when persistent video processing and its manager are available. Discovery +never exposes sample buffers, thread names, executable paths, or raw process +output. diff --git a/docs/storage.md b/docs/storage.md index 3bc5fd9..900f46d 100644 --- a/docs/storage.md +++ b/docs/storage.md @@ -1,6 +1,6 @@ # Persistent media storage -GlitchCraft 0.2.0 uses a configurable managed data root: +GlitchCraft 0.2.1 uses a configurable managed data root: ```text data/ @@ -58,3 +58,8 @@ diagnostics whose values and granularity may vary by platform. Video completion installs the final MP4 and marks its job completed in one manifest mutation. Intermediate job files remain temporary and are never served. +Manifest schema remains v2 in application 0.2.1. Telemetry contract v2 adds only +backward-compatible job defaults: a phase, at most 32 durable milestones, a +terminal timing summary, and coarse last-known frame counters. High-frequency +FPS, ETA, encoded-time, queue-position, and stale calculations are runtime-only +and never cause per-frame manifest writes. diff --git a/docs/testing.md b/docs/testing.md index f512063..46fb5f0 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -22,6 +22,7 @@ python -m pytest tests/test_media.py python -m pytest tests/test_image_assets.py tests/test_image_workflow.py python -m pytest tests/test_storage_manifest.py tests/test_service_contract.py python -m pytest tests/test_video_workflow.py +python -m pytest tests/test_telemetry.py ``` Tests use synthetic NumPy frames and temporary directories. The narrowly marked @@ -55,3 +56,23 @@ and legacy media locations with test-owned temporary directories. The video suite also covers FFprobe translation, finalizer cancellation, persistent job transitions/recovery, queue bounds, timestamp previews, and full, closed, open-ended, suffix, HEAD, and unsatisfiable byte-range responses. + +Telemetry tests use an injectable fake monotonic/UTC clock—never real sleeps—to +cover rolling and average FPS, ETA gating, unknown totals, stalled/stale work, +progress clamping, queue repositioning, restart reconstruction, terminal +summaries, and invalid/nonfinite inputs. A 1,000-callback test confirms runtime +publication remains near four Hz and the rolling sample deque stays bounded. +FFmpeg fixtures cover progress blocks, CRLF/LF, malformed fields, time +fallbacks, interleaved progress/diagnostics, bounded stderr, cancellation, +failure, and reader shutdown. + +A local 10,000-callback microbenchmark on the development machine measured +approximately 0.66 microseconds of telemetry-store overhead per frame callback. +This is diagnostic rather than a cross-platform performance guarantee; the +bounded/throttled behavior is the enforced contract. + +The Playwright video fixture walks through queued, processing, stale, +finalizing, verifying, saving, completed, cancel-pending, and canceled states. +It checks concise formatting, keyboard-accessible details, ordered effects, +previous-result retention, responsive containment, and Axe results without +requiring a long render. diff --git a/docs/video-jobs.md b/docs/video-jobs.md index 9aab126..3553603 100644 --- a/docs/video-jobs.md +++ b/docs/video-jobs.md @@ -20,3 +20,29 @@ sets `recoveredAfterRestart`, removes partial temporary artifacts through the temporary-job lifecycle, and requeues below the attempt limit. Missing source media and exhausted attempts produce controlled failures. A completed output and job are committed in one manifest mutation. + +## Render telemetry contract v2 + +Application 0.2.1 keeps the persistent states above and adds more precise phase +codes: `waiting`, `preparing_source`, `applying_effects`, +`finalizing_output`, `verifying_output`, `saving_output`, `completed`, +`failed`, and `canceled`. + +The individual job endpoint includes a stable `telemetry` object. It reports +queue position, jobs ahead, active/concurrent workers, enabled effect types, +frame counters, recent and average processing speed, elapsed time, estimated +remaining time, normalized FFmpeg progress, output bytes, attempt, and recovery +status. Unknown values are `null`; frame totals, FPS, ETA, and FFmpeg fields are +never fabricated. ETA is approximate and unavailable until enough timing data +exists. + +High-frequency values live in a lock-protected runtime store. Elapsed +calculations use a monotonic clock, public timestamps use UTC, recent-speed +samples are bounded, and publication is throttled to at most about four updates +per second. Queue positions are runtime-only and do not rewrite queued records. + +The manifest remains schema v2. Jobs created before 0.2.1 load with safe +defaults. Only bounded phase milestones, coarse frame progress, and a terminal +timing summary are durable. FFmpeg finalization uses program-progress output and +separately drains bounded diagnostics. Unknown keys and malformed optional +values are ignored; neither diagnostics nor reader-thread details enter the API. diff --git a/docs/video-workflow.md b/docs/video-workflow.md index 49ac2c5..8cd4d16 100644 --- a/docs/video-workflow.md +++ b/docs/video-workflow.md @@ -17,6 +17,21 @@ cancellation between frames and while FFmpeg runs. Clients poll `GET /api/video-jobs/{jobId}` or cancel with `POST /api/video-jobs/{jobId}/cancel`. +The interface polls detailed telemetry instead of showing only a generic +percentage. Its concise line distinguishes waiting, preparation, applying the +complete enabled effect stack to each frame, finalizing MP4, verification, +transactional saving, and completion. A semantic “Processing details” +disclosure contains source, recipe, ordered enabled effects, queue, timing, +attempt, restart, audio, and output-profile information. + +Frame totals may be unavailable for unusual media. FFmpeg builds also vary in +which progress fields they emit, so missing values remain unknown. ETA is an +estimate and remains unavailable early or while work is stalled. Telemetry is +polling-based; GlitchCraft does not use WebSockets or server-sent events. +Screen readers hear phase changes, meaningful progress thresholds, periodic +long-running updates, and terminal results rather than every frame. A previous +valid output stays visible while a new render is pending. + Completed metadata is available at `GET /api/video-outputs/{outputId}/metadata`. The output resource supports GET, HEAD, and one byte range: closed (`bytes=0-99`), open-ended (`bytes=100-`), or diff --git a/glitchcraft/jobs/manager.py b/glitchcraft/jobs/manager.py index 89e6c0c..bdc81c9 100644 --- a/glitchcraft/jobs/manager.py +++ b/glitchcraft/jobs/manager.py @@ -3,12 +3,20 @@ from __future__ import annotations import logging +import time from collections import deque from contextlib import suppress from pathlib import Path from threading import Condition, Thread from glitchcraft.errors import ProcessingCanceled, QueueCapacityError +from glitchcraft.jobs.telemetry import ( + FinalizationProgressSnapshot, + FrameProgressSnapshot, + RuntimeVideoTelemetryStore, + VideoJobPhase, + VideoJobTelemetry, +) from glitchcraft.media.ffmpeg import finalize_video from glitchcraft.media.probe import probe_video from glitchcraft.media.video import process_video @@ -35,6 +43,7 @@ def __init__( self._active: set[str] = set() self._workers: list[Thread] = [] self._stopping = False + self.telemetry_store = RuntimeVideoTelemetryStore(monotonic=time.monotonic) def start(self) -> None: with self._condition: @@ -61,14 +70,26 @@ def enqueue(self, job_id: str) -> None: raise QueueCapacityError("The video processing queue is full.") self._queue.append(job_id) self._queued.add(job_id) + self._ensure_telemetry(job_id) + self._refresh_queue_telemetry() self._condition.notify() def cancel(self, job_id: str) -> None: - self.repository.request_video_job_cancellation(job_id) + current = self.repository.request_video_job_cancellation(job_id) with self._condition: with suppress(ValueError): self._queue.remove(job_id) self._queued.discard(job_id) + self._refresh_queue_telemetry() + if current.state == VideoJobState.CANCELED: + self._ensure_telemetry(job_id) + summary = self.telemetry_store.finish(job_id, VideoJobPhase.CANCELED) + self.repository.update_video_job( + job_id, + phase=VideoJobPhase.CANCELED, + timing_summary=summary, + last_frames_processed=summary.frames_processed, + ) def shutdown(self, *, timeout: float = 5) -> None: with self._condition: @@ -79,6 +100,7 @@ def shutdown(self, *, timeout: float = 5) -> None: worker.join(timeout) with self._condition: self._workers.clear() + self._refresh_queue_telemetry() def status(self) -> dict[str, int]: with self._condition: @@ -90,6 +112,43 @@ def status(self) -> dict[str, int]: "active": len(self._active), } + def telemetry(self, job_id: str) -> VideoJobTelemetry: + self._ensure_telemetry(job_id) + snapshot = self.telemetry_store.snapshot(job_id) + if snapshot is None: + raise RuntimeError("Video job telemetry could not be initialized.") + return snapshot + + def _ensure_telemetry(self, job_id: str) -> None: + if self.telemetry_store.snapshot(job_id) is not None: + return + job = self.repository.get_video_job(job_id) + source = self.repository.get_video_source(job.source_id) + self.telemetry_store.create( + job.id, + phase=job.phase, + progress=job.progress, + enabled_effect_types=( + effect.type.value for effect in job.recipe.effects if effect.enabled + ), + audio_mode=job.audio_mode.value, + attempt=job.attempt, + recovered_after_restart=job.recovered_after_restart, + source_frame_rate=source.frame_rate, + source_duration_seconds=source.duration_seconds, + frames_processed=job.last_frames_processed, + total_frames=job.last_total_frames or source.frame_count, + timing_summary=job.timing_summary, + ) + + def _refresh_queue_telemetry(self) -> None: + self.telemetry_store.set_queue( + self._queue, + active_workers=len(self._active), + worker_concurrency=self.concurrency, + queue_capacity=self.capacity, + ) + def _worker(self) -> None: while True: with self._condition: @@ -100,11 +159,13 @@ def _worker(self) -> None: job_id = self._queue.popleft() self._queued.discard(job_id) self._active.add(job_id) + self._refresh_queue_telemetry() try: self._run(job_id) finally: with self._condition: self._active.discard(job_id) + self._refresh_queue_telemetry() def _run(self, job_id: str) -> None: intermediate: Path | None = None @@ -113,12 +174,21 @@ def _run(self, job_id: str) -> None: job = self.repository.get_video_job(job_id) if job.state != VideoJobState.QUEUED: return + self._ensure_telemetry(job_id) job = self.repository.update_video_job( job_id, state=VideoJobState.PREPARING, + phase=VideoJobPhase.PREPARING_SOURCE, + progress=1, stage="preparing", increment_attempt=not job.recovered_after_restart, ) + self.telemetry_store.transition( + job_id, + VideoJobPhase.PREPARING_SOURCE, + progress=1, + attempt=job.attempt, + ) with self.repository.lease_video_source(job.source_id) as (source, source_path): intermediate = self.repository.new_video_job_temporary_path(job_id, ".mp4") staged_output = self.repository.new_video_job_temporary_path(job_id, ".mp4") @@ -129,15 +199,26 @@ def canceled() -> bool: self.repository.update_video_job( job_id, state=VideoJobState.PROCESSING, + phase=VideoJobPhase.APPLYING_EFFECTS, progress=5, stage="processing", ) + self.telemetry_store.transition(job_id, VideoJobPhase.APPLYING_EFFECTS, progress=5) - def progress(done: int, total: int) -> None: - percent = min(89, max(5, int(done * 84 / total) + 5)) if total else 5 + def progress(snapshot: FrameProgressSnapshot) -> None: + if not self.telemetry_store.update_frames(job_id, snapshot): + return + runtime = self.telemetry_store.snapshot(job_id) + if runtime is None: + return current = self.repository.get_video_job(job_id) - if percent >= current.progress + 10: - self.repository.update_video_job(job_id, progress=percent) + if runtime.progress >= current.progress + 10: + self.repository.update_video_job( + job_id, + progress=runtime.progress, + last_frames_processed=snapshot.frames_processed, + last_total_frames=snapshot.total_frames, + ) process_video( source_path, @@ -149,16 +230,32 @@ def progress(done: int, total: int) -> None: self.repository.update_video_job( job_id, state=VideoJobState.MUXING, + phase=VideoJobPhase.FINALIZING_OUTPUT, progress=90, stage="muxing", ) + self.telemetry_store.transition( + job_id, VideoJobPhase.FINALIZING_OUTPUT, progress=90 + ) + + def finalization_progress(snapshot: FinalizationProgressSnapshot) -> None: + self.telemetry_store.update_finalization(job_id, snapshot) + finalize_video( intermediate, source_path, staged_output, preserve_audio=job.audio_mode.value == "preserve" and source.has_audio, cancellation_check=canceled, + progress_hook=finalization_progress, ) + self.repository.update_video_job( + job_id, + phase=VideoJobPhase.VERIFYING_OUTPUT, + progress=98, + stage="muxing", + ) + self.telemetry_store.transition(job_id, VideoJobPhase.VERIFYING_OUTPUT, progress=98) metadata = probe_video(staged_output) expected_audio = job.audio_mode.value == "preserve" and source.has_audio if ( @@ -168,6 +265,20 @@ def progress(done: int, total: int) -> None: or (metadata.has_audio and metadata.audio_codec != "aac") ): raise RuntimeError("The finalized video does not match the output profile.") + self.repository.update_video_job( + job_id, + phase=VideoJobPhase.SAVING_OUTPUT, + progress=99, + stage="muxing", + ) + self.telemetry_store.transition(job_id, VideoJobPhase.SAVING_OUTPUT, progress=99) + runtime = self.telemetry_store.snapshot(job_id) + output_size = staged_output.stat().st_size + summary = self.telemetry_store.finish( + job_id, + VideoJobPhase.COMPLETED, + output_bytes=output_size, + ) self.repository.complete_video_job( job_id=job_id, staged_path=staged_output, @@ -177,29 +288,55 @@ def progress(done: int, total: int) -> None: duration_seconds=metadata.duration_seconds, frame_rate=metadata.frame_rate, has_audio=metadata.has_audio, - file_size=staged_output.stat().st_size, + file_size=output_size, + timing_summary=summary, + last_frames_processed=( + runtime.frames_processed if runtime is not None else None + ), + last_total_frames=runtime.total_frames if runtime is not None else None, ) staged_output = None except ProcessingCanceled: current = self.repository.get_video_job(job_id) if not current.state.terminal: + self._ensure_telemetry(job_id) + summary = self.telemetry_store.finish(job_id, VideoJobPhase.CANCELED) self.repository.update_video_job( job_id, state=VideoJobState.CANCELED, + phase=VideoJobPhase.CANCELED, stage="canceled", cancellation_requested=True, + timing_summary=summary, + last_frames_processed=summary.frames_processed, ) except Exception: logger.exception("Persistent video job %s failed", job_id) try: current = self.repository.get_video_job(job_id) if not current.state.terminal: + self._ensure_telemetry(job_id) + runtime = self.telemetry_store.snapshot(job_id) + if runtime is not None and runtime.phase.terminal: + self.telemetry_store.transition( + job_id, + current.phase, + progress=current.progress, + ) + summary = self.telemetry_store.finish( + job_id, + VideoJobPhase.FAILED, + error_code="video_processing_failed", + ) self.repository.update_video_job( job_id, state=VideoJobState.FAILED, + phase=VideoJobPhase.FAILED, stage="failed", error_code="video_processing_failed", error_message="Video processing failed.", + timing_summary=summary, + last_frames_processed=summary.frames_processed, ) except Exception: logger.exception("Could not persist failure for video job %s", job_id) diff --git a/glitchcraft/jobs/telemetry.py b/glitchcraft/jobs/telemetry.py new file mode 100644 index 0000000..eaa820f --- /dev/null +++ b/glitchcraft/jobs/telemetry.py @@ -0,0 +1,514 @@ +"""Typed, bounded runtime telemetry for persistent video jobs.""" + +from __future__ import annotations + +from collections import deque +from collections.abc import Callable, Iterable +from dataclasses import dataclass, field +from datetime import UTC, datetime +from enum import StrEnum +from math import isfinite +from threading import Lock + +from pydantic import BaseModel, ConfigDict, Field, field_validator + + +def utc_now() -> datetime: + return datetime.now(UTC) + + +class TelemetryModel(BaseModel): + model_config = ConfigDict(extra="forbid", populate_by_name=True) + + +class VideoJobPhase(StrEnum): + WAITING = "waiting" + PREPARING_SOURCE = "preparing_source" + APPLYING_EFFECTS = "applying_effects" + FINALIZING_OUTPUT = "finalizing_output" + VERIFYING_OUTPUT = "verifying_output" + SAVING_OUTPUT = "saving_output" + COMPLETED = "completed" + FAILED = "failed" + CANCELED = "canceled" + + @property + def label(self) -> str: + return { + self.WAITING: "Waiting", + self.PREPARING_SOURCE: "Preparing source", + self.APPLYING_EFFECTS: "Applying effects", + self.FINALIZING_OUTPUT: "Finalizing MP4", + self.VERIFYING_OUTPUT: "Verifying output", + self.SAVING_OUTPUT: "Saving output", + self.COMPLETED: "Complete", + self.FAILED: "Failed", + self.CANCELED: "Canceled", + }[self] + + @property + def terminal(self) -> bool: + return self in {self.COMPLETED, self.FAILED, self.CANCELED} + + +class FrameProgressSnapshot(TelemetryModel): + frames_processed: int = Field(alias="framesProcessed", ge=0) + total_frames: int | None = Field(default=None, alias="totalFrames", ge=1) + current_frame_index: int = Field(alias="currentFrameIndex", ge=0) + source_frame_rate: float | None = Field(default=None, alias="sourceFrameRate", gt=0) + processed_duration_seconds: float | None = Field( + default=None, alias="processedDurationSeconds", ge=0 + ) + + @field_validator("source_frame_rate", "processed_duration_seconds") + @classmethod + def finite_optional(cls, value: float | None) -> float | None: + if value is not None and not isfinite(value): + raise ValueError("telemetry values must be finite") + return value + + +class FinalizationProgressSnapshot(TelemetryModel): + encoded_frame: int | None = Field(default=None, alias="encodedFrame", ge=0) + encoded_frames_per_second: float | None = Field( + default=None, alias="encodedFramesPerSecond", ge=0 + ) + encoded_duration_seconds: float | None = Field( + default=None, alias="encodedDurationSeconds", ge=0 + ) + encode_speed: float | None = Field(default=None, alias="encodeSpeed", gt=0) + output_bytes: int | None = Field(default=None, alias="outputBytes", ge=0) + + @field_validator("encoded_frames_per_second", "encoded_duration_seconds", "encode_speed") + @classmethod + def finite_optional(cls, value: float | None) -> float | None: + if value is not None and not isfinite(value): + raise ValueError("telemetry values must be finite") + return value + + +class VideoJobMilestone(TelemetryModel): + phase: VideoJobPhase + timestamp: datetime + progress: int = Field(ge=0, le=100) + + @field_validator("timestamp") + @classmethod + def utc_timestamp(cls, value: datetime) -> datetime: + if value.tzinfo is None or value.utcoffset() is None: + raise ValueError("milestone timestamps must include a UTC offset") + return value.astimezone(UTC) + + +class VideoJobTimingSummary(TelemetryModel): + frames_processed: int | None = Field(default=None, alias="framesProcessed", ge=0) + processing_duration_seconds: float | None = Field( + default=None, alias="processingDurationSeconds", ge=0 + ) + average_processing_frames_per_second: float | None = Field( + default=None, alias="averageProcessingFramesPerSecond", ge=0 + ) + finalization_duration_seconds: float | None = Field( + default=None, alias="finalizationDurationSeconds", ge=0 + ) + verification_duration_seconds: float | None = Field( + default=None, alias="verificationDurationSeconds", ge=0 + ) + total_duration_seconds: float | None = Field(default=None, alias="totalDurationSeconds", ge=0) + output_bytes: int | None = Field(default=None, alias="outputBytes", ge=0) + terminal_phase: VideoJobPhase = Field(alias="terminalPhase") + ended_during_phase: VideoJobPhase | None = Field(default=None, alias="endedDuringPhase") + error_code: str | None = Field(default=None, alias="errorCode", max_length=64) + + @field_validator( + "processing_duration_seconds", + "average_processing_frames_per_second", + "finalization_duration_seconds", + "verification_duration_seconds", + "total_duration_seconds", + ) + @classmethod + def finite_optional(cls, value: float | None) -> float | None: + if value is not None and not isfinite(value): + raise ValueError("summary values must be finite") + return value + + +class VideoJobTelemetry(TelemetryModel): + phase: VideoJobPhase + phase_label: str = Field(alias="phaseLabel") + progress: int = Field(ge=0, le=100) + updated_at: datetime = Field(alias="updatedAt") + stale: bool + enabled_effect_count: int = Field(alias="enabledEffectCount", ge=0) + enabled_effect_types: list[str] = Field(alias="enabledEffectTypes") + audio_mode: str = Field(alias="audioMode") + attempt: int = Field(ge=0) + recovered_after_restart: bool = Field(alias="recoveredAfterRestart") + queue_position: int | None = Field(default=None, alias="queuePosition", ge=1) + jobs_ahead: int | None = Field(default=None, alias="jobsAhead", ge=0) + active_workers: int = Field(alias="activeWorkers", ge=0) + worker_concurrency: int = Field(alias="workerConcurrency", ge=1) + queue_capacity: int = Field(alias="queueCapacity", ge=1) + frames_processed: int | None = Field(default=None, alias="framesProcessed", ge=0) + total_frames: int | None = Field(default=None, alias="totalFrames", ge=1) + source_frame_rate: str | None = Field(default=None, alias="sourceFrameRate") + processed_duration_seconds: float | None = Field( + default=None, alias="processedDurationSeconds", ge=0 + ) + source_duration_seconds: float | None = Field(default=None, alias="sourceDurationSeconds", ge=0) + current_frames_per_second: float | None = Field( + default=None, alias="currentFramesPerSecond", ge=0 + ) + average_frames_per_second: float | None = Field( + default=None, alias="averageFramesPerSecond", ge=0 + ) + stage_elapsed_seconds: float = Field(alias="stageElapsedSeconds", ge=0) + total_elapsed_seconds: float = Field(alias="totalElapsedSeconds", ge=0) + estimated_remaining_seconds: float | None = Field( + default=None, alias="estimatedRemainingSeconds", ge=0 + ) + encoded_frame: int | None = Field(default=None, alias="encodedFrame", ge=0) + encoded_frames_per_second: float | None = Field( + default=None, alias="encodedFramesPerSecond", ge=0 + ) + encoded_duration_seconds: float | None = Field( + default=None, alias="encodedDurationSeconds", ge=0 + ) + encode_speed: float | None = Field(default=None, alias="encodeSpeed", gt=0) + output_bytes: int | None = Field(default=None, alias="outputBytes", ge=0) + timing_summary: VideoJobTimingSummary | None = Field(default=None, alias="timingSummary") + + +@dataclass +class _RuntimeEntry: + job_id: str + phase: VideoJobPhase + progress: int + enabled_effect_types: tuple[str, ...] + audio_mode: str + attempt: int + recovered_after_restart: bool + source_frame_rate: str | None + source_duration_seconds: float | None + created_monotonic: float + stage_started_monotonic: float + updated_monotonic: float + updated_at: datetime + queue_position: int | None = None + jobs_ahead: int | None = None + active_workers: int = 0 + worker_concurrency: int = 1 + queue_capacity: int = 1 + frames_processed: int | None = None + total_frames: int | None = None + processed_duration_seconds: float | None = None + current_frames_per_second: float | None = None + average_frames_per_second: float | None = None + estimated_remaining_seconds: float | None = None + encoded_frame: int | None = None + encoded_frames_per_second: float | None = None + encoded_duration_seconds: float | None = None + encode_speed: float | None = None + output_bytes: int | None = None + timing_summary: VideoJobTimingSummary | None = None + samples: deque[tuple[float, int]] = field(default_factory=lambda: deque(maxlen=32)) + phase_durations: dict[VideoJobPhase, float] = field(default_factory=dict) + last_public_update: float = -1.0 + + +class RuntimeVideoTelemetryStore: + """Small lock-protected telemetry snapshots, independent of manifest writes.""" + + def __init__( + self, + *, + monotonic: Callable[[], float], + now: Callable[[], datetime] = utc_now, + update_interval_seconds: float = 0.25, + stale_after_seconds: float = 5.0, + ) -> None: + self._monotonic = monotonic + self._now = now + self._update_interval = update_interval_seconds + self._stale_after = stale_after_seconds + self._lock = Lock() + self._entries: dict[str, _RuntimeEntry] = {} + + def create( + self, + job_id: str, + *, + phase: VideoJobPhase, + progress: int, + enabled_effect_types: Iterable[str], + audio_mode: str, + attempt: int, + recovered_after_restart: bool, + source_frame_rate: str | None, + source_duration_seconds: float | None, + frames_processed: int | None = None, + total_frames: int | None = None, + timing_summary: VideoJobTimingSummary | None = None, + ) -> None: + now_mono = self._monotonic() + with self._lock: + self._entries[job_id] = _RuntimeEntry( + job_id=job_id, + phase=phase, + progress=max(0, min(100, progress)), + enabled_effect_types=tuple(enabled_effect_types), + audio_mode=audio_mode, + attempt=attempt, + recovered_after_restart=recovered_after_restart, + source_frame_rate=source_frame_rate, + source_duration_seconds=source_duration_seconds, + created_monotonic=now_mono, + stage_started_monotonic=now_mono, + updated_monotonic=now_mono, + updated_at=self._now(), + frames_processed=frames_processed, + total_frames=total_frames, + timing_summary=timing_summary, + ) + + def remove(self, job_id: str) -> None: + with self._lock: + self._entries.pop(job_id, None) + + def set_queue( + self, + job_ids: Iterable[str], + *, + active_workers: int, + worker_concurrency: int, + queue_capacity: int, + ) -> None: + queued = tuple(job_ids) + now_mono = self._monotonic() + now_utc = self._now() + with self._lock: + queued_set = set(queued) + for job_id, entry in self._entries.items(): + entry.active_workers = active_workers + entry.worker_concurrency = worker_concurrency + entry.queue_capacity = queue_capacity + if job_id in queued_set: + position = queued.index(job_id) + 1 + if entry.queue_position != position: + entry.updated_monotonic = now_mono + entry.updated_at = now_utc + entry.queue_position = position + entry.jobs_ahead = position - 1 + else: + entry.queue_position = None + entry.jobs_ahead = None + + def transition( + self, + job_id: str, + phase: VideoJobPhase, + *, + progress: int | None = None, + attempt: int | None = None, + ) -> None: + now_mono = self._monotonic() + with self._lock: + entry = self._entries[job_id] + elapsed = max(0.0, now_mono - entry.stage_started_monotonic) + entry.phase_durations[entry.phase] = entry.phase_durations.get(entry.phase, 0) + elapsed + entry.phase = phase + entry.stage_started_monotonic = now_mono + entry.updated_monotonic = now_mono + entry.updated_at = self._now() + entry.last_public_update = now_mono + if progress is not None: + entry.progress = max(0, min(100, progress)) + if attempt is not None: + entry.attempt = attempt + if phase != VideoJobPhase.WAITING: + entry.queue_position = None + entry.jobs_ahead = None + + def update_frames(self, job_id: str, snapshot: FrameProgressSnapshot) -> bool: + now_mono = self._monotonic() + with self._lock: + entry = self._entries[job_id] + if not entry.samples: + entry.samples.append((now_mono, snapshot.frames_processed)) + force = ( + snapshot.total_frames is not None + and snapshot.frames_processed >= snapshot.total_frames + ) + if not force and now_mono - entry.last_public_update < self._update_interval: + return False + if entry.samples[-1][0] != now_mono: + entry.samples.append((now_mono, snapshot.frames_processed)) + while len(entry.samples) > 1 and now_mono - entry.samples[0][0] > 5: + entry.samples.popleft() + stage_elapsed = max(0.0, now_mono - entry.stage_started_monotonic) + current_fps: float | None = None + if len(entry.samples) >= 2: + sample_elapsed = entry.samples[-1][0] - entry.samples[0][0] + frame_delta = entry.samples[-1][1] - entry.samples[0][1] + if sample_elapsed >= 0.5 and frame_delta >= 0: + current_fps = frame_delta / sample_elapsed + average_fps = snapshot.frames_processed / stage_elapsed if stage_elapsed > 0 else None + eta: float | None = None + speed = current_fps or average_fps + if ( + speed is not None + and speed > 0 + and stage_elapsed >= 0.5 + and snapshot.total_frames is not None + ): + eta = max(0.0, (snapshot.total_frames - snapshot.frames_processed) / speed) + ratio = ( + snapshot.frames_processed / snapshot.total_frames if snapshot.total_frames else 0 + ) + entry.progress = max(5, min(89, 5 + int(max(0.0, min(1.0, ratio)) * 84))) + entry.frames_processed = snapshot.frames_processed + entry.total_frames = snapshot.total_frames + entry.processed_duration_seconds = snapshot.processed_duration_seconds + entry.current_frames_per_second = current_fps + entry.average_frames_per_second = average_fps + entry.estimated_remaining_seconds = eta + entry.updated_monotonic = now_mono + entry.updated_at = self._now() + entry.last_public_update = now_mono + return True + + def update_finalization( + self, + job_id: str, + snapshot: FinalizationProgressSnapshot, + ) -> None: + now_mono = self._monotonic() + with self._lock: + entry = self._entries[job_id] + ratio: float | None = None + if ( + snapshot.encoded_duration_seconds is not None + and entry.source_duration_seconds + and entry.source_duration_seconds > 0 + ): + ratio = max( + 0.0, + min(1.0, snapshot.encoded_duration_seconds / entry.source_duration_seconds), + ) + eta: float | None = None + if ( + snapshot.encode_speed is not None + and ratio is not None + and snapshot.encoded_duration_seconds is not None + ): + remaining_media = max( + 0.0, (entry.source_duration_seconds or 0) - snapshot.encoded_duration_seconds + ) + eta = remaining_media / snapshot.encode_speed + if ratio is not None: + entry.progress = 90 + int(ratio * 7) + entry.encoded_frame = snapshot.encoded_frame + entry.encoded_frames_per_second = snapshot.encoded_frames_per_second + entry.encoded_duration_seconds = snapshot.encoded_duration_seconds + entry.encode_speed = snapshot.encode_speed + entry.output_bytes = snapshot.output_bytes + entry.estimated_remaining_seconds = eta + entry.updated_monotonic = now_mono + entry.updated_at = self._now() + entry.last_public_update = now_mono + + def finish( + self, + job_id: str, + phase: VideoJobPhase, + *, + output_bytes: int | None = None, + error_code: str | None = None, + ) -> VideoJobTimingSummary: + if not phase.terminal: + raise ValueError("terminal telemetry requires a terminal phase") + now_mono = self._monotonic() + with self._lock: + entry = self._entries[job_id] + ended_during = entry.phase + elapsed = max(0.0, now_mono - entry.stage_started_monotonic) + entry.phase_durations[entry.phase] = entry.phase_durations.get(entry.phase, 0) + elapsed + total = max(0.0, now_mono - entry.created_monotonic) + processing = entry.phase_durations.get(VideoJobPhase.APPLYING_EFFECTS) + finalization = entry.phase_durations.get(VideoJobPhase.FINALIZING_OUTPUT) + verification = entry.phase_durations.get(VideoJobPhase.VERIFYING_OUTPUT) + average = ( + entry.frames_processed / processing + if processing and entry.frames_processed is not None + else None + ) + summary = VideoJobTimingSummary( + frames_processed=entry.frames_processed, + processing_duration_seconds=processing, + average_processing_frames_per_second=average, + finalization_duration_seconds=finalization, + verification_duration_seconds=verification, + total_duration_seconds=total, + output_bytes=output_bytes if output_bytes is not None else entry.output_bytes, + terminal_phase=phase, + ended_during_phase=(ended_during if phase != VideoJobPhase.COMPLETED else None), + error_code=error_code, + ) + entry.phase = phase + entry.progress = 100 if phase == VideoJobPhase.COMPLETED else entry.progress + entry.stage_started_monotonic = now_mono + entry.updated_monotonic = now_mono + entry.updated_at = self._now() + entry.estimated_remaining_seconds = None + entry.timing_summary = summary + return summary + + def snapshot(self, job_id: str) -> VideoJobTelemetry | None: + now_mono = self._monotonic() + with self._lock: + entry = self._entries.get(job_id) + if entry is None: + return None + stage_elapsed = ( + 0.0 if entry.phase.terminal else max(0.0, now_mono - entry.stage_started_monotonic) + ) + total_elapsed = max(0.0, now_mono - entry.created_monotonic) + stale = ( + not entry.phase.terminal + and entry.phase != VideoJobPhase.WAITING + and now_mono - entry.updated_monotonic >= self._stale_after + ) + return VideoJobTelemetry( + phase=entry.phase, + phase_label=entry.phase.label, + progress=entry.progress, + updated_at=entry.updated_at, + stale=stale, + enabled_effect_count=len(entry.enabled_effect_types), + enabled_effect_types=list(entry.enabled_effect_types), + audio_mode=entry.audio_mode, + attempt=entry.attempt, + recovered_after_restart=entry.recovered_after_restart, + queue_position=entry.queue_position, + jobs_ahead=entry.jobs_ahead, + active_workers=entry.active_workers, + worker_concurrency=entry.worker_concurrency, + queue_capacity=entry.queue_capacity, + frames_processed=entry.frames_processed, + total_frames=entry.total_frames, + source_frame_rate=entry.source_frame_rate, + processed_duration_seconds=entry.processed_duration_seconds, + source_duration_seconds=entry.source_duration_seconds, + current_frames_per_second=entry.current_frames_per_second, + average_frames_per_second=entry.average_frames_per_second, + stage_elapsed_seconds=stage_elapsed, + total_elapsed_seconds=total_elapsed, + estimated_remaining_seconds=entry.estimated_remaining_seconds, + encoded_frame=entry.encoded_frame, + encoded_frames_per_second=entry.encoded_frames_per_second, + encoded_duration_seconds=entry.encoded_duration_seconds, + encode_speed=entry.encode_speed, + output_bytes=entry.output_bytes, + timing_summary=entry.timing_summary, + ) diff --git a/glitchcraft/media/ffmpeg.py b/glitchcraft/media/ffmpeg.py index 5d0c778..e7b0f4f 100644 --- a/glitchcraft/media/ffmpeg.py +++ b/glitchcraft/media/ffmpeg.py @@ -1,16 +1,107 @@ """FFmpeg invocation isolated from route and effect code.""" import subprocess -import tempfile import time -from collections.abc import Callable +from collections.abc import Callable, Iterator +from contextlib import suppress from pathlib import Path +from queue import Empty, Full, Queue +from threading import Thread +from typing import TextIO from glitchcraft.errors import ExternalToolError, ProcessingCanceled +from glitchcraft.jobs.telemetry import FinalizationProgressSnapshot DIAGNOSTIC_LIMIT = 4096 +def _finite_nonnegative(value: str) -> float | None: + try: + parsed = float(value) + except ValueError: + return None + if parsed < 0 or parsed != parsed or parsed in {float("inf"), float("-inf")}: + return None + return parsed + + +def _positive(value: str) -> float | None: + parsed = _finite_nonnegative(value) + return parsed if parsed is not None and parsed > 0 else None + + +def _encoded_seconds(values: dict[str, str]) -> float | None: + out_time = values.get("out_time") + if out_time: + try: + hours, minutes, seconds = out_time.split(":", 2) + parsed = int(hours) * 3600 + int(minutes) * 60 + float(seconds) + if parsed >= 0: + return parsed + except (TypeError, ValueError): + pass + for key in ("out_time_us", "out_time_ms"): + raw = values.get(key) + if raw is not None: + fallback = _finite_nonnegative(raw) + if fallback is not None: + # FFmpeg's historical out_time_ms key is expressed in microseconds. + return fallback / 1_000_000 + return None + + +def parse_ffmpeg_progress(values: dict[str, str]) -> FinalizationProgressSnapshot: + """Normalize one FFmpeg program-progress block without exposing raw fields.""" + + frame_value = _finite_nonnegative(values.get("frame", "")) + size_value = _finite_nonnegative(values.get("total_size", "")) + speed_text = values.get("speed", "") + speed = _positive(speed_text[:-1]) if speed_text.endswith("x") else None + return FinalizationProgressSnapshot( + encoded_frame=int(frame_value) if frame_value is not None else None, + encoded_frames_per_second=_finite_nonnegative(values.get("fps", "")), + encoded_duration_seconds=_encoded_seconds(values), + encode_speed=speed, + output_bytes=int(size_value) if size_value is not None else None, + ) + + +def iter_ffmpeg_progress(lines: Iterator[str]) -> Iterator[FinalizationProgressSnapshot]: + """Yield normalized snapshots for complete progress blocks.""" + + block: dict[str, str] = {} + for raw_line in lines: + line = raw_line.strip() + if not line or "=" not in line: + continue + key, value = line.split("=", 1) + if key == "progress": + yield parse_ffmpeg_progress(block) + block = {} + else: + block[key] = value + + +def _read_progress( + stream: TextIO, + updates: Queue[FinalizationProgressSnapshot], +) -> None: + for snapshot in iter_ffmpeg_progress(iter(stream.readline, "")): + try: + updates.put_nowait(snapshot) + except Full: + with suppress(Empty): + updates.get_nowait() + updates.put_nowait(snapshot) + + +def _drain_diagnostics(stream: TextIO, diagnostics: list[str]) -> None: + retained = "" + while chunk := stream.read(1024): + retained = (retained + chunk)[-DIAGNOSTIC_LIMIT:] + diagnostics.append(retained) + + def finalize_video( intermediate_path: Path, source_path: Path, @@ -18,6 +109,7 @@ def finalize_video( *, preserve_audio: bool, cancellation_check: Callable[[], bool], + progress_hook: Callable[[FinalizationProgressSnapshot], None] | None = None, ) -> None: """Create the browser-safe H.264/AAC MP4 while allowing prompt cancellation.""" @@ -29,36 +121,86 @@ def finalize_video( command.extend(["-c:a", "aac", "-b:a", "192k", "-shortest"]) else: command.append("-an") - command.extend(["-movflags", "+faststart", str(output_path)]) - with tempfile.TemporaryFile() as diagnostics: - try: - process = subprocess.Popen( - command, - stdout=subprocess.DEVNULL, - stderr=diagnostics, + command.extend( + [ + "-movflags", + "+faststart", + "-progress", + "pipe:1", + "-nostats", + str(output_path), + ] + ) + try: + process = subprocess.Popen( + command, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + encoding="utf-8", + errors="replace", + bufsize=1, + ) + except OSError as exc: + raise ExternalToolError("FFmpeg could not be started.") from exc + updates: Queue[FinalizationProgressSnapshot] = Queue(maxsize=32) + diagnostics: list[str] = [] + readers: list[Thread] = [] + stdout = getattr(process, "stdout", None) + stderr = getattr(process, "stderr", None) + if stdout is not None: + readers.append( + Thread( + target=_read_progress, + args=(stdout, updates), + name="glitchcraft-ffmpeg-progress", + daemon=True, ) - except OSError as exc: - raise ExternalToolError("FFmpeg could not be started.") from exc - try: - while process.poll() is None: - if cancellation_check(): - process.terminate() - try: - process.wait(timeout=3) - except subprocess.TimeoutExpired: - process.kill() - raise ProcessingCanceled("Video finalization was canceled.") - time.sleep(0.05) - if process.returncode: - diagnostics.seek(0, 2) - diagnostics.seek(max(0, diagnostics.tell() - DIAGNOSTIC_LIMIT)) - diagnostics.read(DIAGNOSTIC_LIMIT) - raise ExternalToolError("Video finalization failed.") - finally: - if process.poll() is None: - process.kill() - if process.returncode: - output_path.unlink(missing_ok=True) + ) + if stderr is not None: + readers.append( + Thread( + target=_drain_diagnostics, + args=(stderr, diagnostics), + name="glitchcraft-ffmpeg-diagnostics", + daemon=True, + ) + ) + for reader in readers: + reader.start() + try: + while process.poll() is None: + while True: + try: + snapshot = updates.get_nowait() + except Empty: + break + if progress_hook is not None: + progress_hook(snapshot) + if cancellation_check(): + process.terminate() + try: + process.wait(timeout=3) + except subprocess.TimeoutExpired: + process.kill() + raise ProcessingCanceled("Video finalization was canceled.") + time.sleep(0.05) + for reader in readers: + reader.join(timeout=1) + while True: + try: + snapshot = updates.get_nowait() + except Empty: + break + if progress_hook is not None: + progress_hook(snapshot) + if process.returncode: + raise ExternalToolError("Video finalization failed.") + finally: + if process.poll() is None: + process.kill() + if process.returncode: + output_path.unlink(missing_ok=True) def reencode_for_browser(input_path: Path) -> None: diff --git a/glitchcraft/media/video.py b/glitchcraft/media/video.py index 78e9205..7762bc8 100644 --- a/glitchcraft/media/video.py +++ b/glitchcraft/media/video.py @@ -11,10 +11,11 @@ from glitchcraft.contracts.effects import Recipe from glitchcraft.effects.engine import apply_effect_stack from glitchcraft.errors import MediaReadError, MediaWriteError, ProcessingCanceled +from glitchcraft.jobs.telemetry import FrameProgressSnapshot from glitchcraft.media.color import bgr_to_rgb, rgb_to_bgr from glitchcraft.media.image_io import save_image_rgb -ProgressHook = Callable[[int, int], None] +ProgressHook = Callable[[FrameProgressSnapshot], None] CancellationCheck = Callable[[], bool] @@ -98,7 +99,15 @@ def process_video( writer.write(rgb_to_bgr(processed)) frame_index += 1 if progress_hook is not None: - progress_hook(frame_index, total) + progress_hook( + FrameProgressSnapshot( + frames_processed=frame_index, + total_frames=total or None, + current_frame_index=frame_index - 1, + source_frame_rate=fps, + processed_duration_seconds=frame_index / fps, + ) + ) if frame_index == 0: raise MediaReadError("The video contained no readable frames.") finally: diff --git a/glitchcraft/service_contract.py b/glitchcraft/service_contract.py index 69eb87f..8681d58 100644 --- a/glitchcraft/service_contract.py +++ b/glitchcraft/service_contract.py @@ -19,6 +19,7 @@ "persistent-video-library", "bounded-video-jobs", "video-job-cancellation", + "video-render-telemetry", "timestamp-video-preview", "audio-preserving-video-export", "http-range-video-streaming", @@ -57,7 +58,9 @@ def effect_metadata() -> list[dict[str, Any]]: ] -def capability_details(*, storage_available: bool) -> list[dict[str, Any]]: +def capability_details( + *, storage_available: bool, video_manager_running: bool = True +) -> list[dict[str, Any]]: video_ready = ffmpeg_available() and ffprobe_available() and storage_available return [ {"slug": "image-effects", "exists": True, "available": True, "optional": False}, @@ -110,6 +113,12 @@ def capability_details(*, storage_available: bool) -> list[dict[str, Any]]: "available": video_ready, "optional": True, }, + { + "slug": "video-render-telemetry", + "exists": True, + "available": video_ready and video_manager_running, + "optional": True, + }, { "slug": "timestamp-video-preview", "exists": True, diff --git a/glitchcraft/storage/contracts.py b/glitchcraft/storage/contracts.py index aa0ac94..7f7a867 100644 --- a/glitchcraft/storage/contracts.py +++ b/glitchcraft/storage/contracts.py @@ -6,11 +6,16 @@ from enum import StrEnum from math import isfinite from pathlib import PurePosixPath -from typing import Annotated, Literal +from typing import Annotated, Any, Literal from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator from glitchcraft.contracts.effects import Recipe +from glitchcraft.jobs.telemetry import ( + VideoJobMilestone, + VideoJobPhase, + VideoJobTimingSummary, +) PositiveDimension = Annotated[int, Field(ge=1, le=1_000_000)] OpaqueId = Annotated[str, Field(min_length=16, max_length=256, pattern=r"^[A-Za-z0-9_-]+$")] @@ -314,6 +319,7 @@ class VideoJobRecord(StorageModel): recipe: Recipe audio_mode: AudioMode = AudioMode.PRESERVE state: VideoJobState = VideoJobState.QUEUED + phase: VideoJobPhase = VideoJobPhase.WAITING progress: Annotated[int, Field(ge=0, le=100)] = 0 stage: Annotated[str, Field(min_length=1, max_length=64)] = "queued" created_at: datetime @@ -323,9 +329,32 @@ class VideoJobRecord(StorageModel): attempt: Annotated[int, Field(ge=0, le=100)] = 0 recovered_after_restart: bool = False cancellation_requested: bool = False + milestones: tuple[VideoJobMilestone, ...] = () + timing_summary: VideoJobTimingSummary | None = None + last_frames_processed: Annotated[int, Field(ge=0)] | None = None + last_total_frames: Annotated[int, Field(ge=1)] | None = None error_code: Annotated[str, Field(min_length=1, max_length=64)] | None = None error_message: Annotated[str, Field(min_length=1, max_length=256)] | None = None + @model_validator(mode="before") + @classmethod + def infer_legacy_phase(cls, value: Any) -> Any: + if isinstance(value, dict) and "phase" not in value: + state = value.get("state", VideoJobState.QUEUED) + value = { + **value, + "phase": { + VideoJobState.QUEUED: VideoJobPhase.WAITING, + VideoJobState.PREPARING: VideoJobPhase.PREPARING_SOURCE, + VideoJobState.PROCESSING: VideoJobPhase.APPLYING_EFFECTS, + VideoJobState.MUXING: VideoJobPhase.FINALIZING_OUTPUT, + VideoJobState.COMPLETED: VideoJobPhase.COMPLETED, + VideoJobState.FAILED: VideoJobPhase.FAILED, + VideoJobState.CANCELED: VideoJobPhase.CANCELED, + }[VideoJobState(state)], + } + return value + @field_validator("created_at", "updated_at", "started_at", "completed_at") @classmethod def validate_timestamp(cls, value: datetime | None) -> datetime | None: @@ -333,6 +362,14 @@ def validate_timestamp(cls, value: datetime | None) -> datetime | None: @model_validator(mode="after") def validate_state_fields(self) -> VideoJobRecord: + if len(self.milestones) > 32: + raise ValueError("video job milestone history is bounded to 32 entries") + if ( + self.last_frames_processed is not None + and self.last_total_frames is not None + and self.last_frames_processed > self.last_total_frames + ): + raise ValueError("processed frames cannot exceed total frames") if self.state == VideoJobState.COMPLETED and ( self.output_id is None or self.progress != 100 or self.completed_at is None ): diff --git a/glitchcraft/storage/repository.py b/glitchcraft/storage/repository.py index d399933..e47d922 100644 --- a/glitchcraft/storage/repository.py +++ b/glitchcraft/storage/repository.py @@ -17,6 +17,11 @@ from pydantic import ValidationError from glitchcraft.contracts.effects import Recipe +from glitchcraft.jobs.telemetry import ( + VideoJobMilestone, + VideoJobPhase, + VideoJobTimingSummary, +) from glitchcraft.storage.contracts import ( CleanupRequest, CleanupResult, @@ -256,6 +261,7 @@ def _reconcile(self, *, recovered: bool) -> None: job.model_copy( update={ "state": VideoJobState.COMPLETED, + "phase": VideoJobPhase.COMPLETED, "output_id": matching_output.id, "progress": 100, "stage": "completed", @@ -263,6 +269,14 @@ def _reconcile(self, *, recovered: bool) -> None: "completed_at": now, "error_code": None, "error_message": None, + "milestones": ( + *job.milestones[-31:], + VideoJobMilestone( + phase=VideoJobPhase.COMPLETED, + timestamp=now, + progress=100, + ), + ), } ).model_dump() ) @@ -275,12 +289,22 @@ def _reconcile(self, *, recovered: bool) -> None: job.model_copy( update={ "state": VideoJobState.FAILED, + "phase": VideoJobPhase.FAILED, "output_id": None, "stage": "failed", "updated_at": now, "completed_at": now, "error_code": "reconciliation_missing_media", "error_message": "Required persistent media is unavailable.", + "timing_summary": None, + "milestones": ( + *job.milestones[-31:], + VideoJobMilestone( + phase=VideoJobPhase.FAILED, + timestamp=now, + progress=job.progress, + ), + ), } ).model_dump() ) @@ -727,6 +751,13 @@ def create_video_job( audio_mode=audio_mode, created_at=now, updated_at=now, + milestones=( + VideoJobMilestone( + phase=VideoJobPhase.WAITING, + timestamp=now, + progress=0, + ), + ), ) next_manifest = self._validated_manifest(jobs={**self._manifest.jobs, job_id: record}) self._writer.write(next_manifest) @@ -755,6 +786,7 @@ def update_video_job( job_id: str, *, state: VideoJobState | None = None, + phase: VideoJobPhase | None = None, progress: int | None = None, stage: str | None = None, cancellation_requested: bool | None = None, @@ -762,6 +794,9 @@ def update_video_job( error_message: str | None = None, increment_attempt: bool = False, recovered_after_restart: bool | None = None, + timing_summary: VideoJobTimingSummary | None = None, + last_frames_processed: int | None = None, + last_total_frames: int | None = None, ) -> VideoJobRecord: with self._lock: current = self.get_video_job(job_id) @@ -775,6 +810,18 @@ def update_video_job( changes["started_at"] = now if state.terminal: changes["completed_at"] = now + if phase is not None: + changes["phase"] = phase + if phase != current.phase: + milestone_progress = progress if progress is not None else current.progress + changes["milestones"] = ( + *current.milestones[-31:], + VideoJobMilestone( + phase=phase, + timestamp=now, + progress=milestone_progress, + ), + ) if progress is not None: changes["progress"] = progress if stage is not None: @@ -789,6 +836,12 @@ def update_video_job( changes["attempt"] = current.attempt + 1 if recovered_after_restart is not None: changes["recovered_after_restart"] = recovered_after_restart + if timing_summary is not None: + changes["timing_summary"] = timing_summary + if last_frames_processed is not None: + changes["last_frames_processed"] = last_frames_processed + if last_total_frames is not None: + changes["last_total_frames"] = last_total_frames record = VideoJobRecord.model_validate(current.model_copy(update=changes).model_dump()) next_manifest = self._validated_manifest(jobs={**self._manifest.jobs, job_id: record}) self._writer.write(next_manifest) @@ -804,6 +857,7 @@ def request_video_job_cancellation(self, job_id: str) -> VideoJobRecord: return self.update_video_job( job_id, state=VideoJobState.CANCELED, + phase=VideoJobPhase.CANCELED, progress=current.progress, stage="canceled", cancellation_requested=True, @@ -833,6 +887,9 @@ def complete_video_job( frame_rate: str, has_audio: bool, file_size: int, + timing_summary: VideoJobTimingSummary | None = None, + last_frames_processed: int | None = None, + last_total_frames: int | None = None, ) -> VideoOutputRecord: with self._lock: job = self.get_video_job(job_id) @@ -862,11 +919,23 @@ def complete_video_job( job.model_copy( update={ "state": VideoJobState.COMPLETED, + "phase": VideoJobPhase.COMPLETED, "progress": 100, "stage": "completed", "output_id": output_id, "updated_at": _utc_now(), "completed_at": _utc_now(), + "timing_summary": timing_summary, + "last_frames_processed": last_frames_processed, + "last_total_frames": last_total_frames, + "milestones": ( + *job.milestones[-31:], + VideoJobMilestone( + phase=VideoJobPhase.COMPLETED, + timestamp=_utc_now(), + progress=100, + ), + ), } ).model_dump() ) @@ -1006,7 +1075,13 @@ def recover_video_jobs(self) -> list[VideoJobRecord]: self.update_video_job( job.id, state=VideoJobState.CANCELED, + phase=VideoJobPhase.CANCELED, stage="canceled", + timing_summary=VideoJobTimingSummary( + frames_processed=job.last_frames_processed, + terminal_phase=VideoJobPhase.CANCELED, + ended_during_phase=job.phase, + ), ) ) continue @@ -1015,9 +1090,16 @@ def recover_video_jobs(self) -> list[VideoJobRecord]: self.update_video_job( job.id, state=VideoJobState.FAILED, + phase=VideoJobPhase.FAILED, stage="failed", error_code="restart_attempts_exhausted", error_message="The job exceeded its restart recovery limit.", + timing_summary=VideoJobTimingSummary( + frames_processed=job.last_frames_processed, + terminal_phase=VideoJobPhase.FAILED, + ended_during_phase=job.phase, + error_code="restart_attempts_exhausted", + ), ) ) continue @@ -1028,11 +1110,20 @@ def recover_video_jobs(self) -> list[VideoJobRecord]: job.model_copy( update={ "state": VideoJobState.QUEUED, + "phase": VideoJobPhase.WAITING, "stage": "queued", "progress": 0, "updated_at": _utc_now(), "attempt": job.attempt + 1, "recovered_after_restart": True, + "milestones": ( + *job.milestones[-31:], + VideoJobMilestone( + phase=VideoJobPhase.WAITING, + timestamp=_utc_now(), + progress=0, + ), + ), } ).model_dump() ) diff --git a/glitchcraft/version.py b/glitchcraft/version.py index 202dd9f..36ee1ad 100644 --- a/glitchcraft/version.py +++ b/glitchcraft/version.py @@ -3,8 +3,8 @@ APP_ID = "glitchcraft" APP_NAME = "GlitchCraft" APP_DESCRIPTOR = "Local visual-effects workspace" -APP_VERSION = "0.2.0" +APP_VERSION = "0.2.1" MANIFEST_SCHEMA_VERSION = 2 RECIPE_SCHEMA_VERSION = 1 STORAGE_SCHEMA_VERSION = 2 -VIDEO_JOB_SCHEMA_VERSION = 1 +VIDEO_JOB_SCHEMA_VERSION = 2 diff --git a/glitchcraft/web/routes.py b/glitchcraft/web/routes.py index d1408b4..4840198 100644 --- a/glitchcraft/web/routes.py +++ b/glitchcraft/web/routes.py @@ -32,6 +32,7 @@ from glitchcraft.errors import ExternalToolError, GlitchCraftError, QueueCapacityError from glitchcraft.jobs.contracts import VideoJobRequest, VideoPreviewRequest from glitchcraft.jobs.manager import VideoJobManager +from glitchcraft.jobs.telemetry import FrameProgressSnapshot, VideoJobPhase from glitchcraft.media.ffmpeg import reencode_for_browser from glitchcraft.media.image_io import ( SUPPORTED_IMAGE_FORMATS, @@ -286,12 +287,14 @@ def _video_output_json(output: VideoOutputRecord) -> dict[str, Any]: } -def _video_job_json(job: VideoJobRecord) -> dict[str, Any]: +def _video_job_json(job: VideoJobRecord, *, full_telemetry: bool = True) -> dict[str, Any]: + telemetry = _video_manager().telemetry(job.id) value: dict[str, Any] = { "jobId": job.id, "sourceId": job.source_id, "state": job.state, - "progress": job.progress, + "progress": telemetry.progress, + "phase": telemetry.phase, "stage": job.stage, "attempt": job.attempt, "audioMode": job.audio_mode, @@ -300,7 +303,7 @@ def _video_job_json(job: VideoJobRecord) -> dict[str, Any]: "recoveredAfterRestart": job.recovered_after_restart, "cancellationRequested": job.cancellation_requested, "createdAt": job.created_at.isoformat().replace("+00:00", "Z"), - "updatedAt": job.updated_at.isoformat().replace("+00:00", "Z"), + "updatedAt": telemetry.updated_at.isoformat().replace("+00:00", "Z"), "startedAt": ( job.started_at.isoformat().replace("+00:00", "Z") if job.started_at else None ), @@ -309,7 +312,20 @@ def _video_job_json(job: VideoJobRecord) -> dict[str, Any]: ), "statusUrl": url_for("glitchcraft.get_video_job", job_id=job.id), "cancelUrl": url_for("glitchcraft.cancel_video_job", job_id=job.id), + "queuePosition": telemetry.queue_position, + "framesProcessed": telemetry.frames_processed, + "totalFrames": telemetry.total_frames, } + if full_telemetry: + value["telemetry"] = telemetry.model_dump(mode="json", by_alias=True) + value["milestones"] = [ + milestone.model_dump(mode="json", by_alias=True) for milestone in job.milestones + ] + value["timingSummary"] = ( + job.timing_summary.model_dump(mode="json", by_alias=True) + if job.timing_summary + else None + ) if job.output_id: value["outputId"] = job.output_id try: @@ -332,8 +348,9 @@ def _video_worker( ) -> None: try: - def update_progress(done: int, total: int) -> None: - percent = min(99, int((done / max(total, done)) * 100)) + def update_progress(snapshot: FrameProgressSnapshot) -> None: + total = snapshot.total_frames or snapshot.frames_processed + percent = min(99, int((snapshot.frames_processed / total) * 100)) store.update(task_id, progress=percent) process_video(input_path, output_path, recipe, update_progress) @@ -401,6 +418,11 @@ def metadata() -> Response | tuple[Response, int]: videoJobs={ **_video_manager().status(), "states": [state.value for state in VideoJobState], + "telemetryContractVersion": VIDEO_JOB_SCHEMA_VERSION, + "telemetryPhases": [phase.value for phase in VideoJobPhase], + "pollingModel": "client-polling", + "runtimeUpdateFrequencyHz": 4, + "finalizationTelemetry": "ffmpeg-program-progress", }, standardVideoOutput={ "container": "mp4", @@ -515,10 +537,16 @@ def readiness() -> tuple[Response, int]: @bp.get("/api/capabilities") def capabilities() -> Response: repository = _repository() + manager_status = _video_manager().status() return jsonify( schemaVersion=1, service=APP_ID, - capabilities=capability_details(storage_available=repository.available), + capabilities=capability_details( + storage_available=repository.available, + video_manager_running=( + manager_status["workers"] > 0 or bool(current_app.config.get("TESTING")) + ), + ), ffmpeg={"available": ffmpeg_available()}, ffprobe={"available": ffprobe_available(), "required": False}, imageFormats=image_formats(), @@ -987,6 +1015,7 @@ def create_video_job(source_id: str) -> tuple[Response, int]: _repository().update_video_job( job.id, state=VideoJobState.FAILED, + phase=VideoJobPhase.FAILED, stage="failed", error_code="queue_full", error_message="The video processing queue is full.", @@ -1020,7 +1049,7 @@ def list_video_jobs() -> Response | tuple[Response, int]: jobs = _repository().list_video_jobs() if state is not None: jobs = [job for job in jobs if job.state == state] - return jsonify(jobs=[_video_job_json(job) for job in jobs]) + return jsonify(jobs=[_video_job_json(job, full_telemetry=False) for job in jobs]) except ValueError as exc: return _validation_error(exc) except StorageUnavailableError as exc: diff --git a/pyproject.toml b/pyproject.toml index 09aa43d..c4ef64f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "glitchcraft" -version = "0.2.0" +version = "0.2.1" description = "A local workspace for reproducible glitch and signal effects." readme = "README.md" requires-python = ">=3.11" diff --git a/static/app-manifest.json b/static/app-manifest.json index d737339..03542ba 100644 --- a/static/app-manifest.json +++ b/static/app-manifest.json @@ -10,6 +10,7 @@ "persistent-video-library", "bounded-video-jobs", "video-job-cancellation", + "video-render-telemetry", "timestamp-video-preview", "audio-preserving-video-export", "http-range-video-streaming" @@ -28,5 +29,5 @@ "id": "glitchcraft", "name": "GlitchCraft", "schemaVersion": 1, - "version": "0.2.0" + "version": "0.2.1" } diff --git a/static/app.js b/static/app.js index a86688c..eb7373c 100644 --- a/static/app.js +++ b/static/app.js @@ -33,6 +33,12 @@ videoSource: null, videoJobId: null, videoPollTimer: null, + videoPollController: null, + videoPollRevision: 0, + videoCancelPending: false, + lastAnnouncedPhase: null, + lastAnnouncedBucket: -1, + lastAnnouncementAt: 0, videoPreviewUrl: null, }; @@ -386,9 +392,7 @@ throw new Error(await errorMessage(response, "Video preview failed.")); } const blob = await response.blob(); - if (imageState.videoPreviewUrl) { - URL.revokeObjectURL(imageState.videoPreviewUrl); - } + releaseVideoPreview(); imageState.videoPreviewUrl = URL.createObjectURL(blob); document.querySelector("#preview-image").src = imageState.videoPreviewUrl; } catch (error) { @@ -396,12 +400,291 @@ } } + function releaseVideoPreview() { + if (imageState.videoPreviewUrl) { + URL.revokeObjectURL(imageState.videoPreviewUrl); + imageState.videoPreviewUrl = null; + } + } + + const effectNames = { + noise: "Noise", + pixelation: "Pixelation", + horizontal_glitch: "Horizontal glitch", + frame_shift: "Frame shift", + color_bleed: "Color bleed", + scan_lines: "Scan lines", + static: "Static", + flicker: "Flicker", + }; + + const formatInteger = (value) => + Number.isFinite(value) ? Math.round(value).toLocaleString("en-US") : "Unknown"; + + const formatFps = (value) => + Number.isFinite(value) && value >= 0 ? `${value.toFixed(1)} fps` : null; + + function formatDuration(value) { + if (!Number.isFinite(value) || value < 0) { + return "Unknown"; + } + const seconds = Math.round(value); + if (seconds < 60) { + return `${seconds}s`; + } + const minutes = Math.floor(seconds / 60); + return `${minutes}m ${seconds % 60}s`; + } + + function formatClock(value) { + if (!Number.isFinite(value) || value < 0) { + return null; + } + const seconds = Math.round(value); + return `${String(Math.floor(seconds / 60)).padStart(2, "0")}:${String(seconds % 60).padStart(2, "0")}`; + } + + const formatEta = (value) => + Number.isFinite(value) && value >= 0 + ? `about ${formatDuration(Math.round(value / 5) * 5)} remaining` + : "estimating time remaining"; + + function formatBytes(value) { + if (!Number.isFinite(value) || value < 0) { + return "Unknown"; + } + const units = ["B", "KB", "MB", "GB"]; + let size = value; + let unit = 0; + while (size >= 1024 && unit < units.length - 1) { + size /= 1024; + unit += 1; + } + return `${unit === 0 ? Math.round(size) : size.toFixed(1)} ${units[unit]}`; + } + + const effectPhrase = (count) => `Applying ${count} ${count === 1 ? "effect" : "effects"}`; + + function queuePhrase(telemetry) { + if (telemetry.recoveredAfterRestart) { + return "Restarted job — waiting to resume"; + } + if (telemetry.jobsAhead === 0) { + return "Waiting — next in queue"; + } + if (Number.isFinite(telemetry.jobsAhead)) { + return `Waiting — ${telemetry.jobsAhead} jobs ahead`; + } + return "Waiting"; + } + + function normalizedTelemetry(job) { + if (job.telemetry) { + return job.telemetry; + } + const phaseByState = { + queued: "waiting", + preparing: "preparing_source", + processing: "applying_effects", + muxing: "finalizing_output", + completed: "completed", + failed: "failed", + canceled: "canceled", + }; + const activeEffects = buildRecipe().effects.filter((effect) => effect.enabled); + return { + phase: job.phase || phaseByState[job.state] || "waiting", + phaseLabel: job.stage || job.state || "Waiting", + progress: job.progress || 0, + enabledEffectCount: activeEffects.length, + enabledEffectTypes: activeEffects.map((effect) => effect.type), + audioMode: document.querySelector("#audio-mode").value, + attempt: job.attempt || 0, + recoveredAfterRestart: Boolean(job.recoveredAfterRestart), + stale: false, + totalElapsedSeconds: null, + stageElapsedSeconds: null, + }; + } + + function setDefinitionList(element, entries) { + element.replaceChildren(); + entries.forEach(([term, value]) => { + const dt = document.createElement("dt"); + const dd = document.createElement("dd"); + dt.textContent = term; + dd.textContent = value ?? "Unknown"; + element.append(dt, dd); + }); + } + + function renderTelemetryDetails(job, telemetry) { + const source = imageState.videoSource || {}; + setDefinitionList(document.querySelector("#telemetry-source"), [ + ["Filename", source.originalName], + ["Resolution", source.width && source.height ? `${source.width} × ${source.height}` : null], + ["Duration", formatDuration(source.durationSeconds)], + ["Frame rate", telemetry.sourceFrameRate], + ["Expected frames", formatInteger(telemetry.totalFrames)], + ]); + setDefinitionList(document.querySelector("#telemetry-recipe"), [ + ["Seed", formatInteger(job.seed ?? imageState.seed)], + ["Effects", `${telemetry.enabledEffectCount} enabled`], + ["Audio", telemetry.audioMode === "remove" ? "Remove audio" : "Preserve source audio"], + ]); + const effects = document.querySelector("#telemetry-effects"); + effects.replaceChildren(); + telemetry.enabledEffectTypes.forEach((type) => { + const item = document.createElement("li"); + item.textContent = effectNames[type] || type; + effects.append(item); + }); + setDefinitionList(document.querySelector("#telemetry-work"), [ + ["Phase", telemetry.phaseLabel], + ["Current frame", formatInteger(telemetry.framesProcessed)], + ["Current speed", formatFps(telemetry.currentFramesPerSecond)], + ["Average speed", formatFps(telemetry.averageFramesPerSecond)], + ["Source time", formatDuration(telemetry.processedDurationSeconds)], + ["Stage elapsed", formatDuration(telemetry.stageElapsedSeconds)], + ["Total elapsed", formatDuration(telemetry.totalElapsedSeconds)], + ["ETA", formatEta(telemetry.estimatedRemainingSeconds)], + ["Queue position", telemetry.queuePosition ? String(telemetry.queuePosition) : null], + ]); + setDefinitionList(document.querySelector("#telemetry-job"), [ + ["Attempt", String(telemetry.attempt)], + ["Restart recovery", telemetry.recoveredAfterRestart ? "Yes" : "No"], + ["Output", "MP4 · H.264 · yuv420p"], + ["Audio output", telemetry.audioMode === "remove" ? "No audio" : "AAC, first stream"], + ["Updated", telemetry.updatedAt || job.updatedAt], + ]); + } + + function announceVideoStatus(job, telemetry, primary) { + const now = Date.now(); + const bucket = Math.floor((telemetry.progress || 0) / 10); + const phaseChanged = telemetry.phase !== imageState.lastAnnouncedPhase; + const thresholdChanged = bucket > imageState.lastAnnouncedBucket; + const timedUpdate = now - imageState.lastAnnouncementAt >= 30000; + if ( + phaseChanged || + thresholdChanged || + timedUpdate || + ["completed", "failed", "canceled"].includes(job.state) + ) { + document.querySelector("#video-progress-announcement").textContent = primary; + imageState.lastAnnouncedPhase = telemetry.phase; + imageState.lastAnnouncedBucket = bucket; + imageState.lastAnnouncementAt = now; + } + } + + function renderVideoStatus(job) { + const telemetry = normalizedTelemetry(job); + const primary = document.querySelector("#video-progress-heading"); + const metrics = document.querySelector("#video-progress-metrics"); + const context = document.querySelector("#video-progress-stage"); + const effectCount = telemetry.enabledEffectCount || 0; + let primaryText = telemetry.phaseLabel; + let metricParts = []; + let contextText = ""; + if (telemetry.phase === "waiting") { + primaryText = queuePhrase(telemetry); + contextText = `${telemetry.activeWorkers || 0} active of ${telemetry.workerConcurrency || 1} worker`; + } else if (telemetry.phase === "preparing_source") { + primaryText = telemetry.recoveredAfterRestart + ? "Resumed after restart — preparing source" + : "Preparing source"; + } else if (telemetry.phase === "applying_effects") { + primaryText = `${effectPhrase(effectCount)} — frame ${formatInteger(telemetry.framesProcessed)}`; + if (Number.isFinite(telemetry.totalFrames)) { + primaryText += ` of ${formatInteger(telemetry.totalFrames)}`; + } + metricParts = [ + `${Math.round(telemetry.progress)}%`, + formatFps(telemetry.currentFramesPerSecond), + Number.isFinite(telemetry.totalElapsedSeconds) + ? `${formatDuration(telemetry.totalElapsedSeconds)} elapsed` + : null, + formatEta(telemetry.estimatedRemainingSeconds), + ]; + } else if (telemetry.phase === "finalizing_output") { + primaryText = "Finalizing MP4"; + const encoded = formatClock(telemetry.encodedDurationSeconds); + const duration = formatClock(telemetry.sourceDurationSeconds); + if (encoded) { + primaryText += ` — ${encoded}${duration ? ` of ${duration}` : ""}`; + } + metricParts = [ + Number.isFinite(telemetry.encodeSpeed) ? `${telemetry.encodeSpeed.toFixed(1)}x` : null, + Number.isFinite(telemetry.stageElapsedSeconds) + ? `${formatDuration(telemetry.stageElapsedSeconds)} elapsed` + : null, + formatEta(telemetry.estimatedRemainingSeconds), + ]; + } else if (telemetry.phase === "verifying_output") { + primaryText = "Verifying output"; + contextText = "Checking codec, pixel format, audio, duration, and playback compatibility"; + } else if (telemetry.phase === "saving_output") { + primaryText = "Saving output"; + contextText = "Committing the processed video and completed job together"; + } else if (telemetry.phase === "completed") { + const summary = telemetry.timingSummary || job.timingSummary || {}; + primaryText = `Complete — ${formatInteger(summary.framesProcessed ?? telemetry.framesProcessed)} frames`; + if (Number.isFinite(summary.totalDurationSeconds)) { + primaryText += ` in ${formatDuration(summary.totalDurationSeconds)}`; + } + metricParts = [ + "H.264 MP4", + telemetry.audioMode === "remove" ? "audio removed" : "AAC audio preserved", + Number.isFinite(summary.outputBytes) ? formatBytes(summary.outputBytes) : null, + ]; + } else if (telemetry.phase === "canceled") { + primaryText = "Canceled — partial files removed"; + } else if (telemetry.phase === "failed") { + primaryText = "Video processing failed"; + contextText = job.error?.message || "The job could not be completed."; + } + if ( + imageState.videoCancelPending && + !["completed", "failed", "canceled"].includes(job.state) + ) { + contextText = + telemetry.phase === "finalizing_output" + ? "Stopping finalization…" + : "Canceling after the current frame…"; + } else if (telemetry.stale) { + contextText = "Waiting for a processing update…"; + } else if (telemetry.recoveredAfterRestart && telemetry.phase === "applying_effects") { + contextText = "Resumed after restart — runtime speed and ETA are being recalculated"; + } + primary.textContent = primaryText; + metrics.textContent = metricParts.filter(Boolean).join(" · "); + context.textContent = contextText; + progressBar.value = telemetry.progress; + progressBar.setAttribute("aria-valuenow", String(Math.round(telemetry.progress))); + progressBar.setAttribute("aria-valuetext", `${primaryText}. ${metrics.textContent}`); + renderTelemetryDetails(job, telemetry); + announceVideoStatus(job, telemetry, primaryText); + } + + function stopVideoPolling() { + window.clearTimeout(imageState.videoPollTimer); + imageState.videoPollTimer = null; + imageState.videoPollController?.abort(); + imageState.videoPollController = null; + } + async function processFullVideo() { if (!imageState.videoSource) { return; } const processButton = document.querySelector("#process-full"); const cancelButton = document.querySelector("#cancel-processing"); + stopVideoPolling(); + imageState.videoPollRevision += 1; + imageState.videoCancelPending = false; + imageState.lastAnnouncedPhase = null; + imageState.lastAnnouncedBucket = -1; processButton.disabled = true; cancelButton.disabled = false; try { @@ -423,7 +706,10 @@ imageState.videoJobId = job.jobId; progressPanel.hidden = false; progressBar.value = 0; - pollVideoProgress(); + if (!document.querySelector("#video-result").hidden) { + document.querySelector("#video-result-heading").textContent = "Previous processed video"; + } + pollVideoProgress(imageState.videoPollRevision, job.jobId); } catch (error) { processButton.disabled = false; cancelButton.disabled = true; @@ -431,41 +717,68 @@ } } - async function pollVideoProgress() { - if (!imageState.videoJobId) { + async function pollVideoProgress(revision, jobId) { + if ( + imageState.videoPollController || + revision !== imageState.videoPollRevision || + jobId !== imageState.videoJobId + ) { return; } + const controller = new AbortController(); + imageState.videoPollController = controller; try { const response = await fetch( - `/api/video-jobs/${encodeURIComponent(imageState.videoJobId)}`, + `/api/video-jobs/${encodeURIComponent(jobId)}`, + {signal: controller.signal}, ); if (!response.ok) { throw new Error(await errorMessage(response, "Job status is unavailable.")); } const job = await response.json(); - progressBar.value = job.progress; - document.querySelector("#video-progress-stage").textContent = - `${job.stage} · ${job.progress}%`; + if (revision !== imageState.videoPollRevision || jobId !== imageState.videoJobId) { + return; + } + renderVideoStatus(job); if (job.state === "completed") { - progressPanel.hidden = true; + stopVideoPolling(); document.querySelector("#process-full").disabled = false; document.querySelector("#cancel-processing").disabled = true; document.querySelector("#processed-video").src = job.outputUrl; document.querySelector("#download-link").href = `${job.outputUrl}/download`; + document.querySelector("#video-result-heading").textContent = "Processed video"; document.querySelector("#video-result").hidden = false; return; } if (job.state === "failed" || job.state === "canceled") { - progressPanel.hidden = true; + stopVideoPolling(); document.querySelector("#process-full").disabled = false; document.querySelector("#cancel-processing").disabled = true; - setNotice(job.error?.message || `Video job ${job.state}.`, "error"); + imageState.videoCancelPending = false; + if (job.state === "failed") { + setNotice(job.error?.message || "Video processing failed.", "error"); + } return; } - imageState.videoPollTimer = window.setTimeout(pollVideoProgress, 750); + imageState.videoPollController = null; + imageState.videoPollTimer = window.setTimeout( + () => pollVideoProgress(revision, jobId), + job.state === "queued" ? 1200 : 750, + ); } catch (error) { - progressPanel.hidden = true; - setNotice(error.message, "error"); + if (error.name !== "AbortError" && revision === imageState.videoPollRevision) { + document.querySelector("#video-progress-stage").textContent = + "Waiting for a processing update…"; + imageState.videoPollController = null; + imageState.videoPollTimer = window.setTimeout( + () => pollVideoProgress(revision, jobId), + 1200, + ); + } + } finally { + if (imageState.videoPollController === controller) { + imageState.videoPollController = null; + } } } @@ -508,23 +821,48 @@ document.querySelector("#process-full").addEventListener("click", processFullVideo); document.querySelector("#video-preview-time").addEventListener("input", schedulePreview); document.querySelector("#cancel-processing").addEventListener("click", async () => { - if (imageState.videoJobId) { - await fetch(`/api/video-jobs/${encodeURIComponent(imageState.videoJobId)}/cancel`, { - method: "POST", - }); - pollVideoProgress(); + if (imageState.videoJobId && !imageState.videoCancelPending) { + imageState.videoCancelPending = true; + const cancelButton = document.querySelector("#cancel-processing"); + cancelButton.disabled = true; + document.querySelector("#video-progress-stage").textContent = + "Canceling after the current frame…"; + try { + const response = await fetch( + `/api/video-jobs/${encodeURIComponent(imageState.videoJobId)}/cancel`, + { + method: "POST", + }, + ); + if (!response.ok) { + throw new Error(await errorMessage(response, "Cancellation could not be requested.")); + } + } catch { + imageState.videoCancelPending = false; + cancelButton.disabled = false; + } } }); document.querySelector("#cancel-preview").addEventListener("click", () => { document.querySelector("#preview-section").hidden = true; + stopVideoPolling(); + releaseVideoPreview(); + imageState.videoPollRevision += 1; + imageState.videoJobId = null; imageState.videoSource = null; fileInput.value = ""; fileInput.focus(); }); document.querySelector("#process-again").addEventListener("click", () => { + stopVideoPolling(); + releaseVideoPreview(); document.querySelector("#video-result").hidden = true; document.querySelector("#preview-section").hidden = true; fileInput.value = ""; fileInput.focus(); }); + window.addEventListener("pagehide", () => { + stopVideoPolling(); + releaseVideoPreview(); + }); })(); diff --git a/static/style.css b/static/style.css index a35f72e..88994cd 100644 --- a/static/style.css +++ b/static/style.css @@ -422,6 +422,74 @@ progress { accent-color: var(--accent); } +.progress-primary { + margin-bottom: 0.25rem; + color: var(--text); + font-size: 1.1rem; + font-weight: 700; +} + +.progress-metrics, +.progress-context { + margin-block: 0.35rem var(--space-2); +} + +.processing-details { + margin-top: var(--space-3); + border-top: 1px solid var(--border); + padding-top: var(--space-2); +} + +.processing-details summary { + width: fit-content; + cursor: pointer; + font-weight: 700; +} + +.telemetry-details-grid { + display: grid; + grid-template-columns: repeat(auto-fit, minmax(13rem, 1fr)); + gap: var(--space-3); + margin-top: var(--space-3); +} + +.telemetry-details-grid h3 { + margin-bottom: var(--space-2); + font-size: 0.95rem; +} + +.telemetry-details-grid dl { + display: grid; + grid-template-columns: max-content minmax(0, 1fr); + gap: 0.25rem 0.75rem; + margin: 0; +} + +.telemetry-details-grid dt { + color: var(--muted); +} + +.telemetry-details-grid dd { + min-width: 0; + margin: 0; + overflow-wrap: anywhere; +} + +.telemetry-details-grid ol { + margin-block: var(--space-2) 0; + padding-left: 1.25rem; +} + +.visually-hidden { + position: absolute; + width: 1px; + height: 1px; + overflow: hidden; + clip: rect(0 0 0 0); + clip-path: inset(50%); + white-space: nowrap; +} + @media (max-width: 767px) { .workspace-shell { width: min(100% - 1rem, 1440px); diff --git a/templates/index.html b/templates/index.html index e27e0da..16051c7 100644 --- a/templates/index.html +++ b/templates/index.html @@ -242,7 +242,7 @@

Review a processed frame

-