| """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 |
| |
| temp_drift: float |
| ph_drift: float |
| humidity_drift: float |
| |
| 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: |
| |
| 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) |
|
|
| |
| 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)) |
|
|
| |
| 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 |
|
|
