Source code for annie.pages.settings

"""Settings tab — UI preferences, render cache, review export/import, maintenance.

Surfaces the session-only UI preferences (Browse / Annotator row height), the
render-cache controls, the review-status export/import actions, and a Maintenance
block that lets you reclaim disk space by removing old logs, session databases, and
rendered clips.
"""

from __future__ import annotations

import contextlib
from dataclasses import dataclass
from typing import TYPE_CHECKING

from fastapi import Response
from nicegui import app, run, ui

from annie.app_state import state
from annie.core import logbook, theme
from annie.core.config import settings
from annie.pages import annotator, browse, download
from annie.pages.lazy import schedule
from annie.pages.refresh import client_refreshable

if TYPE_CHECKING:
    from collections.abc import Callable
    from pathlib import Path


@app.get("/annie/export/review/{fmt}")
def _review_export_route(fmt: str) -> Response:
    """Serve the review-status export's text; the browser saves it via the Save dialog."""
    if fmt == "json":
        return Response(content=state.store.export_json_text(), media_type="application/json")
    if fmt == "csv":
        return Response(content=state.store.export_csv_text(), media_type="text/csv")
    return Response(status_code=204)


# ── helpers ───────────────────────────────────────────────────────────────────


def _human_size(n_bytes: int) -> str:
    """Format a byte count as a human-readable string."""
    if n_bytes < 1024:
        return f"{n_bytes} B"
    if n_bytes < 1024**2:
        return f"{n_bytes / 1024:.1f} KB"
    return f"{n_bytes / 1024**2:.1f} MB"


def _dir_size(directory: Path) -> int:
    """Sum of all file sizes (non-recursive) in ``directory``, 0 if missing."""
    if not directory.is_dir():
        return 0
    return sum(f.stat().st_size for f in directory.iterdir() if f.is_file())


def _old_logs() -> list[Path]:
    """Log files in logs_dir that are not the current session's log."""
    if not settings.logs_dir.is_dir():
        return []
    current = logbook.LOG.log_path
    return [
        f
        for f in settings.logs_dir.iterdir()
        if f.is_file() and f.suffix == ".log" and f != current
    ]


def _old_session_dbs() -> list[Path]:
    """Session DB files that are not the currently active database."""
    if not settings.sessions_dir.is_dir():
        return []
    active = state.store.db_path.resolve()
    return [
        f
        for f in settings.sessions_dir.iterdir()
        if f.is_file() and f.suffix == ".db" and f.resolve() != active
    ]


# ── maintenance actions ───────────────────────────────────────────────────────


def _clear_logs() -> str:
    """Delete old log files and return a summary string."""
    targets = _old_logs()
    freed = sum(f.stat().st_size for f in targets)
    for f in targets:
        f.unlink(missing_ok=True)
    return f"Deleted {len(targets)} log file(s) — {_human_size(freed)} freed"


def _clear_session_dbs() -> str:
    """Delete old session DB files and return a summary string."""
    targets = _old_session_dbs()
    freed = sum(f.stat().st_size for f in targets)
    for f in targets:
        f.unlink(missing_ok=True)
    return f"Deleted {len(targets)} session DB(s) — {_human_size(freed)} freed"


def _clear_render_cache() -> str:
    """Delete all render clips (blocking; run off the loop) and return a summary.

    The Browse and Annotator render boxes are reset afterwards by :func:`_run_cleanup`.
    """
    _jobs, files = state.renderer.clear_all()
    return f"Cleared {files} rendered clip(s) — render boxes reset to default"


def _clean_up(actions: list[Callable[[], str]]) -> str:
    """Run the chosen clean-ups (blocking file deletes) and join their summaries."""
    parts = [action() for action in actions]
    return " · ".join(parts) if parts else "Nothing to clean up."


