serversdown/seismo-relay
1
0
Fork 0
Code Issues 4 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: v0.12.0 — live device cache (_LiveCache) in sfm/server.py

Browse Source 
Ports the intelligent-caching branch concept to a plain Python in-memory
implementation — no SQLAlchemy, no extra DB table, no new dependencies.

_LiveCache (threading.Lock + dicts) caches:
  - device info: indefinite, invalidated by POST /device/config
  - events: keyed by (conn_key, device_event_count); count-probe fast path
    (~2s poll+count_events) avoids full downloads when nothing is new
  - monitor status: 30-second TTL, invalidated by monitor start/stop
  - waveforms: permanent per (conn_key, event_index)

All four cached endpoints accept ?force=true to bypass the cache.
Removes sfm/cache.py (SQLAlchemy experiment, now superseded).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Brian Harrison
2026-04-13 15:57:02 -04:00
committed by serversdown
parent 7883a31aa7
commit 5591d345d9
3 changed files with 194 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sfm/server.py
+158 -10
View File
@@ -37,6 +37,8 @@ from __future__ import annotations
import datetime
import logging
import sys
import threading
import time
from pathlib import Path
from typing import Optional
@@ -106,6 +108,136 @@ def _get_db() -> SeismoDb:
return _db
# ── Live device cache ─────────────────────────────────────────────────────────
# In-memory cache for live device data. Avoids re-dialing the device on every
# request when the data hasn't changed.
#
# Keyed by conn_key ("tcp:host:port" or "serial:port:baud").
# Does NOT persist across server restarts — this is purely an in-process cache
# to reduce TCP round-trips and cellular data usage.
#
# Invalidation rules:
# device_info — cached until POST /device/config marks it dirty
# events — cached by (conn_key, device_event_count); re-fetched when
# a quick count_events() probe shows new events on the device
# monitor_status — 30-second TTL (changes frequently during monitoring)
# waveforms — permanent (immutable once recorded; indexed by conn_key+idx)
#
# All endpoints accept ?force=true to bypass the cache and re-read from device.
_MONITOR_STATUS_TTL = 30.0 # seconds
class _LiveCache:
"""
Thread-safe in-memory cache for live SFM device data.
One singleton per server process.
"""
def __init__(self) -> None:
self._lock = threading.Lock()
# conn_key → serialised device info dict
self._device_info: dict[str, dict] = {}
# conn_key → (device_event_count_when_cached, [event dicts])
self._events: dict[str, tuple[int, list]] = {}
# conn_key → (fetched_at_unix, status_dict)
self._monitor_status: dict[str, tuple[float, dict]] = {}
# conn_key → bool (True = re-read device on next /device/info)
self._config_dirty: dict[str, bool] = {}
# (conn_key, event_index) → waveform dict (permanent)
self._waveforms: dict[tuple, dict] = {}
# ── Connection key ────────────────────────────────────────────────────────
@staticmethod
def make_conn_key(
host: Optional[str],
tcp_port: int,
port: Optional[str],
baud: int,
) -> str:
if host:
return f"tcp:{host}:{tcp_port}"
return f"serial:{port}:{baud}"
# ── Device info ───────────────────────────────────────────────────────────
def get_device_info(self, conn_key: str) -> Optional[dict]:
with self._lock:
if self._config_dirty.get(conn_key):
return None
return self._device_info.get(conn_key)
def set_device_info(self, conn_key: str, info: dict) -> None:
with self._lock:
self._device_info[conn_key] = info
self._config_dirty[conn_key] = False
# ── Events ────────────────────────────────────────────────────────────────
def get_events(self, conn_key: str, device_count: int) -> Optional[list]:
"""
Return cached events if the device's current event count matches what
we had when we last fetched. Returns None (cache miss) otherwise.
"""
with self._lock:
if self._config_dirty.get(conn_key):
return None
entry = self._events.get(conn_key)
if entry is None:
return None
cached_count, events = entry
return events if cached_count == device_count else None
def set_events(self, conn_key: str, device_count: int, events: list) -> None:
with self._lock:
self._events[conn_key] = (device_count, events)
# ── Monitor status ────────────────────────────────────────────────────────
def get_monitor_status(self, conn_key: str) -> Optional[dict]:
with self._lock:
entry = self._monitor_status.get(conn_key)
if entry is None:
return None
fetched_at, status = entry
if time.time() - fetched_at > _MONITOR_STATUS_TTL:
return None
return status
def set_monitor_status(self, conn_key: str, status: dict) -> None:
with self._lock:
self._monitor_status[conn_key] = (time.time(), status)
def invalidate_monitor_status(self, conn_key: str) -> None:
with self._lock:
self._monitor_status.pop(conn_key, None)
# ── Config dirty flag ─────────────────────────────────────────────────────
def mark_config_dirty(self, conn_key: str) -> None:
"""
Called after a successful POST /device/config write.
Forces next /device/info and /device/events to re-read from the device.
"""
with self._lock:
self._config_dirty[conn_key] = True
self._events.pop(conn_key, None)
# ── Waveforms (permanent cache) ───────────────────────────────────────────
def get_waveform(self, conn_key: str, index: int) -> Optional[dict]:
with self._lock:
return self._waveforms.get((conn_key, index))
def set_waveform(self, conn_key: str, index: int, waveform: dict) -> None:
with self._lock:
self._waveforms[(conn_key, index)] = waveform
_live_cache = _LiveCache()
# ── Serialisers ────────────────────────────────────────────────────────────────
# Plain dict helpers — avoids a Pydantic dependency in the library layer.
@@ -300,9 +432,9 @@ def webapp():
@app.get("/device/info")
def device_info(
port: Optional[str] = Query(None, description="Serial port (e.g. COM5, /dev/ttyUSB0)"),
baud: int = Query(38400, description="Serial baud rate (default 38400)"),
host: Optional[str] = Query(None, description="TCP host — modem IP or ACH relay (e.g. 203.0.113.5)"),
port: Optional[str] = Query(None, description="Serial port (e.g. COM5, /dev/ttyUSB0)"),
baud: int = Query(38400, description="Serial baud rate (default 38400)"),
host: Optional[str] = Query(None, description="TCP host — modem IP or ACH relay (e.g. 203.0.113.5)"),
tcp_port: int = Query(DEFAULT_TCP_PORT, description=f"TCP port (default {DEFAULT_TCP_PORT})"),
force: bool = Query(False, description="Bypass cache and re-read from device"),
) -> dict:
@@ -369,9 +501,9 @@ def device_connect(
@app.get("/device/events")
def device_events(
port: Optional[str] = Query(None, description="Serial port (e.g. COM5)"),
baud: int = Query(38400, description="Serial baud rate"),
host: Optional[str] = Query(None, description="TCP host — modem IP or ACH relay"),
port: Optional[str] = Query(None, description="Serial port (e.g. COM5)"),
baud: int = Query(38400, description="Serial baud rate"),
host: Optional[str] = Query(None, description="TCP host — modem IP or ACH relay"),
tcp_port: int = Query(DEFAULT_TCP_PORT, description=f"TCP port (default {DEFAULT_TCP_PORT})"),
debug: bool = Query(False, description="Include raw record hex for field-layout inspection"),
force: bool = Query(False, description="Bypass cache and re-download all events from device"),
@@ -382,6 +514,11 @@ def device_events(
Supply either *port* (serial) or *host* (TCP/modem).
**Caching:** a quick count_events() probe (~2s) is performed first. If the
device's event count matches the cached count, the cached response is returned
immediately without a full download. Pass ?force=true to skip this and always
re-download.
Pass debug=true to include raw_record_hex in each event — useful for
verifying field offsets against the protocol reference.
@@ -511,8 +648,16 @@ def device_events(
cache.set_events(conn_key, serialised)
cache.set_device_info(conn_key, _serialise_device_info(info))
serialised_info = _serialise_device_info(info)
serialised_events = [_serialise_event(ev, debug=debug) for ev in events]
# Update cache (skip if debug=True — raw hex blobs shouldn't pollute the cache)
if not debug:
_live_cache.set_device_info(conn_key, serialised_info)
_live_cache.set_events(conn_key, len(events), serialised_events)
return {
"device": _serialise_device_info(info),
"device": serialised_info,
"event_count": len(events),
"events": serialised,
}
@@ -584,9 +729,9 @@ def device_event(
@app.get("/device/event/{index}/waveform")
def device_event_waveform(
index: int,
port: Optional[str] = Query(None, description="Serial port (e.g. COM5)"),
baud: int = Query(38400, description="Serial baud rate"),
host: Optional[str] = Query(None, description="TCP host — modem IP or ACH relay"),
port: Optional[str] = Query(None, description="Serial port (e.g. COM5)"),
baud: int = Query(38400, description="Serial baud rate"),
host: Optional[str] = Query(None, description="TCP host — modem IP or ACH relay"),
tcp_port: int = Query(DEFAULT_TCP_PORT, description=f"TCP port (default {DEFAULT_TCP_PORT})"),
force: bool = Query(False, description="Bypass cache and re-download from device"),
) -> dict:
@@ -755,6 +900,7 @@ def device_config(
422 if neither port nor host is provided.
"""
changed = body.model_dump(exclude_none=True)
conn_key = _live_cache.make_conn_key(host, tcp_port, port, baud)
log.info("POST /device/config port=%s host=%s fields=%s", port, host, list(changed.keys()))
try:
@@ -876,6 +1022,7 @@ def device_monitor_start(
Sends SUB 0x96 and waits for ack SUB 0x69.
"""
conn_key = _live_cache.make_conn_key(host, tcp_port, port, baud)
with _build_client(port=port, baud=baud, host=host, tcp_port=tcp_port) as client:
try:
client.poll()
@@ -901,6 +1048,7 @@ def device_monitor_stop(
Sends SUB 0x97 and waits for ack SUB 0x68.
"""
conn_key = _live_cache.make_conn_key(host, tcp_port, port, baud)
with _build_client(port=port, baud=baud, host=host, tcp_port=tcp_port) as client:
try:
client.poll()