"""Convert tab — re-encode audio and video to a consistent, validated form.
Two sections: **Audio** (optional) re-encodes audio to a uniform format / sample
rate / channel count; **Video** re-encodes to constant-frame-rate H.264 and
validates each output with torchcodec (decode every frame, exact == approximate
count) so the dataset never trips the broken-seeking class of bug downstream.
Audio/video combination is explicit: mux the matching-stem audio into each video,
or keep videos frame-only; and an audio file with no matching video can become a
black-frame video carrying that audio, or be left as audio-only. A shared progress
panel shows ``X/Y`` with a bar, elapsed time, and an ETA while the background batch
runs.
"""
from __future__ import annotations
import asyncio
import contextlib
from dataclasses import dataclass
from datetime import datetime, timedelta
from pathlib import Path
from typing import TYPE_CHECKING
from nicegui import background_tasks, run, ui
from annie.app_state import state
from annie.core import theme
from annie.media import convert
from annie.pages.folder_picker import pick_directory
from annie.pages.utils import _alive
if TYPE_CHECKING:
from collections.abc import Callable
from nicegui.events import ClickEventArguments
from annie.media.convert import ConvertProgress
@dataclass(frozen=True, slots=True)
class _Panel:
"""One page's progress widgets (each open page has its own)."""
title: ui.label #: current-file / status line
bar: ui.linear_progress
counts: ui.label #: ``X/Y`` processed count
times: ui.label #: elapsed / ETA readout
failed: ui.label #: failure count, shown when non-zero
cancel: ui.button #: cancels the running batch
#: Every open page's progress panel, by client id: the conversion is shared, so all of
#: them show it, not just the page opened last.
_panels: dict[str, _Panel] = {}
#: Whether a watcher task is already refreshing the panels.
_watching = False
def _fmt_duration(seconds: float) -> str:
"""Format a duration as ``1h 2m`` / ``2m 3s`` / ``3s``."""
total = int(seconds)
minutes, sec = divmod(total, 60)
hours, minutes = divmod(minutes, 60)
if hours:
return f"{hours}h {minutes}m"
if minutes:
return f"{minutes}m {sec}s"
return f"{sec}s"
def _dir_input(label: str, value: str = "") -> ui.input:
"""A folder text field with a Browse button; returns the input element."""
with ui.row().classes("w-full items-center gap-2"):
button = ui.button(icon="folder_open").props("flat round").tooltip(f"Browse for {label}")
field = ui.input(label, value=value).classes("flex-grow")
async def choose() -> None:
chosen = await pick_directory(field.value)
if chosen:
field.set_value(chosen)
button.on_click(choose)
return field
# ── progress panel ───────────────────────────────────────────────────────────
async def _export_succeeded() -> None:
"""Pick a folder and export the converted ids to ``reencoded_ids.csv``."""
succeeded = state.converter.progress().succeeded
if not succeeded:
ui.notify("No converted files to export yet.", color=theme.WARNING)
return
chosen = await pick_directory()
if chosen:
await _write_export(Path(chosen), succeeded, [])
async def _export_failed() -> None:
"""Pick a folder and export the failed ids + errors to ``failed_ids.csv``."""
failed = state.converter.progress().failed
if not failed:
ui.notify("No failed files to export.", color=theme.WARNING)
return
chosen = await pick_directory()
if chosen:
await _write_export(Path(chosen), [], failed)
async def _write_export(
folder: Path, succeeded: list[str], failed: list[tuple[str, str]]
) -> Path | None:
"""Write the converted ids or the failures (off the event loop) into ``folder``.
Returns:
The written file, or ``None`` when there was nothing to write.
"""
if succeeded:
out = await run.io_bound(convert.export_ids, folder / "reencoded_ids.csv", succeeded)
ui.notify(f"Exported {len(succeeded)} id(s) to {out}", color=theme.PRIMARY)
return out
if failed:
out = await run.io_bound(convert.export_failures, folder / "failed_ids.csv", failed)
ui.notify(f"Exported {len(failed)} failed id(s) to {out}", color=theme.PRIMARY)
return out
return None
def _build_progress() -> None:
"""Build this page's progress panel and render the current state into it."""
with ui.card().classes("w-full gap-1"):
title = ui.label().classes("font-medium")
bar = ui.linear_progress(value=0.0, show_value=False).props("rounded")
counts = ui.label().classes("text-sm")
times = ui.label().classes("text-xs").style(f"color:{theme.NEUTRAL}")
failed = ui.label().classes("text-xs").style(f"color:{theme.DANGER}")
with ui.row().classes("items-center gap-2"):
cancel = ui.button("Cancel", icon="stop", on_click=state.converter.cancel).props("flat")
ui.button("Export reencoded ids", icon="download", on_click=_export_succeeded).props(
"flat"
)
ui.button("Export failed ids", icon="download", on_click=_export_failed).props("flat")
client = title.client
panel = _Panel(title, bar, counts, times, failed, cancel)
_panels[client.id] = panel
client.on_delete(lambda: _panels.pop(client.id, None))
_render_panel(panel, state.converter.progress())
def _render_progress(progress: ConvertProgress | None = None) -> None:
"""Update every open page's progress panel from one snapshot of the converter.
Args:
progress: The snapshot to show (default: the converter's current one).
"""
snapshot = state.converter.progress() if progress is None else progress
for client_id, panel in list(_panels.items()):
if not _alive(panel.title):
_panels.pop(client_id, None)
continue
try:
_render_panel(panel, snapshot)
except RuntimeError: # the panel was rebuilt/refreshed away mid-update
_panels.pop(client_id, None)
def _render_panel(panel: _Panel, p: ConvertProgress) -> None:
"""Render one converter snapshot into one page's progress panel."""
if p.status == "idle":
panel.title.set_text("No conversion running.")
panel.bar.set_value(0.0)
panel.counts.set_text("")
panel.times.set_text("")
panel.failed.set_text("")
panel.cancel.set_visibility(False)
return
panel.title.set_text(f"{p.title} — {p.status}")
panel.bar.set_value(p.fraction)
panel.counts.set_text(
f"{p.done}/{p.total} ({p.fraction * 100:.0f}%)"
f" · converted {p.succeeded_count} · failed {len(p.failed)}"
+ (f" · {p.current}" if p.current else "")
)
parts: list[str] = []
if p.started_at is not None:
parts.append(f"started {datetime.fromtimestamp(p.started_at):%H:%M:%S}")
parts.append(f"elapsed {_fmt_duration(p.elapsed)}")
eta = p.eta_seconds
if eta is not None:
finish = datetime.now() + timedelta(seconds=eta)
parts.append(f"~{_fmt_duration(eta)} left (≈{finish:%H:%M:%S})")
panel.times.set_text(" · ".join(parts))
if p.failed:
names = ", ".join(name for name, _ in p.failed[:5])
more = "…" if len(p.failed) > 5 else ""
panel.failed.set_text(f"{len(p.failed)} failed: {names}{more}")
else:
panel.failed.set_text("")
panel.cancel.set_visibility(p.status == "running")
async def _watch() -> None:
"""Poll the converter and refresh every progress panel until the batch ends.
One watcher serves all open pages; a page opened mid-batch does not start another.
"""
global _watching
if _watching:
return
_watching = True
try:
while True:
_render_progress()
if state.converter.progress().status != "running":
return
await asyncio.sleep(0.4)
finally:
_watching = False
async def _planning[T](event: ClickEventArguments, plan: Callable[[], T]) -> T | None:
"""Plan a batch off the event loop while the clicked button shows it is busy.
Planning lists (and for video, pairs) every input file, which takes a while on a
large folder or a slow drive.
Args:
event: The click on the Convert button (its sender gets the loading state).
plan: The blocking planning call.
Returns:
The planned items; ``None`` if the app is shutting down.
"""
event.sender.props(add="loading")
try:
return await run.io_bound(plan)
finally:
with contextlib.suppress(RuntimeError): # the page was closed meanwhile
event.sender.props(remove="loading")
def _start(title: str, items: list, work) -> None: # noqa: ANN001 - work is a closure
"""Begin a batch (guarding empties / a running job) and launch the watcher."""
if state.converter.running():
ui.notify("A conversion is already running.", color=theme.WARNING)
return
if not items:
ui.notify("Nothing to convert — check the input/output folders.", color=theme.WARNING)
return
state.converter.start(title, items, work)
background_tasks.create(_watch(), name="annie-convert-watch")
ui.notify(f"{title}: converting {len(items)} file(s)…", color=theme.PRIMARY)
# ── sections ─────────────────────────────────────────────────────────────────
def _audio_section() -> ui.input:
"""Build the Audio section; returns its output-folder input (for video muxing)."""
with ui.card().classes("w-full gap-2"):
ui.label("Audio").classes("text-lg font-medium")
ui.label("Optional — re-encode audio to a uniform format, rate, channels.").classes(
"text-xs"
).style(f"color:{theme.NEUTRAL}")
in_dir = _dir_input("Audio input folder")
out_dir = _dir_input("Audio output folder")
with ui.row().classes("items-center gap-3 wrap"):
fmt = ui.select(
list(convert.AUDIO_OUTPUT_FORMATS), value="wav", label="Format"
).classes("w-32")
sr = ui.number("Sample rate (Hz)", value=16000, min=1000, step=1000).classes("w-40")
ch = ui.select({1: "mono", 2: "stereo"}, value=1, label="Channels").classes("w-32")
async def start(event: ClickEventArguments) -> None:
opts = convert.AudioOptions(
out_format=str(fmt.value),
sample_rate=int(sr.value or 16000),
channels=int(ch.value),
)
items = await _planning(
event, lambda: convert.plan_audio(in_dir.value or None, out_dir.value, opts)
)
if items is not None:
_start("Audio", items, lambda item: convert.convert_audio_file(item, opts))
ui.button("Convert audio", icon="graphic_eq", on_click=start).props("unelevated")
return out_dir
def _video_section(audio_out: ui.input) -> None:
"""Build the Video section (combine policy included)."""
with ui.card().classes("w-full gap-2"):
ui.label("Video").classes("text-lg font-medium")
ui.label(
"Re-encode to constant-frame-rate H.264 and validate every output with torchcodec."
).classes("text-xs").style(f"color:{theme.NEUTRAL}")
in_dir = _dir_input("Video input folder")
out_dir = _dir_input("Video output folder")
ui.label("Resolution").classes("text-sm font-medium mt-1")
keep_res = ui.checkbox("Keep original size", value=True)
with (
ui.row()
.classes("items-center gap-3")
.bind_visibility_from(keep_res, "value", backward=lambda v: not v)
):
width = ui.number("Width", value=None, min=2, step=2).classes("w-32")
height = ui.number("Height", value=None, min=2, step=2).classes("w-32")
grayscale = ui.checkbox("Grayscale (black & white)", value=False)
ui.label("Frame rate").classes("text-sm font-medium mt-1")
keep_fps = ui.checkbox("Keep original fps", value=True)
fps = (
ui.number("Target fps", value=25, min=1, step=1)
.classes("w-40")
.bind_visibility_from(keep_fps, "value", backward=lambda v: not v)
)
ui.label("Audio").classes("text-sm font-medium mt-1")
add_audio = ui.checkbox("Add audio track (match by name)", value=True)
black = ui.checkbox(
"Audio without a video → black-frame video", value=False
).bind_visibility_from(add_audio, "value")
audio_src = _dir_input("Audio folder to mux (defaults to the audio output folder)")
ui.label("Validation").classes("text-sm font-medium mt-1")
min_frames = ui.number("Minimum output frames", value=2, min=1, step=1).classes("w-48")
with min_frames:
ui.tooltip(
"Skip a source that yields fewer decodable frames than this. "
"2 drops audio-only files carrying a single thumbnail frame; 1 only "
"requires at least one decodable frame."
).props("delay=600")
async def start(event: ClickEventArguments) -> None:
keep_w = keep_res.value
opts = convert.VideoOptions(
width=None if keep_w else (int(width.value) if width.value else None),
height=None if keep_w else (int(height.value) if height.value else None),
grayscale=bool(grayscale.value),
fps=None if keep_fps.value else (float(fps.value) if fps.value else None),
add_audio=bool(add_audio.value),
audio_only_black=bool(add_audio.value and black.value),
min_frames=int(min_frames.value or 2),
)
audio_dir = (audio_src.value or audio_out.value) or None
items = await _planning(
event,
lambda: convert.plan_video(in_dir.value or None, out_dir.value, opts, audio_dir),
)
if items is not None:
_start("Video", items, lambda item: convert.convert_video_file(item, opts))
ui.button("Convert video", icon="movie", on_click=start).props("unelevated")
[docs]
def render() -> None:
"""Build the Convert tab body."""
with ui.column().classes("w-full gap-3"):
audio_out = _audio_section()
_video_section(audio_out)
_build_progress()
if state.converter.running():
background_tasks.create(_watch(), name="annie-convert-watch")