1"""Abstract data source for the anomaly detector.
2
3The detector doesn't read from InfluxDB or any other remote store; it asks a
4:class:`DataProvider` for the last N hours of sensor data.
5
6Window shape returned by :meth:`get_window`:
7
8 Dict[str, np.ndarray] — one numpy array per parameter code, all the
9 same length. Indexed by parameter (``"g1"``, ``"hum"``, ``"t"``, ...).
10 The ``"t"`` column is ms-epoch as ``int64``; every other column is
11 ``float64`` with ``NaN`` for missing samples. Row count is
12 ``len(window["t"])``. Returned ``None`` when the cache is unavailable,
13 empty, or holds no records inside the requested window.
14
15The production implementation is :class:`OzCacheProvider`, which wraps the
16Redis-backed payload buffer that ``Manager.sendData`` populates as a side
17effect of normal publishing.
18"""
19
20from __future__ import annotations
21
22import logging
23import time
24from abc import ABC, abstractmethod
25from typing import TYPE_CHECKING, Dict, Optional
26
27import numpy as np
28
29if TYPE_CHECKING: # pragma: no cover — type-only import to avoid Manager dep at runtime
30 from Manager.OzCache import OzCache
31
32logger = logging.getLogger(__name__)
33
34# Type alias used across the module for clarity.
35Window = Dict[str, np.ndarray]
36
37
[docs]
38def is_empty(window: Optional[Window]) -> bool:
39 """Return True when the window has no rows or doesn't exist.
40
41 Used by every check function as the first guard, replacing the
42 pandas ``df is None or df.empty`` idiom.
43 """
44 return window is None or len(window.get("t", [])) == 0
45
46
47class DataProvider(ABC):
48 """Supplies recent sensor data to the anomaly detector."""
49
50 @abstractmethod
51 def get_window(self, hours: int) -> Optional[Window]:
52 """Return the last ``hours`` of sensor data as a column-oriented dict.
53
54 Args:
55 hours: Length of the window to return (typically 24 or 48).
56
57 Returns:
58 Dict mapping each parameter code to a numpy array of the same
59 length, or ``None`` if insufficient data is available.
60 """
61
62
63class OzCacheProvider(DataProvider):
64 """Read recent sensor payloads from the Manager-owned :class:`OzCache`.
65
66 OzCache is a Redis-backed rolling list of the same ``data`` dicts that
67 ``Manager.sendData`` publishes to the gateway. The cache is populated by
68 two paths:
69
70 1. **Live ticks** — every successful publish writes its payload to Redis
71 (``Manager.sendData`` calls ``self.cache.push(payload)`` before the
72 HTTP POST).
73 2. **Cold-start refill** — :meth:`OzCache.refill` paginates through the
74 gateway's ``/ppb/device/`` endpoint at boot to seed the cache before
75 live ticks have caught up.
76
77 OzCache normalises every payload's ``t`` field to **milliseconds** —
78 live ticks are already ms (``lastPublishMillis``), and refilled PPB
79 rows (which arrive in seconds) are multiplied by 1000 inside
80 ``OzCache.refill``. So this provider can apply a single ms-based
81 cutoff without inspecting where each row came from.
82 """
83
84 def __init__(self, cache: "OzCache | None") -> None:
85 """Args:
86 cache: The ``OzCache`` instance owned by ``Manager`` (typically
87 ``manager.cache``). May be ``None`` or unconfigured —
88 :meth:`get_window` handles both and returns ``None``.
89 """
90 self.cache = cache
91
92 def get_window(self, hours: int) -> Optional[Window]:
93 """Return the last ``hours`` of cached payloads as a column dict.
94
95 Returns ``None`` when the cache is unavailable, empty, or holds no
96 records inside the requested window.
97 """
98 logger.info("OzCacheProvider: get_window(hours=%d) requested", hours)
99
100 if self.cache is None or not getattr(self.cache, "enabled", False):
101 logger.warning("OzCacheProvider: cache disabled or missing — returning None")
102 return None
103
104 rows = self.cache.window()
105 raw_count = len(rows)
106 if not rows:
107 logger.warning("OzCacheProvider: cache empty — returning None")
108 return None
109
110 # Normalise t from seconds to ms when needed. Live ticks push
111 # seconds-epoch t (the variable is misnamed `lastPublishMillis`
112 # but holds seconds); refilled rows already have ms because
113 # OzCache.refill() multiplies by 1000 before LPUSH. Per-row
114 # check: if t looks like seconds (< 10^11), upgrade to ms.
115 for r in rows:
116 t = r.get("t")
117 if isinstance(t, (int, float)) and t and t < 10 ** 11:
118 r["t"] = int(t) * 1000
119
120 cutoff_ms = (int(time.time()) - hours * 3600) * 1000
121 rows = [r for r in rows if int(r.get("t", 0)) >= cutoff_ms]
122 filtered_count = len(rows)
123 logger.info(
124 "OzCacheProvider: cache_size=%d in_window=%d cutoff_ms=%d",
125 raw_count, filtered_count, cutoff_ms,
126 )
127 if not rows:
128 logger.warning(
129 "OzCacheProvider: no records in last %dh (cache had %d total)",
130 hours, raw_count,
131 )
132 return None
133
134 # Sort ascending — original cloud checks were written against
135 # ascending traces, and the constant-instance check needs adjacency.
136 rows.sort(key=lambda r: int(r.get("t", 0)))
137
138 # Column-orient: gather every key seen across rows. `t` stays int64;
139 # everything else becomes float64 with NaN for missing/non-numeric.
140 all_keys = set()
141 for r in rows:
142 all_keys.update(r.keys())
143
144 window: Window = {}
145 for k in all_keys:
146 if k == "t":
147 window[k] = np.array(
148 [int(r.get("t", 0)) for r in rows], dtype=np.int64,
149 )
150 else:
151 vals = []
152 for r in rows:
153 v = r.get(k)
154 try:
155 vals.append(float(v) if v is not None else np.nan)
156 except (TypeError, ValueError):
157 vals.append(np.nan)
158 window[k] = np.array(vals, dtype=np.float64)
159
160 n_rows = len(rows)
161 logger.info(
162 "OzCacheProvider: window shape=(%d rows, %d cols) time=[%d -> %d ms]",
163 n_rows, len(window),
164 int(window["t"][0]), int(window["t"][-1]),
165 )
166 logger.info("OzCacheProvider: columns=%s", sorted(window.keys()))
167 # Sample the newest row for sanity (skip 't' since it's printed above).
168 latest = {
169 k: float(window[k][-1])
170 for k in sorted(window.keys()) if k != "t"
171 }
172 latest["t"] = int(window["t"][-1])
173 logger.info("OzCacheProvider: latest row=%s", latest)
174 return window