refactor: single camera registry

tests/test_cameras.py

@@ -0,0 +1,121 @@
+"""Camera registry invariants + derivation pins.
+
+The registry (viofosync_lib/cameras.py) is the single source of
+truth; naming.py derives the export-type tables from it. These
+tests pin the derived values to the exact literals that used to
+be hardcoded, so a registry edit that silently changes behavior
+fails here first.
+"""
+from __future__ import annotations
+
+from viofosync_lib._archive import downloaded_filename_glob
+from viofosync_lib.cameras import (
+    CAMERA_LETTERS,
+    CAMERAS,
+    CHANNEL_FOR_LETTER,
+    pair_slot_of,
+)
+from web.services import naming
+
+
+def test_registry_invariants():
+    letters = [c.letter for c in CAMERAS]
+    channels = [c.channel for c in CAMERAS]
+    assert len(set(letters)) == len(letters), "duplicate letter"
+    assert len(set(channels)) == len(channels), "duplicate channel"
+    assert all(len(c.letter) == 1 and c.letter.isupper() for c in CAMERAS)
+    assert "other" not in channels, '"other" is the fallback, not a camera'
+
+
+def test_letters_and_front_rear_positions():
+    # Front and rear are load-bearing: front is ffmpeg input 0 /
+    # audio source, and both always render in the archive grid.
+    assert CAMERA_LETTERS == "FRTI"
+    assert CAMERAS[0].channel == "front"
+    assert CAMERAS[1].channel == "rear"
+
+
+def test_channel_for_letter_matches_registry():
+    assert CHANNEL_FOR_LETTER == {
+        "F": "front", "R": "rear", "T": "tele", "I": "interior",
+    }
+
+
+def test_pair_slot_of_rear_fallback():
+    # channel_of sends unknowns to "other"; the pairers' slot
+    # helper keeps their historical rear fallback instead.
+    assert pair_slot_of("F") == "front"
+    assert pair_slot_of("PT") == "tele"
+    assert pair_slot_of("X") == "rear"
+    assert pair_slot_of("") == "rear"
+    assert pair_slot_of(None) == "rear"
+
+
+def test_glob_derivation():
+    assert downloaded_filename_glob.endswith(f"_*[{CAMERA_LETTERS}].MP4")
+
+
+def test_js_mirror_matches_registry():
+    """app.js hand-mirrors the registry (no bundler, so it can't
+    import it). Extract the JS CAMERAS table and compare, so the
+    two can't drift apart silently."""
+    import re
+    from pathlib import Path
+
+    repo = Path(__file__).resolve().parents[1]
+    js = (repo / "web" / "static" / "app.js").read_text()
+    rows = re.findall(
+        r'\{\s*letter:\s*"(\w)",\s*channel:\s*"(\w+)",\s*'
+        r'label:\s*"(\w+)"\s*\}',
+        js,
+    )
+    assert rows == [
+        (c.letter, c.channel, c.label) for c in CAMERAS
+    ], "web/static/app.js CAMERAS table is out of sync with the registry"
+
+
+# --- Derived export-type tables pin the pre-registry literals ----
+
+
+def test_export_job_types():
+    assert set(naming.EXPORT_JOB_TYPES) == {
+        "join_front", "join_rear", "join_tele", "join_interior",
+        "pip", "pip_rear", "pip_tele", "pip_interior",
+    }
+
+
+def test_join_letter_for_type():
+    assert naming.JOIN_LETTER_FOR_TYPE == {
+        "join_front": "F",
+        "join_rear": "R",
+        "join_tele": "T",
+        "join_interior": "I",
+    }
+
+
+def test_pip_partner_and_main_for_type():
+    assert naming.PIP_PARTNER_FOR_TYPE == {
+        "pip": "rear",
+        "pip_rear": "rear",
+        "pip_tele": "tele",
+        "pip_interior": "interior",
+    }
+    assert naming.PIP_MAIN_FOR_TYPE == {
+        "pip": "front",
+        "pip_rear": "rear",
+        "pip_tele": "tele",
+        "pip_interior": "interior",
+    }
+
+
+def test_label_for_type():
+    assert naming.LABEL_FOR_TYPE == {
+        "join_front": "front",
+        "join_rear": "rear",
+        "join_tele": "tele",
+        "join_interior": "interior",
+        "pip": "pip-front",
+        "pip_rear": "pip-rear",
+        "pip_tele": "pip-tele",
+        "pip_interior": "pip-interior",
+    }

