diff --git a/custom_components/hass_datapoints/__init__.py b/custom_components/hass_datapoints/__init__.py index 0265d94..a5a5026 100644 --- a/custom_components/hass_datapoints/__init__.py +++ b/custom_components/hass_datapoints/__init__.py @@ -258,7 +258,8 @@ async def async_unload_entry(hass: HomeAssistant, entry: DatapointsConfigEntry) """Unload a config entry.""" unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS) hass.services.async_remove(DOMAIN, SERVICE_RECORD) - hass.data[DOMAIN].pop("store", None) + if store := hass.data[DOMAIN].pop("store", None): + await hass.async_add_executor_job(store.close) hass.data[DOMAIN].pop("anomaly_cache", None) hass.data[DOMAIN].pop("in_flight", None) hass.data[DOMAIN].pop("scan_semaphore", None) diff --git a/custom_components/hass_datapoints/hass-datapoints-cards.js b/custom_components/hass_datapoints/hass-datapoints-cards.js index 1c30119..ef69ce8 100644 --- a/custom_components/hass_datapoints/hass-datapoints-cards.js +++ b/custom_components/hass_datapoints/hass-datapoints-cards.js @@ -42252,7 +42252,7 @@ ].forEach((card) => { if (!registeredTypes.has(card.type)) window.customCards?.push(card); }); - console.groupCollapsed(`%c hass-datapoints %c v0.6.3 loaded `, "color:#fff;background:#03a9f4;font-weight:bold;padding:2px 6px;border-radius:3px 0 0 3px", "color:#03a9f4;background:#fff;font-weight:bold;padding:2px 6px;border:1px solid #03a9f4;border-radius:0 3px 3px 0", ...[]); + console.groupCollapsed(`%c hass-datapoints %c v0.6.4 loaded `, "color:#fff;background:#03a9f4;font-weight:bold;padding:2px 6px;border-radius:3px 0 0 3px", "color:#03a9f4;background:#fff;font-weight:bold;padding:2px 6px;border:1px solid #03a9f4;border-radius:0 3px 3px 0", ...[]); console.log("Enable debug logging by setting %cwindow.__HASS_DATAPOINTS_DEV__ = true", "color:#333;background:#eee;border:1px solid #777;padding:2px 6px;border-radius:5px; font-family: Courier"); console.groupEnd(); //#endregion diff --git a/custom_components/hass_datapoints/manifest.json b/custom_components/hass_datapoints/manifest.json index 11c3b1e..9d114c0 100644 --- a/custom_components/hass_datapoints/manifest.json +++ b/custom_components/hass_datapoints/manifest.json @@ -8,5 +8,5 @@ "iot_class": "local_push", "issue_tracker": "https://github.com/buggedcom/hass-datapoints/issues", "requirements": [], - "version": "0.6.3" + "version": "0.6.4" } diff --git a/custom_components/hass_datapoints/sensor.py b/custom_components/hass_datapoints/sensor.py index 94b340c..9e08400 100644 --- a/custom_components/hass_datapoints/sensor.py +++ b/custom_components/hass_datapoints/sensor.py @@ -95,6 +95,11 @@ class _DatapointsSensorBase(SensorEntity): """Shared base for all Hass Data Points sensor entities.""" _attr_has_entity_name = True + # Subclasses whose _compute() performs blocking I/O (SQLite reads) set this + # to True so refreshes run in the executor instead of on the event loop. + # Sensors that compute from in-memory monitor state leave it False and + # refresh inline (offloading those would race with loop-side mutations). + _compute_blocks_io = False def __init__(self, entry: ConfigEntry, store: DatapointsStore) -> None: """Initialise with config entry and data store.""" @@ -118,7 +123,19 @@ async def async_added_to_hass(self) -> None: def _handle_store_update(self) -> None: """Refresh state after any store mutation.""" - self._attr_native_value = self._compute() + self._refresh_value() + + def _refresh_value(self) -> None: + """Recompute and write state — off the loop when _compute() blocks.""" + if self._compute_blocks_io: + self.hass.async_create_task(self._async_refresh_value()) + else: + self._attr_native_value = self._compute() + self.async_write_ha_state() + + async def _async_refresh_value(self) -> None: + """Run the blocking _compute() in the executor, then write state.""" + self._attr_native_value = await self.hass.async_add_executor_job(self._compute) self.async_write_ha_state() def _compute(self) -> Any: @@ -148,8 +165,7 @@ async def async_added_to_hass(self) -> None: @callback def _handle_time_interval(self, now: datetime) -> None: """Refresh state on every timer tick.""" - self._attr_native_value = self._compute() - self.async_write_ha_state() + self._refresh_value() # --------------------------------------------------------------------------- @@ -161,6 +177,7 @@ class DatapointsCountSensor(_DatapointsSensorBase): """Expose the total number of recorded datapoints.""" _attr_icon = "mdi:counter" + _compute_blocks_io = True def __init__(self, entry: ConfigEntry, store: DatapointsStore) -> None: """Initialise the datapoint count sensor.""" @@ -178,6 +195,7 @@ class DatapointsLastTimestampSensor(_DatapointsSensorBase): _attr_device_class = SensorDeviceClass.TIMESTAMP _attr_icon = "mdi:clock-outline" + _compute_blocks_io = True def __init__(self, entry: ConfigEntry, store: DatapointsStore) -> None: """Initialise the last recorded timestamp sensor.""" @@ -206,6 +224,7 @@ class DatapointsLastMessageSensor(_DatapointsSensorBase): """Expose the message of the most recently recorded datapoint.""" _attr_icon = "mdi:text" + _compute_blocks_io = True def __init__(self, entry: ConfigEntry, store: DatapointsStore) -> None: """Initialise the last recorded message sensor.""" @@ -228,6 +247,7 @@ class DatapointsTimeSinceLastSensor(_DatapointsPeriodicSensorBase): _attr_native_unit_of_measurement = UnitOfTime.HOURS _attr_suggested_display_precision = 1 _attr_icon = "mdi:timer-outline" + _compute_blocks_io = True def __init__( self, entry: ConfigEntry, store: DatapointsStore, hass: HomeAssistant @@ -259,6 +279,7 @@ class DatapointsTodayCountSensor(_DatapointsPeriodicSensorBase): """Expose the count of datapoints recorded since the start of today (local time).""" _attr_icon = "mdi:calendar-today" + _compute_blocks_io = True def __init__( self, entry: ConfigEntry, store: DatapointsStore, hass: HomeAssistant @@ -278,6 +299,7 @@ class DatapointsWeekCountSensor(_DatapointsPeriodicSensorBase): """Expose the count of datapoints recorded since the start of this week (Mon, local time).""" _attr_icon = "mdi:calendar-week" + _compute_blocks_io = True def __init__( self, entry: ConfigEntry, store: DatapointsStore, hass: HomeAssistant @@ -298,6 +320,7 @@ class DatapointsAutomationCountSensor(_DatapointsSensorBase): """Expose the count of automation-triggered datapoints.""" _attr_icon = "mdi:robot" + _compute_blocks_io = True def __init__(self, entry: ConfigEntry, store: DatapointsStore) -> None: """Initialise the automation count sensor.""" @@ -315,6 +338,7 @@ class DatapointsManualCountSensor(_DatapointsSensorBase): """Expose the count of manually recorded datapoints.""" _attr_icon = "mdi:hand-back-right" + _compute_blocks_io = True def __init__(self, entry: ConfigEntry, store: DatapointsStore) -> None: """Initialise the manual count sensor.""" diff --git a/custom_components/hass_datapoints/store.py b/custom_components/hass_datapoints/store.py index 8c76143..12cd4d5 100644 --- a/custom_components/hass_datapoints/store.py +++ b/custom_components/hass_datapoints/store.py @@ -5,6 +5,7 @@ import json import logging import sqlite3 +import threading import uuid from collections.abc import Callable from datetime import UTC, datetime @@ -89,26 +90,36 @@ def _event_to_params(event: dict[str, Any]) -> tuple: class _EventDb: - """Synchronous SQLite backend for event storage.""" + """Synchronous SQLite backend for event storage. + + Holds a single long-lived connection guarded by a lock rather than opening + (and leaking) a fresh connection per operation. ``check_same_thread=False`` + lets Home Assistant's executor threads reuse it; the lock serialises access + so the shared connection is never touched concurrently. All access is + blocking and MUST be dispatched from an executor, never the event loop. + """ def __init__(self, db_path: str) -> None: self._db_path = db_path + self._lock = threading.Lock() + self._conn = sqlite3.connect(db_path, check_same_thread=False) + self._conn.row_factory = sqlite3.Row self._init_db() - def _connect(self) -> sqlite3.Connection: - conn = sqlite3.connect(self._db_path, check_same_thread=False) - conn.row_factory = sqlite3.Row - return conn - def _init_db(self) -> None: - with self._connect() as conn: - conn.execute("PRAGMA journal_mode=WAL") - conn.executescript(_SCHEMA_SQL) + with self._lock, self._conn: + self._conn.execute("PRAGMA journal_mode=WAL") + self._conn.executescript(_SCHEMA_SQL) + + def close(self) -> None: + """Close the underlying connection. Called once when the entry unloads.""" + with self._lock: + self._conn.close() def insert(self, event: dict[str, Any]) -> None: """Insert a new event row.""" - with self._connect() as conn: - conn.execute( + with self._lock, self._conn: + self._conn.execute( """ INSERT INTO events (id, timestamp, message, annotation, @@ -122,8 +133,8 @@ def insert(self, event: dict[str, Any]) -> None: def insert_many(self, events: list[dict[str, Any]]) -> None: """Bulk-insert events using INSERT OR IGNORE (idempotent for migration).""" params = [_event_to_params(e) for e in events] - with self._connect() as conn: - conn.executemany( + with self._lock, self._conn: + self._conn.executemany( """ INSERT OR IGNORE INTO events (id, timestamp, message, annotation, @@ -141,20 +152,20 @@ def update(self, event_id: str, fields: dict[str, Any]) -> bool: } cols = ", ".join(f"{k}=?" for k in encoded) params = list(encoded.values()) + [event_id] - with self._connect() as conn: - cursor = conn.execute(f"UPDATE events SET {cols} WHERE id=?", params) + with self._lock, self._conn: + cursor = self._conn.execute(f"UPDATE events SET {cols} WHERE id=?", params) return cursor.rowcount > 0 def delete(self, event_id: str) -> bool: """Delete an event by ID. Returns True if the row existed.""" - with self._connect() as conn: - cursor = conn.execute("DELETE FROM events WHERE id=?", (event_id,)) + with self._lock, self._conn: + cursor = self._conn.execute("DELETE FROM events WHERE id=?", (event_id,)) return cursor.rowcount > 0 def delete_dev(self) -> int: """Delete all dev-flagged events. Returns the number of rows deleted.""" - with self._connect() as conn: - cursor = conn.execute("DELETE FROM events WHERE dev=1") + with self._lock, self._conn: + cursor = self._conn.execute("DELETE FROM events WHERE dev=1") return cursor.rowcount def query( @@ -165,17 +176,26 @@ def query( limit: int | None, offset: int, ) -> list[dict[str, Any]]: - """Return events filtered by time range, then by entity IDs in Python.""" - with self._connect() as conn: - rows = conn.execute( - """ - SELECT * FROM events - WHERE (? IS NULL OR timestamp >= ?) - AND (? IS NULL OR timestamp <= ?) - ORDER BY timestamp ASC - """, - (start, start, end, end), - ).fetchall() + """Return events filtered by time range (and, in Python, by entity IDs). + + With no entity filter, LIMIT/OFFSET are pushed into SQL so the whole + table is never materialised. With an entity filter the JSON-array + intersection has to run in Python, so pagination is applied after it. + """ + sql = [ + "SELECT * FROM events", + "WHERE (? IS NULL OR timestamp >= ?)", + "AND (? IS NULL OR timestamp <= ?)", + "ORDER BY timestamp ASC", + ] + params: list[Any] = [start, start, end, end] + push_pagination = entity_ids is None + if push_pagination and (limit is not None or offset): + # SQLite uses LIMIT -1 to mean "no limit" when only an offset is set. + sql.append("LIMIT ? OFFSET ?") + params.extend([limit if limit is not None else -1, offset]) + with self._lock: + rows = self._conn.execute("\n".join(sql), params).fetchall() events = [_row_to_dict(r) for r in rows] @@ -191,24 +211,23 @@ def query( seen.add(ev["id"]) filtered.append(ev) events = filtered - - if offset: - events = events[offset:] - if limit is not None: - events = events[:limit] + if offset: + events = events[offset:] + if limit is not None: + events = events[:limit] return events def count(self) -> int: """Return total event count.""" - with self._connect() as conn: - row = conn.execute("SELECT COUNT(*) FROM events").fetchone() + with self._lock: + row = self._conn.execute("SELECT COUNT(*) FROM events").fetchone() return row[0] def count_in_range(self, start: str, end: str | None) -> int: """Return event count within a time range.""" - with self._connect() as conn: - row = conn.execute( + with self._lock: + row = self._conn.execute( "SELECT COUNT(*) FROM events WHERE timestamp >= ? AND (? IS NULL OR timestamp <= ?)", (start, end, end), ).fetchone() @@ -216,24 +235,24 @@ def count_in_range(self, start: str, end: str | None) -> int: def bounds(self) -> tuple[str | None, str | None]: """Return (earliest_timestamp, latest_timestamp) or (None, None) if empty.""" - with self._connect() as conn: - row = conn.execute( + with self._lock: + row = self._conn.execute( "SELECT MIN(timestamp), MAX(timestamp) FROM events" ).fetchone() return row[0], row[1] def last(self) -> dict[str, Any] | None: """Return the most recently timestamped event, or None if empty.""" - with self._connect() as conn: - row = conn.execute( + with self._lock: + row = self._conn.execute( "SELECT * FROM events ORDER BY timestamp DESC LIMIT 1" ).fetchone() return _row_to_dict(row) if row else None def automation_manual_counts(self) -> tuple[int, int]: """Return (automation_count, manual_count).""" - with self._connect() as conn: - row = conn.execute( + with self._lock: + row = self._conn.execute( """ SELECT COUNT(CASE WHEN automation_id IS NOT NULL THEN 1 END), @@ -245,8 +264,8 @@ def automation_manual_counts(self) -> tuple[int, int]: def get_by_id(self, event_id: str) -> dict[str, Any] | None: """Fetch a single event by ID, or None if not found.""" - with self._connect() as conn: - row = conn.execute( + with self._lock: + row = self._conn.execute( "SELECT * FROM events WHERE id=?", (event_id,) ).fetchone() return _row_to_dict(row) if row else None @@ -301,6 +320,10 @@ async def async_load(self) -> None: if "last_resolved_clusters_summary" not in m: m["last_resolved_clusters_summary"] = [] + def close(self) -> None: + """Release the SQLite connection. Called when the config entry unloads.""" + self._event_db.close() + async def async_record( self, message: str, diff --git a/custom_components/hass_datapoints/websocket_api.py b/custom_components/hass_datapoints/websocket_api.py index 692b5ad..678a21c 100644 --- a/custom_components/hass_datapoints/websocket_api.py +++ b/custom_components/hass_datapoints/websocket_api.py @@ -190,12 +190,14 @@ async def ws_get_events( ] limit: int = msg.get("limit", 200) offset: int = msg.get("offset", 0) - events = store.get_events( - start=msg.get("start_time"), - end=msg.get("end_time"), - entity_ids=entity_ids, - limit=limit, - offset=offset, + # SQLite reads block, so run them in the executor rather than on the loop. + events = await hass.async_add_executor_job( + store.get_events, + msg.get("start_time"), + msg.get("end_time"), + entity_ids, + limit, + offset, ) connection.send_result(msg["id"], {"events": events}) @@ -219,7 +221,9 @@ async def ws_get_event_bounds( _get_global_history_bounds, recorder ) if start_time is None and end_time is None: - start_time, end_time = store.get_event_bounds() + start_time, end_time = await hass.async_add_executor_job( + store.get_event_bounds + ) source = "datapoints_store_fallback" connection.send_result( msg["id"], diff --git a/package.json b/package.json index a73f502..a9e5fe7 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "hass-datapoints", - "version": "0.6.3", + "version": "0.6.4", "private": true, "description": "Build and development entrypoints for the hass-datapoints Home Assistant integration.", "packageManager": "pnpm@10.33.0", diff --git a/tests/conftest.py b/tests/conftest.py index 1a463df..1a44083 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -9,6 +9,7 @@ from __future__ import annotations +import contextlib import sys from datetime import UTC from datetime import datetime as _datetime @@ -29,6 +30,25 @@ def _stub(name: str, obj: object | None = None) -> None: sys.modules[name] = obj if obj is not None else MagicMock() +def _make_sync_hass() -> MagicMock: + """Return a hass mock that runs executor jobs and created tasks inline.""" + + async def _run_executor(fn, *args): + return fn(*args) + + def _run_task(coro): + # The offloaded refresh only awaits the inline executor above, so it + # never suspends — a single send() runs the coroutine to completion. + with contextlib.suppress(StopIteration): + coro.send(None) + return MagicMock() + + hass = MagicMock() + hass.async_add_executor_job = AsyncMock(side_effect=_run_executor) + hass.async_create_task = _run_task + return hass + + # -- voluptuous --------------------------------------------------------------- # voluptuous is a real dependency listed in requirements_dev.txt; do NOT stub # it so that tests exercising vol.Match / vol.In / vol.Range work correctly. @@ -156,6 +176,12 @@ def __new__(cls, *args, **kwargs): # Install per-instance mocks regardless of whether __init__ calls super() obj.async_write_ha_state = MagicMock() obj._removed_callbacks = [] + # A hass whose executor runs inline and whose create_task drives the + # coroutine to completion synchronously — sensors that offload their + # blocking _compute() to the executor thus refresh synchronously under + # test (the refresh only awaits the inline executor, which never + # suspends). + obj.hass = _make_sync_hass() return obj def async_on_remove(self, callback) -> None: diff --git a/tests/test_websocket_api.py b/tests/test_websocket_api.py index 3521a9e..e4f2768 100644 --- a/tests/test_websocket_api.py +++ b/tests/test_websocket_api.py @@ -100,8 +100,13 @@ def test_GIVEN_integer_zero_WHEN_called_THEN_is_unix_epoch(self): def _make_hass(store: object) -> MagicMock: + async def _run_executor(fn, *args): + return fn(*args) + hass = MagicMock() hass.data = {DOMAIN: {"store": store}} + # ws handlers offload blocking store reads to the executor; run them inline. + hass.async_add_executor_job = AsyncMock(side_effect=_run_executor) return hass @@ -173,11 +178,11 @@ async def test_GIVEN_time_filters_WHEN_called_THEN_passes_filters_to_store(self) await ws_get_events(hass, connection, msg) store.get_events.assert_called_once_with( - start="2024-01-01T00:00:00+00:00", - end="2024-12-31T00:00:00+00:00", - entity_ids=None, - limit=200, - offset=0, + "2024-01-01T00:00:00+00:00", + "2024-12-31T00:00:00+00:00", + None, + 200, + 0, ) @@ -222,11 +227,11 @@ async def test_GIVEN_entity_filter_WHEN_called_THEN_passes_entity_ids_to_store( await ws_get_events(hass, connection, msg) store.get_events.assert_called_once_with( - start=None, - end=None, - entity_ids=["sensor.a"], - limit=200, - offset=0, + None, + None, + ["sensor.a"], + 200, + 0, ) @@ -661,11 +666,11 @@ async def test_GIVEN_admin_user_with_entity_filter_WHEN_called_THEN_passes_filte await ws_get_events(hass, connection, msg) store.get_events.assert_called_once_with( - start=None, - end=None, - entity_ids=["sensor.a", "sensor.b"], - limit=200, - offset=0, + None, + None, + ["sensor.a", "sensor.b"], + 200, + 0, ) async def test_GIVEN_non_admin_user_WHEN_entity_ids_include_forbidden_THEN_forbidden_stripped( @@ -684,9 +689,7 @@ async def test_GIVEN_non_admin_user_WHEN_entity_ids_include_forbidden_THEN_forbi await ws_get_events(hass, connection, msg) - store.get_events.assert_called_once_with( - start=None, end=None, entity_ids=["sensor.a"], limit=200, offset=0 - ) + store.get_events.assert_called_once_with(None, None, ["sensor.a"], 200, 0) async def test_GIVEN_non_admin_user_WHEN_all_entity_ids_forbidden_THEN_empty_filter_passed( self, @@ -703,9 +706,7 @@ async def test_GIVEN_non_admin_user_WHEN_all_entity_ids_forbidden_THEN_empty_fil await ws_get_events(hass, connection, msg) - store.get_events.assert_called_once_with( - start=None, end=None, entity_ids=[], limit=200, offset=0 - ) + store.get_events.assert_called_once_with(None, None, [], 200, 0) async def test_GIVEN_non_admin_user_WHEN_no_entity_filter_THEN_store_called_with_none( self, @@ -719,9 +720,7 @@ async def test_GIVEN_non_admin_user_WHEN_no_entity_filter_THEN_store_called_with await ws_get_events(hass, connection, msg) - store.get_events.assert_called_once_with( - start=None, end=None, entity_ids=None, limit=200, offset=0 - ) + store.get_events.assert_called_once_with(None, None, None, 200, 0) # ---------------------------------------------------------------------------