[docs] def cleanup_actions(space: _Space) -> list[Callable[[], str]]: """The clean-ups "Clean up all" runs: only those with something to remove.""" actions: list[Callable[[], str]] = [] if space.old_logs: actions.append(_clear_logs) if space.old_dbs: actions.append(_clear_session_dbs) if space.render_size > 0: actions.append(_clear_render_cache) return actions
async def _run_cleanup(actions: list[Callable[[], str]]) -> None: """Run clean-ups off the event loop, report them, and re-measure the card.""" msg = await run.io_bound(_clean_up, actions) if msg is None: return # the app is shutting down logbook.report(msg, level="info") ui.notify(msg, color=theme.PRIMARY) if _clear_render_cache in actions: # every tab's render boxes pointed at the clips browse.refresh(everywhere=True) annotator.refresh(everywhere=True) _maintenance_section.refresh(None) # None: measure again (not the old sizes) # ── UI ──────────────────────────────────────────────────────────────────────── @dataclass(frozen=True, slots=True) class _Space: """What the Maintenance card can reclaim, measured off the event loop. Attributes: old_logs: Log files other than the current one. old_dbs: Session databases other than the active one. log_size: Their total size in bytes. db_size: Their total size in bytes. render_size: Bytes held by rendered clips in the temp dir. """ old_logs: list[Path] old_dbs: list[Path] log_size: int db_size: int render_size: int def _measure_space() -> _Space: """List and size what can be reclaimed (blocking: walks folders; run off the loop).""" old_logs = _old_logs() old_dbs = _old_session_dbs() return _Space( old_logs, old_dbs, sum(f.stat().st_size for f in old_logs), sum(f.stat().st_size for f in old_dbs), _dir_size(settings.temp_dir), ) async def _measure_then_show() -> None: """Measure the reclaimable space off the loop, then fill the Maintenance card.""" space = await run.io_bound(_measure_space) if space is not None: with contextlib.suppress(RuntimeError): # the tab was closed meanwhile _maintenance_section.refresh(space) def _maintenance_header() -> None: """The Maintenance card's title row.""" with ui.row().classes("items-center gap-2"): ui.icon("build", color=theme.PRIMARY) ui.label("Maintenance").classes("font-medium") @client_refreshable def _maintenance_section(space: _Space | None = None) -> None: """Maintenance card: space summary + targeted cleanup buttons. Built first with a spinner while the folders are measured off the event loop (``space`` ``None``), then rebuilt with the measured :class:`_Space`. """ if space is None: with ui.card().classes("w-full"), ui.column().classes("w-full gap-3") as body: _maintenance_header() with ui.row().classes("items-center gap-2"): ui.spinner(size="1.5em", color=theme.PRIMARY) ui.label("Measuring reclaimable space…").classes("text-sm").style( f"color:{theme.NEUTRAL}" ) schedule(body, _measure_then_show) return old_logs, old_dbs = space.old_logs, space.old_dbs log_size, db_size, render_size = space.log_size, space.db_size, space.render_size with ui.card().classes("w-full"): with ui.column().classes("w-full gap-3"): _maintenance_header() # ── space summary ───────────────────────────────────────────── with ui.column().classes("w-full gap-1"): ui.label("Reclaimable space").classes("text-sm").style(f"color:{theme.NEUTRAL}") rows = [ ( "receipt_long", f"Old log files ({len(old_logs)})", _human_size(log_size), settings.logs_dir, ), ( "storage", f"Old session databases ({len(old_dbs)})", _human_size(db_size), settings.sessions_dir, ), ( "movie", "Rendered clips (temp dir)", _human_size(render_size), settings.temp_dir, ), ] for icon, label, size, path in rows: with ui.row().classes("items-center gap-2 w-full"): ui.icon(icon, size="sm").style(f"color:{theme.NEUTRAL}") ui.label(label).classes("text-sm flex-grow") ui.label(size).classes("text-sm font-mono").style(f"color:{theme.NEUTRAL}") ui.label(str(path)).classes("text-xs ml-6 break-all").style( f"color:{theme.NEUTRAL}" ) ttl = ( ui.number( "Auto-delete clips older than (seconds)", value=settings.temp_ttl_seconds, min=0, ) .classes("w-72") .props("dense") ) ttl.on("blur", lambda _: setattr(settings, "temp_ttl_seconds", int(ttl.value or 0))) ui.separator() # ── individual actions ──────────────────────────────────────── with ui.row().classes("items-center gap-2 flex-wrap"): ui.button( "Delete old logs", icon="receipt_long", on_click=lambda: _run_cleanup([_clear_logs]), ).props("flat").tooltip( f"Remove {len(old_logs)} log file(s) from {settings.logs_dir}" ).set_enabled(bool(old_logs)) ui.button( "Delete old session DBs", icon="storage", on_click=lambda: _run_cleanup([_clear_session_dbs]), ).props("flat").tooltip( f"Remove {len(old_dbs)} session database(s) from {settings.sessions_dir}" ).set_enabled(bool(old_dbs)) ui.button( "Clear render cache", icon="movie", on_click=lambda: _run_cleanup([_clear_render_cache]), ).props("flat").tooltip( "Delete all rendered clips and reset Browse/Annotator render boxes" ).set_enabled(render_size > 0) ui.separator() total = log_size + db_size + render_size async def _clean_all() -> None: await _run_cleanup(cleanup_actions(space)) ui.button( f"Clean up all ({_human_size(total)})", icon="cleaning_services", on_click=_clean_all, ).props("unelevated").set_enabled(total > 0)
[docs] def render() -> None: """Build the Settings tab body.""" with ui.column().classes("w-full gap-3"): with ui.card().classes("w-full"): ui.label("Browse & Annotator").classes("font-medium") ui.label("Row height sizes every thumbnail, strip frame, and render box.").classes( "text-xs" ).style(f"color:{theme.NEUTRAL}") browse_h = ui.number( "Browse row height (px)", value=state.ui.browse_row_height, min=72, max=400, step=15, ).classes("w-64") def set_browse_height() -> None: state.ui.browse_row_height = int(browse_h.value or state.ui.browse_row_height) browse.refresh() browse_h.on("blur", lambda _: set_browse_height()) annot_h = ui.number( "Annotator row max height (px)", value=state.ui.annotator_row_height, min=72, max=600, step=15, ).classes("w-64") def set_annot_height() -> None: state.ui.annotator_row_height = int(annot_h.value or state.ui.annotator_row_height) annotator.refresh() annot_h.on("blur", lambda _: set_annot_height()) ui.separator() # Read live by annie.pages.paging on every sentinel event, so toggling # this takes effect on the open lists without rebuilding them. ui.checkbox( "Auto scroll", value=state.ui.auto_scroll, on_change=lambda e: setattr(state.ui, "auto_scroll", bool(e.value)), ) ui.label("Reveal the next rows on reaching the bottom, without clicking.").classes( "text-xs" ).style(f"color:{theme.NEUTRAL}") rows_per_load = ui.number( "Rows per load", value=state.ui.page_size, min=1, step=1, precision=0, ).classes("w-64") def set_page_size() -> None: state.ui.page_size = max(1, int(rows_per_load.value or state.ui.page_size)) rows_per_load.set_value(state.ui.page_size) browse.refresh() annotator.refresh() rows_per_load.on("blur", lambda _: set_page_size()) embed_ttl = ui.number( "Unembed played clips after (seconds)", value=state.ui.embed_ttl_seconds, min=5, step=5, precision=0, ).classes("w-64") ui.label( "A played clip reverts to its placeholder, releasing the video " "buffered in this browser tab." ).classes("text-xs").style(f"color:{theme.NEUTRAL}") def set_embed_ttl() -> None: state.ui.embed_ttl_seconds = max( 5, int(embed_ttl.value or state.ui.embed_ttl_seconds) ) embed_ttl.set_value(state.ui.embed_ttl_seconds) embed_ttl.on("blur", lambda _: set_embed_ttl()) unload_after = ui.number( "Unload off-screen rows after (seconds)", value=state.ui.unload_after_seconds, min=5, step=5, precision=0, ).classes("w-64") ui.label( "A row scrolled well past drops its preview frames and decodes them " "again on the way back. Raise it to trade memory for fewer redecodes." ).classes("text-xs").style(f"color:{theme.NEUTRAL}") def set_unload_after() -> None: state.ui.unload_after_seconds = max( 5, int(unload_after.value or state.ui.unload_after_seconds) ) unload_after.set_value(state.ui.unload_after_seconds) unload_after.on("blur", lambda _: set_unload_after()) with ui.card().classes("w-full"): ui.label("Review status export / import").classes("font-medium") ui.label(f"Database: {state.store.db_path}").classes("text-xs").style( f"color:{theme.NEUTRAL}" ) with ui.row().classes("gap-2"): for fmt in ("json", "csv"): button = ui.button(f"Export {fmt.upper()}", icon="download").props("flat") download.attach_save( button, filename=f"annie_review.{fmt}", url=f"/annie/export/review/{fmt}", extension=fmt, ) _maintenance_section() with ui.card().classes("w-full"): ui.label("About").classes("font-medium") ui.label("Annie — by Ádám Fodor").classes("text-sm") with ui.row().classes("items-center gap-2"): ui.icon("bug_report", color=theme.PRIMARY) ui.link( "Report an issue", "https://github.com/fodorad/Annie/issues", new_tab=True ).classes("text-sm") with ui.row().classes("items-center gap-2"): ui.icon("mail", color=theme.PRIMARY) ui.link("fodorad201@gmail.com", "mailto:fodorad201@gmail.com").classes("text-sm")