feat: camera control — read and adjust dashcam settings over Wi-Fi

README.md

@@ -14,6 +14,7 @@ Self-hosted web app for syncing, browsing, and exporting recordings from a Viofo
 - **Video editor** - trim clips and cut between the front and rear cameras, then export a single video.
 - **Flexible exports** - original clips, joined front/rear, picture-in-picture, or edited cuts; hardware-accelerated where your system supports it.
 - **Storage management** - set a size or age limit and the oldest footage is pruned to fit; optional auto-delete clears the camera's SD card once a clip is safely saved.
+- **Camera control** - read and adjust the dashcam's own settings (resolution-adjacent options, parking mode, watermarks, HDR, LEDs, GPS…) from a Camera tab, with destructive actions hard-blocked.
 - **Easy browser-based setup** - a first-run wizard, then a settings page.
 - **Home Assistant support** - over MQTT, with sync status, alerts, and action buttons.
 
@@ -25,6 +26,7 @@ Self-hosted web app for syncing, browsing, and exporting recordings from a Viofo
 - [Getting started](#getting-started)
 - [Configuration](#configuration)
 - [Home Assistant](#home-assistant-via-mqtt)
+- [Camera control](#camera-control)
 - [Reference](#reference)
 - [About](#about)
 
@@ -154,6 +156,29 @@ For "prioritize the last N hours", publish to `{node_id}/cmd/prioritize_recent`
 
 - The MQTT password is stored in `config.json` in plaintext, alongside the bcrypt hash of the admin password and the session secret. The same access controls already apply to that file.
 
+## Camera control
+
+The **Camera** tab reads the dashcam's current settings and lets you change them over Wi-Fi — parking mode, watermarks, HDR, LEDs, GPS, beeps, time/date, loop length, bitrate, and so on. On/off settings are toggles; multi-choice settings are drop-downs populated with the camera's own option labels. Each change is validated, sent, and read back to confirm it applied.
+
+![The Camera tab](screenshots/camera_control.webp)
+
+This drives the undocumented Novatek **netapp** HTTP interface (`http://<cam>/?custom=1&cmd=<id>&par=<value>`). Because that protocol has no schema, the option labels and value enumerations come from a derived per-model command map (`viofosync_lib/data/command_map.json`); see [Command map data](#command-map-data).
+
+**Safety model.** On this protocol a bare command is *not* always a harmless read — some ids are destructive actions. The control layer:
+
+- **Hard-blocks destructive ids** (format SD, factory reset, firmware update, delete file, reboot, restart-Wi-Fi, SSD format/delete) — they are refused before any request is built and are never shown in the UI.
+- **Allow-lists writes** to enumerated settings only, validates the value against the camera's option list, and **verifies** by reading the setting back.
+- **Is gentle**: one request at a time with a short timeout, so it doesn't overrun the camera's single-threaded daemon.
+- **Auto-pauses recording** for the few settings the camera only accepts when stopped (loop length, bitrate), then resumes — flagged in the UI as "paused recording".
+
+**What it won't change.** Settings the camera refuses over Wi-Fi are shown read-only with the reason — e.g. *recording resolution* (changeable on the camera, not in station mode) and *exposure*. Settings for a lens that isn't attached (rear/interior HDR, video-merge, …) are read-only with "needs the rear/interior camera" and light up automatically once that lens is connected (detected via the live sensor count).
+
+Only the **A329S** has been validated against real hardware; other models are mapped from the app data but untested.
+
+### Command map data
+
+`viofosync_lib/data/command_map.json` is a plain reformatting of the *factual API data* (command ids, English keys, descriptions, and option enumerations) found in the official VIOFO Android app's `device-cmd-manager.db` asset. No app code or resources are included, and the original `.db` is not redistributed. Regenerate or extend it with `scripts/build_command_map.py` (see the script header for how to pull the asset from the APK).
+
 ## Reference
 
 ### Reverse geocoding

scripts/build_command_map.py

@@ -0,0 +1,88 @@
+#!/usr/bin/env python3
+"""Build ``viofosync_lib/data/command_map.json`` from the VIOFO app's
+``device-cmd-manager.db``.
+
+The Camera-control feature needs a semantic map for the Novatek netapp HTTP
+protocol (command id -> key/options), which the bare protocol doesn't provide.
+That mapping is derived from the official VIOFO Android app, which ships a
+SQLite asset ``assets/device-cmd-manager.db``.
+
+This script reformats only the *factual API data* (command ids, English keys,
+descriptions and option enumerations) into a plain JSON file we can ship and
+diff. It does not copy any app code or resources. We do not redistribute the
+``.db`` itself.
+
+To regenerate (e.g. for a newer app version, or to add a model):
+
+    # obtain the app's base APK, then:
+    unzip -o com.viofo.dashcam.apk 'assets/device-cmd-manager.db' -d /tmp/viofo
+    python3 scripts/build_command_map.py \
+        --db /tmp/viofo/assets/device-cmd-manager.db \
+        --out viofosync_lib/data/command_map.json
+
+Only the A329S has been validated against real hardware; other models are
+reformatted as-is and untested.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sqlite3
+
+
+def build(db_path: str) -> dict:
+    db = sqlite3.connect(db_path)
+    db.row_factory = sqlite3.Row
+    models = [r[0] for r in db.execute(
+        "SELECT DISTINCT DEVICE_MODEL FROM CMD_DEVICE_MANAGER ORDER BY DEVICE_MODEL")]
+    out: dict = {
+        "_provenance": (
+            "Command ids, keys and option enumerations reformatted from the "
+            "VIOFO dashcam Android app asset (device-cmd-manager.db). Factual "
+            "API data only. A329S validated against hardware; other models "
+            "reformatted as-is and untested. See scripts/build_command_map.py."
+        ),
+        "models": {},
+    }
+    for m in models:
+        commands: dict = {}
+        rows = db.execute(
+            "SELECT m.CMD cmd, m.CMD_KEY key, m.DESCRIPTION descr "
+            "FROM CMD_MANAGER m JOIN CMD_DEVICE_MANAGER d ON d.CMD_ID=m._ID "
+            "WHERE d.DEVICE_MODEL=? GROUP BY m.CMD, m.CMD_KEY "
+            "ORDER BY CAST(m.CMD AS INTEGER)", (m,))
+        for r in rows:
+            opts = [
+                {"index": int(o["_INDEX"]), "value": o["_VALUE"],
+                 "camera_tag": o["CAMERA_TAG"]}
+                for o in db.execute(
+                    "SELECT o._INDEX, o._VALUE, o.CAMERA_TAG "
+                    "FROM DASHCAM_MENU_OPTION_INFO o "
+                    "JOIN CMD_MANAGER m ON o.CMD_ID=m._ID "
+                    "WHERE o.DEVICE_MODEL=? AND m.CMD=? "
+                    "ORDER BY CAST(o._INDEX AS INTEGER)", (m, r["cmd"]))
+                if str(o["_INDEX"]).lstrip("-").isdigit()]
+            commands.setdefault(str(r["cmd"]), {
+                "key": r["key"], "description": r["descr"], "options": opts})
+        out["models"][m] = {"commands": commands}
+    return out
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("--db", required=True, help="path to device-cmd-manager.db")
+    ap.add_argument("--out", default=os.path.join(
+        os.path.dirname(__file__), "..", "viofosync_lib", "data",
+        "command_map.json"))
+    args = ap.parse_args()
+    data = build(args.db)
+    os.makedirs(os.path.dirname(os.path.abspath(args.out)), exist_ok=True)
+    with open(args.out, "w", encoding="utf-8") as f:
+        json.dump(data, f, ensure_ascii=False, indent=1)
+    print(f"wrote {args.out}: {len(data['models'])} models")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_control.py

@@ -0,0 +1,124 @@
+"""Unit tests for the camera-control layer (viofosync_lib._control).
+
+All hardware I/O is monkeypatched, so these run without a camera. They cover
+the parts that protect the device: the destructive denylist, value validation,
+the support classification (lens / Wi-Fi limits), the read-back verification,
+and the record-gated retry.
+"""
+from __future__ import annotations
+
+import pytest
+
+from viofosync_lib import _control as control
+
+# --------------------------------------------------------------------------- #
+# Pure data layer (uses the shipped command_map.json; no camera)
+# --------------------------------------------------------------------------- #
+
+def test_detect_model_longest_match():
+    assert control.detect_model("VIOFO_A329S_V2.0_260313") == "A329S"
+    # 'A329S' must beat the shorter 'A329'
+    assert control.detect_model("A329") == "A329"
+    assert control.detect_model(None) == control.DEFAULT_MODEL
+
+
+def test_resolve_cmd_by_key_and_number():
+    assert control._resolve_cmd("A329S", "CMD_GPS_SWITCH")[0] == 8208
+    assert control._resolve_cmd("A329S", 8208)[1] == "CMD_GPS_SWITCH"
+
+
+def test_resolve_cmd_unknown_raises():
+    with pytest.raises(control.ValidationError):
+        control._resolve_cmd("A329S", "CMD_NOPE")
+    with pytest.raises(control.ValidationError):
+        control._resolve_cmd("A329S", 999999)
+
+
+def test_options_have_unique_sorted_indices():
+    opts = control._options("A329S", 8222, "F")
+    idx = [int(o["index"]) for o in opts]
+    assert idx == sorted(idx)
+    assert len(idx) == len(set(idx))
+
+
+def test_support_station_locked_and_lenses():
+    # Resolution is not changeable over Wi-Fi.
+    ok, reason = control._support(8222, "F")
+    assert not ok and "Wi-Fi" in reason
+    # Rear-HDR needs the rear lens: blocked on 'F', allowed on 'F+R'.
+    assert control._support(9319, "F") == (False, "Needs the rear camera")
+    assert control._support(9319, "F+R") == (True, None)
+    # A plain setting is supported.
+    assert control._support(8208, "F") == (True, None)
+
+
+# --------------------------------------------------------------------------- #
+# Transport helpers
+# --------------------------------------------------------------------------- #
+
+def test_base_url_normalisation():
+    assert control.base_url("192.168.1.254") == "http://192.168.1.254"
+    assert control.base_url("http://cam/") == "http://cam"
+
+
+def test_parser_reads_pairs_and_leaf_tags():
+    xml = ('<?xml version="1.0"?><Function><Cmd>3012</Cmd><Status>0</Status>'
+           '<String>VIOFO_A329S_V2.0</String></Function>')
+    assert control._flat_tags(xml)["String"] == "VIOFO_A329S_V2.0"
+    dump = "<Cmd>2001</Cmd><Status>1</Status><Cmd>8208</Cmd><Status>0</Status>"
+    pairs = control._PAIR_RE.findall(dump)
+    assert [(int(c), int(s)) for c, s in pairs] == [(2001, 1), (8208, 0)]
+
+
+# --------------------------------------------------------------------------- #
+# Write path (monkeypatched transport)
+# --------------------------------------------------------------------------- #
+
+def test_set_setting_refuses_destructive():
+    with pytest.raises(control.DestructiveCommandError):
+        control.set_setting("cam", 3010, 0, model="A329S")  # format SD
+
+
+def test_set_setting_validates_value():
+    with pytest.raises(control.ValidationError):
+        control.set_setting("cam", "CMD_GPS_SWITCH", "banana", model="A329S")
+
+
+def test_set_setting_happy_path(monkeypatch):
+    sent = {}
+
+    def fake_send(address, cmd, par=None, s=None):
+        sent["cmd"], sent["par"] = cmd, par
+        return {"Cmd": str(cmd), "Status": "0"}
+
+    # Read-back reports the value we just wrote -> verified.
+    monkeypatch.setattr(control, "_send", fake_send)
+    monkeypatch.setattr(control, "read_status_pairs",
+                        lambda addr: [(8208, 1)])
+    monkeypatch.setattr(control, "VERIFY_SETTLE", 0)
+
+    r = control.set_setting("cam", "CMD_GPS_SWITCH", "1", model="A329S")
+    assert sent == {"cmd": 8208, "par": 1}
+    assert r["ok"] and r["verified"] is True and r["applied_index"] == 1
+    assert r["record_cycled"] is False
+
+
+def test_record_gated_retry_cycles_recording(monkeypatch):
+    calls = []
+
+    def fake_send(address, cmd, par=None, s=None):
+        calls.append((cmd, par))
+        return {"Status": "0"}
+
+    # First verify fails, second (after stopping recording) succeeds.
+    results = iter([(True, False, 3, "0"), (True, True, 2, "0")])
+    monkeypatch.setattr(control, "_send", fake_send)
+    monkeypatch.setattr(control, "_apply_and_verify",
+                        lambda a, c, i: next(results))
+    monkeypatch.setattr(control, "_record_state", lambda a: 1)
+    monkeypatch.setattr(control, "RECORD_SETTLE", 0)
+
+    r = control.set_setting("cam", 8222, 2, model="A329S")  # resolution (gated)
+    assert r["record_cycled"] is True and r["verified"] is True
+    # Recording was stopped (par=0) and restarted (par=1).
+    assert (control.RECORD_CMD, 0) in calls and (control.RECORD_CMD, 1) in calls