From 86d8a04f2a39ae52802e0a7e1af2aedef78461dc Mon Sep 17 00:00:00 2001 From: oleost <5670710+oleost@users.noreply.github.com> Date: Sun, 25 Oct 2026 20:35:05 +0100 Subject: [PATCH 1/9] Back to the beta channel after 0.10.1 Co-Authored-By: Claude Opus 5.5 --- visionstate/config.yaml | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/visionstate/config.yaml b/visionstate/config.yaml index fdc4bae..f924542 100644 --- a/visionstate/config.yaml +++ b/visionstate/config.yaml @@ -1,5 +1,5 @@ -name: VisionState -version: "0.10.1" +name: VisionState (beta) +version: "0.10.1b4" slug: visionstate description: Teach a local AI to recognise states (open, closed, …), find objects (people, cars, pets) or read numbers in your camera images url: https://github.com/oleost/VisionState @@ -16,7 +16,7 @@ ingress_port: 8099 # a request into memory and refuses one over 16 MB. ingress_stream: true panel_icon: mdi:eye-check-outline -panel_title: VisionState +panel_title: VisionState β homeassistant_api: true hassio_api: true services: @@ -39,3 +39,5 @@ schema: mqtt_port: port? mqtt_username: str? mqtt_password: password? +environment: + VISIONSTATE_MEDIA: /media/visionstate_beta From 8d14065738c54c8bc16969d7dcd1e16281687cdb Mon Sep 17 00:00:00 2001 From: oleost <5670710+oleost@users.noreply.github.com> Date: Mon, 26 Oct 2026 10:56:42 +0100 Subject: [PATCH 2/9] Under the hood: fewer recorder rows, nothing blocking the event loop, faster checks - Home Assistant stored a state at every check: no timestamp in the attributes, the confidence and probabilities in whole percent, and pictures and object attributes (boxes, last seen) sent when what they show changes, otherwise every PUBLISH["refresh_s"]. On the test installation: 4380 -> about 500-750 rows an hour. - No database work directly in async code (35 places): a busy SQLite no longer stops every sensor, MQTT and the API; tests/test_event_loop.py guards it. - History clean-up removes files only after their rows are committed, reads only the columns it needs, and removes frames no row lists; images are written whole or not at all; a partial index for the review queue; SQLite synchronous=NORMAL (WAL). - ONNX Runtime threads no longer spin: about 30 % less CPU per model run. - RTSP: grabs on threads of their own, FFmpeg's socket timeout under its current name, and a stream grabbed again within RUNTIME["rtsp_keep_open_s"] is kept open (about 30 ms a frame instead of 1-3 s during a burst or the live view). - A Home Assistant camera's picture has the same size limit as a snapshot URL. Co-Authored-By: Claude Opus 5.5 --- docs/ARCHITECTURE.md | 8 +- docs/SCOPE.md | 49 +++- visionstate/DOCS.md | 19 +- visionstate/backend/tests/test_cpu_only.py | 4 + visionstate/backend/tests/test_event_loop.py | 71 ++++++ visionstate/backend/tests/test_publish.py | 89 +++++++- visionstate/backend/tests/test_robustness.py | 53 +++++ visionstate/backend/tests/test_storage.py | 112 ++++++++- visionstate/backend/tests/test_uploads.py | 50 ++++ visionstate/backend/visionstate/api/backup.py | 2 +- visionstate/backend/visionstate/api/common.py | 6 + visionstate/backend/visionstate/api/faces.py | 107 +++++---- .../backend/visionstate/api/imports.py | 9 +- visionstate/backend/visionstate/api/review.py | 214 ++++++++++-------- .../backend/visionstate/api/samples.py | 90 +++++--- .../backend/visionstate/api/sensors.py | 119 +++++----- visionstate/backend/visionstate/api/system.py | 4 +- visionstate/backend/visionstate/api/teach.py | 109 +++++---- visionstate/backend/visionstate/backbones.py | 19 +- visionstate/backend/visionstate/db.py | 12 +- .../backend/visionstate/engine/checks.py | 12 +- .../backend/visionstate/engine/history.py | 119 +++++++--- .../backend/visionstate/engine/logic.py | 6 + .../backend/visionstate/engine/models.py | 2 +- .../backend/visionstate/engine/objects.py | 19 +- .../backend/visionstate/engine/publishing.py | 27 +++ .../backend/visionstate/engine/reading.py | 12 +- .../backend/visionstate/engine/runtime.py | 29 ++- .../backend/visionstate/engine/state.py | 2 + visionstate/backend/visionstate/settings.py | 18 ++ visionstate/backend/visionstate/sources.py | 143 ++++++++++-- visionstate/backend/visionstate/storage.py | 10 +- 32 files changed, 1170 insertions(+), 375 deletions(-) create mode 100644 visionstate/backend/tests/test_event_loop.py diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 884b162..c5b46b3 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -34,7 +34,10 @@ one event loop: - **I/O** (camera, MQTT, Home Assistant) is async. - **Database and file work** runs in worker threads (`asyncio.to_thread`): SQLite with WAL, one - short session per unit of work (`Database.session()`). + short session per unit of work (`Database.session()`). Never directly in an `async def`: + `tests/test_event_loop.py` fails on a session (or `load_sensor`) opened there. +- **RTSP grabs** run on threads of their own (`sources.FrameGrabber`, `RUNTIME["rtsp_workers"]`), + so a camera that does not answer holds none of the default pool's. - **AI inference and training** runs in worker threads too, at most `RUNTIME["max_concurrent_inferences"]` at a time (the semaphore `Runtime._sem`). - **Long-running loops** (one per sensor, MQTT, the Home Assistant event stream, clean-up) catch @@ -133,7 +136,8 @@ error is kept in `LiveState.error` and shown in the UI. the move kept training images in `/samples/`: they are read there until `engine.HistoryMixin.move_samples` has copied them (`settings.SAMPLES_MOVE`). A file is removed only after the row that lists it (`db.Database.after_commit`), and an image that is gone anyway - is left out of training and of the taught boxes instead of stopping the sensor. + is left out of training and of the taught boxes instead of stopping the sensor. The history clean-up + (`engine.HistoryMixin.cleanup_history`) also removes frames no row lists, once they are old. - **Object sensors' taught boxes** are `sample` rows with `object_label`, `detected`, `box` and `score`; own labels live in `sensor.objects["custom"]`. Readings and object events are `prediction` rows (see `db.py`). diff --git a/docs/SCOPE.md b/docs/SCOPE.md index 84f6c6b..3a8503f 100644 --- a/docs/SCOPE.md +++ b/docs/SCOPE.md @@ -95,6 +95,8 @@ so retraining only runs the backbone on new or changed samples. `backbones.trim_memory`, as glibc kept the freed memory as well. With the default models the app is about 190 MB smaller (a test installation with nine sensors and faces: 943 → 677 MB) for about 6 % more time per check. The footprints of the detectors were measured again this way. + ONNX Runtime's threads do not spin while they wait for work (`backbones.cpu_session`): by + default they keep cores busy between runs, CPU taken from Home Assistant on the same machine. - DINOv2 base, evaluated (2026-10-11) with the crops of the 14 objects in the CC0 test photos, each also changed in light, framing, blur, JPEG quality and night colours: the gap between the same object changed (5th percentile) and other objects of its kind (90th percentile) was 0.12 for base @@ -429,7 +431,7 @@ results: [`tools/wheelreader/README.md`](../tools/wheelreader/README.md); how it |---|---| | HA camera entity (default) | `camera_proxy` via the Supervisor/HA API. Works with any `camera.*` entity. | | HTTP snapshot URL | HTTP GET of a JPEG/PNG. | -| RTSP stream | First decoded frame via PyAV (bundled FFmpeg). | +| RTSP stream | First decoded frame via PyAV (bundled FFmpeg); kept open while frames are asked for often (below). | | Upload | Images (JPG/PNG/WebP/BMP), ZIP archives, video (MP4/MKV/MOV/AVI/WebM). | Video upload: one frame every *N* seconds (default 5), near-duplicates skipped by perceptual @@ -438,9 +440,18 @@ hash, then the current model suggests a label for each frame. Upload and ZIP siz Before a frame is fetched the address is checked (`sources.check_source`): a snapshot URL must be `http(s)://`, a stream a network address (FFmpeg would also open local files and its own -pseudo-protocols), a camera an entity ID. A snapshot is read up to `RUNTIME["max_frame_mb"]`, and -every image is refused above `RUNTIME["max_image_megapixels"]` — checked from its header, before -the pixels are decoded. +pseudo-protocols), a camera an entity ID. A snapshot (a URL's or a Home Assistant camera's) is read +up to `RUNTIME["max_frame_mb"]`, and every image is refused above `RUNTIME["max_image_megapixels"]` +— checked from its header, before the pixels are decoded. An RTSP grab blocks a thread for seconds +(connecting, waiting for a key frame; FFmpeg's socket `timeout`), so it runs on threads of its +own (`RUNTIME["rtsp_workers"]`): cameras that do not answer never hold up the database or the +models, which share the default thread pool. Connecting for one frame took 1–2.8 s with a real camera (an iPhone +camera app, 2026-10-26: connecting, waiting for a key frame), so a stream grabbed again within +`RUNTIME["rtsp_keep_open_s"]` (a burst of checks, the live view) is kept open +(`sources.RtspStream`: a thread decodes it and keeps the newest frame; a grab takes a frame decoded +after it asked) — about 30 ms a frame, for 4–6 % of one core while it is open. It closes once +nobody asked for that long, so a check at the regular interval still connects for one frame and +nothing decodes in between. ## 7. When and how a sensor decides @@ -533,7 +544,7 @@ match: `m3` with `water` made the value entity disappear without a word), so `mq unit as Home Assistant does when that is clear (`m3` → `m³`, `kwh` → `kWh`) and otherwise leaves the device class out (a warning in the log). The lists are `settings.HA_DEVICE_CLASS_UNITS`. Units without a device class are never touched: those entities exist, with history in their unit. -Attributes: `read_text`, `rejected`, `last_update`, `last_trigger`. The button is "Read now". +Attributes: `read_text`, `rejected`, `last_trigger`. The button is "Read now". Four more entities with `enabled_by_default: false` (nothing changes for users who do not turn them on): *Raw reading* (`raw`, diagnostic), *Problem* (`problem`, enum of `settings.READING_PROBLEMS`, diagnostic), *Accepted (24 h)* (`accepted`, %, diagnostic, kept in @@ -551,8 +562,20 @@ Plus one app-wide **VisionState** device with `sensor.visionstate_review_queue` for review; attributes: per-sensor breakdown, `oldest_waiting_since`). The review reminder (§7) uses Home Assistant's REST API (`persistent_notification`, `notify`), not MQTT. -Attributes on the state entity: `probabilities`, `top_state`, `last_update`, `trained`, -`last_trigger`. Availability: app-wide LWT plus per-sensor camera availability. Removing a sensor +Attributes on the state entity: `probabilities`, `top_state`, `trained`, `last_trigger`. + +**Nothing changes at every check unless the result does** (decided 2026-10-26): the recorder of +Home Assistant writes a state row for every `state_changed` event, which an MQTT entity fires when +its state or attributes differ from the last message; an image entity's state is the time of its +last picture, so every picture, even an identical one, is a new row. Until 0.10.1 a state sensor +on the default interval wrote several rows a minute (a `last_update` timestamp in the attributes, +the confidence to one decimal, the picture). Now: no timestamps in attributes (Home Assistant +keeps `last_reported`), the confidence and the probabilities in whole percent +(`engine.logic.percent`), and the pictures and an object's attributes (boxes, last seen) through +`engine.publishing.PublishingMixin._send_on_change`: at once when what they are about changes (the published +state, the objects and their counts, the value), otherwise every `settings.PUBLISH["refresh_s"]`. + +Availability: app-wide LWT plus per-sensor camera availability. Removing a sensor removes its entities. **Send to Home Assistant** (`sensor.publish`, default `settings.SENSOR_PUBLISH_DEFAULT`, DB migration 9 keeps existing sensors on): when off, discovery stays (the entities and the IDs the user gave them are kept), but no value is published (the @@ -683,7 +706,17 @@ setting edited in Settings → Storage (`/api/v1/storage`, which also reports di the former app option `history_retention_days` was removed in 0.6.1 without carrying its value over (early days; the release notes say so). The clean-up runs every 10 minutes and right after the limits change. Training images are never removed -automatically. +automatically. A frame is removed only once the removal of its row is committed, and frames no row +lists (the app stopped between writing a frame and its row) are removed once they are older than +`HISTORY_CLEANUP["orphan_age_s"]` — only in the history folders. Every image is written next to its +name and then moved there (`storage.Storage._write`), so a power cut or a full disk never leaves +half a picture under a name a row lists. + +The database is SQLite in WAL mode with `synchronous=NORMAL`: safe from corruption, and a power +cut can lose only the last commit (a crash of the app loses nothing); it saves a sync of the log +at every commit, which matters on SD cards. Nothing on the event loop waits for it: every query of +an `async` function runs through `asyncio.to_thread` (`tests/test_event_loop.py` fails otherwise), +as a busy database would otherwise stop every sensor, MQTT and the API with it. ## 12. Import / export diff --git a/visionstate/DOCS.md b/visionstate/DOCS.md index c63eef2..e19fafd 100644 --- a/visionstate/DOCS.md +++ b/visionstate/DOCS.md @@ -49,8 +49,8 @@ The sensor appears in Home Assistant as a device named after the sensor, with th | Entity | What it is | |---|---| | `sensor.` | The state (`open`, `closed`, …, or `unknown` when unsure) | -| `sensor._confidence` | How sure the AI is, in % | -| `image._last_frame` | The image region that was classified | +| `sensor._confidence` | How sure the AI is, in whole % | +| `image._last_frame` | The image region that was classified — sent when the state changes, otherwise every few minutes | | `button._classify_now` | Classify right now (use it in automations) | | `switch._enabled` | Pause / resume the sensor | @@ -63,8 +63,13 @@ In addition, the **VisionState** device has `sensor.visionstate_review_queue`: t frames waiting for review, with a per-sensor breakdown and `oldest_waiting_since` (when the oldest of them came) as attributes. -The state entity also has a `probabilities` attribute with the score of every state and a -`last_trigger` attribute telling what caused the last check. +The state entity also has a `probabilities` attribute with the score of every state (in whole +percent, as a fraction) and a `last_trigger` attribute telling what caused the last check. + +Nothing VisionState sends changes at every check unless the result does: Home Assistant stores +each new state in its database, so values that only move a little (a timestamp, a confidence to +the decimal, the same picture again) would fill it. The pictures, and an object's boxes, follow +the result at once and are otherwise sent again every few minutes. **Tuning a sensor before Home Assistant uses it:** switch off **Send to Home Assistant** (in the last step of the wizard, or under **Settings → General**). The sensor then runs as usual — its @@ -92,8 +97,10 @@ Where it gets your camera wrong, you can correct it later (see *Teaching an obje | `sensor.__count` | How many there are | plus `image.…_last_frame` (the region with boxes drawn), `button.…_detect_now` and - `switch.…_enabled`, like every sensor. The binary sensor's attributes list the confidence - and the boxes. Removing an object from the list removes its entities. + `switch.…_enabled`, like every sensor. The binary sensor's attributes list the confidence, + the boxes and when it was last seen; they follow at once when the object comes or goes or its + count changes, and every few minutes while it stays. Removing an object from the list removes + its entities. - **Settings tab → Sensor output**: | Setting | Default | Meaning | diff --git a/visionstate/backend/tests/test_cpu_only.py b/visionstate/backend/tests/test_cpu_only.py index 486d296..b25d96a 100644 --- a/visionstate/backend/tests/test_cpu_only.py +++ b/visionstate/backend/tests/test_cpu_only.py @@ -18,6 +18,10 @@ def test_backbone_and_detector_use_only_the_cpu(): ) assert embedder.session.get_providers() == ["CPUExecutionProvider"] assert detector.session.get_providers() == ["CPUExecutionProvider"] + # Threads do not spin while they wait for work (that takes CPU from Home Assistant). + options = detector.session.get_session_options() + assert options.get_session_config_entry("session.intra_op.allow_spinning") == "0" + assert options.get_session_config_entry("session.inter_op.allow_spinning") == "0" @requires_reader diff --git a/visionstate/backend/tests/test_event_loop.py b/visionstate/backend/tests/test_event_loop.py new file mode 100644 index 0000000..2447ecb --- /dev/null +++ b/visionstate/backend/tests/test_event_loop.py @@ -0,0 +1,71 @@ +"""Nothing waits on the database on the event loop: a busy SQLite (``timeout`` 30 s) would stop +every sensor, MQTT and the API with it. Database work in an ``async def`` goes through +``asyncio.to_thread`` (a function defined inside it and run that way is fine).""" + +import ast +from pathlib import Path + +import visionstate + +PACKAGE = Path(visionstate.__file__).parent + +# Methods that open a database session themselves. +BLOCKING = {"session", "get_setting", "set_setting", "get_text", "get_dict"} # on .db +BLOCKING_RUNTIME = {"load_sensor", "frame_for_view", "heads_load"} + + +def _own_calls(fn: ast.AsyncFunctionDef) -> list[ast.Call]: + """The calls in a function's own body, not in functions or lambdas defined inside it.""" + calls, stack = [], list(fn.body) + while stack: + node = stack.pop() + if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef | ast.Lambda): + continue + if isinstance(node, ast.Call): + calls.append(node) + stack.extend(ast.iter_child_nodes(node)) + return calls + + +def _blocking(call: ast.Call) -> bool: + func = call.func + if not isinstance(func, ast.Attribute): + return False + target = func.value + on_db = ( + isinstance(target, ast.Attribute) and target.attr == "db" or isinstance(target, ast.Name) and target.id == "db" + ) + return on_db and func.attr in BLOCKING or func.attr in BLOCKING_RUNTIME + + +def blocking_calls(source: str, name: str = "") -> list[str]: + found = [] + for fn in ast.walk(ast.parse(source)): + if isinstance(fn, ast.AsyncFunctionDef): + found += [f"{name}:{c.lineno} {fn.name}: {ast.unparse(c.func)}" for c in _own_calls(fn) if _blocking(c)] + return found + + +def test_the_check_finds_a_blocking_call(): + planted = """ +async def endpoint(rt): + def inside(): + with rt.db.session() as s: # fine: run in a thread + pass + await asyncio.to_thread(inside) + await asyncio.to_thread(rt.load_sensor, 1) # fine: passed, not called + with rt.db.session() as s: # blocks the loop + pass + rt.load_sensor(1) # blocks the loop +""" + assert sorted(blocking_calls(planted)) == [ + ":10 endpoint: rt.load_sensor", + ":8 endpoint: rt.db.session", + ] + + +def test_no_database_work_on_the_event_loop(): + found = [] + for path in sorted(PACKAGE.rglob("*.py")): + found += blocking_calls(path.read_text(encoding="utf-8"), str(path.relative_to(PACKAGE))) + assert not found, "Run these through asyncio.to_thread:\n" + "\n".join(found) diff --git a/visionstate/backend/tests/test_publish.py b/visionstate/backend/tests/test_publish.py index ae83daa..d012c7d 100644 --- a/visionstate/backend/tests/test_publish.py +++ b/visionstate/backend/tests/test_publish.py @@ -1,4 +1,5 @@ -"""A sensor that is not sent to Home Assistant: it runs, but its entities stay unavailable.""" +"""What is sent to Home Assistant: a sensor that is not sent (it runs, its entities stay +unavailable), and nothing that would make Home Assistant store an entity again at every check.""" import io import json @@ -100,3 +101,89 @@ def test_migration_keeps_existing_sensors_sent(tmp_path): db.init() with db.session() as s: assert [row.publish for row in s.query(Sensor)] == [True] + + +class CountingMqtt(RecordingMqtt): + def __init__(self, real): + super().__init__(real) + self.count: dict[str, int] = {} + + async def publish(self, topic, payload, retain=False): + self.sent[topic] = payload + self.count[topic] = self.count.get(topic, 0) + 1 + + +def test_a_picture_is_sent_when_it_shows_something_new(tmp_path, monkeypatch): + """Home Assistant stores every picture an image entity gets: the same picture is sent again + only every PUBLISH["refresh_s"], a new one (another state) at once.""" + import asyncio + from types import SimpleNamespace + + from visionstate.engine import Runtime, publishing + from visionstate.settings import PUBLISH + + settings = Settings( + data_dir=tmp_path / "data", media_dir=tmp_path / "media", frontend_dir=tmp_path, bundled_models_dir=MODEL_DIR + ) + settings.ensure_dirs() + db = Database(settings.db_path) + db.init() + rt = Runtime(settings, db) + mqtt = CountingMqtt(rt.mqtt) + rt.mqtt = mqtt # type: ignore[assignment] + now = [1000.0] + monkeypatch.setattr(publishing.time, "time", lambda: now[0]) + encoded = [] + + def send(cfg, shows): # noqa: ANN001 + asyncio.run(rt._send_on_change(cfg, "t/image", shows, lambda: encoded.append(shows) or b"jpeg")) + + cfg = SimpleNamespace(id=1, publish=True) + send(cfg, "closed") + now[0] += 10 + send(cfg, "closed") # the same: not sent, not even encoded + assert mqtt.count["t/image"] == 1 and encoded == ["closed"] + send(cfg, "open") # something new: at once + now[0] += PUBLISH["refresh_s"] - 1 + send(cfg, "open") + assert mqtt.count["t/image"] == 2 + now[0] += 2 + send(cfg, "open") # the same for long enough: sent again, so a dashboard is not left behind + assert mqtt.count["t/image"] == 3 + rt.live_state(1).sent_on_change.clear() # what publish_discovery does (MQTT connected again) + send(cfg, "open") + assert mqtt.count["t/image"] == 4 + send(SimpleNamespace(id=2, publish=False), "open") # not sent to Home Assistant + assert mqtt.count["t/image"] == 4 + db.engine.dispose() + + +@requires_reader +def test_values_that_change_at_every_check_are_not_sent(tmp_path): + """No timestamp in the attributes, the confidence in whole percent, the pictures once while the + value stays: nothing makes Home Assistant store the entities again at every check.""" + settings = Settings( + data_dir=tmp_path / "data", media_dir=tmp_path / "media", frontend_dir=tmp_path, bundled_models_dir=MODEL_DIR + ) + with TestClient(create_app(settings)) as client: + rt = client.app.state.runtime + rt.grabber = DisplayCamera("0012345.6") + mqtt = CountingMqtt(rt.mqtt) + rt.mqtt = mqtt + body = { + "name": "Meter", + "kind": "reading", + "source_type": "http", + "source": "x", + "reading": {"mode": "counter", "decimals": 1, "change_limit": "off"}, + "interval_s": 1, + "debounce": 1, + } + client.post("/api/v1/sensors", json=body) + base = "visionstate/meter" + assert wait_for(lambda: mqtt.count.get(f"{base}/confidence", 0) >= 3, timeout=60) + attributes = mqtt.sent[f"{base}/attributes"] + attributes = json.loads(attributes) if isinstance(attributes, str) else attributes + assert "last_update" not in attributes + assert mqtt.sent[f"{base}/confidence"].isdigit() # whole percent + assert mqtt.count[f"{base}/image"] == 1 and mqtt.count[f"{base}/reader_image"] == 1 diff --git a/visionstate/backend/tests/test_robustness.py b/visionstate/backend/tests/test_robustness.py index 1b363d9..c241909 100644 --- a/visionstate/backend/tests/test_robustness.py +++ b/visionstate/backend/tests/test_robustness.py @@ -104,6 +104,59 @@ def test_image_pixel_limit(monkeypatch): imaging.decode(data) +def test_home_assistant_snapshot_size_limit(tmp_path, monkeypatch): + """A camera through Home Assistant has the same limit as a snapshot URL.""" + monkeypatch.setitem(vs_settings.RUNTIME, "max_frame_mb", 0.001) # 1 kB + settings = make_settings(tmp_path) + settings.ha_url, settings.ha_token = "http://ha:8123", "token" + + def ha_with(content: bytes) -> FrameGrabber: + grabber = FrameGrabber(HomeAssistant(settings)) + grabber.ha._client = httpx.AsyncClient( + transport=httpx.MockTransport(lambda r: httpx.Response(200, content=content)) + ) + return grabber + + assert asyncio.run(ha_with(b"x" * 500).grab("ha_camera", "camera.door")) == b"x" * 500 + with pytest.raises(SourceError, match="larger than"): + asyncio.run(ha_with(b"x" * 5000).grab("ha_camera", "camera.door")) + + +def test_rtsp_grabs_run_on_their_own_threads(tmp_path, monkeypatch): + """A camera that hangs holds an RTSP thread, never one of the default executor's (database, + models); FFmpeg's socket timeout has its current name.""" + import threading + + from visionstate import sources + + seen: dict = {} + + def fake_grab(url: str) -> bytes: + seen["thread"] = threading.current_thread().name + return b"jpeg" + + monkeypatch.setattr(sources, "grab_rtsp", fake_grab) + grabber = FrameGrabber(HomeAssistant(make_settings(tmp_path))) + assert asyncio.run(grabber.grab("rtsp", "rtsp://cam/s1")) == b"jpeg" + assert seen["thread"].startswith("rtsp") + asyncio.run(grabber.close()) + + opened: dict = {} + + class Refused(Exception): + pass + + def fake_open(url, options=None, timeout=None): # noqa: ANN001 + opened.update(options) + raise Refused + + monkeypatch.undo() # the real grab_rtsp again + monkeypatch.setattr("av.open", fake_open) + with pytest.raises(Refused): + sources.grab_rtsp("rtsp://cam/s1") + assert opened["timeout"] == "10000000" and "stimeout" not in opened + + def test_ha_client_without_api_raises_a_clear_error(tmp_path): ha = HomeAssistant(make_settings(tmp_path)) assert not ha.enabled diff --git a/visionstate/backend/tests/test_storage.py b/visionstate/backend/tests/test_storage.py index e286290..a743c5c 100644 --- a/visionstate/backend/tests/test_storage.py +++ b/visionstate/backend/tests/test_storage.py @@ -1,4 +1,5 @@ -"""History limits: removal by age and by total size (oldest first, review queue last), and the API.""" +"""History limits: removal by age and by total size (oldest first, review queue last), and the API; +files written whole and removed with their rows; frames no row lists.""" from datetime import timedelta @@ -337,3 +338,112 @@ def locked(): with rt.db.session() as s: assert s.get(Sample, sample_id) is None assert "Could not remove a file" in caplog.text + + +def test_history_files_go_only_when_the_rows_are_committed(runtime, monkeypatch): + """The clean-up removes a frame together with its row: a commit that fails keeps both.""" + rt = runtime + sid = state_sensor(rt) + (row_id,) = add_frames(rt, sid, [(400, True)]) + with rt.db.session() as s: + path = rt.storage.history_path(sid, s.get(Prediction, row_id).frame) + + real_session = rt.db.session + + def failing_session(): + from contextlib import contextmanager + + @contextmanager + def session(): + with real_session() as s: + yield s + s.flush() + raise OSError("database is locked") + + return session() + + monkeypatch.setattr(rt.db, "session", failing_session) + with pytest.raises(OSError): + rt._remove_history([(row_id, sid, path.name)]) + monkeypatch.undo() + assert path.exists() and remaining(rt) == {row_id} + rt.cleanup_history() + assert not path.exists() and remaining(rt) == set() + + +def test_frames_no_row_lists_are_removed_once_they_are_old(runtime): + import os + import time + + from visionstate.settings import HISTORY_CLEANUP + + rt = runtime + sid = state_sensor(rt) + (row_id,) = add_frames(rt, sid, [(0, True)]) + with rt.db.session() as s: + kept = rt.storage.history_path(sid, s.get(Prediction, row_id).frame) + kept_thumb = rt.storage.thumbnail("history", row_id, kept) + old_orphan = rt.storage.history_path(sid, rt.storage.save_history(sid, noise(2))) + young_orphan = rt.storage.history_path(sid, rt.storage.save_history(sid, noise(3))) + old_thumb = rt.storage.thumb_path("history", row_id + 100) + old_thumb.write_bytes(kept_thumb.read_bytes()) + half = old_orphan.with_name("x.jpg.part") # a write cut off by a power cut + half.write_bytes(b"half") + other = rt.settings.history_dir / "notes.txt" # not ours: left alone + other.write_text("x") + long_ago = time.time() - HISTORY_CLEANUP["orphan_age_s"] - 60 + for path in (kept, kept_thumb, old_orphan, old_thumb, half, other): + os.utime(path, (long_ago, long_ago)) + + assert rt.remove_orphan_frames() == 3 + assert kept.exists() and kept_thumb.exists() and young_orphan.exists() and other.exists() + assert not old_orphan.exists() and not old_thumb.exists() and not half.exists() + + +def test_image_files_are_written_whole_or_not_at_all(runtime, monkeypatch): + from visionstate import imaging + + rt = runtime + sid = state_sensor(rt) + name = rt.storage.save_history(sid, noise(1)) + folder = rt.storage.history_path(sid, name).parent + assert [p.name for p in folder.iterdir()] == [name] + + def cut_off(image, quality=90): # noqa: ANN001 + raise OSError("No space left on device") + + monkeypatch.setattr(imaging, "encode_jpeg", cut_off) + with pytest.raises(OSError): + rt.storage.save_history(sid, noise(2)) + assert [p.name for p in folder.iterdir()] == [name] # no half picture, no leftover + + +def test_the_review_queue_is_counted_from_an_index(runtime): + """The count after every check that flags a frame reads the partial index, not the history.""" + from sqlalchemy import func, select, text + + rt = runtime + query = select(Prediction.sensor_id, func.count(Prediction.id)).where(Prediction.reviewed.is_(False)) + sql = str(query.group_by(Prediction.sensor_id).compile(rt.db.engine, compile_kwargs={"literal_binds": True})) + with rt.db.engine.connect() as conn: + assert conn.execute(text("PRAGMA synchronous")).scalar() == 1 # NORMAL (with WAL) + plan = " ".join(str(row[-1]) for row in conn.execute(text("EXPLAIN QUERY PLAN " + sql))) + assert "ix_prediction_waiting" in plan, plan + + +def test_an_existing_database_gets_the_index_at_start(settings): + """Indexes added later are made when the app starts on an older database (Database.init).""" + from sqlalchemy import text + + from visionstate.db import Database + + settings.ensure_dirs() + db = Database(settings.db_path) + db.init() + with db.engine.begin() as conn: + conn.execute(text("DROP INDEX ix_prediction_waiting")) # as a database of an older version + db.init() + with db.engine.connect() as conn: + names = {row[0] for row in conn.execute(text("SELECT name FROM sqlite_master WHERE type = 'index'"))} + db.engine.dispose() + assert "ix_prediction_waiting" in names diff --git a/visionstate/backend/tests/test_uploads.py b/visionstate/backend/tests/test_uploads.py index cf2b3d1..0fc1435 100644 --- a/visionstate/backend/tests/test_uploads.py +++ b/visionstate/backend/tests/test_uploads.py @@ -70,3 +70,53 @@ def test_stream_grab_error_is_redacted(): with pytest.raises(sources.SourceError) as err: sources.grab_rtsp("rtsp://admin:secret@127.0.0.1:9/stream") assert "secret" not in str(err.value) + + +def test_a_stream_grabbed_often_is_kept_open(tmp_path, monkeypatch): + """One check connects for one frame; frames asked for again soon (a burst, the live view) + come from a stream kept open, which goes once it ends or nobody asks for a while.""" + from visionstate import settings as vs_settings + + video = tmp_path / "clip.mp4" + make_video(video, [("open", 30)]) + one_shots: list[str] = [] + real_grab = sources.grab_rtsp + + def counting_grab(url: str) -> bytes: + one_shots.append(url) + return real_grab(url) + + monkeypatch.setattr(sources, "grab_rtsp", counting_grab) + grabber = sources.FrameGrabber(sources.HomeAssistant(_settings(tmp_path))) + url = str(video) + assert Image.open(io.BytesIO(grabber._grab_stream(url))).size == (320, 180) + assert one_shots == [url] and not grabber._streams # a single check: no stream left open + assert Image.open(io.BytesIO(grabber._grab_stream(url))).size == (320, 180) # soon again: kept open + stream = grabber._streams[url] + assert one_shots == [url] + stream._thread.join(timeout=30) # a file ends (a camera that stops sending) + assert not stream.alive + grabber._grab_stream(url) # a stream that ended is opened again + assert grabber._streams[url] is not stream + grabber._streams[url].close() + + monkeypatch.setitem(vs_settings.RUNTIME, "rtsp_keep_open_s", 0) # 0: always one frame per connection + other = sources.FrameGrabber(sources.HomeAssistant(_settings(tmp_path))) + other._grab_stream(url) + other._grab_stream(url) + assert len(one_shots) == 3 and not other._streams + + +def test_a_stream_that_cannot_be_opened_says_why(tmp_path): + stream = sources.RtspStream("rtsp://admin:secret@127.0.0.1:9/stream", keep_open=5) + with pytest.raises(sources.SourceError) as err: + stream.grab() + assert "secret" not in str(err.value) + + +def _settings(tmp_path): + from visionstate.settings import Settings + + return Settings( + data_dir=tmp_path / "d", media_dir=tmp_path / "m", frontend_dir=tmp_path, bundled_models_dir=tmp_path + ) diff --git a/visionstate/backend/visionstate/api/backup.py b/visionstate/backend/visionstate/api/backup.py index 1e9ef0e..9eb44b1 100644 --- a/visionstate/backend/visionstate/api/backup.py +++ b/visionstate/backend/visionstate/api/backup.py @@ -227,7 +227,7 @@ async def _start(app, settings: Settings, on_connect: Callable[[MqttBridge], Awa """A runtime on the data there is now; ``on_connect``: done once its MQTT is connected.""" db = Database(settings.db_path) try: - db.init() + await asyncio.to_thread(db.init) # an older backup's database is upgraded here rt = Runtime(settings, db) if on_connect is not None: rt.on_first_connect.append(on_connect) diff --git a/visionstate/backend/visionstate/api/common.py b/visionstate/backend/visionstate/api/common.py index 85077ad..5d17da1 100644 --- a/visionstate/backend/visionstate/api/common.py +++ b/visionstate/backend/visionstate/api/common.py @@ -99,6 +99,12 @@ def get_sensor(session: Session, sensor_id: int, kind: str | None = None) -> Sen return sensor +def check_sensor(rt: Runtime, sensor_id: int, kind: str | None = None) -> None: + """404 (or 400, see get_sensor) unless the sensor exists; for a worker thread (asyncio.to_thread).""" + with rt.db.session() as s: + get_sensor(s, sensor_id, kind) + + def state_id_for(sensor: Sensor, key: str | None) -> int | None: if key is None: return None diff --git a/visionstate/backend/visionstate/api/faces.py b/visionstate/backend/visionstate/api/faces.py index 58ccfb7..10c326c 100644 --- a/visionstate/backend/visionstate/api/faces.py +++ b/visionstate/backend/visionstate/api/faces.py @@ -193,31 +193,40 @@ async def teach_face(sensor_id: int, body: FaceTeachIn, request: Request) -> dic cfg = await asyncio.to_thread(_config, rt, sensor_id) if not body.person and not body.new_person: raise HTTPException(400, "Say who it is: a person or a new person") - with rt.db.session() as s: - if body.person: - person = person_by_key(s, body.person) - count = s.scalar(select(func.count()).select_from(Face).where(Face.person_id == person.id)) or 0 - if count >= FACES["max_faces_per_person"]: - raise HTTPException(400, f"A person can have at most {FACES['max_faces_per_person']} faces") - else: - # Checked now, made only once the face is good. - if (s.scalar(select(func.count()).select_from(Person)) or 0) >= FACES["max_people"]: - raise HTTPException(400, f"VisionState can know at most {FACES['max_people']} people") - _check_name(s, body.new_person or "") + + def check() -> None: + with rt.db.session() as s: + if body.person: + person = person_by_key(s, body.person) + count = s.scalar(select(func.count()).select_from(Face).where(Face.person_id == person.id)) or 0 + if count >= FACES["max_faces_per_person"]: + raise HTTPException(400, f"A person can have at most {FACES['max_faces_per_person']} faces") + else: + # Checked now, made only once the face is good. + if (s.scalar(select(func.count()).select_from(Person)) or 0) >= FACES["max_people"]: + raise HTTPException(400, f"VisionState can know at most {FACES['max_people']} people") + _check_name(s, body.new_person or "") + + await asyncio.to_thread(check) image, origin, face, problem = await _find(rt, sensor_id, body, cfg) if problem or face is None: raise HTTPException(400, PROBLEMS.get(problem or "none", PROBLEMS["none"])) crop, local = faces.face_crop(image, face) filename = await asyncio.to_thread(rt.storage.save_face, crop) - with rt.db.session() as s: - person = person_by_key(s, body.person) if body.person else new_person(s, body.new_person or "") - face_id = add_face_row(s, person.id, sensor_id, filename, crop, faces.face_to_dict(local), origin) - key, name = person.key, person.name - n = s.scalar(select(func.count()).select_from(Face).where(Face.person_id == person.id)) or 0 + + def add() -> tuple[int, str, str, int]: + with rt.db.session() as s: + person = person_by_key(s, body.person) if body.person else new_person(s, body.new_person or "") + face_id = add_face_row(s, person.id, sensor_id, filename, crop, faces.face_to_dict(local), origin) + key, name = person.key, person.name + n = s.scalar(select(func.count()).select_from(Face).where(Face.person_id == person.id)) or 0 + return face_id, key, name, n + + face_id, key, name, n = await asyncio.to_thread(add) if body.new_person: await rt.people_changed() # the person's entities on every sensor that recognises faces else: - rt.faces_changed() + await asyncio.to_thread(rt.faces_changed) if body.history_id is not None: await asyncio.to_thread(remember_taught, rt, sensor_id, body.history_id, body.box, key, "person", True) await rt.person_last_seen_changed(key) # Home Assistant gets that last_seen too @@ -348,13 +357,17 @@ async def add_photos( raise HTTPException(400, "Say who it is: a person or a new person") if not await asyncio.to_thread(rt.face_sensor_ids): # the face model is only fetched after a sensor's confirmation raise HTTPException(400, "Turn on face recognition on a sensor first (its Settings → Faces)") - with rt.db.session() as s: - if person: - person_by_key(s, person) - else: - if (s.scalar(select(func.count()).select_from(Person)) or 0) >= FACES["max_people"]: - raise HTTPException(400, f"VisionState can know at most {FACES['max_people']} people") - _check_name(s, new_name or "") + + def check() -> None: + with rt.db.session() as s: + if person: + person_by_key(s, person) + else: + if (s.scalar(select(func.count()).select_from(Person)) or 0) >= FACES["max_people"]: + raise HTTPException(400, f"VisionState can know at most {FACES['max_people']} people") + _check_name(s, new_name or "") + + await asyncio.to_thread(check) model = await _face_model(rt) min_px = rt.face_limits(None)["min_face_px"] found: list[tuple[Image.Image, faces.Face]] = [] @@ -382,19 +395,26 @@ async def add_photos( skipped.append({"file": name, "message": PHOTO_PROBLEMS.get(problems[0], PHOTO_PROBLEMS["none"])}) if not found: return {"label": person, "name": None, "added": 0, "skipped": skipped, "faces": None} - with rt.db.session() as s: - who = person_by_key(s, person) if person else new_person(s, new_name or "") - count = s.scalar(select(func.count()).select_from(Face).where(Face.person_id == who.id)) or 0 - added = 0 - for image, face in found: - if count >= FACES["max_faces_per_person"]: - skipped.append({"file": "", "message": f"A person has at most {FACES['max_faces_per_person']} faces."}) - break - crop, local = faces.face_crop(image, face) - filename = rt.storage.save_face(crop) - add_face_row(s, who.id, None, filename, crop, faces.face_to_dict(local), "upload") - count, added = count + 1, added + 1 - result = {"label": who.key, "name": who.name, "added": added, "skipped": skipped, "faces": count} + + def add() -> dict: + with rt.db.session() as s: + who = person_by_key(s, person) if person else new_person(s, new_name or "") + count = s.scalar(select(func.count()).select_from(Face).where(Face.person_id == who.id)) or 0 + added = 0 + for image, face in found: + if count >= FACES["max_faces_per_person"]: + skipped.append( + {"file": "", "message": f"A person has at most {FACES['max_faces_per_person']} faces."} + ) + break + crop, local = faces.face_crop(image, face) + filename = rt.storage.save_face(crop) + add_face_row(s, who.id, None, filename, crop, faces.face_to_dict(local), "upload") + count, added = count + 1, added + 1 + result = {"label": who.key, "name": who.name, "added": added, "skipped": skipped, "faces": count} + return result + + result = await asyncio.to_thread(add) if not person: await rt.people_changed() else: @@ -421,10 +441,15 @@ def _pictures(upload_file: UploadFile, name: str, limit: int) -> list[Image.Imag async def rename_person(key: str, body: PersonIn, request: Request) -> dict: """A person's name (their entities' names follow; their key and entity IDs stay).""" rt = runtime(request) - with rt.db.session() as s: - person = person_by_key(s, key) - person.name = _check_name(s, body.name, keep=person.id) - result = {"key": person.key, "name": person.name} + + def rename() -> dict: + with rt.db.session() as s: + person = person_by_key(s, key) + person.name = _check_name(s, body.name, keep=person.id) + result = {"key": person.key, "name": person.name} + return result + + result = await asyncio.to_thread(rename) await rt.people_changed() return result diff --git a/visionstate/backend/visionstate/api/imports.py b/visionstate/backend/visionstate/api/imports.py index b6eca1d..793a9fb 100644 --- a/visionstate/backend/visionstate/api/imports.py +++ b/visionstate/backend/visionstate/api/imports.py @@ -4,7 +4,6 @@ import asyncio import logging -import tempfile import zipfile from pathlib import Path from typing import TYPE_CHECKING @@ -30,7 +29,7 @@ unique_slug, ) from .faces import add_face_row, new_person, taken_keys -from .samples import copy_limited +from .samples import spool_upload if TYPE_CHECKING: from .sensors import SensorIn @@ -43,10 +42,8 @@ async def import_bundle(request: Request, file: UploadFile = File(...)) -> dict: """A new sensor from an exported bundle (validated like the API); a state sensor is trained.""" rt = runtime(request) - with tempfile.NamedTemporaryFile(delete=False, suffix=".zip") as tmp: - path = Path(tmp.name) - within_limit = copy_limited(file.file, tmp, UPLOAD_LIMITS["max_file_mb"] * 1_000_000) - if not within_limit: + path, too_large = await asyncio.to_thread(spool_upload, file, "bundle.zip") + if too_large: path.unlink(missing_ok=True) raise HTTPException(413, f"Bundle is larger than {UPLOAD_LIMITS['max_file_mb']} MB") try: diff --git a/visionstate/backend/visionstate/api/review.py b/visionstate/backend/visionstate/api/review.py index 130efcd..2872286 100644 --- a/visionstate/backend/visionstate/api/review.py +++ b/visionstate/backend/visionstate/api/review.py @@ -102,14 +102,18 @@ async def review_dismiss_all(sensor_id: int, request: Request) -> dict: Nothing is learned from them; answers already given and the reading counts stay. """ rt = runtime(request) - with rt.db.session() as s: - get_sensor(s, sensor_id) - result = s.execute( - update(Prediction) - .where(Prediction.sensor_id == sensor_id, Prediction.reviewed.is_(False)) - .values(reviewed=True) - ) - dismissed = cast(CursorResult, result).rowcount + + def dismiss() -> int: + with rt.db.session() as s: + get_sensor(s, sensor_id) + result = s.execute( + update(Prediction) + .where(Prediction.sensor_id == sensor_id, Prediction.reviewed.is_(False)) + .values(reviewed=True) + ) + return cast(CursorResult, result).rowcount + + dismissed = await asyncio.to_thread(dismiss) await rt.publish_review_count() return {"dismissed": dismissed} @@ -142,77 +146,88 @@ def _newest_rejected(s: Session, row: Prediction) -> bool: @router.post("/review/{prediction_id}") async def review_answer(prediction_id: int, body: ReviewIn, request: Request) -> dict: rt = runtime(request) - with rt.db.session() as s: - row = s.get(Prediction, prediction_id) - if row is None: - raise HTTPException(404, "Review item not found") - sensor = get_sensor(s, row.sensor_id) - kind = sensor.kind + + def kind_of() -> str: + with rt.db.session() as s: + row = s.get(Prediction, prediction_id) + if row is None: + raise HTTPException(404, "Review item not found") + return get_sensor(s, row.sensor_id).kind + + kind = await asyncio.to_thread(kind_of) if kind == KIND_OBJECTS: return await _answer_object(rt, request, prediction_id, body) - anchor: tuple[int, str, datetime] | None = None - with rt.db.session() as s: - row = s.get(Prediction, prediction_id) - if row is None: - raise HTTPException(404, "Review item not found") - sensor = get_sensor(s, row.sensor_id) - if sensor.kind == KIND_READING: - if body.action not in ("read_ok", "misread", "skip"): - raise HTTPException(400, "Answer read_ok, misread or skip for a reading") - if body.action != "skip": - row.read_ok = body.action == "read_ok" - row.correct_value = None - if body.action == "misread" and body.value and body.value.strip(): - settings = merge_reading(sensor.reading) - number = readers.right_value(body.value, settings) # digits only: placed like the reader - if number is None: - raise HTTPException(400, f"Not a number: {body.value!r}") - row.correct_value = readers.format_value(number, settings) - # A reading with the wrong number of digits is no value, even when read as it is shown. - probs = row.probs or {} - kept = probs.get("reason") in ANCHOR_REASONS - right = row.correct_value if body.action == "misread" else probs.get("value") if kept else None - if right is not None and _newest_rejected(s, row): - anchor = (row.sensor_id, right, row.created_at) - row.reviewed = True - elif body.action not in ("confirm", "label", "skip"): - raise HTTPException(400, "Answer confirm, label or skip") - if sensor.kind == KIND_READING: + + def answer_reading() -> tuple[int, str, datetime] | None: + """Checks that the answer fits the kind; stores a reading's. Returns the reading to anchor on, if any.""" + anchor: tuple[int, str, datetime] | None = None + with rt.db.session() as s: + row = s.get(Prediction, prediction_id) + if row is None: + raise HTTPException(404, "Review item not found") + sensor = get_sensor(s, row.sensor_id) + if sensor.kind == KIND_READING: + if body.action not in ("read_ok", "misread", "skip"): + raise HTTPException(400, "Answer read_ok, misread or skip for a reading") + if body.action != "skip": + row.read_ok = body.action == "read_ok" + row.correct_value = None + if body.action == "misread" and body.value and body.value.strip(): + settings = merge_reading(sensor.reading) + number = readers.right_value(body.value, settings) # digits only: placed like the reader + if number is None: + raise HTTPException(400, f"Not a number: {body.value!r}") + row.correct_value = readers.format_value(number, settings) + # A reading with the wrong number of digits is no value, even when read as it is shown. + probs = row.probs or {} + kept = probs.get("reason") in ANCHOR_REASONS + right = row.correct_value if body.action == "misread" else probs.get("value") if kept else None + if right is not None and _newest_rejected(s, row): + anchor = (row.sensor_id, right, row.created_at) + row.reviewed = True + elif body.action not in ("confirm", "label", "skip"): + raise HTTPException(400, "Answer confirm, label or skip") + return anchor + + anchor = await asyncio.to_thread(answer_reading) + if kind == KIND_READING: if anchor is not None: await rt.anchor_reading(*anchor) await rt.publish_review_count() return {"ok": True} - with rt.db.session() as s: - row = s.get(Prediction, prediction_id) - if row is None: # removed in the meantime (history clean-up) - raise HTTPException(404, "Review item not found") - sensor = get_sensor(s, row.sensor_id) - key = row.state_key if body.action == "confirm" else body.state_key - state_id = state_id_for(sensor, key) if body.action in ("confirm", "label") else None - sensor_id, frame = row.sensor_id, row.frame - row.reviewed = True - # Answered before: change (or, when skipped now, remove) the sample that answer added. - sample = s.get(Sample, row.sample_id) if row.sample_id is not None else None - if sample is not None and sample.sensor_id == sensor_id: - if state_id is None: - rt.db.after_commit(s, partial(rt.storage.delete_sample, sensor_id, sample.id, sample.filename)) - s.delete(sample) - row.sample_id = None - else: - s.execute(delete(SampleLabel).where(SampleLabel.sample_id == sample.id)) - s.add(SampleLabel(sample_id=sample.id, state_id=state_id)) - sample.verified = False - rt.schedule_retrain(sensor_id) - state_id = None # nothing more to add + + def answer_state() -> tuple[int, int | None, str | None]: + with rt.db.session() as s: + row = s.get(Prediction, prediction_id) + if row is None: # removed in the meantime (history clean-up) + raise HTTPException(404, "Review item not found") + sensor = get_sensor(s, row.sensor_id) + key = row.state_key if body.action == "confirm" else body.state_key + state_id = state_id_for(sensor, key) if body.action in ("confirm", "label") else None + sensor_id, frame = row.sensor_id, row.frame + row.reviewed = True + # Answered before: change (or, when skipped now, remove) the sample that answer added. + sample = s.get(Sample, row.sample_id) if row.sample_id is not None else None + if sample is not None and sample.sensor_id == sensor_id: + if state_id is None: + rt.db.after_commit(s, partial(rt.storage.delete_sample, sensor_id, sample.id, sample.filename)) + s.delete(sample) + row.sample_id = None + else: + s.execute(delete(SampleLabel).where(SampleLabel.sample_id == sample.id)) + s.add(SampleLabel(sample_id=sample.id, state_id=state_id)) + sample.verified = False + rt.schedule_retrain(sensor_id) + state_id = None # nothing more to add + return sensor_id, state_id, frame + + sensor_id, state_id, frame = await asyncio.to_thread(answer_state) if state_id is not None and frame: path = rt.storage.history_path(sensor_id, frame) if path.exists(): image = await asyncio.to_thread(lambda: imaging.load(path)) sample_id = await asyncio.to_thread(rt.add_sample, sensor_id, image, "review", state_id) - with rt.db.session() as s: - answered = s.get(Prediction, prediction_id) - if answered is not None: - answered.sample_id = sample_id + await asyncio.to_thread(_link_answer, rt, prediction_id, sample_id) rt.schedule_retrain(sensor_id) await rt.publish_review_count() return {"ok": True} @@ -227,30 +242,42 @@ async def _answer_object(rt: Runtime, request: Request, prediction_id: int, body ``state_key``, new as a new person called ``value``; no and skip teach nothing (a face is never taught as "not someone").""" action = body.action - with rt.db.session() as s: - row = s.get(Prediction, prediction_id) - is_face = bool(row is not None and ((row.probs or {}).get("ask") or {}).get("face")) + + def about_face() -> bool: + with rt.db.session() as s: + row = s.get(Prediction, prediction_id) + return bool(row is not None and ((row.probs or {}).get("ask") or {}).get("face")) + + is_face = await asyncio.to_thread(about_face) if action not in (("yes", "no", "skip", "person", "new") if is_face else ("yes", "no", "skip")): raise HTTPException(400, "Answer yes, no or skip") - with rt.db.session() as s: - row = s.get(Prediction, prediction_id) - if row is None: - raise HTTPException(404, "Review item not found") - question = (row.probs or {}).get("ask") - if not question: - raise HTTPException(400, "Not a question about an own label") - sensor_id, score, taught = row.sensor_id, row.confidence, row.sample_id - row.reviewed = True - row.sample_id = None - if taught is not None and is_face and (face := s.get(Face, taught)) is not None: - rt.db.after_commit(s, partial(rt.storage.delete_face, face.filename)) # the face the first answer taught - s.delete(face) - elif taught is not None and not is_face and (sample := s.get(Sample, taught)) is not None: - rt.db.after_commit(s, partial(rt.storage.delete_sample, sensor_id, sample.id, sample.filename)) - s.delete(sample) + + def take_out() -> tuple[dict, int, float, int | None]: + """Out of the queue; the box or face an earlier answer taught goes.""" + with rt.db.session() as s: + row = s.get(Prediction, prediction_id) + if row is None: + raise HTTPException(404, "Review item not found") + question = (row.probs or {}).get("ask") + if not question: + raise HTTPException(400, "Not a question about an own label") + sensor_id, score, taught = row.sensor_id, row.confidence, row.sample_id + row.reviewed = True + row.sample_id = None + if taught is not None and is_face and (face := s.get(Face, taught)) is not None: + rt.db.after_commit( + s, partial(rt.storage.delete_face, face.filename) + ) # the face the first answer taught + s.delete(face) + elif taught is not None and not is_face and (sample := s.get(Sample, taught)) is not None: + rt.db.after_commit(s, partial(rt.storage.delete_sample, sensor_id, sample.id, sample.filename)) + s.delete(sample) + return question, sensor_id, score, taught + + question, sensor_id, score, taught = await asyncio.to_thread(take_out) if taught is not None: if is_face: - rt.faces_changed() + await asyncio.to_thread(rt.faces_changed) else: rt.taught_changed(sensor_id) if is_face: @@ -262,9 +289,7 @@ async def _answer_object(rt: Runtime, request: Request, prediction_id: int, body history_id=prediction_id, box=question["box"], label=label, detected=question["detected"], score=score ) result = await teach_box(sensor_id, taught_box, request) - with rt.db.session() as s: - if (answered := s.get(Prediction, prediction_id)) is not None: - answered.sample_id = result["id"] + await asyncio.to_thread(_link_answer, rt, prediction_id, result["id"]) await rt.publish_review_count() return {"ok": True} @@ -286,6 +311,11 @@ async def _answer_face( return face = FaceTeachIn(history_id=prediction_id, box=question["box"], score=score, person=person, new_person=new_person) result = await teach_face(sensor_id, face, request) + await asyncio.to_thread(_link_answer, rt, prediction_id, result["id"]) + + +def _link_answer(rt: Runtime, prediction_id: int, sample_id: int) -> None: + """Remember what an answer taught (a sample, a taught box or a face), so answering again replaces it.""" with rt.db.session() as s: if (answered := s.get(Prediction, prediction_id)) is not None: - answered.sample_id = result["id"] + answered.sample_id = sample_id diff --git a/visionstate/backend/visionstate/api/samples.py b/visionstate/backend/visionstate/api/samples.py index 7e5bc90..dd84244 100644 --- a/visionstate/backend/visionstate/api/samples.py +++ b/visionstate/backend/visionstate/api/samples.py @@ -47,6 +47,18 @@ def copy_limited(source: IO[bytes], target: IO[bytes], max_bytes: int, chunk: in return True +def _state_id(rt: Runtime, sensor_id: int, key: str | None) -> int | None: + """The id of a state sensor's state ``key`` (None: no label); 404/400 as get_sensor.""" + with rt.db.session() as s: + return state_id_for(get_sensor(s, sensor_id, KIND_STATES), key) + + +def spool_upload(upload_file: UploadFile, name: str) -> tuple[Path, bool]: + """An uploaded file copied to a temporary file; True when it was larger than the limit.""" + with tempfile.NamedTemporaryFile(delete=False, suffix=Path(name).suffix) as tmp: + return Path(tmp.name), not copy_limited(upload_file.file, tmp, UPLOAD_LIMITS["max_file_mb"] * 1_000_000) + + def _retrain(request: Request, sensor_id: int) -> None: runtime(request).schedule_retrain(sensor_id) @@ -55,8 +67,7 @@ def _retrain(request: Request, sensor_id: int) -> None: async def capture(sensor_id: int, body: CaptureIn, request: Request) -> dict: """Label the frame the user is looking at (by frame id), or a fresh one.""" rt = runtime(request) - with rt.db.session() as s: - state_id = state_id_for(get_sensor(s, sensor_id, KIND_STATES), body.state_key) + state_id = await asyncio.to_thread(_state_id, rt, sensor_id, body.state_key) data = rt.live_state(sensor_id).frame(body.frame_id) if body.frame_id else None if data is None: shown = await asyncio.to_thread(rt.frame_for_view, sensor_id) @@ -85,16 +96,13 @@ async def upload( """Training images from uploaded files (images, ZIP archives, videos), labelled with ``state_key`` when given; returns the new samples and an error per file that failed.""" rt = runtime(request) - with rt.db.session() as s: - state_id = state_id_for(get_sensor(s, sensor_id, KIND_STATES), state_key or None) + state_id = await asyncio.to_thread(_state_id, rt, sensor_id, state_key or None) created: list[int] = [] errors: list[str] = [] for upload_file in files: name = upload_file.filename or "upload" - with tempfile.NamedTemporaryFile(delete=False, suffix=Path(name).suffix) as tmp: - tmp_path = Path(tmp.name) - too_large = not copy_limited(upload_file.file, tmp, UPLOAD_LIMITS["max_file_mb"] * 1_000_000) + tmp_path, too_large = await asyncio.to_thread(spool_upload, upload_file, name) if too_large: tmp_path.unlink(missing_ok=True) errors.append(f"{name}: larger than {UPLOAD_LIMITS['max_file_mb']} MB") @@ -145,22 +153,29 @@ async def list_samples( """A page of a sensor's training images, newest first, with the model's suggestion for unlabelled ones.""" rt = runtime(request) - with rt.db.session() as s: - sensor = get_sensor(s, sensor_id) - key_by_state = {st.id: st.key for st in sensor.states} - labelled_ids = select(SampleLabel.sample_id) - query = select(Sample).where(Sample.sensor_id == sensor_id) - if filter == "labelled": - query = query.where(Sample.id.in_(labelled_ids)) - elif filter == "unlabelled": - query = query.where(Sample.id.not_in(labelled_ids)) - if state: - query = query.where( - Sample.id.in_(select(SampleLabel.sample_id).where(SampleLabel.state_id == state_id_for(sensor, state))) - ) - total = s.scalar(select(func.count()).select_from(query.subquery())) - samples = s.scalars(query.order_by(Sample.created_at.desc()).limit(min(limit, 1000)).offset(offset)).all() - cfg = SensorConfig.from_row(sensor) + + def page() -> tuple[dict[int, str], int | None, list[Sample], SensorConfig]: + with rt.db.session() as s: + sensor = get_sensor(s, sensor_id) + key_by_state = {st.id: st.key for st in sensor.states} + labelled_ids = select(SampleLabel.sample_id) + query = select(Sample).where(Sample.sensor_id == sensor_id) + if filter == "labelled": + query = query.where(Sample.id.in_(labelled_ids)) + elif filter == "unlabelled": + query = query.where(Sample.id.not_in(labelled_ids)) + if state: + query = query.where( + Sample.id.in_( + select(SampleLabel.sample_id).where(SampleLabel.state_id == state_id_for(sensor, state)) + ) + ) + total = s.scalar(select(func.count()).select_from(query.subquery())) + samples = s.scalars(query.order_by(Sample.created_at.desc()).limit(min(limit, 1000)).offset(offset)).all() + cfg = SensorConfig.from_row(sensor) + return key_by_state, total, list(samples), cfg + + key_by_state, total, samples, cfg = await asyncio.to_thread(page) unlabelled = [x for x in samples if not x.labels] suggestions = await asyncio.to_thread(rt.suggest, cfg, unlabelled) if unlabelled else {} return { @@ -187,18 +202,27 @@ def label_samples(sensor_id: int, body: LabelIn, request: Request) -> dict: async def accept_suggestions(sensor_id: int, body: IdsIn, request: Request) -> dict: """Label the given unlabelled images with what the model suggests for them.""" rt = runtime(request) - with rt.db.session() as s: - sensor = get_sensor(s, sensor_id) - cfg = SensorConfig.from_row(sensor) - samples = [ - x - for x in s.scalars(select(Sample).where(Sample.sensor_id == sensor_id, Sample.id.in_(body.sample_ids))) - if not x.labels - ] + + def unlabelled() -> tuple[SensorConfig, list[Sample]]: + with rt.db.session() as s: + sensor = get_sensor(s, sensor_id) + cfg = SensorConfig.from_row(sensor) + samples = [ + x + for x in s.scalars(select(Sample).where(Sample.sensor_id == sensor_id, Sample.id.in_(body.sample_ids))) + if not x.labels + ] + return cfg, samples + + cfg, samples = await asyncio.to_thread(unlabelled) suggestions = await asyncio.to_thread(rt.suggest, cfg, samples) state_ids = {st["key"]: st["id"] for st in cfg.states} - with rt.db.session() as s: - s.add_all(SampleLabel(sample_id=i, state_id=state_ids[key]) for i, (key, _) in suggestions.items()) + + def label() -> None: + with rt.db.session() as s: + s.add_all(SampleLabel(sample_id=i, state_id=state_ids[key]) for i, (key, _) in suggestions.items()) + + await asyncio.to_thread(label) if suggestions: _retrain(request, sensor_id) return {"updated": len(suggestions)} diff --git a/visionstate/backend/visionstate/api/sensors.py b/visionstate/backend/visionstate/api/sensors.py index 1dee9e6..9d3aa15 100644 --- a/visionstate/backend/visionstate/api/sensors.py +++ b/visionstate/backend/visionstate/api/sensors.py @@ -43,6 +43,7 @@ StateIn, Triggers, check_faces, + check_sensor, check_teach, get_sensor, runtime, @@ -179,13 +180,17 @@ def list_sensors(request: Request) -> list[dict]: async def create_sensor(body: SensorIn, request: Request) -> dict: rt = runtime(request) body.checked() - with rt.db.session() as s: - sensor = body.new_sensor(unique_slug(s, body.name)) - s.add(sensor) - s.flush() - sensor_id = sensor.id + + def create() -> int: + with rt.db.session() as s: + sensor = body.new_sensor(unique_slug(s, body.name)) + s.add(sensor) + s.flush() + return sensor.id + + sensor_id = await asyncio.to_thread(create) await rt.sensor_created(sensor_id) - return get_one(sensor_id, request) + return await asyncio.to_thread(get_one, sensor_id, request) @router.get("/{sensor_id}") @@ -202,44 +207,48 @@ async def update_sensor(sensor_id: int, body: SensorPatch, request: Request) -> _check_source_type(body.source_type) if body.states is not None: validate_states(body.states) - with rt.db.session() as s: - sensor = get_sensor(s, sensor_id) - if sensor.kind != KIND_STATES and body.states is not None: - raise HTTPException(400, "Only state sensors have states") - if sensor.kind != KIND_OBJECTS and body.objects is not None: - raise HTTPException(400, "Only object sensors have objects") - if sensor.kind != KIND_READING and body.reading is not None: - raise HTTPException(400, "Only reading sensors have reading settings") - old_roi, old_keys, old_source = sensor.roi, [st.key for st in sensor.states], sensor.source - for field in ("name", "source_type", "source", "interval_s", "threshold", "debounce", "enabled", "publish"): - value = getattr(body, field) - if value is not None: - setattr(sensor, field, value) - if body.review is not None: - sensor.review = body.review.stored() - if body.triggers is not None: - sensor.triggers = body.triggers.model_dump() - if body.clear_roi: - sensor.roi = None - elif body.roi is not None: - sensor.roi = body.roi.normalised() - if body.states is not None: - _apply_states(sensor, body.states) - if body.objects is not None: - # Merged, so own labels (made by teaching) and fields an older UI does not send stay. - sensor.objects = {**(sensor.objects or {}), **body.objects.stored(exclude_unset=True)} - merged = merge_objects(sensor.objects) - check_teach(merge_teach(rt.backbone_id, rt.global_teach, merged["teach"])) - check_faces(faces.merge_limits(rt.face_choice, rt.global_faces, merged["faces"])) - if body.reading is not None: - sensor.reading = body.reading.model_dump() - if sensor.roi != old_roi or sensor.source != old_source: - sensor.frame_size = None # the next frame's size is the one the region is drawn on - s.flush() - changed = sensor.roi != old_roi or [st.key for st in sensor.states] != old_keys - retrain = changed and sensor.kind == KIND_STATES + + def change() -> bool: + with rt.db.session() as s: + sensor = get_sensor(s, sensor_id) + if sensor.kind != KIND_STATES and body.states is not None: + raise HTTPException(400, "Only state sensors have states") + if sensor.kind != KIND_OBJECTS and body.objects is not None: + raise HTTPException(400, "Only object sensors have objects") + if sensor.kind != KIND_READING and body.reading is not None: + raise HTTPException(400, "Only reading sensors have reading settings") + old_roi, old_keys, old_source = sensor.roi, [st.key for st in sensor.states], sensor.source + for field in ("name", "source_type", "source", "interval_s", "threshold", "debounce", "enabled", "publish"): + value = getattr(body, field) + if value is not None: + setattr(sensor, field, value) + if body.review is not None: + sensor.review = body.review.stored() + if body.triggers is not None: + sensor.triggers = body.triggers.model_dump() + if body.clear_roi: + sensor.roi = None + elif body.roi is not None: + sensor.roi = body.roi.normalised() + if body.states is not None: + _apply_states(sensor, body.states) + if body.objects is not None: + # Merged, so own labels (made by teaching) and fields an older UI does not send stay. + sensor.objects = {**(sensor.objects or {}), **body.objects.stored(exclude_unset=True)} + merged = merge_objects(sensor.objects) + check_teach(merge_teach(rt.backbone_id, rt.global_teach, merged["teach"])) + check_faces(faces.merge_limits(rt.face_choice, rt.global_faces, merged["faces"])) + if body.reading is not None: + sensor.reading = body.reading.model_dump() + if sensor.roi != old_roi or sensor.source != old_source: + sensor.frame_size = None # the next frame's size is the one the region is drawn on + s.flush() + changed = sensor.roi != old_roi or [st.key for st in sensor.states] != old_keys + return changed and sensor.kind == KIND_STATES + + retrain = await asyncio.to_thread(change) await rt.sensor_updated(sensor_id, retrain=retrain) - return get_one(sensor_id, request) + return await asyncio.to_thread(get_one, sensor_id, request) @router.delete("/{sensor_id}", status_code=204) @@ -250,8 +259,12 @@ async def delete_sensor(sensor_id: int, request: Request) -> Response: raise HTTPException(404, "Sensor not found") if rt.live_state(sensor_id).camera_changed is not None: await rt.camera_notice(cfg, None) - with rt.db.session() as s: - s.delete(get_sensor(s, sensor_id)) + + def delete() -> None: + with rt.db.session() as s: + s.delete(get_sensor(s, sensor_id)) + + await asyncio.to_thread(delete) await rt.sensor_deleted(sensor_id, cfg.descriptor) return Response(status_code=204) @@ -279,7 +292,7 @@ async def live_frame(sensor_id: int, request: Request, cached: bool = False, fra raise HTTPException(404, "Frame no longer cached") latest_id, data = live.frames[-1] return Response(data, media_type="image/jpeg", headers={"X-Frame-Id": latest_id, "Cache-Control": "no-store"}) - shown = rt.frame_for_view(sensor_id) + shown = await asyncio.to_thread(rt.frame_for_view, sensor_id) if cached and live.frames: new_id, data = live.frames[-1] elif shown is not None: @@ -312,8 +325,12 @@ async def accept_picture(sensor_id: int, request: Request) -> dict: """A reading sensor's camera changed its picture size and the region is still right: read again, with the next frame's size as the one the region is drawn on.""" rt = runtime(request) - with rt.db.session() as s: - get_sensor(s, sensor_id, KIND_READING).frame_size = None + + def accept() -> None: + with rt.db.session() as s: + get_sensor(s, sensor_id, KIND_READING).frame_size = None + + await asyncio.to_thread(accept) rt.wake(sensor_id, force=True, paused_too=True) return {"ok": True} @@ -321,8 +338,7 @@ async def accept_picture(sensor_id: int, request: Request) -> dict: @router.post("/{sensor_id}/retrain") async def retrain(sensor_id: int, request: Request) -> dict: rt = runtime(request) - with rt.db.session() as s: - get_sensor(s, sensor_id, KIND_STATES) + await asyncio.to_thread(check_sensor, rt, sensor_id, KIND_STATES) rt.schedule_retrain(sensor_id, delay=0) return {"ok": True} @@ -519,8 +535,7 @@ async def reading_export( GitHub's file size: only the region of each, what was read, the answer. ``meter``: what the user says the meter is.""" rt = runtime(request) - with rt.db.session() as s: - get_sensor(s, sensor_id, KIND_READING) + await asyncio.to_thread(check_sensor, rt, sensor_id, KIND_READING) reader = await asyncio.to_thread(rt.db.get_text, "reader", readers.DEFAULT_READER) fd, name = tempfile.mkstemp(suffix=".zip") os.close(fd) diff --git a/visionstate/backend/visionstate/api/system.py b/visionstate/backend/visionstate/api/system.py index 4f96d17..f821b1d 100644 --- a/visionstate/backend/visionstate/api/system.py +++ b/visionstate/backend/visionstate/api/system.py @@ -306,7 +306,7 @@ async def put_settings(body: SettingsIn, request: Request) -> dict: raise HTTPException(400, "Unknown reader") if body.face_model is not None and body.face_model not in faces.RECOGNISERS: raise HTTPException(400, "Unknown face model") - current = get_settings(request) + current = await asyncio.to_thread(get_settings, request) # Only reload what changed: a new backbone retrains every state sensor. if body.backbone != current["backbone"]: try: @@ -325,7 +325,7 @@ async def put_settings(body: SettingsIn, request: Request) -> dict: raise HTTPException(500, f"Could not load reader: {redact(str(err))}") from err if body.face_model is not None and body.face_model != current["face_model"]: await rt.set_face_model(body.face_model) # loads in the background when a sensor uses faces - return get_settings(request) + return await asyncio.to_thread(get_settings, request) @router.post("/faces/retry") diff --git a/visionstate/backend/visionstate/api/teach.py b/visionstate/backend/visionstate/api/teach.py index a787798..7390742 100644 --- a/visionstate/backend/visionstate/api/teach.py +++ b/visionstate/backend/visionstate/api/teach.py @@ -5,6 +5,7 @@ import asyncio from datetime import UTC, timedelta from functools import partial +from pathlib import Path from fastapi import APIRouter, HTTPException, Request, Response from PIL import Image @@ -105,11 +106,15 @@ async def frame_image( ) -> tuple[Image.Image, str]: """The frame a box was drawn on, and where it came from ("live" or "history").""" if history_id is not None: - with rt.db.session() as s: - row = s.get(Prediction, history_id) - if row is None or row.sensor_id != sensor_id or not row.frame: - raise HTTPException(404, "History frame not found") - path = rt.storage.history_path(sensor_id, row.frame) + + def history_frame() -> Path: + with rt.db.session() as s: + row = s.get(Prediction, history_id) + if row is None or row.sensor_id != sensor_id or not row.frame: + raise HTTPException(404, "History frame not found") + return rt.storage.history_path(sensor_id, row.frame) + + path = await asyncio.to_thread(history_frame) if not path.exists(): raise HTTPException(404, "History frame file missing") return await asyncio.to_thread(imaging.load, path), "history" @@ -129,41 +134,51 @@ async def teach_box(sensor_id: int, body: TeachIn, request: Request) -> dict: if not, it cannot be found there later either. """ rt = runtime(request) - with rt.db.session() as s: - sensor = get_sensor(s, sensor_id, KIND_OBJECTS) - objects = merge_objects(sensor.objects) - count = s.scalar( - select(func.count()) - .select_from(Sample) - .where(Sample.sensor_id == sensor_id, Sample.object_label.is_not(None)) - ) - if (count or 0) >= TEACH["max_examples"]: - raise HTTPException(400, f"A sensor can be taught at most {TEACH['max_examples']} boxes") - if body.detected is not None and body.detected not in detectors.LABELS.by_key: - raise HTTPException(400, f"Unknown object {body.detected!r}") - new = None - if body.new_label: - new = own_label( - objects, body.new_label, body.parent or body.detected or "", set(s.scalars(select(Person.key))) + + def check() -> tuple[dict | None, str, dict | None]: + """The label the box is taught as (and a new own label, if it makes one); 400 when it cannot be.""" + with rt.db.session() as s: + sensor = get_sensor(s, sensor_id, KIND_OBJECTS) + objects = merge_objects(sensor.objects) + count = s.scalar( + select(func.count()) + .select_from(Sample) + .where(Sample.sensor_id == sensor_id, Sample.object_label.is_not(None)) ) - label = new["key"] - else: - active = {c["key"] for c in active_custom(objects)} - label = body.label or "" - if label == NONE_LABEL and body.detected is None: - raise HTTPException(400, "A box the detector missed needs a label") - if label != NONE_LABEL and label not in objects["classes"] and label not in active: - raise HTTPException(400, f"Not one of the sensor's objects or labels: {label!r}") - roi = sensor.roi + if (count or 0) >= TEACH["max_examples"]: + raise HTTPException(400, f"A sensor can be taught at most {TEACH['max_examples']} boxes") + if body.detected is not None and body.detected not in detectors.LABELS.by_key: + raise HTTPException(400, f"Unknown object {body.detected!r}") + new = None + if body.new_label: + new = own_label( + objects, body.new_label, body.parent or body.detected or "", set(s.scalars(select(Person.key))) + ) + label = new["key"] + else: + active = {c["key"] for c in active_custom(objects)} + label = body.label or "" + if label == NONE_LABEL and body.detected is None: + raise HTTPException(400, "A box the detector missed needs a label") + if label != NONE_LABEL and label not in objects["classes"] and label not in active: + raise HTTPException(400, f"Not one of the sensor's objects or labels: {label!r}") + roi = sensor.roi + return new, label, roi + + new, label, roi = await asyncio.to_thread(check) x1, _, x2, y2 = body.box if not imaging.in_region((x1 + x2) / 2, y2, roi): raise HTTPException(400, "That box is outside the sensor's region, so it never counts") image, origin = await frame_image(rt, sensor_id, body.frame_id, body.history_id) if new is not None: - with rt.db.session() as s: - row = get_sensor(s, sensor_id) - stored = merge_objects(row.objects) - row.objects = {**(row.objects or {}), "custom": [*stored["custom"], new]} + + def add_label() -> None: + with rt.db.session() as s: + row = get_sensor(s, sensor_id) + stored = merge_objects(row.objects) + row.objects = {**(row.objects or {}), "custom": [*stored["custom"], new]} + + await asyncio.to_thread(add_label) example_id = await asyncio.to_thread( rt.add_object_example, sensor_id, image, body.box, label, body.detected, body.score, origin ) @@ -262,15 +277,23 @@ def forget_box(sensor_id: int, example_id: int, request: Request) -> Response: async def remove_labels(rt: Runtime, sensor_id: int, keys: set[str] | None) -> None: """Remove own labels (``None``: all of them, and every taught box) with their entities.""" - with rt.db.session() as s: - sensor = get_sensor(s, sensor_id, KIND_OBJECTS) - objects = merge_objects(sensor.objects) - gone = [c["key"] for c in objects["custom"] if keys is None or c["key"] in keys] - if keys is not None and not gone: - raise HTTPException(404, "Label not found") - delete_examples(rt, s, sensor_id, [] if keys is None else [Sample.object_label.in_(gone)]) - sensor.objects = {**(sensor.objects or {}), "custom": [c for c in objects["custom"] if c["key"] not in gone]} - slug = sensor.slug + + def remove() -> tuple[list[str], str]: + with rt.db.session() as s: + sensor = get_sensor(s, sensor_id, KIND_OBJECTS) + objects = merge_objects(sensor.objects) + gone = [c["key"] for c in objects["custom"] if keys is None or c["key"] in keys] + if keys is not None and not gone: + raise HTTPException(404, "Label not found") + delete_examples(rt, s, sensor_id, [] if keys is None else [Sample.object_label.in_(gone)]) + sensor.objects = { + **(sensor.objects or {}), + "custom": [c for c in objects["custom"] if c["key"] not in gone], + } + slug = sensor.slug + return gone, slug + + gone, slug = await asyncio.to_thread(remove) for key in gone: await rt.mqtt.remove_object_class(slug, key) await rt.sensor_updated(sensor_id, retrain=False) diff --git a/visionstate/backend/visionstate/backbones.py b/visionstate/backend/visionstate/backbones.py index 1272cd5..dafda83 100644 --- a/visionstate/backend/visionstate/backbones.py +++ b/visionstate/backend/visionstate/backbones.py @@ -9,6 +9,7 @@ from __future__ import annotations import ctypes +import functools import gc import hashlib import json @@ -25,6 +26,7 @@ from PIL import Image from . import imaging +from .settings import RUNTIME log = logging.getLogger(__name__) @@ -165,11 +167,18 @@ def cpu_session(model_path: Path): No memory arena: with it, each session keeps the largest working memory any image ever needed (D-FINE S alone held about 265 MB after a few detections) for the rest of the run; without it that memory is freed after every run, for about 6 % more time per check. + + No spinning: by default ONNX Runtime's threads keep a core busy while they wait for work, which + on a small machine takes CPU from Home Assistant itself; RUNTIME["inference_threads"] threads + per model (0: ONNX Runtime's choice, one per physical core). """ import onnxruntime as ort options = ort.SessionOptions() options.enable_cpu_mem_arena = False + options.add_session_config_entry("session.intra_op.allow_spinning", "0") + options.add_session_config_entry("session.inter_op.allow_spinning", "0") + options.intra_op_num_threads = RUNTIME["inference_threads"] options.log_severity_level = 3 options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL return ort.InferenceSession(str(model_path), sess_options=options, providers=["CPUExecutionProvider"]) @@ -190,10 +199,16 @@ def trim_memory() -> None: Without it glibc keeps what the largest image needed: on Home Assistant OS the app stayed about 190 MB larger with the default models. Only glibc has it; elsewhere this does nothing.""" + if (trim := _malloc_trim()) is not None: + trim(0) + + +@functools.cache +def _malloc_trim(): # noqa: ANN202 - a ctypes function, or None try: - ctypes.CDLL("libc.so.6").malloc_trim(0) + return ctypes.CDLL("libc.so.6").malloc_trim except OSError, AttributeError: - pass + return None class Embedder: diff --git a/visionstate/backend/visionstate/db.py b/visionstate/backend/visionstate/db.py index 1c00a55..5300551 100644 --- a/visionstate/backend/visionstate/db.py +++ b/visionstate/backend/visionstate/db.py @@ -255,8 +255,13 @@ class Prediction(Base): """ __tablename__ = "prediction" - # The history of some sensors, newest first, paged by (time, id): see api/history.py. - __table_args__ = (Index("ix_prediction_sensor_time", "sensor_id", "created_at", "id"),) + # The history of some sensors, newest first, paged by (time, id): see api/history.py. The frames + # waiting for review (counted after every check that adds one): the condition is written as + # SQLAlchemy writes ``reviewed.is_(False)``, as SQLite only uses a partial index for that text. + __table_args__ = ( + Index("ix_prediction_sensor_time", "sensor_id", "created_at", "id"), + Index("ix_prediction_waiting", "sensor_id", "created_at", sqlite_where=text("reviewed IS 0")), + ) id: Mapped[int] = mapped_column(primary_key=True) sensor_id: Mapped[int] = mapped_column(ForeignKey("sensor.id", ondelete="CASCADE"), index=True) @@ -379,6 +384,9 @@ def __init__(self, path: Path): def _pragmas(dbapi_conn, _record): # noqa: ANN001 cur = dbapi_conn.cursor() cur.execute("PRAGMA journal_mode=WAL") + # Safe from corruption with WAL; a power cut may lose the last commit, never a crash of the + # app. Saves a sync of the log at every commit (SD cards). + cur.execute("PRAGMA synchronous=NORMAL") cur.execute("PRAGMA foreign_keys=ON") cur.close() diff --git a/visionstate/backend/visionstate/engine/checks.py b/visionstate/backend/visionstate/engine/checks.py index 0237f6f..d32b492 100644 --- a/visionstate/backend/visionstate/engine/checks.py +++ b/visionstate/backend/visionstate/engine/checks.py @@ -7,7 +7,6 @@ import math import random import time -from datetime import UTC, datetime from PIL import Image @@ -18,7 +17,7 @@ from ..settings import RUNTIME, UNKNOWN_STATE from ..sources import SourceError from .base import RuntimeBase -from .logic import next_check_at, review_reason +from .logic import next_check_at, percent, review_reason from .objects import ObjectChecksMixin from .reading import ReadingChecksMixin from .state import SensorConfig @@ -284,20 +283,21 @@ async def _run_states(self, cfg: SensorConfig, image: Image.Image) -> None: crop = imaging.crop(image, cfg.roi) await self._send(cfg, t["state"], live.debouncer.published or UNKNOWN_STATE, retain=True) - await self._send(cfg, t["confidence"], f"{confidence * 100:.1f}", retain=True) + await self._send(cfg, t["confidence"], percent(confidence), retain=True) await self._send( cfg, t["attributes"], { - "probabilities": {k: round(v, 4) for k, v in probs.items()}, + # Whole percent, like the confidence: a value that changes at every check makes Home + # Assistant store the state again each time. + "probabilities": {k: round(v, 2) for k, v in probs.items()}, "top_state": top, - "last_update": datetime.now(UTC).isoformat(), "trained": bool(probs), "last_trigger": live.last_trigger, }, retain=True, ) - await self._send(cfg, t["image"], await asyncio.to_thread(imaging.encode_jpeg, crop, 80), retain=True) + await self._send_on_change(cfg, t["image"], live.debouncer.published, lambda: imaging.encode_jpeg(crop, 80)) reason = None if probs: diff --git a/visionstate/backend/visionstate/engine/history.py b/visionstate/backend/visionstate/engine/history.py index 5be749b..bc62e1c 100644 --- a/visionstate/backend/visionstate/engine/history.py +++ b/visionstate/backend/visionstate/engine/history.py @@ -5,17 +5,23 @@ import asyncio import logging import shutil +import time +from collections.abc import Sequence from datetime import datetime, timedelta +from functools import partial +from pathlib import Path -from sqlalchemy import func, select +from sqlalchemy import delete, func, select from ..db import Prediction, Sample, utcnow -from ..settings import RUNTIME, SAMPLES_MOVE, STORAGE_DEFAULTS +from ..settings import HISTORY_CLEANUP, RUNTIME, SAMPLES_MOVE, STORAGE_DEFAULTS from .base import RuntimeBase from .publishing import PublishingMixin log = logging.getLogger(__name__) +HistoryRow = tuple[int, int, str | None] # a history row's id, sensor_id and frame + class HistoryMixin(PublishingMixin, RuntimeBase): async def _cleanup_loop(self) -> None: @@ -102,41 +108,58 @@ def set_storage_limits(self, rules: dict) -> None: self.history_trimmed = False # the next clean-up tells whether the new size limit still bites self.db.set_setting("storage", self.storage_rules) - def _history_file_size(self, row: Prediction) -> int: + def _history_file_size(self, prediction_id: int, sensor_id: int, frame: str | None) -> int: size = 0 for path in ( - self.storage.history_path(row.sensor_id, row.frame) if row.frame else None, - self.storage.thumb_path("history", row.id), + self.storage.history_path(sensor_id, frame) if frame else None, + self.storage.thumb_path("history", prediction_id), ): if path is not None and path.exists(): size += path.stat().st_size return size + def _remove_history(self, rows: Sequence[HistoryRow]) -> None: + """Remove history rows (id, sensor_id, frame); their files go once that is committed + (db.Database.after_commit), so a failed commit never leaves rows without their frames.""" + for start in range(0, len(rows), HISTORY_CLEANUP["batch"]): + batch = rows[start : start + HISTORY_CLEANUP["batch"]] + with self.db.session() as s: + s.execute(delete(Prediction).where(Prediction.id.in_([row[0] for row in batch]))) + self.db.after_commit(s, partial(self._delete_history_files, batch)) + + def _delete_history_files(self, rows: Sequence[HistoryRow]) -> None: + for prediction_id, sensor_id, frame in rows: + self.storage.delete_history(sensor_id, prediction_id, frame) + def cleanup_history(self) -> None: """Remove history older than the limit (frames waiting for review get twice as long), - then the oldest frames while the history is larger than its size limit. Readings - verified by hand are kept.""" + then the oldest frames while the history is larger than its size limit, then frames no + row lists any more. Readings verified by hand are kept.""" limits = self.storage_limits() days = limits["history_days"] cutoff = utcnow() - timedelta(days=days) review_cutoff = utcnow() - timedelta(days=days * 2) with self.db.session() as s: - old = s.scalars( - select(Prediction).where( - Prediction.created_at < cutoff, - (Prediction.reviewed.is_(True)) | (Prediction.created_at < review_cutoff), - Prediction.read_ok.is_(None), # verified readings are kept, like training images + old: list[HistoryRow] = [ + (row.id, row.sensor_id, row.frame) + for row in s.execute( + select(Prediction.id, Prediction.sensor_id, Prediction.frame).where( + Prediction.created_at < cutoff, + (Prediction.reviewed.is_(True)) | (Prediction.created_at < review_cutoff), + Prediction.read_ok.is_(None), # verified readings are kept, like training images + ) ) - ).all() - for row in old: - self.storage.delete_history(row.sensor_id, row.id, row.frame) - s.delete(row) + ] + self._remove_history(old) if old: log.info("Removed %d old history frames", len(old)) trimmed = self._trim_history(limits["history_max_gb"]) if trimmed: self.history_trimmed = True log.info("Removed %d history frames to stay under %.1f GB", trimmed, limits["history_max_gb"]) + orphans = self.remove_orphan_frames() + if orphans: + log.info("Removed %d history frames that no history row listed", orphans) def _trim_history(self, max_gb: float) -> int: """Remove the oldest history frames until all of them fit in ``max_gb`` (0 = no limit). @@ -151,21 +174,59 @@ def _trim_history(self, max_gb: float) -> int: if used <= budget: return 0 removed = 0 + for waiting in (False, True): + with self.db.session() as s: + rows: list[HistoryRow] = [ + (row.id, row.sensor_id, row.frame) + for row in s.execute( + select(Prediction.id, Prediction.sensor_id, Prediction.frame) + .where(Prediction.reviewed.is_(not waiting), Prediction.read_ok.is_(None)) + .order_by(Prediction.created_at) + ) + ] + chosen: list[HistoryRow] = [] + for row in rows: + if used <= budget: + break + used -= self._history_file_size(*row) + chosen.append(row) + self._remove_history(chosen) + removed += len(chosen) + if used <= budget: + break + return removed + + def remove_orphan_frames(self) -> int: + """History frames and thumbnails no row lists (the app stopped between writing a frame and + its row, or a row went while its file could not be removed), once they are older than + HISTORY_CLEANUP["orphan_age_s"]: a frame is written just before its row. Only the history + folders are looked at. Returns how many files were removed.""" with self.db.session() as s: - for waiting in (False, True): - rows = s.scalars( - select(Prediction) - .where(Prediction.reviewed.is_(not waiting), Prediction.read_ok.is_(None)) - .order_by(Prediction.created_at) - ).all() - for row in rows: - if used <= budget: - return removed - used -= self._history_file_size(row) - self.storage.delete_history(row.sensor_id, row.id, row.frame) - s.delete(row) + listed = { + (sensor_id, frame) for sensor_id, frame in s.execute(select(Prediction.sensor_id, Prediction.frame)) + } + ids = set(s.scalars(select(Prediction.id))) + before = time.time() - HISTORY_CLEANUP["orphan_age_s"] + removed = 0 + candidates: list[tuple[Path, bool]] = [] + history = self.settings.history_dir + if history.is_dir(): + for folder in history.iterdir(): + if folder.is_dir() and folder.name.isdigit(): + sensor_id = int(folder.name) + candidates += [(f, (sensor_id, f.name) in listed) for f in folder.iterdir() if f.is_file()] + thumbs = self.settings.thumbs_dir / "history" + if thumbs.is_dir(): + candidates += [(f, f.stem.isdigit() and int(f.stem) in ids) for f in thumbs.iterdir() if f.is_file()] + for path, known in candidates: + if known: + continue + try: + if path.stat().st_mtime < before: + path.unlink() removed += 1 - s.flush() + except OSError: # gone meanwhile, or cannot be removed: tried again next time + pass return removed def storage_usage(self) -> dict: diff --git a/visionstate/backend/visionstate/engine/logic.py b/visionstate/backend/visionstate/engine/logic.py index f276d12..e9d7705 100644 --- a/visionstate/backend/visionstate/engine/logic.py +++ b/visionstate/backend/visionstate/engine/logic.py @@ -240,3 +240,9 @@ def reading_verdict( if reason == "went down" and readers.settling(value, last, settings): return None, True return reason, False + + +def percent(share: float) -> str: + """A confidence for Home Assistant, in whole percent: a value that changes at every check + makes Home Assistant store the entity's state again each time.""" + return f"{share * 100:.0f}" diff --git a/visionstate/backend/visionstate/engine/models.py b/visionstate/backend/visionstate/engine/models.py index 832c8bd..0c8b7f0 100644 --- a/visionstate/backend/visionstate/engine/models.py +++ b/visionstate/backend/visionstate/engine/models.py @@ -42,7 +42,7 @@ async def load_embedder(self, backbone_id: str) -> None: async def set_backbone(self, backbone_id: str) -> None: await self.load_embedder(backbone_id) await asyncio.to_thread(backbones.release_memory) - self.db.set_setting("backbone", backbone_id) + await asyncio.to_thread(self.db.set_setting, "backbone", backbone_id) for sensor_id in await asyncio.to_thread(self._sensor_ids, KIND_STATES): self.schedule_retrain(sensor_id, delay=0) diff --git a/visionstate/backend/visionstate/engine/objects.py b/visionstate/backend/visionstate/engine/objects.py index 1c343ac..7b58f76 100644 --- a/visionstate/backend/visionstate/engine/objects.py +++ b/visionstate/backend/visionstate/engine/objects.py @@ -70,7 +70,7 @@ async def _run_objects(self, cfg: SensorConfig, image: Image.Image, frame_id: st track = live.tracks[key] ot = object_topics(cfg.slug, key) attributes = { - "confidence": round(track.score, 4), + "confidence": round(track.score, 2), "boxes": [d["box"] for d in found if teach.counts_for(d, key)], "last_seen": datetime.fromtimestamp(track.last_seen, UTC).isoformat() if track.last_seen else None, "last_trigger": live.last_trigger, @@ -79,18 +79,21 @@ async def _run_objects(self, cfg: SensorConfig, image: Image.Image, frame_id: st attributes["filtered"] = sum(1 for d in found if d.get("filtered") and d["key"] == key) await self._send(cfg, ot["state"], "ON" if track.on else "OFF", retain=True) await self._send(cfg, ot["count"], str(track.count), retain=True) - await self._send(cfg, ot["attributes"], attributes, retain=True) + shows = (track.on, track.count, live.last_trigger, attributes.get("filtered")) + await self._send_on_change(cfg, ot["attributes"], shows, lambda attributes=attributes: attributes) names = {key: label.name for key, label in detectors.LABELS.by_key.items()} names.update({c["key"]: c["name"] for c in cfg.custom_labels}) names[FACE_UNKNOWN_KEY] = "Unknown person" drawn = [{**d, "key": d.get("label") or d["key"]} for d in counted] - annotated = await asyncio.to_thread( - lambda: imaging.crop_box( - imaging.draw_detections(image, drawn, names, OBJECT_BOX_COLOR), imaging.region_box(cfg.roi) + + def picture() -> bytes: + box = imaging.region_box(cfg.roi) + return imaging.encode_jpeg( + imaging.crop_box(imaging.draw_detections(image, drawn, names, OBJECT_BOX_COLOR), box), 80 ) - ) - jpeg = await asyncio.to_thread(imaging.encode_jpeg, annotated, 80) - await self._send(cfg, topics(cfg.slug)["image"], jpeg, retain=True) + + shows = tuple((key, live.tracks[key].on, live.tracks[key].count) for key in keys) + await self._send_on_change(cfg, topics(cfg.slug)["image"], shows, picture) for key in changed: track = live.tracks[key] published = "on" if track.on else "off" diff --git a/visionstate/backend/visionstate/engine/publishing.py b/visionstate/backend/visionstate/engine/publishing.py index a9012d6..32ce119 100644 --- a/visionstate/backend/visionstate/engine/publishing.py +++ b/visionstate/backend/visionstate/engine/publishing.py @@ -4,12 +4,15 @@ import asyncio import logging +import time +from collections.abc import Callable from datetime import UTC, datetime from sqlalchemy import func, select from ..db import Prediction, Sensor from ..mqtt import REVIEW_TOPICS, object_topics, topics +from ..settings import PUBLISH from .base import RuntimeBase from .state import SensorConfig @@ -22,6 +25,28 @@ async def _send(self, cfg: SensorConfig, topic: str, payload: str | bytes | dict if cfg.publish: await self.mqtt.publish(topic, payload, retain=retain) + async def _send_on_change( + self, cfg: SensorConfig, topic: str, shows: object, make: Callable[[], bytes | dict] + ) -> None: + """A picture, or attributes that change a little at every check (an object's boxes and + when it was last seen): sent when what they are about changed (``shows``: the state, the + objects, the value), otherwise only every PUBLISH["refresh_s"]. + + Home Assistant stores the state again for every picture an image entity gets (its state + is the time of the picture) and for attributes that changed, so sending them at every + check filled its database. ``make`` builds the payload only when it is sent. After MQTT + (re)connects, publish_discovery forgets what was sent, so the next check sends it again. + """ + if not cfg.publish: + return + live = self.live_state(cfg.id) + now = time.time() + sent = live.sent_on_change.get(topic) + if sent is not None and sent[0] == shows and now - sent[1] < PUBLISH["refresh_s"]: + return + await self.mqtt.publish(topic, await asyncio.to_thread(make), retain=True) + live.sent_on_change[topic] = (shows, now) + async def publish_discovery(self, cfg: SensorConfig) -> None: """Tell Home Assistant about a sensor's entities (again), and repeat what it last knew. @@ -39,6 +64,8 @@ async def publish_discovery(self, cfg: SensorConfig) -> None: await self.mqtt.remove_rate(cfg.slug) # only counters have a rate await self.mqtt.publish_discovery(cfg.descriptor) self._registry_wanted.set() + if cfg.id in self.live: + self.live[cfg.id].sent_on_change.clear() # sent again at the next check (e.g. publishing switched on) await self.mqtt.publish(t["enabled"], "ON" if cfg.enabled else "OFF", retain=True) live = self.live.get(cfg.id) # Not sent to Home Assistant: its entities are unavailable (no values, no statistics). Sent diff --git a/visionstate/backend/visionstate/engine/reading.py b/visionstate/backend/visionstate/engine/reading.py index 73b96d5..0924112 100644 --- a/visionstate/backend/visionstate/engine/reading.py +++ b/visionstate/backend/visionstate/engine/reading.py @@ -18,7 +18,7 @@ from ..redact import redact from ..settings import COUNTER_LIMIT, READING, REMINDER, VERSION from .base import RuntimeBase -from .logic import reading_verdict +from .logic import percent, reading_verdict from .models import ModelsMixin from .publishing import PublishingMixin from .state import SensorConfig @@ -188,21 +188,21 @@ async def _publish_reading( t = topics(cfg.slug) if live.debouncer.published is not None: await self._send(cfg, t["state"], live.debouncer.published, retain=True) - await self._send(cfg, t["confidence"], f"{text.score * 100:.1f}", retain=True) + await self._send(cfg, t["confidence"], percent(text.score), retain=True) await self._send( cfg, t["attributes"], { "read_text": text.text, - "last_update": datetime.now(UTC).isoformat(), "rejected": reason, "last_trigger": live.last_trigger, }, retain=True, ) + shows = (live.debouncer.published, text.text, reason) crop = imaging.crop_box(image, imaging.region_box(cfg.roi)) - await self._send(cfg, t["image"], await asyncio.to_thread(imaging.encode_jpeg, crop, 80), retain=True) - await self._send(cfg, t["reader_image"], reader_jpeg, retain=True) + await self._send_on_change(cfg, t["image"], shows, lambda: imaging.encode_jpeg(crop, 80)) + await self._send_on_change(cfg, t["reader_image"], shows, lambda: reader_jpeg) await self._send_reading_diagnostics(cfg, text.text, shown, reason, now) async def _keep_reading( @@ -351,7 +351,7 @@ async def anchor_reading(self, sensor_id: int, value: str, at: datetime) -> None watch.budget_s, watch.held = max(0.0, now - watch.accepted_at), False watch.seen_at, watch.sample = now, None # nothing learned across it await asyncio.to_thread(self._save_watch, sensor_id, watch, value) - cfg = self.load_sensor(sensor_id) + cfg = await asyncio.to_thread(self.load_sensor, sensor_id) if cfg is not None: await self._send(cfg, topics(cfg.slug)["state"], value, retain=True) diff --git a/visionstate/backend/visionstate/engine/runtime.py b/visionstate/backend/visionstate/engine/runtime.py index 5d628d6..389c0c7 100644 --- a/visionstate/backend/visionstate/engine/runtime.py +++ b/visionstate/backend/visionstate/engine/runtime.py @@ -18,6 +18,7 @@ from .history import HistoryMixin from .logic import entity_triggers from .reminders import RemindersMixin +from .state import SensorConfig log = logging.getLogger(__name__) @@ -54,14 +55,14 @@ async def start(self) -> None: self.face_choice = await asyncio.to_thread(self.db.get_text, "face_model", faces.DEFAULT_RECOGNISER) if self.face_choice not in faces.RECOGNISERS: self.face_choice = faces.DEFAULT_RECOGNISER - backbone_id = self.db.get_text("backbone", backbones.DEFAULT_BACKBONE) + backbone_id = await asyncio.to_thread(self.db.get_text, "backbone", backbones.DEFAULT_BACKBONE) try: await self.load_embedder(backbone_id) except Exception as err: # noqa: BLE001 log.exception("Could not load backbone %s", backbone_id) self.embedder_error = str(err) for sensor_id in await asyncio.to_thread(self._sensor_ids, KIND_STATES): - if self.heads_load(sensor_id): + if await asyncio.to_thread(self.heads_load, sensor_id): log.info("Sensor %s: model from another version, retraining from stored images", sensor_id) self.schedule_retrain(sensor_id, delay=0) await asyncio.to_thread(self.restore_last_seen) # "last seen" survives an update or restart @@ -211,24 +212,20 @@ async def refresh_trigger_entities(self) -> None: self._entity_index = await asyncio.to_thread(self._trigger_index) self.ha_events.set_entities(set(self._entity_index)) - def start_burst(self, sensor_id: int, source: str, detail: str) -> None: - """Check now, then keep checking at the burst pace for the configured duration.""" - cfg = self.load_sensor(sensor_id) - if cfg is None or not cfg.enabled: + def start_burst(self, cfg: SensorConfig, source: str, detail: str) -> None: + """Check now, then keep checking at the burst pace for the configured duration (on the loop).""" + if not cfg.enabled: return - live = self.live_state(sensor_id) + live = self.live_state(cfg.id) now = time.time() live.burst_until = now + cfg.triggers["burst_duration_s"] live.last_trigger = {"source": source, "detail": detail, "at": now} - self.wake(sensor_id, force=True) - - def _entity_changed(self, sensor_id: int, entity_id: str, old: str | None, new: str | None) -> None: - cfg = self.load_sensor(sensor_id) - if cfg is None or not entity_triggers(cfg.triggers, entity_id, new): - return - log.debug("Sensor %s triggered by %s: %s -> %s", sensor_id, entity_id, old, new) - self.start_burst(sensor_id, "entity", f"{entity_id}: {old} → {new}") + self.wake(cfg.id, force=True) async def _on_ha_state(self, entity_id: str, old: str | None, new: str | None) -> None: for sensor_id in self._entity_index.get(entity_id, set()): - await asyncio.to_thread(self._entity_changed, sensor_id, entity_id, old, new) + cfg = await asyncio.to_thread(self.load_sensor, sensor_id) + if cfg is None or not entity_triggers(cfg.triggers, entity_id, new): + continue + log.debug("Sensor %s triggered by %s: %s -> %s", sensor_id, entity_id, old, new) + self.start_burst(cfg, "entity", f"{entity_id}: {old} → {new}") diff --git a/visionstate/backend/visionstate/engine/state.py b/visionstate/backend/visionstate/engine/state.py index 2b62711..65f3b48 100644 --- a/visionstate/backend/visionstate/engine/state.py +++ b/visionstate/backend/visionstate/engine/state.py @@ -241,6 +241,8 @@ class LiveState: # ({"from": [w, h], "to": [w, h], "since": epoch seconds}); readings are held until it is # back or the user accepts it (engine/reading.py). camera_changed: dict | None = None + # Sent to Home Assistant when it changed: topic -> (what it showed, when), see PublishingMixin._send_on_change. + sent_on_change: dict[str, tuple[object, float]] = field(default_factory=dict) def remember(self, data: bytes) -> str: frame_id = f"{time.time_ns():x}" diff --git a/visionstate/backend/visionstate/settings.py b/visionstate/backend/visionstate/settings.py index b1e377e..c305b69 100644 --- a/visionstate/backend/visionstate/settings.py +++ b/visionstate/backend/visionstate/settings.py @@ -543,6 +543,12 @@ "history_days": (1, 365), "history_max_gb": (0.0, 1000.0), } +# The history clean-up (engine/history.py): rows removed per transaction, and how old a frame no row +# lists must be before it is removed (a frame is written just before its row). +HISTORY_CLEANUP = { + "batch": 500, + "orphan_age_s": 3600.0, +} # Training images moved from the media folder to the app's own storage (engine/history.py, # move_samples). The copies left in the media folder are removed this many days after the move: until @@ -575,8 +581,17 @@ # --- Runtime ------------------------------------------------------------------ +# What is sent to Home Assistant (engine/publishing.py). A sensor's picture (its image entity) and +# an object's attributes (boxes, last seen) are sent when what they show changes, and otherwise this +# often, so a dashboard is never far behind: Home Assistant stores every picture it gets, and every +# change of attributes, as a new state. +PUBLISH = { + "refresh_s": 300.0, +} + RUNTIME = { "max_concurrent_inferences": 2, + "inference_threads": 0, # threads per model run (backbones.cpu_session); 0: one per physical core "frame_cache_size": 5, # recent frames kept per sensor so a label hits the frame the user saw "retrain_delay_s": 1.0, # coalesce rapid label clicks into one retrain "cleanup_interval_s": 600, # history clean-up (age and size limits) @@ -584,6 +599,9 @@ # A camera picture larger than this is refused (a misconfigured URL streaming without end, a # decompression bomb): a 12-megapixel JPEG is a few MB, an 8K frame 33 megapixels. "max_frame_mb": 50, + "rtsp_workers": 4, # RTSP grabs at once (on threads of their own, see sources.FrameGrabber) + # An RTSP stream grabbed again within this many seconds is kept open (sources.FrameGrabber); 0: never. + "rtsp_keep_open_s": 5.0, "max_image_megapixels": 60, "loop_retry_s": 30.0, # a sensor's loop failed unexpectedly (database busy): try again after this "night_colorfulness": 4.0, # mean channel difference below this = greyscale/IR image diff --git a/visionstate/backend/visionstate/sources.py b/visionstate/backend/visionstate/sources.py index c373fca..0f6372b 100644 --- a/visionstate/backend/visionstate/sources.py +++ b/visionstate/backend/visionstate/sources.py @@ -6,6 +6,9 @@ import io import logging import re +import threading +import time +from concurrent.futures import ThreadPoolExecutor import httpx @@ -119,10 +122,10 @@ async def entities(self) -> list[dict]: async def snapshot(self, entity_id: str) -> bytes: if not self.enabled: raise SourceError("Home Assistant API is not configured") - resp = await self._client.get(f"{self.base}/camera_proxy/{entity_id}") - if resp.status_code != 200: - raise SourceError(f"{entity_id}: HTTP {resp.status_code}") - return resp.content + async with self._client.stream("GET", f"{self.base}/camera_proxy/{entity_id}") as resp: + if resp.status_code != 200: + raise SourceError(f"{entity_id}: HTTP {resp.status_code}") + return await read_limited(resp) async def state(self, entity_id: str) -> str | None: """The current state of an entity, or None when it does not exist.""" @@ -199,9 +202,39 @@ class FrameGrabber: def __init__(self, ha: HomeAssistant): self.ha = ha self._http = httpx.AsyncClient(timeout=RUNTIME["http_timeout_s"], follow_redirects=True) + # An RTSP grab blocks a thread for seconds (connecting, waiting for a key frame): on threads + # of its own, so cameras that do not answer never hold up the database or the AI models. + self._rtsp = ThreadPoolExecutor(max_workers=RUNTIME["rtsp_workers"], thread_name_prefix="rtsp") + self._streams: dict[str, RtspStream] = {} # streams kept open while frames come often + self._last_grab: dict[str, float] = {} # when each stream address was last grabbed async def close(self) -> None: await self._http.aclose() + for stream in self._streams.values(): + stream.close() + self._rtsp.shutdown(wait=False, cancel_futures=True) + + def _grab_stream(self, url: str) -> bytes: + """A frame of an RTSP stream (on an RTSP thread). + + A check now and then connects for one frame (connecting and waiting for a key frame take a + second or more). Frames asked for again within RUNTIME["rtsp_keep_open_s"] (a burst of + checks, someone watching the live view) come from the stream kept open meanwhile; it is + closed once nobody asked for that long, as decoding it costs CPU all the while. + """ + keep_open = RUNTIME["rtsp_keep_open_s"] + now = time.monotonic() + last, self._last_grab[url] = self._last_grab.get(url), now + stream = self._streams.get(url) + if stream is not None and not stream.alive: + stream = self._streams.pop(url) + stream.close() + stream = None + if stream is None and (keep_open <= 0 or last is None or now - last > keep_open): + return grab_rtsp(url) + if stream is None: + stream = self._streams[url] = RtspStream(url, keep_open) + return stream.grab() async def grab(self, source_type: str, source: str) -> bytes: check_source(source_type, source) @@ -210,22 +243,28 @@ async def grab(self, source_type: str, source: str) -> bytes: return await self.ha.snapshot(source) if source_type == "http": return await self._get_limited(source) - return await asyncio.to_thread(grab_rtsp, source) + return await asyncio.get_running_loop().run_in_executor(self._rtsp, self._grab_stream, source) except httpx.HTTPError as err: raise SourceError(describe(err)) from err async def _get_limited(self, url: str) -> bytes: """The body of a snapshot URL, refused when it is larger than RUNTIME["max_frame_mb"].""" - limit = RUNTIME["max_frame_mb"] * 1_000_000 async with self._http.stream("GET", url) as resp: if resp.status_code != 200: raise SourceError(f"HTTP {resp.status_code}") - body = bytearray() - async for chunk in resp.aiter_bytes(): - body += chunk - if len(body) > limit: - raise SourceError(f"Picture larger than {RUNTIME['max_frame_mb']} MB") - return bytes(body) + return await read_limited(resp) + + +async def read_limited(resp: httpx.Response) -> bytes: + """A picture's body, refused when it is larger than RUNTIME["max_frame_mb"] (a misconfigured + address streaming without end).""" + limit = RUNTIME["max_frame_mb"] * 1_000_000 + body = bytearray() + async for chunk in resp.aiter_bytes(): + body += chunk + if len(body) > limit: + raise SourceError(f"Picture larger than {RUNTIME['max_frame_mb']} MB") + return bytes(body) def describe(err: httpx.HTTPError) -> str: @@ -237,15 +276,87 @@ def describe(err: httpx.HTTPError) -> str: return type(err).__name__ +STREAM_OPTIONS = {"rtsp_transport": "tcp", "timeout": "10000000"} # FFmpeg's socket timeout, in µs + + +def _jpeg(frame) -> bytes: # noqa: ANN001 - an av.VideoFrame + buf = io.BytesIO() + frame.to_image().save(buf, format="JPEG", quality=90) + return buf.getvalue() + + def grab_rtsp(url: str) -> bytes: + """One frame of a stream: connect, decode the first frame, disconnect.""" import av try: - with av.open(url, options={"rtsp_transport": "tcp", "stimeout": "10000000"}, timeout=10) as container: + # "timeout" was "stimeout" before FFmpeg 5 (an unknown option is ignored without a word). + with av.open(url, options=STREAM_OPTIONS, timeout=10) as container: for frame in container.decode(video=0): - buf = io.BytesIO() - frame.to_image().save(buf, format="JPEG", quality=90) - return buf.getvalue() + return _jpeg(frame) except av.FFmpegError as err: raise SourceError(redact(str(err))) from err raise SourceError("No video frame received") + + +class RtspStream: + """An RTSP stream kept open: a thread of its own decodes it and keeps the newest frame, until + no frame was asked for in ``keep_open`` seconds or the stream fails (see FrameGrabber._grab_stream).""" + + def __init__(self, url: str, keep_open: float): + self.url, self.keep_open = url, keep_open + self._changed = threading.Condition() + self._frame = None + self._frame_at = 0.0 + self._error: SourceError | None = None + self._asked_at = time.monotonic() + self._closed = False + self._thread = threading.Thread(target=self._run, name="rtsp-stream", daemon=True) + self._thread.start() + + @property + def alive(self) -> bool: + return self._thread.is_alive() + + def close(self) -> None: + with self._changed: + self._closed = True + self._changed.notify_all() + + def grab(self) -> bytes: + """A frame decoded after this call (never an old one), or the stream's error.""" + asked = time.monotonic() + with self._changed: + self._asked_at = asked + ready = self._changed.wait_for( + lambda: self._frame_at > asked or self._error is not None or not self.alive, + timeout=RUNTIME["http_timeout_s"], + ) + if self._frame_at > asked: + frame = self._frame + elif self._error is not None: + raise self._error + else: + raise SourceError("No video frame received" if ready else "The stream stopped sending frames") + return _jpeg(frame) + + def _run(self) -> None: + import av + + try: + with av.open(self.url, options=STREAM_OPTIONS, timeout=10) as container: + for frame in container.decode(video=0): + with self._changed: + self._frame, self._frame_at = frame, time.monotonic() + self._changed.notify_all() + if self._closed or time.monotonic() - self._asked_at > self.keep_open: + return + except av.FFmpegError as err: + with self._changed: + self._error = SourceError(redact(str(err))) + except Exception as err: # noqa: BLE001 - the thread must end with a reason for the waiting grab + with self._changed: + self._error = SourceError(redact(str(err)) or type(err).__name__) + finally: + with self._changed: + self._changed.notify_all() diff --git a/visionstate/backend/visionstate/storage.py b/visionstate/backend/visionstate/storage.py index d0da5bf..7e67145 100644 --- a/visionstate/backend/visionstate/storage.py +++ b/visionstate/backend/visionstate/storage.py @@ -47,8 +47,16 @@ def new_filename() -> str: return f"{uuid.uuid4().hex}.jpg" def _write(self, path: Path, image: Image.Image) -> None: + """Whole or not at all: written next to it, then put in its place (a power cut or a full + disk never leaves half a picture under the name a row lists).""" path.parent.mkdir(parents=True, exist_ok=True) - path.write_bytes(imaging.encode_jpeg(image)) + part = path.with_name(f"{path.name}.part") + try: + part.write_bytes(imaging.encode_jpeg(image)) + part.replace(path) + except BaseException: + part.unlink(missing_ok=True) + raise def save_sample(self, sensor_id: int, image: Image.Image) -> str: filename = self.new_filename() From 5b6937870d6dfc40c02140abbd920994de08edb9 Mon Sep 17 00:00:00 2001 From: oleost <5670710+oleost@users.noreply.github.com> Date: Mon, 26 Oct 2026 11:32:14 +0100 Subject: [PATCH 3/9] Decode small where that is enough; keep the built UI in the browser - Change detection decodes a JPEG only as large as its signature needs (1/2, 1/4 or 1/8 of its size while decoding): the same signature, four to six times faster. - History and training thumbnails are decoded the same way (a 4K frame three times faster). - The built UI's hashed assets are sent with a year's Cache-Control (index.html stays no-cache), so opening the panel does not ask for them again. - Docs for today's changes; a test that read a sensor's status once now waits for it. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 9 ++++- docs/ARCHITECTURE.md | 10 +++--- docs/SCOPE.md | 5 ++- visionstate/DOCS.md | 7 ++-- visionstate/backend/tests/test_quality.py | 4 ++- visionstate/backend/tests/test_security.py | 9 +++++ visionstate/backend/tests/test_storage.py | 30 +++++++++++++++++ .../backend/visionstate/engine/checks.py | 11 ++++--- visionstate/backend/visionstate/imaging.py | 33 ++++++++++++++++--- visionstate/backend/visionstate/main.py | 16 ++++++++- visionstate/backend/visionstate/settings.py | 3 ++ visionstate/backend/visionstate/storage.py | 7 ++-- 12 files changed, 122 insertions(+), 22 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 504e9a6..a1aed8c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -63,7 +63,11 @@ Every fact has **one** home; other files link to it instead of repeating it. - Sensor kinds (`settings.SENSOR_KINDS`): `single_state`, `objects`, `reading`. A new feature says which kind(s) it applies to. - New engine code goes into the part (mixin) it belongs to. A long-running loop must survive any - exception (log it once, try again) — only `CancelledError` may end it. + exception (log it once, try again) — only `CancelledError` may end it. Database work in an + `async def` goes through `asyncio.to_thread` (`tests/test_event_loop.py` fails otherwise). +- What is sent to Home Assistant must not change at every check unless the result does (no + timestamps, no needless decimals, pictures through `_send_on_change`): Home Assistant stores + every change as a new row (`docs/SCOPE.md` §8). - Model registries (`backbones.json`, `detectors.json`, `readers.json`, `faces.json`): entries are never changed or removed once released (their texts, measured footprint and similarity limits may be); a new model gets a new id, pinned by Hugging Face revision and SHA-256, and is checked against real inputs in the tests (one public ONNX conversion turned out to be @@ -184,3 +188,6 @@ Many users run Home Assistant on phones and tablets. like `measure` in `DetectionBoxes.svelte` (not the next animation frame: none comes in a hidden tab). - The Chrome tab Claude drives counts as hidden (`document.visibilityState`); views that only poll while visible look fine there. Use `setVisibility` (e2e) or override it by hand. +- A backend test that swaps in a fake camera after `TestClient(create_app(...))` has started may + see the first check use the real one (and the sensor go *unavailable*): wait for the state you + expect with `wait_for`, never read it once. diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index c5b46b3..a6fdb26 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -50,13 +50,13 @@ API as `request.app.state.runtime`. | File | What it does | |---|---| -| `__main__.py`, `main.py` | Entry point (logging, uvicorn), FastAPI app, Ingress-only guard, static UI | +| `__main__.py`, `main.py` | Entry point (logging, uvicorn), FastAPI app, Ingress-only guard, static UI (hashed assets kept by the browser: `HashedAssets`) | | `settings.py` | **Every** default, limit and tunable, the app options (`Settings`), the `merge_*` helpers that fill a stored setting with defaults | | `db.py` | SQLAlchemy tables and the schema migrations (`MIGRATIONS`, `SCHEMA_VERSION`) | -| `storage.py` | Image files on disk: training samples, history frames, thumbnails | +| `storage.py` | Image files on disk: training samples, history frames, thumbnails; written whole or not at all | | `engine/` | The runtime — see below | | `api/` | The REST API (`/api/v1/…`), one file per topic: `sensors.py` (CRUD, quality, export), `history.py` (the history: one filter for the list, its count and facets; history frames), `samples.py` (training images), `teach.py` (taught boxes; what is taught from a history frame is written into its row: `remember_taught`), `faces.py` (teaching a face on a sensor; the people and their faces, shared by every sensor), `cameras.py` (camera and entity lists, the light, previews), `review.py` (review queue and answers), `imports.py` (import), `backup.py` (backup of everything, restore), `system.py` (UI config, status, `alive` for Home Assistant's watchdog, AI models and deleting downloaded ones, storage, review rules, review reminder, teaching and face limits), `common.py` (input models and views shared by them) | -| `sources.py` | Camera sources: Home Assistant camera, HTTP snapshot, RTSP; address checks and size limits; the Home Assistant REST client (states, services, notify services, the app's page) | +| `sources.py` | Camera sources: Home Assistant camera, HTTP snapshot, RTSP (on threads of its own; a stream grabbed often is kept open: `RtspStream`); address checks and size limits; the Home Assistant REST client (states, services, notify services, the app's page) | | `mqtt.py` | MQTT topics, Home Assistant discovery messages, the bridge (connect, publish, commands) | | `ha_events.py` | Home Assistant WebSocket: state changes of trigger entities, entity registry | | `lights.py` | A sensor's light: switched on for checks and while someone looks, shared by both | @@ -65,7 +65,7 @@ API as `request.app.state.runtime`. | `faces.py` | Faces in person boxes (registry `faces.json`, never bundled): finding them (YuNet), the quality gate, aligning, evening out the light (`even_light`), embedding, telling who it is | | `classifier.py` | The small per-sensor classifier ("head") on top of the backbone | | `teach.py` | Object sensors: comparing detected boxes with the boxes the user taught | -| `imaging.py` | Decoding, regions (ROI), crops, change signatures, drawing boxes | +| `imaging.py` | Decoding (also decoded small where that is enough: change signatures, thumbnails), regions (ROI), crops, change signatures, drawing boxes | | `uploads.py`, `bundle.py` | Uploaded files (images, ZIP, video) → frames; sensor export/import bundles | | `backup.py` | A backup of everything (database copy + files, by the parts of `settings.BACKUP_PARTS`) and restoring one: staged, swapped in, undone on failure | | `redact.py` | Masks credentials in URLs, messages and log lines | @@ -96,7 +96,7 @@ Runtime ── ChecksMixin ──┬─ ObjectChecksMixin ── FacesMixin, Tea | `teaching.py` | Taught boxes of object sensors (index, embeddings) | | `faces.py` | Faces on object sensors: the face model (downloaded and loaded in the background, let go when unused), the index of the people's taught faces (shared by every sensor), the face step of a check (`apply_faces`), each person's own entities (where they are seen, when they were last seen) | | `training.py` | Training the state sensors' heads; embeddings of samples | -| `publishing.py` | Discovery and values to Home Assistant, the review queue entity, commands from it | +| `publishing.py` | Discovery and values to Home Assistant (pictures and object attributes only when they change, or now and then: `_send_on_change`), the review queue entity, commands from it | | `history.py` | History clean-up by age and size, disk use | | `reminders.py` | The review reminder: a notification in Home Assistant (and a push) when frames waited long | diff --git a/docs/SCOPE.md b/docs/SCOPE.md index 3a8503f..d89a710 100644 --- a/docs/SCOPE.md +++ b/docs/SCOPE.md @@ -463,7 +463,10 @@ nothing decodes in between. and `unavailable`/`unknown` ignored). `only_states` limits an entity to one new state (compared without regard to case). - Optional region change detection: a small greyscale copy of the ROI is compared every - *N* seconds; the AI only runs when the difference exceeds a threshold. + *N* seconds; the AI only runs when the difference exceeds a threshold. The frame is decoded + only as large as that copy needs (`imaging.decode_for_signature`: a JPEG decodes at 1/2, 1/4 or + 1/8 of its size), four to six times faster than whole, with the same result; history + thumbnails are decoded the same way. - Every trigger starts a *burst* (default every 2 s for 30 s) to catch the final state. - A sensor that recognises faces starts one by itself when a check finds a person nobody is named for (`engine.logic.steer_face_burst`, tunables `FACES["burst_*"]`): someone walking up diff --git a/visionstate/DOCS.md b/visionstate/DOCS.md index e19fafd..1efd1f6 100644 --- a/visionstate/DOCS.md +++ b/visionstate/DOCS.md @@ -15,7 +15,9 @@ number** from a display, without any training. Everything runs on this machine; VisionState finds the broker automatically. - **A camera** in Home Assistant (any `camera.*` entity: Frigate, ESP32-CAM, Reolink, generic camera…), or a direct HTTP snapshot URL (`http://` or `https://`) or stream URL - (`rtsp://`, `rtsps://`, also an `http://` video stream). + (`rtsp://`, `rtsps://`, also an `http://` video stream). A stream is kept open while frames + are asked for often (a burst of checks, the live view), so those come at once instead of after + a new connection each time. - amd64 (Intel/AMD) or aarch64 (Raspberry Pi 4/5, 64-bit OS). Any 64-bit Intel/AMD CPU works, also in a virtual machine with a generic CPU type (for example Proxmox's `kvm64`). @@ -572,7 +574,8 @@ a known point: reached first; the oldest frames are removed first, frames waiting for review last (they are kept twice as many days). Set either limit there (`0` GB = no size limit). Training images, and readings you verified on a reading sensor's Quality tab or in the review queue, are never - removed automatically. + removed automatically. History frames that no history entry lists any more (the app was + stopped while it saved one) are removed after an hour. - **Export** (on a sensor) downloads a ZIP with the sensor's settings (region, states, objects or reading settings, triggers, review overrides) and all its images with labels — for an object sensor, the boxes it was taught and its own labels. Camera passwords are removed from the file. diff --git a/visionstate/backend/tests/test_quality.py b/visionstate/backend/tests/test_quality.py index 9c139b4..58e8ea0 100644 --- a/visionstate/backend/tests/test_quality.py +++ b/visionstate/backend/tests/test_quality.py @@ -155,7 +155,9 @@ def test_outdated_heads_are_retrained_on_startup(tmp_path, model_dir, outdated): with TestClient(create_app(settings)) as client: client.app.state.runtime.grabber = camera assert wait_for(lambda: client.get(f"/api/v1/sensors/{sid}").json()["model"]["version"] > version) - assert client.get(f"/api/v1/sensors/{sid}").json()["status"] == "ok" + # The first check may run before the fake camera is in place (source "x": unavailable); the + # check after the retrain uses it. + assert wait_for(lambda: client.get(f"/api/v1/sensors/{sid}").json()["status"] == "ok") head = classifier.load(path) assert head is not None and head.format == classifier.HEAD_FORMAT assert legacy.read_bytes() == classifier.LEGACY_MARKER diff --git a/visionstate/backend/tests/test_security.py b/visionstate/backend/tests/test_security.py index 7cd1a30..807fe2c 100644 --- a/visionstate/backend/tests/test_security.py +++ b/visionstate/backend/tests/test_security.py @@ -101,6 +101,15 @@ def test_static_files_cannot_escape_frontend_dir(client): assert "top secret" not in resp.text +def test_the_built_ui_is_kept_by_the_browser_but_not_its_index(client, tmp_path): + """Asset names carry a hash of their content: kept for a year; index.html names the current ones.""" + (tmp_path / "frontend" / "assets" / "index-AbCd1234.js").write_text("app()") + asset = client.get("/assets/index-AbCd1234.js") + assert asset.status_code == 200 and "immutable" in asset.headers["cache-control"] + assert client.get("/").headers["cache-control"] == "no-cache" + assert "cache-control" not in client.get("/assets/missing.js").headers + + def test_review_rules_global_and_per_sensor(client): assert client.get("/api/v1/review-rules").json()["spot_rate"] == 0.0 rules = client.put("/api/v1/review-rules", json={"below": 0.8, "cooldown_s": 600}).json() diff --git a/visionstate/backend/tests/test_storage.py b/visionstate/backend/tests/test_storage.py index a743c5c..0b0af25 100644 --- a/visionstate/backend/tests/test_storage.py +++ b/visionstate/backend/tests/test_storage.py @@ -447,3 +447,33 @@ def test_an_existing_database_gets_the_index_at_start(settings): names = {row[0] for row in conn.execute(text("SELECT name FROM sqlite_master WHERE type = 'index'"))} db.engine.dispose() assert "ix_prediction_waiting" in names + + +def test_frames_are_decoded_small_where_that_is_enough(tmp_path): + """Change detection and thumbnails decode a JPEG at 1/2, 1/4 or 1/8 of its size while decoding: + the signature stays the same (to far below any threshold), the thumbnail its size.""" + import io + + from visionstate import imaging + from visionstate.settings import CHANGE_SIGNATURE_DECODE, CHANGE_SIGNATURE_SIZE, THUMB_SIZE + + from .conftest import ASSETS + + photo = Image.open(ASSETS / "beach.jpg").convert("RGB").resize((3840, 2160)) + buf = io.BytesIO() + photo.save(buf, format="JPEG", quality=90) + data = buf.getvalue() + for roi in (None, {"x": 0.6, "y": 0.5, "w": 0.1, "h": 0.2}): + small = imaging.decode_for_signature(data, roi) + assert small.width < 3840 # decoded smaller … + region = imaging.crop(small, roi) + assert min(region.size) >= CHANGE_SIGNATURE_SIZE * CHANGE_SIGNATURE_DECODE # … but large enough + whole = imaging.region_signature(imaging.decode(data), roi) + assert imaging.change_score(whole, imaging.region_signature(small, roi)) < 0.005 + path = tmp_path / "frame.jpg" + path.write_bytes(data) + assert imaging.load(path, at_least=(THUMB_SIZE, THUMB_SIZE)).width < 3840 + assert imaging.thumbnail(imaging.load(path, at_least=(THUMB_SIZE, THUMB_SIZE))).size == (THUMB_SIZE, 180) + png = io.BytesIO() + photo.resize((640, 360)).save(png, format="PNG") + assert imaging.decode(png.getvalue(), at_least=(10, 10)).size == (640, 360) # not a JPEG: whole diff --git a/visionstate/backend/visionstate/engine/checks.py b/visionstate/backend/visionstate/engine/checks.py index d32b492..e4e8fec 100644 --- a/visionstate/backend/visionstate/engine/checks.py +++ b/visionstate/backend/visionstate/engine/checks.py @@ -173,7 +173,8 @@ async def probe(self, cfg: SensorConfig) -> None: return try: frame_id, data = await self.grab(cfg) - image = await asyncio.to_thread(imaging.decode, data) + # Only for the signature: decoded small (a check that follows decodes it whole). + small = await asyncio.to_thread(imaging.decode_for_signature, data, cfg.roi) except (SourceError, OSError) as err: live.available, live.error = False, redact(str(err)) return @@ -184,11 +185,11 @@ async def probe(self, cfg: SensorConfig) -> None: # No baseline (first run or the region changed). With a light this frame is the # baseline (checks in the light never set one); without, check it, which sets one. if lit: - live.signature = await asyncio.to_thread(imaging.region_signature, image, cfg.roi) + live.signature = await asyncio.to_thread(imaging.region_signature, small, cfg.roi) else: - await self.run_once(cfg, data=data, image=image, frame_id=frame_id) + await self.run_once(cfg, data=data, frame_id=frame_id) return - signature = await asyncio.to_thread(imaging.region_signature, image, cfg.roi) + signature = await asyncio.to_thread(imaging.region_signature, small, cfg.roi) score = imaging.change_score(live.signature, signature) live.change_score = score if score >= cfg.triggers["change_threshold"]: @@ -199,7 +200,7 @@ async def probe(self, cfg: SensorConfig) -> None: live.signature = signature await self.run_once(cfg) else: - await self.run_once(cfg, data=data, image=image, frame_id=frame_id) + await self.run_once(cfg, data=data, frame_id=frame_id) async def grab(self, cfg: SensorConfig) -> tuple[str, bytes]: data = await self.grabber.grab(cfg.source_type, cfg.source) diff --git a/visionstate/backend/visionstate/imaging.py b/visionstate/backend/visionstate/imaging.py index fce8e50..b183a45 100644 --- a/visionstate/backend/visionstate/imaging.py +++ b/visionstate/backend/visionstate/imaging.py @@ -4,12 +4,21 @@ import hashlib import io +import math from pathlib import Path import numpy as np from PIL import Image, ImageDraw, ImageOps -from .settings import CHANGE_SIGNATURE_SIZE, JPEG_QUALITY, NEUTRAL_FILL, ROI_MAX_POINTS, RUNTIME, THUMB_SIZE +from .settings import ( + CHANGE_SIGNATURE_DECODE, + CHANGE_SIGNATURE_SIZE, + JPEG_QUALITY, + NEUTRAL_FILL, + ROI_MAX_POINTS, + RUNTIME, + THUMB_SIZE, +) FULL_FRAME_KEY = "full" @@ -18,18 +27,34 @@ class ImageTooLarge(OSError): """An image with more pixels than RUNTIME["max_image_megapixels"]; handled like an unreadable one.""" -def decode(data: bytes) -> Image.Image: +def decode(data: bytes, at_least: tuple[int, int] | None = None) -> Image.Image: + """An image (RGB). ``at_least``: a JPEG may be decoded smaller, down to this size (its decoder + scales by 1/2, 1/4 or 1/8 while decoding, several times faster than decoding it whole).""" image = Image.open(io.BytesIO(data)) # Checked from the header, before the pixels are decoded (they could take gigabytes). if image.width * image.height > RUNTIME["max_image_megapixels"] * 1_000_000: raise ImageTooLarge(f"Image too large ({image.width}×{image.height})") + if at_least is not None: + image.draft("RGB", at_least) # only JPEG does anything with it image = ImageOps.exif_transpose(image) return image.convert("RGB") -def load(path: Path) -> Image.Image: - """A stored image (RGB), with its file closed right away (it may be deleted meanwhile).""" +def decode_for_signature(data: bytes, roi: dict | None) -> Image.Image: + """A frame decoded just large enough for region_signature (change detection): the region keeps + CHANGE_SIGNATURE_DECODE times the signature's size on each side.""" + roi = normalise_roi(roi) + side = CHANGE_SIGNATURE_SIZE * CHANGE_SIGNATURE_DECODE + w, h = (roi["w"], roi["h"]) if roi else (1.0, 1.0) + return decode(data, at_least=(math.ceil(side / w), math.ceil(side / h))) + + +def load(path: Path, at_least: tuple[int, int] | None = None) -> Image.Image: + """A stored image (RGB), with its file closed right away (it may be deleted meanwhile); + ``at_least`` as for decode.""" with Image.open(path) as image: + if at_least is not None: + image.draft("RGB", at_least) return image.convert("RGB") diff --git a/visionstate/backend/visionstate/main.py b/visionstate/backend/visionstate/main.py index 05450fa..3b725be 100644 --- a/visionstate/backend/visionstate/main.py +++ b/visionstate/backend/visionstate/main.py @@ -9,6 +9,8 @@ from fastapi import FastAPI, Request from fastapi.responses import FileResponse, JSONResponse from fastapi.staticfiles import StaticFiles +from starlette.responses import Response +from starlette.types import Scope from . import backup from .api import backup as backup_api @@ -23,6 +25,18 @@ WATCHDOG_PATH = f"{API_PREFIX}/alive" # config.yaml "watchdog" +class HashedAssets(StaticFiles): + """The built UI's scripts, styles and fonts: Vite puts a hash of the content in each name, so + a browser keeps them instead of asking again every time the panel opens (index.html is never + kept: it names the current ones).""" + + async def get_response(self, path: str, scope: Scope) -> Response: + response = await super().get_response(path, scope) + if response.status_code == 200: + response.headers["Cache-Control"] = "public, max-age=31536000, immutable" + return response + + def create_app(settings: Settings | None = None) -> FastAPI: settings = settings or load_settings() settings.ensure_dirs() @@ -83,7 +97,7 @@ async def ingress_only(request: Request, call_next): frontend = settings.frontend_dir.resolve() index = frontend / "index.html" if index.exists(): - app.mount("/assets", StaticFiles(directory=frontend / "assets"), name="assets") + app.mount("/assets", HashedAssets(directory=frontend / "assets"), name="assets") @app.get("/{path:path}", include_in_schema=False) async def spa(path: str): diff --git a/visionstate/backend/visionstate/settings.py b/visionstate/backend/visionstate/settings.py index c305b69..8de2977 100644 --- a/visionstate/backend/visionstate/settings.py +++ b/visionstate/backend/visionstate/settings.py @@ -425,6 +425,9 @@ # New entity states that are ignored (the entity going offline is not a real event). TRIGGER_IGNORED_STATES = {"unavailable", "unknown"} CHANGE_SIGNATURE_SIZE = 48 # edge length of the greyscale thumbnail used for change detection +# A probe decodes a JPEG only this many times larger than the signature (over its region), not +# whole: the same signature (to about 0.001), four to six times faster (imaging.decode_for_signature). +CHANGE_SIGNATURE_DECODE = 4 # --- Region of interest ------------------------------------------------------------ diff --git a/visionstate/backend/visionstate/storage.py b/visionstate/backend/visionstate/storage.py index 7e67145..373f36d 100644 --- a/visionstate/backend/visionstate/storage.py +++ b/visionstate/backend/visionstate/storage.py @@ -8,7 +8,7 @@ from PIL import Image from . import imaging -from .settings import Settings +from .settings import THUMB_SIZE, Settings class Storage: @@ -80,7 +80,10 @@ def thumbnail(self, kind: str, item_id: int, source: Path) -> Path: """Cached thumbnail for a sample or history frame.""" target = self.thumb_path(kind, item_id) if not target.exists() or target.stat().st_mtime < source.stat().st_mtime: - self._write(target, imaging.thumbnail(imaging.load(source))) + # Decoded just large enough (a JPEG decodes at 1/2, 1/4 or 1/8 of its size): three times + # faster for a 4K frame, and a page of history asks for dozens at once. + size = (THUMB_SIZE, THUMB_SIZE) + self._write(target, imaging.thumbnail(imaging.load(source, at_least=size))) return target def delete_sample(self, sensor_id: int, sample_id: int, filename: str) -> None: From ad02e84d996f5ec1414db822db8a791695001c98 Mon Sep 17 00:00:00 2001 From: oleost <5670710+oleost@users.noreply.github.com> Date: Mon, 26 Oct 2026 11:32:48 +0100 Subject: [PATCH 4/9] 0.11.0b1: less in Home Assistant's database, faster checks, quicker RTSP Co-Authored-By: Claude Opus 5.5 --- visionstate/CHANGELOG.md | 30 ++++++++++++++++++++++++++++++ visionstate/config.yaml | 2 +- 2 files changed, 31 insertions(+), 1 deletion(-) diff --git a/visionstate/CHANGELOG.md b/visionstate/CHANGELOG.md index 8c415d6..731c9f3 100644 --- a/visionstate/CHANGELOG.md +++ b/visionstate/CHANGELOG.md @@ -1,5 +1,35 @@ # Changelog +## 0.11.0b1 + +### Better + +- **Much less in Home Assistant's database.** VisionState no longer sends values that change at + every check when the result does not: on a test installation with nine sensors, Home Assistant + stored about 500–750 states an hour instead of about 4,400. What changed for you: + - the confidence is in whole percent, and the `probabilities` of a state sensor have two + decimals; + - the `last_update` attribute is gone (Home Assistant shows when a sensor last reported); + - the *Last frame* pictures, and an object's boxes and *last seen*, are sent at once when the + state, the objects or the value change, and otherwise every few minutes. + If an automation reads `last_update`, or the boxes while an object stays, look at it. +- **Faster checks with less CPU**: the AI models no longer keep processor cores busy while they + wait — about 30 % less CPU per check on the test machine, which leaves more for Home Assistant. +- **RTSP cameras are much quicker during a burst of checks and in the live view**: the stream is + kept open while frames are asked for often (about 30 ms a frame instead of 1–3 s), and closed + again a few seconds after. +- Change detection and the history's thumbnails decode camera pictures only as large as they + need — several times faster for large pictures. +- The app no longer stalls while its database is busy, and the browser keeps the app's files + instead of loading them every time the panel opens. + +### Fixed + +- History pictures are written whole or not at all, and removed only together with their + history entries; pictures left behind (the app was stopped while saving one) are removed. +- A camera from Home Assistant has the same size limit for its pictures as a snapshot URL. +- An RTSP stream's connection timeout works again (FFmpeg had renamed the option). + ## 0.10.1 ### New diff --git a/visionstate/config.yaml b/visionstate/config.yaml index f924542..8c99080 100644 --- a/visionstate/config.yaml +++ b/visionstate/config.yaml @@ -1,5 +1,5 @@ name: VisionState (beta) -version: "0.10.1b4" +version: "0.11.0b1" slug: visionstate description: Teach a local AI to recognise states (open, closed, …), find objects (people, cars, pets) or read numbers in your camera images url: https://github.com/oleost/VisionState From 721224a5fb5c67e4a95b7537c01f0dfd6ec15147 Mon Sep 17 00:00:00 2001 From: oleost <5670710+oleost@users.noreply.github.com> Date: Mon, 26 Oct 2026 12:36:29 +0100 Subject: [PATCH 5/9] History keeps the camera's own JPEG; no copies when decoding; nothing sent twice; pages rest while hidden - A frame decoded whole and upright is saved to the history as the camera sent it (imaging.original_jpeg): no encoding again (17 ms for a 4K frame on a desktop), nothing lost, usually smaller. Training images are still encoded again (what an uploaded photo carries besides its pixels stays out of exports). - decode no longer copies a frame twice (exif_transpose and convert): about 20 % faster and one frame less in memory at every check. - A sensor's MQTT values are sent only when they differ from the last, and again every PUBLISH["refresh_s"] and after MQTT reconnects. - The status, the dashboard and a sensor's page poll only while the page is visible (lib/poll.ts), and update at once when it is shown again. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 5 +- docs/ARCHITECTURE.md | 1 + docs/SCOPE.md | 9 +++- visionstate/backend/tests/test_publish.py | 28 +++++++++-- visionstate/backend/tests/test_storage.py | 48 +++++++++++++++++++ .../backend/visionstate/engine/publishing.py | 16 +++++-- visionstate/backend/visionstate/imaging.py | 34 +++++++++++-- visionstate/backend/visionstate/storage.py | 13 +++-- visionstate/frontend/e2e/ui.spec.ts | 11 +++-- visionstate/frontend/src/App.svelte | 4 +- visionstate/frontend/src/lib/poll.ts | 22 +++++++++ .../frontend/src/pages/Dashboard.svelte | 5 +- .../frontend/src/pages/SensorPage.svelte | 7 +-- 13 files changed, 172 insertions(+), 31 deletions(-) create mode 100644 visionstate/frontend/src/lib/poll.ts diff --git a/CLAUDE.md b/CLAUDE.md index a1aed8c..2b359ca 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -186,8 +186,9 @@ Many users run Home Assistant on phones and tablets. them resizes what is observed, WebKit reports "ResizeObserver loop completed with undelivered notifications" (seen once in CI on the dashboard, never locally). Measure in a timeout instead, like `measure` in `DetectionBoxes.svelte` (not the next animation frame: none comes in a hidden tab). -- The Chrome tab Claude drives counts as hidden (`document.visibilityState`); views that only poll - while visible look fine there. Use `setVisibility` (e2e) or override it by hand. +- The Chrome tab Claude drives counts as hidden (`document.visibilityState`), and every page polls + only while visible (`lib/poll.ts`, live frames, history): there nothing updates by itself. Use + `setVisibility` (e2e) or override it by hand. - A backend test that swaps in a fake camera after `TestClient(create_app(...))` has started may see the first check use the real one (and the sensor go *unavailable*): wait for the state you expect with `wait_for`, never read it once. diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index a6fdb26..cfdb571 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -154,6 +154,7 @@ error is kept in `LiveState.error` and shown in the UI. | `lib/app.svelte.ts` | Shared state: config from `GET /api/v1/config`, status polling, the last restored backup (Settings is made again with it) | | `lib/router.svelte.ts` | Hash routes (`paths`), a page's query (`route.query`, `setQuery`) | | `lib/history.ts` | The history filter: from and to the page URL, and as the API takes it | +| `lib/poll.ts` | Polling that rests while the page is hidden (`pollWhileVisible`: status, dashboard, sensor page) | | `lib/ui.ts` | UI constants (tabs per sensor kind, wizard texts); `tokens.css` colours, type, spacing | | `lib/objects.ts`, `lib/reading.ts`, `lib/roi.ts`, `lib/format.ts` | Helpers: names and colours of objects and own labels, reading values and units, region geometry, formatting | | `pages/` | Dashboard, the new sensor wizard (`NewSensor.svelte`; its Detect step per kind in `pages/wizard/`), Review, History, People, Settings, and `SensorPage` with one file per tab in `pages/sensor/` (its History tab is `HistoryView`) | diff --git a/docs/SCOPE.md b/docs/SCOPE.md index d89a710..55995de 100644 --- a/docs/SCOPE.md +++ b/docs/SCOPE.md @@ -577,6 +577,10 @@ keeps `last_reported`), the confidence and the probabilities in whole percent (`engine.logic.percent`), and the pictures and an object's attributes (boxes, last seen) through `engine.publishing.PublishingMixin._send_on_change`: at once when what they are about changes (the published state, the objects and their counts, the value), otherwise every `settings.PUBLISH["refresh_s"]`. +Every other value is sent only when it differs from what that topic got last +(`engine.publishing.PublishingMixin._send`), and again after the same time: Home Assistant would +only throw it away. What was sent is forgotten when discovery is published (MQTT connected again, +a sensor changed), so nothing waits for a change after the broker or Home Assistant restarted. Availability: app-wide LWT plus per-sensor camera availability. Removing a sensor removes its entities. **Send to Home Assistant** (`sensor.publish`, default @@ -702,7 +706,10 @@ the old copies stay for `SAMPLES_MOVE["keep_old_days"]` after the first copy, so backup made before the update still gives the older version its images, then they are removed. Files no sensor lists are left in the media folder. -History frames are full camera frames (JPEG 90). They are removed by age (`history_days`; +History frames are full camera frames: the camera's own JPEG when the frame is one and needs no +turning (`imaging.original_jpeg`; not encoded again, so no time spent and nothing lost, and usually +smaller), otherwise JPEG 90. Training images are always encoded again: that leaves out what an +uploaded photo carries besides its pixels (where it was taken), as they go into exports. They are removed by age (`history_days`; frames waiting for review get twice as long) and by total size (`history_max_gb`, 0 = no limit): oldest first, frames waiting for review only when that is not enough. Both limits are a DB setting edited in Settings → Storage (`/api/v1/storage`, which also reports disk use); diff --git a/visionstate/backend/tests/test_publish.py b/visionstate/backend/tests/test_publish.py index d012c7d..ace806c 100644 --- a/visionstate/backend/tests/test_publish.py +++ b/visionstate/backend/tests/test_publish.py @@ -160,14 +160,16 @@ def send(cfg, shows): # noqa: ANN001 @requires_reader def test_values_that_change_at_every_check_are_not_sent(tmp_path): - """No timestamp in the attributes, the confidence in whole percent, the pictures once while the - value stays: nothing makes Home Assistant store the entities again at every check.""" + """No timestamp in the attributes, the confidence in whole percent, and nothing sent again + while it stays the same (the pictures included): nothing makes Home Assistant store the + entities again at every check. A new value is sent at once.""" settings = Settings( data_dir=tmp_path / "data", media_dir=tmp_path / "media", frontend_dir=tmp_path, bundled_models_dir=MODEL_DIR ) with TestClient(create_app(settings)) as client: rt = client.app.state.runtime - rt.grabber = DisplayCamera("0012345.6") + camera = CountingCamera("0012345.6") + rt.grabber = camera mqtt = CountingMqtt(rt.mqtt) rt.mqtt = mqtt body = { @@ -181,9 +183,25 @@ def test_values_that_change_at_every_check_are_not_sent(tmp_path): } client.post("/api/v1/sensors", json=body) base = "visionstate/meter" - assert wait_for(lambda: mqtt.count.get(f"{base}/confidence", 0) >= 3, timeout=60) + assert wait_for(lambda: mqtt.sent.get(f"{base}/state") == "12345.6", timeout=60) + grabs = camera.grabs + assert wait_for(lambda: camera.grabs >= grabs + 3, timeout=60) # three more checks of the same attributes = mqtt.sent[f"{base}/attributes"] attributes = json.loads(attributes) if isinstance(attributes, str) else attributes assert "last_update" not in attributes assert mqtt.sent[f"{base}/confidence"].isdigit() # whole percent - assert mqtt.count[f"{base}/image"] == 1 and mqtt.count[f"{base}/reader_image"] == 1 + for entity in ("state", "confidence", "attributes", "availability", "image", "reader_image"): + assert mqtt.count[f"{base}/{entity}"] == 1, (entity, mqtt.count) + camera.text = "0012346.6" + assert wait_for(lambda: mqtt.sent.get(f"{base}/state") == "12346.6", timeout=60) + assert wait_for(lambda: mqtt.count[f"{base}/image"] == 2, timeout=10) + + +class CountingCamera(DisplayCamera): + def __init__(self, text: str): + super().__init__(text) + self.grabs = 0 + + async def grab(self, source_type, source): + self.grabs += 1 + return await super().grab(source_type, source) diff --git a/visionstate/backend/tests/test_storage.py b/visionstate/backend/tests/test_storage.py index 0b0af25..cb1a5ca 100644 --- a/visionstate/backend/tests/test_storage.py +++ b/visionstate/backend/tests/test_storage.py @@ -477,3 +477,51 @@ def test_frames_are_decoded_small_where_that_is_enough(tmp_path): png = io.BytesIO() photo.resize((640, 360)).save(png, format="PNG") assert imaging.decode(png.getvalue(), at_least=(10, 10)).size == (640, 360) # not a JPEG: whole + + +def test_the_history_keeps_the_cameras_own_jpeg(runtime): + """A frame decoded whole and upright is saved as the camera sent it (not encoded again); + anything else — a crop, a turned picture, a PNG, a frame decoded small — is encoded. A training + image is always encoded again (what an uploaded photo carries besides its pixels stays out).""" + import io + + from visionstate import imaging + + from .conftest import ASSETS + + rt = runtime + sid = state_sensor(rt) + photo = Image.open(ASSETS / "beach.jpg").convert("RGB") + buf = io.BytesIO() + photo.save(buf, format="JPEG", quality=75) + data = buf.getvalue() + + frame = imaging.decode(data) + assert frame.mode == "RGB" and imaging.original_jpeg(frame) == data + assert rt.storage.history_path(sid, rt.storage.save_history(sid, frame)).read_bytes() == data + crop = imaging.crop(frame, {"x": 0.1, "y": 0.1, "w": 0.5, "h": 0.5}) + assert imaging.original_jpeg(crop) is None and imaging.original_jpeg(frame.copy()) is None + assert rt.storage.history_path(sid, rt.storage.save_history(sid, crop)).read_bytes() != data + assert imaging.original_jpeg(imaging.decode(data, at_least=(64, 64))) is None + sample = rt.add_sample(sid, frame, "upload", None) + with rt.db.session() as s: + from visionstate.db import Sample + + assert rt.storage.sample_path(sid, s.get(Sample, sample).filename).read_bytes() != data + + exif = Image.Exif() + exif[imaging.EXIF_ORIENTATION] = 6 # turned: the camera's file is not the picture as shown + turned = io.BytesIO() + photo.save(turned, format="JPEG", quality=75, exif=exif) + upright = imaging.decode(turned.getvalue()) + assert upright.size == (photo.height, photo.width) and imaging.original_jpeg(upright) is None + png = io.BytesIO() + photo.save(png, format="PNG") + assert imaging.original_jpeg(imaging.decode(png.getvalue())) is None + + key = id(frame) + del frame, crop + import gc + + gc.collect() + assert key not in imaging._ORIGINALS # forgotten with its image diff --git a/visionstate/backend/visionstate/engine/publishing.py b/visionstate/backend/visionstate/engine/publishing.py index 32ce119..3eb554b 100644 --- a/visionstate/backend/visionstate/engine/publishing.py +++ b/visionstate/backend/visionstate/engine/publishing.py @@ -21,9 +21,19 @@ class PublishingMixin(RuntimeBase): async def _send(self, cfg: SensorConfig, topic: str, payload: str | bytes | dict, retain: bool = False) -> None: - """Publish one of a sensor's values — unless it is not to be sent to Home Assistant.""" - if cfg.publish: - await self.mqtt.publish(topic, payload, retain=retain) + """Publish one of a sensor's values — unless it is not to be sent to Home Assistant, or it + is what was sent last (sent again every PUBLISH["refresh_s"] all the same, and after MQTT + (re)connects: publish_discovery forgets what was sent). Home Assistant would only throw an + unchanged value away.""" + if not cfg.publish: + return + live = self.live_state(cfg.id) + now = time.time() + sent = live.sent_on_change.get(topic) + if sent is not None and sent[0] == payload and now - sent[1] < PUBLISH["refresh_s"]: + return + await self.mqtt.publish(topic, payload, retain=retain) + live.sent_on_change[topic] = (payload, now) async def _send_on_change( self, cfg: SensorConfig, topic: str, shows: object, make: Callable[[], bytes | dict] diff --git a/visionstate/backend/visionstate/imaging.py b/visionstate/backend/visionstate/imaging.py index b183a45..5be7595 100644 --- a/visionstate/backend/visionstate/imaging.py +++ b/visionstate/backend/visionstate/imaging.py @@ -5,6 +5,7 @@ import hashlib import io import math +import weakref from pathlib import Path import numpy as np @@ -27,17 +28,44 @@ class ImageTooLarge(OSError): """An image with more pixels than RUNTIME["max_image_megapixels"]; handled like an unreadable one.""" +# The camera's own JPEG of a frame decode() made, by id() of that image while it lives (see +# original_jpeg); an entry goes with its image (weakref.finalize), so an id is never reused here. +_ORIGINALS: dict[int, bytes] = {} + +EXIF_ORIENTATION = 0x0112 + + def decode(data: bytes, at_least: tuple[int, int] | None = None) -> Image.Image: """An image (RGB). ``at_least``: a JPEG may be decoded smaller, down to this size (its decoder - scales by 1/2, 1/4 or 1/8 while decoding, several times faster than decoding it whole).""" + scales by 1/2, 1/4 or 1/8 while decoding, several times faster than decoding it whole). + + Without copies: a frame already upright and RGB (most camera pictures) is the decoded image + itself — exif_transpose and convert would each copy it whole. A JPEG decoded whole and + upright is remembered as it came (original_jpeg), so the history keeps the camera's own file. + """ image = Image.open(io.BytesIO(data)) # Checked from the header, before the pixels are decoded (they could take gigabytes). if image.width * image.height > RUNTIME["max_image_megapixels"] * 1_000_000: raise ImageTooLarge(f"Image too large ({image.width}×{image.height})") if at_least is not None: image.draft("RGB", at_least) # only JPEG does anything with it - image = ImageOps.exif_transpose(image) - return image.convert("RGB") + upright = image.getexif().get(EXIF_ORIENTATION, 1) == 1 + jpeg = image.format == "JPEG" + if not upright: + ImageOps.exif_transpose(image, in_place=True) + image.load() + if image.mode != "RGB": + image = image.convert("RGB") + if jpeg and upright and at_least is None: + _ORIGINALS[id(image)] = data + weakref.finalize(image, _ORIGINALS.pop, id(image), None) + return image + + +def original_jpeg(image: Image.Image) -> bytes | None: + """The camera's JPEG ``image`` was decoded from, when it is exactly that picture (not a crop + or another copy of it: those are other images).""" + return _ORIGINALS.get(id(image)) def decode_for_signature(data: bytes, roi: dict | None) -> Image.Image: diff --git a/visionstate/backend/visionstate/storage.py b/visionstate/backend/visionstate/storage.py index 373f36d..99875ba 100644 --- a/visionstate/backend/visionstate/storage.py +++ b/visionstate/backend/visionstate/storage.py @@ -46,13 +46,14 @@ def thumb_path(self, kind: str, item_id: int) -> Path: def new_filename() -> str: return f"{uuid.uuid4().hex}.jpg" - def _write(self, path: Path, image: Image.Image) -> None: + def _write(self, path: Path, image: Image.Image, data: bytes | None = None) -> None: """Whole or not at all: written next to it, then put in its place (a power cut or a full - disk never leaves half a picture under the name a row lists).""" + disk never leaves half a picture under the name a row lists). ``data``: the JPEG to write + as it is (otherwise ``image`` is encoded).""" path.parent.mkdir(parents=True, exist_ok=True) part = path.with_name(f"{path.name}.part") try: - part.write_bytes(imaging.encode_jpeg(image)) + part.write_bytes(data if data is not None else imaging.encode_jpeg(image)) part.replace(path) except BaseException: part.unlink(missing_ok=True) @@ -72,8 +73,12 @@ def delete_face(self, filename: str) -> None: self.face_path(filename).unlink(missing_ok=True) def save_history(self, sensor_id: int, image: Image.Image) -> str: + """A camera frame for the history: the camera's own JPEG when ``image`` is exactly that + (imaging.original_jpeg) — not encoded again, so no time spent and no quality lost, and + usually smaller. Only the history: a training image is always encoded again, which leaves + out what an uploaded photo may carry besides its pixels (where it was taken).""" filename = self.new_filename() - self._write(self.history_path(sensor_id, filename), image) + self._write(self.history_path(sensor_id, filename), image, imaging.original_jpeg(image)) return filename def thumbnail(self, kind: str, item_id: int, source: Path) -> Path: diff --git a/visionstate/frontend/e2e/ui.spec.ts b/visionstate/frontend/e2e/ui.spec.ts index cd12571..dd55512 100644 --- a/visionstate/frontend/e2e/ui.spec.ts +++ b/visionstate/frontend/e2e/ui.spec.ts @@ -111,7 +111,7 @@ test('no page asks the server more often than it polls', async ({ page, request } }); -test('hiding and showing a page with live frames does not start extra polling', async ({ page, request }, info) => { +test('a hidden page asks nothing, and hiding and showing it does not start extra polling', async ({ page, request }, info) => { const id = await seededSensorId(request); // Slow answers, so the page is hidden and shown while a request is on its way. await page.route('**/api/v1/sensors/*/frame*', async (route) => { @@ -130,12 +130,17 @@ test('hiding and showing a page with live frames does not start extra polling', } expectCalm(await apiRequestsDuring(page, POLL_WINDOW_MS)); - // While hidden, nothing is fetched (frames and the light's lease wait for the page). + // While hidden, nothing is fetched at all (frames, the light's lease, the sensor and the + // status wait for the page); shown again, the sensor is brought up to date at once. await setVisibility(page, 'hidden'); await page.waitForTimeout(1_000); // answers already on their way const hidden = await apiRequestsDuring(page, POLL_WINDOW_MS); - expect(Object.keys(hidden).filter((p) => p.endsWith('/frame') || p.endsWith('/lights/hold')), JSON.stringify(hidden)).toEqual([]); + expect(Object.keys(hidden), JSON.stringify(hidden)).toEqual([]); + const refreshed = page.waitForRequest((req) => new URL(req.url()).pathname === `/api/v1/sensors/${id}`, { + timeout: 1_500, + }); await setVisibility(page, 'visible'); + await refreshed; // The light's lease (renewed every 10 s) is not taken more than once either. await page.goto('#/sensors/new'); diff --git a/visionstate/frontend/src/App.svelte b/visionstate/frontend/src/App.svelte index ee3a1cb..05a35e4 100644 --- a/visionstate/frontend/src/App.svelte +++ b/visionstate/frontend/src/App.svelte @@ -3,6 +3,7 @@ import { app, loadConfig, refreshStatus } from './lib/app.svelte'; import Nav from './lib/components/Nav.svelte'; import Toasts from './lib/components/Toasts.svelte'; + import { pollWhileVisible } from './lib/poll'; import { route } from './lib/router.svelte'; import { POLL, SENSOR_TABS, type SensorTab } from './lib/ui'; import Dashboard from './pages/Dashboard.svelte'; @@ -18,8 +19,7 @@ Promise.all([loadConfig(), refreshStatus()]) .then(() => (ready = true)) .catch((err) => (app.error = err.message)); - const timer = setInterval(refreshStatus, POLL.status); - onDestroy(() => clearInterval(timer)); + onDestroy(pollWhileVisible(POLL.status, refreshStatus)); const [section, id, tab] = $derived(route.parts); // '' = the sensor's first tab, which depends on its kind (resolved by SensorPage). diff --git a/visionstate/frontend/src/lib/poll.ts b/visionstate/frontend/src/lib/poll.ts new file mode 100644 index 0000000..aecaffe --- /dev/null +++ b/visionstate/frontend/src/lib/poll.ts @@ -0,0 +1,22 @@ +// Polling that rests while the page is hidden (a background tab, the Home Assistant app put away). + +/** Calls `fn` every `ms` while the page is visible, and when it is shown again after a poll was + * missed (not more often than every `ms`). Returns a function that stops it. */ +export function pollWhileVisible(ms: number, fn: () => void): () => void { + let last = Date.now(); + const run = () => { + last = Date.now(); + fn(); + }; + const timer = setInterval(() => { + if (document.visibilityState === 'visible') run(); + }, ms); + const onVisible = () => { + if (document.visibilityState === 'visible' && Date.now() - last >= ms) run(); + }; + document.addEventListener('visibilitychange', onVisible); + return () => { + clearInterval(timer); + document.removeEventListener('visibilitychange', onVisible); + }; +} diff --git a/visionstate/frontend/src/pages/Dashboard.svelte b/visionstate/frontend/src/pages/Dashboard.svelte index 3b61388..2ecd9b5 100644 --- a/visionstate/frontend/src/pages/Dashboard.svelte +++ b/visionstate/frontend/src/pages/Dashboard.svelte @@ -10,6 +10,7 @@ import Timeline from '../lib/components/Timeline.svelte'; import { plural, pct } from '../lib/format'; import { facesBusy, isObjectSensor, objectName } from '../lib/objects'; + import { pollWhileVisible } from '../lib/poll'; import { isReadingSensor, readingText, readingUnit } from '../lib/reading'; import { READING_MODE_INFO } from '../lib/ui'; import { go, href, paths } from '../lib/router.svelte'; @@ -18,7 +19,6 @@ let sensors = $state(null); let importInput: HTMLInputElement; - let timer: ReturnType; async function load() { try { @@ -29,8 +29,7 @@ } load(); - timer = setInterval(load, POLL.dashboard); - onDestroy(() => clearInterval(timer)); + onDestroy(pollWhileVisible(POLL.dashboard, load)); async function importFile(e: Event) { const file = (e.currentTarget as HTMLInputElement).files?.[0]; diff --git a/visionstate/frontend/src/pages/SensorPage.svelte b/visionstate/frontend/src/pages/SensorPage.svelte index 5c21465..587739d 100644 --- a/visionstate/frontend/src/pages/SensorPage.svelte +++ b/visionstate/frontend/src/pages/SensorPage.svelte @@ -1,5 +1,4 @@