viofosync_lib/__init__.py

@@ -7,6 +7,13 @@
 - :mod:`viofosync_lib._protocol` — HTTP API to the dashcam (XML
   listing, HTML scrape, byte downloader)
 - :mod:`viofosync_lib._gpx` — MP4 atom parsing + GPX generation
+
+plus one deliberately public submodule:
+
+- :mod:`viofosync_lib.cameras` — the camera registry (which
+  lenses exist and how a filename's trailing letter maps to one).
+  Imported directly (``from viofosync_lib.cameras import …``) by
+  both this package and the web layer.
 """
 from __future__ import annotations
 

viofosync_lib/_archive.py

@@ -13,6 +13,8 @@
 import re
 from collections import namedtuple
 
+from .cameras import CAMERA_LETTERS
+
 logger = logging.getLogger("viofosync_lib.archive")
 
 # Recording namedtuple matching Viofo's file information.
@@ -31,12 +33,11 @@
 }
 
 # Downloaded recording filename glob pattern. The trailing
-# letter is the camera: F=front, R=rear, T=telephoto,
-# I=interior.
+# letter is the camera (see cameras.py for the registry).
 downloaded_filename_glob = (
     "[0-9][0-9][0-9][0-9]_[0-9][0-9][0-9][0-9]"
     "_[0-9][0-9][0-9][0-9][0-9][0-9]"
-    "_*[FRTI].MP4"
+    f"_*[{CAMERA_LETTERS}].MP4"
 )
 
 # Downloaded recording filename regular expression.

viofosync_lib/cameras.py

@@ -0,0 +1,64 @@
+"""Camera registry — the single source of truth for Viofo lenses.
+
+A clip's filename ends ``…NNNNN[PE]?<letter>.MP4`` where the
+optional P/E prefix encodes parking/event and ``<letter>`` is the
+camera. 2-channel models record F+R; 3-channel models add either
+T (telephoto, e.g. A329) or I (interior, e.g. A139 / A229 3CH).
+
+Everything camera-shaped derives from :data:`CAMERAS`: the
+download/scan glob, the queue's filename regexes, timeline
+channel keys and labels, archive pair slots, and the per-camera
+export job types (see ``web/services/naming.py`` for the
+app-level derivations). Adding a future lens means adding one
+:class:`Camera` line here plus its UI strings.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class Camera:
+    letter: str   # filename suffix letter (the F in ``…0001F.MP4``)
+    channel: str  # stable machine key: timeline channel / pair slot
+    label: str    # human-facing label
+
+
+# Declaration order is display order (timeline tracks, UI cycles).
+CAMERAS: tuple[Camera, ...] = (
+    Camera("F", "front", "Front"),
+    Camera("R", "rear", "Rear"),
+    Camera("T", "tele", "Tele"),
+    Camera("I", "interior", "Interior"),
+)
+
+# "FRTI" — drops straight into regex/glob character classes.
+CAMERA_LETTERS = "".join(c.letter for c in CAMERAS)
+
+CHANNEL_FOR_LETTER = {c.letter: c.channel for c in CAMERAS}
+
+
+def channel_of(camera: str | None) -> str:
+    """Map a clip's ``camera`` code to its channel key.
+
+    The lens is the trailing letter of the code, so parking/event
+    prefixes (PF, ET, PI, …) resolve to the same channel as the
+    bare letter. Anything unrecognised falls back to ``"other"``
+    so an unexpected code still gets its own track rather than
+    vanishing.
+    """
+    if not camera:
+        return "other"
+    return CHANNEL_FOR_LETTER.get(camera[-1].upper(), "other")
+
+
+def pair_slot_of(camera: str | None) -> str:
+    """Like :func:`channel_of`, but with the pairers' historical
+    ``"rear"`` fallback: when the archive day view and the export
+    pairer group same-capture clips into slots, an unknown letter
+    has always been filed under rear rather than given its own
+    slot."""
+    if not camera:
+        return "rear"
+    return CHANNEL_FOR_LETTER.get(camera[-1].upper(), "rear")

web/db.py

@@ -99,7 +99,7 @@ def migrate_legacy_db_path(new_path: str) -> None:
     basename      TEXT NOT NULL,
     group_name    TEXT,
     timestamp     INTEGER NOT NULL,   -- unix seconds
-    camera        TEXT NOT NULL,      -- 'F' or 'R' or other
+    camera        TEXT NOT NULL,      -- registry letter (F/R/T/I), possibly P/E-prefixed
     sequence      INTEGER NOT NULL,
     event_type    TEXT,               -- 'normal'|'parking'|'ro'
     size_bytes    INTEGER,
@@ -136,7 +136,7 @@ def migrate_legacy_db_path(new_path: str) -> None:
 
 CREATE TABLE IF NOT EXISTS export_jobs (
     id            INTEGER PRIMARY KEY AUTOINCREMENT,
-    type          TEXT NOT NULL,      -- join_front|join_rear|pip|pip_rear|timeline
+    type          TEXT NOT NULL,      -- join_/pip_ per camera (see naming.EXPORT_JOB_TYPES) or timeline
     clip_ids      TEXT NOT NULL,      -- JSON array
     state         TEXT NOT NULL,      -- queued|running|done|failed|cancelled
     progress      REAL NOT NULL DEFAULT 0.0,

web/routers/archive.py

@@ -24,7 +24,13 @@
 from ..services import durations, filmstrip, route_cache, scanner, thumbs
 from ..services import tasks as _tasks
 from ..services import gps as gps_service
-from ..services.naming import CHANNEL_LABELS, CHANNEL_ORDER, channel_of
+from ..services.naming import (
+    CAMERAS,
+    CHANNEL_LABELS,
+    CHANNEL_ORDER,
+    channel_of,
+    pair_slot_of,
+)
 
 log = logging.getLogger("viofosync.archive")
 
@@ -201,25 +207,18 @@ def _in_range(ts: int) -> bool:
     # one capture share a timestamp but get consecutive sequence
     # numbers, so keying on sequence wouldn't pair them.
     # event_type keeps parking (PF/PR) separate from normal (F/R).
-    # Slot is picked from the last letter so PF/EF still = front.
-    # 3-channel models add either T (telephoto) or I (interior)
-    # alongside F+R.
+    # Slot is the registry channel of the last letter, so PF/EF
+    # still = front.
+    slots = [c.channel for c in CAMERAS]
     pairs: dict[tuple[int, str], dict] = defaultdict(
-        lambda: {
-            "front": None, "rear": None,
-            "tele": None, "interior": None,
-            "sequence": None,
-        }
+        lambda: dict.fromkeys([*slots, "sequence"])
     )
     for r in rows:
         if not _in_range(r["timestamp"]):
             continue
-        cam = (r["camera"] or "").upper()
         kind = r["event_type"] or "normal"
         key = (r["timestamp"], kind)
-        slot = {"F": "front", "T": "tele", "I": "interior"}.get(
-            cam[-1:], "rear"
-        )
+        slot = pair_slot_of(r["camera"])
         pairs[key][slot] = dict(r)
         # Prefer the front sequence number for the pair; fall
         # back to any other camera's if there's no front clip.
@@ -234,10 +233,7 @@ def _in_range(ts: int) -> bool:
             "sequence": pair["sequence"],
             "event_type": kind,
             "iso": _dt.datetime.fromtimestamp(ts).isoformat(),
-            "front": pair["front"],
-            "rear": pair["rear"],
-            "tele": pair["tele"],
-            "interior": pair["interior"],
+            **{s: pair[s] for s in slots},
         })
 
     return {"date": date, "clips": clips}

