210 lines
7.7 KiB
Python
210 lines
7.7 KiB
Python
# WashData - Home Assistant integration for appliance cycle monitoring via smart plugs.
|
|
# Copyright (C) 2026 Lukas Bandura
|
|
# SPDX-License-Identifier: AGPL-3.0-or-later
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as published
|
|
# by the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
|
"""Diagnostic ring buffers for WashData - rolling 24-hour window.
|
|
|
|
Each WashDataManager owns one DiagBuffer instance that accumulates three
|
|
independent time-series for the last 24 hours:
|
|
|
|
power_trace - every raw power-sensor reading, before throttling
|
|
state_history - detector state transitions (off/starting/running/...)
|
|
logs - DEBUG-and-above log lines emitted by this integration
|
|
|
|
All buffers are in-memory only (no disk writes) so they impose zero I/O
|
|
overhead and vanish cleanly on HA restart. The 24-hour window caps memory
|
|
at a predictable ceiling regardless of sensor polling rate.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from collections import deque
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any
|
|
|
|
from homeassistant.util import dt as dt_util
|
|
|
|
_WINDOW = timedelta(hours=24)
|
|
|
|
# Hard upper-bound on entries per buffer. At a 1-second sensor interval,
|
|
# 100 000 power entries ~= 27 hours - enough to always cover the window.
|
|
_MAX_POWER = 100_000
|
|
_MAX_LOGS = 5_000
|
|
_MAX_STATES = 2_000
|
|
|
|
_INTEGRATION_LOGGER_NAME = "custom_components.ha_washdata"
|
|
|
|
|
|
def _ts_iso(unix: float) -> str:
|
|
return datetime.fromtimestamp(unix, tz=timezone.utc).isoformat()
|
|
|
|
|
|
class _LogHandler(logging.Handler):
|
|
"""Logging handler that buffers records for a single named device.
|
|
|
|
Installed on the integration root logger so it receives all records
|
|
produced anywhere inside *custom_components.ha_washdata*, then keeps
|
|
only the ones whose formatted message contains ``[device_name]`` -
|
|
the prefix injected by :class:`~.log_utils.DeviceLoggerAdapter`.
|
|
"""
|
|
|
|
def __init__(self, device_name: str) -> None:
|
|
super().__init__()
|
|
# Match the exact prefix added by DeviceLoggerAdapter
|
|
self._tag = f"[{device_name}]"
|
|
# Store (created_float, levelname, message)
|
|
self._buf: deque[tuple[float, str, str]] = deque(maxlen=_MAX_LOGS)
|
|
|
|
def emit(self, record: logging.LogRecord) -> None:
|
|
try:
|
|
msg = record.getMessage()
|
|
if self._tag not in msg:
|
|
return
|
|
self._buf.append((record.created, record.levelname, msg))
|
|
except Exception: # pylint: disable=broad-except
|
|
self.handleError(record)
|
|
|
|
def snapshot(self, cutoff: float) -> list[dict[str, Any]]:
|
|
items = list(self._buf)
|
|
return [
|
|
{"ts": _ts_iso(ts), "lvl": lvl, "msg": msg}
|
|
for ts, lvl, msg in items
|
|
if ts >= cutoff
|
|
]
|
|
|
|
|
|
class DiagBuffer:
|
|
"""Per-device diagnostic ring buffer aggregating power, states, and logs.
|
|
|
|
Lifecycle::
|
|
|
|
# on manager creation
|
|
self.diag_buffer = DiagBuffer(config_entry.title)
|
|
|
|
# on each raw power reading
|
|
self.diag_buffer.record_power(watts, timestamp)
|
|
|
|
# on each detector state transition
|
|
self.diag_buffer.record_state(old, new, program, timestamp)
|
|
|
|
# on manager shutdown
|
|
self.diag_buffer.uninstall()
|
|
|
|
# in diagnostics.py
|
|
snapshot = manager.diag_buffer.snapshot()
|
|
"""
|
|
|
|
def __init__(self, device_name: str) -> None:
|
|
self._device_name = device_name
|
|
|
|
# Raw power readings: (unix_ts_float, watts)
|
|
self._power: deque[tuple[float, float]] = deque(maxlen=_MAX_POWER)
|
|
|
|
# State transitions: (unix_ts_float, from_state, to_state, program)
|
|
self._states: deque[tuple[float, str, str, str]] = deque(maxlen=_MAX_STATES)
|
|
|
|
# Log handler - installed on the integration root logger
|
|
self._log_handler = _LogHandler(device_name)
|
|
logging.getLogger(_INTEGRATION_LOGGER_NAME).addHandler(self._log_handler)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Recording helpers
|
|
# ------------------------------------------------------------------
|
|
|
|
def record_power(self, watts: float, ts: datetime) -> None:
|
|
"""Record one raw power-sensor reading (call *before* any throttling)."""
|
|
self._power.append((ts.timestamp(), watts))
|
|
|
|
def power_samples(self, since_ts: float | None = None) -> list[tuple[float, float]]:
|
|
"""Return raw (unix_ts, watts) readings, optionally only those at/after since_ts."""
|
|
if since_ts is None:
|
|
return list(self._power)
|
|
return [(ts, w) for ts, w in self._power if ts >= since_ts]
|
|
|
|
def record_state(
|
|
self,
|
|
from_state: str,
|
|
to_state: str,
|
|
program: str,
|
|
ts: datetime,
|
|
) -> None:
|
|
"""Record a detector state transition."""
|
|
self._states.append((ts.timestamp(), from_state, to_state, program))
|
|
|
|
# ------------------------------------------------------------------
|
|
# Snapshot
|
|
# ------------------------------------------------------------------
|
|
|
|
def redacted_snapshot(self) -> dict[str, Any]:
|
|
"""Like :meth:`snapshot` but with identifying fields removed.
|
|
|
|
Strips ``device_name`` from the top-level dict and removes the ``msg``
|
|
field from each log entry so raw log text (which contains the device
|
|
name prefix injected by :class:`~.log_utils.DeviceLoggerAdapter`) is
|
|
not included in exported diagnostics.
|
|
"""
|
|
data = self.snapshot()
|
|
data.pop("device_name", None)
|
|
data["logs"] = [
|
|
{k: v for k, v in entry.items() if k != "msg"}
|
|
for entry in data.get("logs", [])
|
|
]
|
|
return data
|
|
|
|
def snapshot(self) -> dict[str, Any]:
|
|
"""Return all three buffers filtered to the last 24 hours.
|
|
|
|
Returned structure::
|
|
|
|
{
|
|
"window_hours": 24,
|
|
"device_name": "...",
|
|
"power_trace": [[iso_ts, watts], ...],
|
|
"state_history": [{"ts": ..., "from": ..., "to": ..., "program": ...}, ...],
|
|
"logs": [{"ts": ..., "lvl": ..., "msg": ...}, ...],
|
|
}
|
|
"""
|
|
cutoff = (dt_util.now() - _WINDOW).timestamp()
|
|
power_items = list(self._power)
|
|
state_items = list(self._states)
|
|
return {
|
|
"window_hours": 24,
|
|
"device_name": self._device_name,
|
|
"power_trace": [
|
|
[_ts_iso(ts), w]
|
|
for ts, w in power_items
|
|
if ts >= cutoff
|
|
],
|
|
"state_history": [
|
|
{"ts": _ts_iso(ts), "from": f, "to": t, "program": prog}
|
|
for ts, f, t, prog in state_items
|
|
if ts >= cutoff
|
|
],
|
|
"logs": self._log_handler.snapshot(cutoff),
|
|
}
|
|
|
|
# ------------------------------------------------------------------
|
|
# Lifecycle
|
|
# ------------------------------------------------------------------
|
|
|
|
def uninstall(self) -> None:
|
|
"""Remove the log handler from the integration logger.
|
|
|
|
Must be called from ``WashDataManager.async_shutdown()`` to avoid
|
|
accumulating stale handlers across config-entry reloads.
|
|
"""
|
|
logging.getLogger(_INTEGRATION_LOGGER_NAME).removeHandler(self._log_handler)
|