1"""
2OzCache — generic device-side payload cache backed by Redis.
3
4Lives inside the Manager package because the cache is a side-effect of
5publishing: Manager.sendData pushes to the cache on the way out. Any
6consumer on the device (ML, anomaly, QC, whatever) reads the same cache
7rather than re-hitting the cloud.
8
9Storage shape:
10 - One Redis LIST per device: cache:{device_id}
11 - Each element is a JSON-serialised copy of the payload that
12 sendData received (the `d` dict — full telemetry for that tick).
13 - LPUSH at the head, LTRIM to a fixed capacity computed from
14 `hours * 3600 / publish_interval_sec`. No time-based rolling —
15 oldest falls off when the list is full.
16
17Config (device-side, lives under `self.config["cache"]`):
18 {
19 "en": 1, # enable / disable
20 "hours": 48, # retention window in hours
21 "refill": 1, # optional; cold-start via ppbanalytics
22 "prefix": "cache" # optional; key namespace
23 }
24"""
25
26from __future__ import annotations
27
28import json
29import math
30import os
31import threading
32import time
33from typing import List, Optional
34
35from utils.oizom_logger import OizomLogger
36
37try:
38 import redis as _redis # type: ignore
39 _REDIS_IMPORT_OK = True
40except ImportError: # pragma: no cover
41 _redis = None
42 _REDIS_IMPORT_OK = False
43
44context_logger = OizomLogger(__name__)
45
46# Refill thread schedule — fires once after boot, then capped backoff
47REFILL_BOOT_DELAY_SEC = 30
48REFILL_BACKOFF_SEC = (60, 120, 300, 900, 1800)
49REFILL_SKIP_SIZE = 500 # skip refill if list already this long
50REFILL_PAGE_LIMIT = 2000
51REFILL_MAX_PAGES = 10
52DEFAULT_CONNECT_TIMEOUT = 2.0
53
54
55class OzCache:
56 """
57 Generic redis-backed rolling cache for outgoing device payloads.
58
59 Construction is cheap; actual Redis connect happens in `configure()`.
60 Safe to leave unconfigured (all methods become no-ops).
61 """
62
63 def __init__(self) -> None:
64 self.enabled = False
65 self.device_id = ""
66 self.hours = 48
67 self.capacity = 0
68 self.prefix = "cache"
69 self._client = None
70 self._push_count = 0
71
72 # ----------------------------------------------------------------- setup
73 def configure(
74 self,
75 config: dict,
76 device_id: str,
77 publish_interval_sec: int,
78 host: Optional[str] = None,
79 port: Optional[int] = None,
80 ) -> bool:
81 """
82 Wire up from the device config. Returns True if the cache is active.
83 A disabled flag, missing device_id, missing redis package, or a
84 failed connect all return False and leave the instance in no-op mode.
85 """
86 if not config or not config.get("en"):
87 self.enabled = False
88 return False
89
90 if not _REDIS_IMPORT_OK:
91 context_logger.warning_with_context(
92 "CACHE", "redis package not installed — cache disabled"
93 )
94 return False
95
96 self.device_id = (device_id or "").strip()
97 if not self.device_id:
98 context_logger.warning_with_context(
99 "CACHE", "device_id empty — cache disabled"
100 )
101 return False
102
103 self.hours = int(config.get("hours", 48))
104 interval = max(int(publish_interval_sec), 1)
105 self.capacity = int(math.ceil(self.hours * 3600 / interval))
106 self.prefix = str(config.get("prefix", "cache"))
107
108 try:
109 self._client = _redis.Redis(
110 host=host or os.getenv("REDIS_HOST", "redis"),
111 port=int(port or os.getenv("REDIS_PORT", "6379")),
112 decode_responses=True,
113 socket_timeout=DEFAULT_CONNECT_TIMEOUT,
114 socket_connect_timeout=DEFAULT_CONNECT_TIMEOUT,
115 )
116 self._client.ping()
117 self.enabled = True
118 context_logger.info_with_context(
119 "CACHE",
120 f"configured: hours={self.hours} capacity={self.capacity} "
121 f"interval={interval}s device={self.device_id}",
122 )
123 return True
124 except Exception as e:
125 self._client = None
126 self.enabled = False
127 context_logger.warning_with_context(
128 "CACHE", f"connect failed ({e}) — cache disabled"
129 )
130 return False
131
132 # ---------------------------------------------------------------- keys
133 def _key(self) -> str:
134 return f"{self.prefix}:{self.device_id}"
135
136 @property
137 def client(self):
138 """Public handle to the underlying redis client.
139
140 Other consumers (e.g. the Anomaly Scheduler) reuse this connection
141 for their own keys instead of opening a second one.
142 """
143 return self._client
144
145 # ---------------------------------------------------------------- writes
146 def push(self, payload: dict) -> bool:
147 """Append one outgoing payload to the head of the list."""
148 if not self.enabled or not isinstance(payload, dict):
149 return False
150 try:
151 data = json.dumps(payload, separators=(",", ":"), default=str)
152 p = self._client.pipeline()
153 p.lpush(self._key(), data)
154 p.ltrim(self._key(), 0, self.capacity - 1)
155 p.expire(self._key(), self.hours * 3600 * 2) # TTL = 2× window
156 results = p.execute()
157 new_size = int(results[0]) if results else self.size()
158 self._push_count += 1
159 if self._push_count == 1:
160 sample = data[:240] + ("..." if len(data) > 240 else "")
161 context_logger.info_with_context(
162 "CACHE",
163 f"FIRST PUSH OK key={self._key()} size={new_size}/{self.capacity}",
164 )
165 context_logger.info_with_context("CACHE", f"sample payload: {sample}")
166 else:
167 sample_keys = list(payload.keys())[:6]
168 context_logger.info_with_context(
169 "CACHE",
170 f"push #{self._push_count} size={new_size}/{self.capacity} "
171 f"keys={sample_keys} t={payload.get('t')}",
172 )
173 return True
174 except Exception as e:
175 context_logger.error_with_context("CACHE", f"push: {e}")
176 return False
177
178 def bulk_load(self, payloads: List[dict]) -> int:
179 """
180 Insert multiple payloads in one pipeline. Caller is responsible for
181 ordering — LPUSH-ing oldest-first keeps the newest entry at list head.
182 """
183 if not self.enabled or not payloads:
184 return 0
185 try:
186 p = self._client.pipeline()
187 for payload in payloads:
188 if not isinstance(payload, dict):
189 continue
190 data = json.dumps(payload, separators=(",", ":"), default=str)
191 p.lpush(self._key(), data)
192 p.ltrim(self._key(), 0, self.capacity - 1)
193 p.expire(self._key(), self.hours * 3600 * 2)
194 p.execute()
195 return len(payloads)
196 except Exception as e:
197 context_logger.error_with_context("CACHE", f"bulk_load: {e}")
198 return 0
199
200 # ---------------------------------------------------------------- reads
201 def window(self) -> List[dict]:
202 """Return every cached payload, newest first."""
203 if not self.enabled:
204 return []
205 try:
206 raw = self._client.lrange(self._key(), 0, -1)
207 return [json.loads(x) for x in raw]
208 except Exception as e:
209 context_logger.error_with_context("CACHE", f"window: {e}")
210 return []
211
212 def size(self) -> int:
213 if not self.enabled:
214 return 0
215 try:
216 return int(self._client.llen(self._key()))
217 except Exception as e:
218 context_logger.error_with_context("CACHE", f"size: {e}")
219 return 0
220
221 def clear(self) -> None:
222 """Drop the entire cache for this device (test/recovery helper)."""
223 if not self.enabled:
224 return
225 try:
226 self._client.delete(self._key())
227 except Exception as e:
228 context_logger.error_with_context("CACHE", f"clear: {e}")
229
230 # ---------------------------------------------------------------- refill
231 def refill(self, manager, hours: Optional[int] = None) -> int:
232 """
233 One-shot refill from ppbanalytics. Returns the number of payloads
234 loaded. Caller controls retry — see `start_background_refill`.
235 """
236 if not self.enabled:
237 return 0
238
239 window = int(hours) if hours else self.hours
240 now_s = int(time.time())
241 gte_s = now_s - window * 3600
242
243 rows: list = []
244 for page in range(REFILL_MAX_PAGES):
245 try:
246 data, status = manager.getPPBanalytics(
247 lte=now_s,
248 gte=gte_s,
249 deviceId=self.device_id,
250 deviceType="",
251 limit=REFILL_PAGE_LIMIT,
252 page=page,
253 )
254 except Exception as e:
255 context_logger.error_with_context(
256 "CACHE", f"refill page {page} error: {e}"
257 )
258 break
259
260 if status != 200 or not isinstance(data, list) or not data:
261 break
262 rows.extend(data)
263 if len(data) < REFILL_PAGE_LIMIT:
264 break
265
266 if not rows:
267 context_logger.warning_with_context(
268 "CACHE", f"refill: no rows returned (window={window}h)"
269 )
270 return 0
271
272 # Sort chronologically so final LPUSH-per-entry leaves the newest
273 # payload at the head of the list.
274 def row_time(r):
275 t = r.get("time")
276 if t is None:
277 t = (r.get("payload") or {}).get("d", {}).get("t", 0)
278 return int(t or 0)
279
280 rows.sort(key=row_time)
281
282 payloads: List[dict] = []
283 for r in rows:
284 d = (r.get("payload") or {}).get("d")
285 if not isinstance(d, dict):
286 continue
287 # ppbanalytics returns `t` in seconds; live ticks write `t` in
288 # ms (lastPublishMillis). Normalise to ms for the cache.
289 payload = dict(d)
290 t = payload.get("t")
291 if isinstance(t, (int, float)) and t < 10**11:
292 payload["t"] = int(t) * 1000
293 payloads.append(payload)
294
295 inserted = self.bulk_load(payloads)
296 context_logger.info_with_context(
297 "CACHE",
298 f"refill: {inserted} payloads loaded (window={window}h, "
299 f"device={self.device_id})",
300 )
301 return inserted
302
303
304# --------------------------------------------------------------------- thread
[docs]
305def _refill_loop(cache: OzCache, manager, window_hours: int) -> None:
306 try:
307 time.sleep(REFILL_BOOT_DELAY_SEC)
308 except Exception:
309 return
310
311 if not cache.enabled:
312 return
313 if cache.size() >= REFILL_SKIP_SIZE:
314 context_logger.info_with_context(
315 "CACHE", f"refill skipped — cache size={cache.size()} already healthy"
316 )
317 return
318
319 attempts = len(REFILL_BACKOFF_SEC) + 1
320 for attempt in range(attempts):
321 try:
322 inserted = cache.refill(manager, hours=window_hours)
323 except Exception as e:
324 inserted = 0
325 context_logger.error_with_context(
326 "CACHE", f"refill attempt {attempt + 1} raised: {e}"
327 )
328
329 if inserted > 0:
330 context_logger.info_with_context(
331 "CACHE",
332 f"refill success on attempt {attempt + 1}: {inserted} payloads",
333 )
334 return
335
336 if attempt == attempts - 1:
337 context_logger.warning_with_context(
338 "CACHE",
339 "refill exhausted retry budget; cache will fill from live ticks",
340 )
341 return
342
343 try:
344 time.sleep(REFILL_BACKOFF_SEC[attempt])
345 except Exception:
346 return
347
348
[docs]
349def start_background_refill(
350 cache: OzCache,
351 manager,
352 window_hours: Optional[int] = None,
353) -> Optional[threading.Thread]:
354 """
355 Spawn the refill daemon. Returns the thread handle, or None when the
356 cache/manager aren't usable. Never blocks the caller.
357 """
358 if cache is None or not cache.enabled:
359 return None
360 if manager is None:
361 context_logger.warning_with_context(
362 "CACHE", "start_background_refill: manager missing, skipping"
363 )
364 return None
365
366 t = threading.Thread(
367 target=_refill_loop,
368 args=(cache, manager, int(window_hours or cache.hours)),
369 name="cache-refill",
370 daemon=True,
371 )
372 t.start()
373 context_logger.info_with_context("CACHE", "refill thread started")
374 return t