web/routers/exports.py

@@ -12,7 +12,12 @@
 
 from ..auth import require_csrf, require_session
 from ..services import export_preview
-from ..services.naming import export_download_name, parse_clip_ids
+from ..services.naming import (
+    CHANNEL_ORDER,
+    EXPORT_JOB_TYPES,
+    export_download_name,
+    parse_clip_ids,
+)
 
 # 1x1 transparent PNG — served when a preview can't be produced (job not done,
 # unknown, or generation failed), so the <img> degrades cleanly.
@@ -55,17 +60,14 @@ def _resolve_default_encoder(app_state) -> str:
 
 
 class Segment(BaseModel):
-    channel: str = Field(pattern="^(front|rear|tele|interior|other)$")
+    channel: str = Field(pattern=f"^({'|'.join(CHANNEL_ORDER)})$")
     start_ts: float
     end_ts: float
 
 
 class CreateExport(BaseModel):
     type: str = Field(
-        pattern=(
-            "^(join_front|join_rear|join_tele|join_interior"
-            "|pip|pip_rear|pip_tele|pip_interior|timeline)$"
-        )
+        pattern=f"^({'|'.join((*EXPORT_JOB_TYPES, 'timeline'))})$"
     )
     clip_ids: List[int] = []
     segments: list[Segment] | None = None