File size: 4,989 Bytes
10795db | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 | """Dev-only sensor simulator.
Drives the production ingest pipeline (`ingest_and_broadcast`) so that the
dashboard has live data without real IoT hardware. Disabled in production
via the ENABLE_SIMULATOR setting.
The simulator deliberately drifts one patient toward an SSI signature so the
'critical' UI states are exercisable on first launch.
"""
from __future__ import annotations
import asyncio
import logging
import random
from dataclasses import dataclass
from sqlalchemy import select
from core.database import SessionLocal
from core.realtime import ingest_and_broadcast
from models import Patient, Sensor
logger = logging.getLogger(__name__)
@dataclass
class _Stream:
patient_id: int
sensor_id: int | None
temperature_c: float
ph: float
humidity: float
# Slow drifts that determine the trajectory
temp_drift: float
ph_drift: float
humidity_drift: float
# Battery + signal state
battery_pct: float
signal_dbm: float
def _seed_streams() -> list[_Stream]:
"""Build one stream per patient using the seeded sensors. The first
patient is given a clearly deteriorating trajectory; the rest stay near
healing baselines with small noise."""
streams: list[_Stream] = []
with SessionLocal() as db:
patients = list(db.scalars(select(Patient).where(Patient.discharged.is_(False))).all())
for idx, p in enumerate(patients):
sensor = db.scalar(select(Sensor).where(Sensor.patient_id == p.id).limit(1))
if idx == 0:
# Deteriorating patient — slow drift toward infection signature
streams.append(
_Stream(
patient_id=p.id,
sensor_id=sensor.id if sensor else None,
temperature_c=37.1,
ph=7.05,
humidity=78.0,
temp_drift=+0.012,
ph_drift=+0.010,
humidity_drift=+0.05,
battery_pct=88.0,
signal_dbm=-62.0,
)
)
else:
streams.append(
_Stream(
patient_id=p.id,
sensor_id=sensor.id if sensor else None,
temperature_c=random.uniform(33.6, 34.8),
ph=random.uniform(5.8, 6.4),
humidity=random.uniform(70.0, 80.0),
temp_drift=0.0,
ph_drift=0.0,
humidity_drift=0.0,
battery_pct=random.uniform(85.0, 99.0),
signal_dbm=random.uniform(-70.0, -50.0),
)
)
return streams
def _step(stream: _Stream) -> None:
stream.temperature_c += stream.temp_drift + random.gauss(0.0, 0.05)
stream.ph += stream.ph_drift + random.gauss(0.0, 0.015)
stream.humidity += stream.humidity_drift + random.gauss(0.0, 0.5)
# Clamp to physically plausible ranges
stream.temperature_c = max(30.0, min(41.0, stream.temperature_c))
stream.ph = max(4.5, min(9.5, stream.ph))
stream.humidity = max(35.0, min(99.0, stream.humidity))
# Slow battery drain + tiny signal jitter
stream.battery_pct = max(5.0, stream.battery_pct - random.uniform(0.0, 0.05))
stream.signal_dbm = max(-90.0, min(-40.0, stream.signal_dbm + random.gauss(0.0, 1.2)))
async def run_simulator(interval_seconds: float = 6.0) -> None:
streams = _seed_streams()
if not streams:
logger.info("Simulator: no patients to drive, exiting")
return
logger.info("Simulator: driving %d streams every %.1fs", len(streams), interval_seconds)
try:
while True:
for stream in streams:
_step(stream)
with SessionLocal() as db:
sensor = db.get(Sensor, stream.sensor_id) if stream.sensor_id else None
if sensor is not None:
sensor.battery_pct = stream.battery_pct
sensor.signal_dbm = stream.signal_dbm
try:
patient = db.get(Patient, stream.patient_id)
await ingest_and_broadcast(
db,
patient_id=stream.patient_id,
temperature_c=stream.temperature_c,
ph=stream.ph,
humidity=stream.humidity,
sensor=sensor,
patient=patient,
)
except Exception:
logger.exception("Simulator ingest failed for patient %s", stream.patient_id)
await asyncio.sleep(interval_seconds)
except asyncio.CancelledError:
logger.info("Simulator: cancelled, exiting")
raise
|