---
-## Architecture
+## Simulated equipment
+
+| ID | Equipment | Rating |
+|---|---|---|
+| UTIL-1 | Utility service, 480 V | 1600 A |
+| GEN-1 | Standby diesel generator | 750 kW |
+| ATS-1 | Automatic transfer switch, open transition | 1600 A |
+| MSB-1 | Main switchboard, 480 V | 1600 A |
+| UPS-A, UPS-B | Double conversion UPS, A and B side | 400 kVA each |
+| PDU-A1, PDU-A2, PDU-B1, PDU-B2 | PDU, 480 to 208/120 V | 150 kVA each |
+| MECH-1 | Mechanical panel (CRAH units, pumps, fans) | 400 A |
-### Backend
-- **FastAPI** serves REST endpoints and a WebSocket broadcast channel.
-- **SQLite** (WAL mode) stores meter readings and alarm records.
-- **Simulator** generates realistic 3-phase power readings using sine-wave load profiles, Gaussian noise, and periodic fault injection (voltage sag, current spike, PF drop, frequency deviation).
-- **Alarm Engine** evaluates every reading against ISA 18.2-aligned thresholds and writes prioritized alarm records to the database.
-- **Collector** runs an async polling loop every 5 seconds, inserting readings and broadcasting to all connected WebSocket clients.
+Total load is about 450 kW with roughly 290 kW of IT load (PUE around 1.6). IT load follows a gentle daily profile, cooling load follows a simulated outdoor temperature, and every value drifts with slow random walks plus measurement noise.
-### Frontend
-- **React 18** single-page application built with **Vite**.
-- **Recharts** for trend charts and sparklines.
-- **Lucide React** for iconography.
-- **React Router v6** for client-side navigation.
-- WebSocket hook with automatic reconnection delivers sub-5-second live updates to all views.
+Demo scenarios:
-### Data Flow
+| Scenario | What happens |
+|---|---|
+| Utility outage | UPS on battery, generator start, ATS transfer, generator frequency dip, cooling restart, retransfer after 30 s of stable utility, generator cooldown |
+| Utility voltage sag | Utility drops to about 85 percent for a few seconds, UPS systems ride through on battery |
+| PDU overload | One PDU ramps past 80 and 90 percent load, which can also push its UPS past the 2N limit, then rebalances |
+| Low power factor | The cooling plant capacitor bank drops out and power factor falls below 0.85 |
+
+---
+
+## Architecture
```
-Simulator
- |
- v (every 5s)
-Collector Loop
- |---> SQLite (readings table)
- |---> Alarm Engine ---> SQLite (alarms table)
- |---> WebSocket Broadcast
- |
- v
- React Frontend
- (readings state)
- |
- Dashboard / One-Line / Meters / Alarms / Trending
+ Simulator or Modbus TCP meters
+ |
+ v every 2 s
+ Collector scan loop ------------> SQLite (history every 10 s, alarm journal)
+ |
+ +--> Alarm manager (ISA 18.2 state machine)
+ |
+ v
+ WebSocket /ws/live ------------> React front end
+ REST /api/* <------------ (acknowledge, shelve, history, demo events)
```
----
+- **Backend:** FastAPI. A single collector owns the scan loop, the latest readings, the alarm manager, and the set of WebSocket clients. On startup in simulation mode it generates 24 hours of history (with past events and simulated operator acknowledgements) in well under a second, so trends and the journal are never empty.
+- **WebSocket feed:** on connect the server sends a `snapshot` (meters, readings, alarms, summary, demo info). After that it pushes `readings` every scan and `alarms` whenever the alarm list changes or an operator acts. Clients can send `{"type": "ping"}` and get a `pong`.
+- **Front end:** React and Vite with hash based routes (`/#/alarms`), so any page can be refreshed or shared on a static host with no rewrite rules. The live feed is a single WebSocket with automatic reconnection and a stale connection watchdog. Chart heavy pages are loaded on demand, so the first load is about 72 kB of gzipped JavaScript.
+- **Cold starts:** the static front end never sleeps. While the API wakes up it shows a progress screen and pings `/api/health` (which is also what wakes the server), then switches to the live view as soon as the first snapshot arrives. If the connection drops later, the page keeps the last data, dims it, and shows "Reconnecting".
-## Tech Stack
+### Stack
-| Layer | Technology | Version |
-|-------------|-----------------------------------|-----------|
-| API Server | FastAPI | 0.111.0 |
-| ASGI Runner | Uvicorn (with standard extras) | 0.29.0 |
-| Database | SQLite (via stdlib sqlite3) | built-in |
-| Frontend | React | 18.3.1 |
-| Build Tool | Vite | 5.2.0 |
-| Charts | Recharts | 2.12.7 |
-| Icons | Lucide React | 0.379.0 |
-| Routing | React Router DOM | 6.23.1 |
+| Layer | Technology |
+|---|---|
+| API and WebSocket | Python 3.13, FastAPI 0.111, Uvicorn |
+| Storage | SQLite (WAL mode) |
+| Field protocol | Modbus TCP via pymodbus (optional) |
+| Front end | React 18, Vite 5, React Router 6 |
+| Charts and icons | Recharts (trend pages), inline SVG, Lucide |
+| Tests | pytest, FastAPI TestClient, Vitest |
+| Hosting | Render (static site plus web service), GitHub Actions CI |
---
-## Installation
+## Run it locally
+
+Requirements: Python 3.11 or newer and Node 18 or newer.
-### Backend
+**Backend** (terminal 1):
```bash
cd backend
+python -m venv .venv
+source .venv/bin/activate # Windows: .venv\Scripts\activate
pip install -r requirements.txt
uvicorn main:app --reload --port 8000
```
-The API will be available at `http://localhost:8000`. Interactive API docs are at `http://localhost:8000/docs`.
+API docs are at http://localhost:8000/docs.
-### Frontend
+**Front end** (terminal 2):
```bash
cd frontend
@@ -107,60 +148,82 @@ npm install
npm run dev
```
-The dashboard will be available at `http://localhost:5173`.
+Open http://localhost:5173. The Vite dev server proxies `/api` and `/ws` to the backend, so no configuration is needed.
----
+**Tests:**
-## Environment Variables
+```bash
+cd backend && pip install -r requirements-dev.txt && python -m pytest -q
+cd frontend && npm test
+```
-### Frontend
+The backend suite covers the simulator (physical consistency, the outage sequence, every scenario), the alarm state machine (every ISA 18.2 transition, deadband, delays, shelving, suppression, KPIs), the WebSocket feed (snapshot, streaming, ping, live alarm updates after operator actions), the REST API, and Modbus decoding. The front end suite covers the feed reducer and URL handling.
-| Variable | Description | Default |
-|----------|-------------|---------|
-| `VITE_API_URL` | Base URL of the backend API (used for HTTP requests and WebSocket connections) | `http://localhost:8000` |
+---
-Create a `frontend/.env` file (or copy `frontend/.env.example`) and set `VITE_API_URL` to your deployed backend URL before running `npm run build`. For example, when deploying to Render:
+## Configuration
-```
-VITE_API_URL=https://power-dashboard-api.onrender.com
-```
+Backend settings are environment variables (see `backend/config.py`):
-The WebSocket URL is derived automatically from `VITE_API_URL` by replacing `http://` with `ws://` (or `https://` with `wss://`).
+| Variable | Default | Purpose |
+|---|---|---|
+| `DATA_SOURCE` | `simulator` | `simulator` for the demo, `modbus` for real meters |
+| `MODBUS_CONFIG` | `modbus_config.json` | Device and register map when `DATA_SOURCE=modbus` |
+| `POLL_INTERVAL` | `2` | Scan interval in seconds |
+| `HISTORY_INTERVAL` | `10` | Seconds between stored history rows |
+| `RETENTION_DAYS` | `7` | Days of history and journal kept |
+| `BACKFILL_HOURS` | `24` | Simulated history generated at startup |
+| `SIM_EVENT_MIN`, `SIM_EVENT_MAX` | `150`, `360` | Seconds between automatic demo events |
+| `SIM_SEED` | random | Fixed seed for repeatable simulation |
+| `DEMO_EVENTS_ENABLED` | `true` | Allow visitors to trigger events from the UI |
+| `CORS_ORIGINS` | `*` | Comma separated allowed origins |
+| `DB_PATH` | `power_dashboard.db` | SQLite file |
----
+Front end: `VITE_API_URL` is the API base URL used at build time. It is optional. A production build without it points at the live demo API, `https://power-dashboard-api.onrender.com`, and the dev server without it proxies to the local backend.
+### Real hardware (Modbus TCP)
-## Features
+```bash
+cd backend
+pip install -r requirements-modbus.txt
+cp modbus_config.example.json modbus_config.json # edit hosts, unit ids, registers
+DATA_SOURCE=modbus uvicorn main:app --port 8000
+```
-1. **Live WebSocket feed** - All meter readings streamed to the browser with under 5-second latency, with automatic reconnection on disconnect.
-2. **8-meter hierarchy simulation** - Utility Feed, Main Switchgear, two distribution panels (IT and Cooling), and four sub-meters (Server Row A1/A2, CRAC Unit 1/2).
-3. **Realistic simulator** - Sine-wave load profiles, Gaussian noise, gradual drift, and stochastic fault injection producing voltage sags, current spikes, power factor drops, and frequency deviations.
-4. **ISA 18.2 alarm engine** - Four-parameter alarm evaluation (voltage, current, power factor, frequency) with four priority levels (Critical, High, Medium, Low) and full acknowledgement workflow.
-5. **One-line SVG diagram** - Interactive electrical distribution single-line diagram with live status color coding and per-node detail panel.
-6. **Per-meter analog gauges** - Arc gauges for voltage, current, power factor, and frequency with threshold-aware color states.
-7. **Historical trending** - Multi-meter, multi-parameter trend chart with 1h/24h/7d period selection and CSV export.
-8. **Alarm management table** - Active and history tabs with inline acknowledgement and ISA 18.2 priority row styling.
-9. **Automatic data pruning** - Readings older than 7 days are removed hourly; alarm history is retained indefinitely.
+Each device entry maps a meter ID to a host, unit ID, and the register for each value (holding, input, coil, or discrete; `float32`, `int16`, `uint16`, `int32`, `uint32`, or `bool`; optional scale and word order). A meter that stops answering is shown as "Comm Fail" and its alarms hold their last state until communication returns. The alarm engine, history, and UI work the same as in the demo.
---
-## Configuration
+## Deployment (Render)
+
+`render.yaml` is a Render Blueprint with two services that deploy from `main` on every merge:
-All tunable parameters are in `backend/config.py`:
+- **power-dashboard-api**: Python web service in `backend/`, started with `uvicorn main:app --host 0.0.0.0 --port $PORT`, health checked at `/api/health`, running the simulator.
+- **power-dashboard**: static site built from `frontend/` with `npm ci && npm run build`, published from `dist`. No dashboard settings are required: hash routes mean refreshes never 404, and the build falls back to the live API URL when `VITE_API_URL` is not set.
-- `POLL_INTERVAL` - Simulator polling rate in seconds (default: 5).
-- `WS_INTERVAL` - WebSocket broadcast interval in seconds (default: 2).
-- `DB_PATH` - SQLite database file path (default: `power_dashboard.db`).
-- `METERS` - Full meter hierarchy definition including nominal voltage, nominal current, CT ratio, and PT ratio.
-- `ALARM_THRESHOLDS` - Per-parameter alarm thresholds expressed as multipliers of nominal values or absolute values for frequency.
+Build filters mean a change under `backend/` only redeploys the API, a change under `frontend/` only redeploys the site, and documentation changes redeploy nothing. GitHub Actions runs both test suites and the production build on every pull request and push to `main`.
---
-## Future Enhancements
+## Project layout
+
+```
+backend/
+ main.py FastAPI app, REST endpoints, WebSocket
+ collector.py Scan loop, backfill, broadcast
+ simulator.py Data center power system model and scenarios
+ alarms.py ISA 18.2 alarm state machine
+ config.py Settings, topology, master alarm database
+ database.py SQLite history and alarm journal
+ modbus_source.py Modbus TCP polling (optional)
+ tests/
+frontend/
+ src/services/ WebSocket feed, REST client
+ src/pages/ Overview, One-Line, Meters, Alarms, Trends, System
+ src/components/ Wake screen, About panel, alarm banner, badges, charts
+render.yaml Render Blueprint
+```
+
+## License
-- **Real Modbus polling** - Replace the simulator with `pymodbus` to connect to physical Modbus TCP/RTU meters.
-- **BACnet integration** - Add BACnet/IP support for building automation system interoperability.
-- **Cloud deployment** - Containerize with Docker and deploy to cloud infrastructure with PostgreSQL replacing SQLite.
-- **ML predictive maintenance** - Apply time-series anomaly detection to identify degrading equipment before alarm thresholds are reached.
-- **User authentication** - Role-based access control for operator, engineer, and administrator tiers.
-- **Email/SMS notifications** - Push alarm notifications via SMTP or SMS gateway on critical priority events.
+MIT, see [LICENSE](LICENSE).
diff --git a/backend/alarm_engine.py b/backend/alarm_engine.py
deleted file mode 100644
index 0ba70b3..0000000
--- a/backend/alarm_engine.py
+++ /dev/null
@@ -1,73 +0,0 @@
-"""Alarm processing engine aligned with ISA 18.2 principles."""
-
-from typing import Dict, Any
-from config import ALARM_THRESHOLDS, METERS
-from database import upsert_alarm, clear_alarm
-
-# Build nominal lookup from config
-_NOMINALS = {m["id"]: m for m in METERS}
-
-
-def _get_nominal(meter_id: int, key: str) -> float:
- return _NOMINALS[meter_id][key]
-
-
-def process_reading(reading: Dict[str, Any]):
- """Evaluate a single reading against alarm thresholds."""
- mid = reading["meter_id"]
- nom_v = _get_nominal(mid, "nominal_voltage")
- nom_i = _get_nominal(mid, "nominal_current")
-
- v = reading["voltage"]
- i = reading["current"]
- pf = reading["power_factor"]
- freq = reading["frequency"]
-
- t = ALARM_THRESHOLDS
-
- # --- Voltage alarms ---
- if v > nom_v * t["voltage_high_critical"]:
- upsert_alarm(mid, "voltage", "High Voltage Critical",
- v, nom_v * t["voltage_high_critical"], priority=1)
- elif v > nom_v * t["voltage_high_warning"]:
- upsert_alarm(mid, "voltage", "High Voltage Warning",
- v, nom_v * t["voltage_high_warning"], priority=2)
- clear_alarm(mid, "voltage") # clear critical if only warning now
- elif v < nom_v * t["voltage_low_critical"]:
- upsert_alarm(mid, "voltage", "Low Voltage Critical",
- v, nom_v * t["voltage_low_critical"], priority=1)
- elif v < nom_v * t["voltage_low_warning"]:
- upsert_alarm(mid, "voltage", "Low Voltage Warning",
- v, nom_v * t["voltage_low_warning"], priority=2)
- else:
- clear_alarm(mid, "voltage")
-
- # --- Current alarms ---
- if i > nom_i * t["current_high_critical"]:
- upsert_alarm(mid, "current", "Overcurrent Critical",
- i, nom_i * t["current_high_critical"], priority=1)
- elif i > nom_i * t["current_high_warning"]:
- upsert_alarm(mid, "current", "High Current Warning",
- i, nom_i * t["current_high_warning"], priority=2)
- else:
- clear_alarm(mid, "current")
-
- # --- Power factor alarms ---
- if pf < t["pf_low_critical"]:
- upsert_alarm(mid, "power_factor", "Low Power Factor Critical",
- pf, t["pf_low_critical"], priority=1)
- elif pf < t["pf_low_warning"]:
- upsert_alarm(mid, "power_factor", "Low Power Factor Warning",
- pf, t["pf_low_warning"], priority=3)
- else:
- clear_alarm(mid, "power_factor")
-
- # --- Frequency alarms ---
- if freq > t["freq_high_critical"] or freq < t["freq_low_critical"]:
- upsert_alarm(mid, "frequency", "Frequency Deviation Critical",
- freq, 60.0, priority=1)
- elif freq > t["freq_high_warning"] or freq < t["freq_low_warning"]:
- upsert_alarm(mid, "frequency", "Frequency Deviation Warning",
- freq, 60.0, priority=3)
- else:
- clear_alarm(mid, "frequency")
diff --git a/backend/alarms.py b/backend/alarms.py
new file mode 100644
index 0000000..2ca21bc
--- /dev/null
+++ b/backend/alarms.py
@@ -0,0 +1,423 @@
+"""Alarm management aligned with ANSI/ISA-18.2.
+
+Each configured alarm point runs the ISA 18.2 alarm state model:
+
+ A NORMAL process normal, nothing to acknowledge
+ B UNACK process abnormal, alarm not yet acknowledged
+ C ACKED process abnormal, alarm acknowledged by the operator
+ D RTN_UNACK process returned to normal before anyone acknowledged it
+ F SHELVED temporarily removed from service by the operator (time limited)
+ G SUPPRESSED suppressed by design, based on plant state
+
+Transitions:
+
+ NORMAL --condition--> UNACK
+ UNACK --acknowledge--> ACKED
+ UNACK --clears--> RTN_UNACK
+ ACKED --clears--> NORMAL
+ RTN_UNACK --acknowledge--> NORMAL
+ RTN_UNACK --condition--> UNACK (re-alarm)
+ any of A to D --shelve--> SHELVED --unshelve or timer--> UNACK or NORMAL
+ any of A to D --design rule--> SUPPRESSED --rule clears--> UNACK or NORMAL
+
+Conditions use a deadband (hysteresis) and on/off delays so a noisy value
+near the setpoint does not chatter. Every transition is written to the
+alarm journal through registered listeners.
+
+This module is pure Python with no I/O so it can be tested in isolation.
+"""
+
+import threading
+from collections import deque
+from datetime import datetime, timezone
+from enum import Enum
+from typing import Any, Callable, Dict, Iterable, List, Optional
+
+from config import ALARM_PRIORITIES, SHELVE_MAX_MIN
+
+
+class AlarmState(str, Enum):
+ NORMAL = "NORMAL"
+ UNACK = "UNACK"
+ ACKED = "ACKED"
+ RTN_UNACK = "RTN_UNACK"
+ SHELVED = "SHELVED"
+ SUPPRESSED = "SUPPRESSED"
+
+
+STATE_LABELS = {
+ AlarmState.NORMAL: "Normal",
+ AlarmState.UNACK: "Unacknowledged",
+ AlarmState.ACKED: "Acknowledged",
+ AlarmState.RTN_UNACK: "Returned to normal, unacknowledged",
+ AlarmState.SHELVED: "Shelved",
+ AlarmState.SUPPRESSED: "Suppressed by design",
+}
+
+# States shown on the operator's active alarm list
+ANNUNCIATED = (AlarmState.UNACK, AlarmState.ACKED, AlarmState.RTN_UNACK)
+# States that still need an acknowledgement
+NEEDS_ACK = (AlarmState.UNACK, AlarmState.RTN_UNACK)
+
+# Journal event types
+EV_ALARM = "ALARM"
+EV_ACK = "ACK"
+EV_RTN = "RTN"
+EV_SHELVE = "SHELVE"
+EV_UNSHELVE = "UNSHELVE"
+EV_SUPPRESS = "SUPPRESS"
+EV_UNSUPPRESS = "UNSUPPRESS"
+
+# ISA 18.2 performance targets for annunciated alarms per operator per 10 minutes
+RATE_ACCEPTABLE = 1
+RATE_MANAGEABLE = 2
+RATE_FLOOD = 10
+RATE_WINDOW_S = 600
+
+
+class AlarmActionError(Exception):
+ """An operator action is not valid in the alarm's current state."""
+
+
+def _iso(epoch: float) -> str:
+ return datetime.fromtimestamp(epoch, timezone.utc).isoformat(timespec="milliseconds")
+
+
+class AlarmPoint:
+ """Runtime state for one configured alarm."""
+
+ def __init__(self, definition: Dict[str, Any], meter_name: str):
+ self.d = definition
+ self.tag: str = definition["tag"]
+ self.meter_name = meter_name
+ self.state = AlarmState.NORMAL
+ self.condition = False
+ self._on_count = 0
+ self._off_count = 0
+ self.value: Optional[float] = None
+ self.alarm_value: Optional[float] = None
+ self.activated_at: Optional[float] = None
+ self.ack_at: Optional[float] = None
+ self.ack_by: Optional[str] = None
+ self.rtn_at: Optional[float] = None
+ self.shelved_at: Optional[float] = None
+ self.shelved_until: Optional[float] = None
+ self.shelved_by: Optional[str] = None
+ self.shelve_reason: Optional[str] = None
+ self.suppress_reason: Optional[str] = None
+ self.evaluated = False
+
+ # Condition with deadband and debounce -----------------------------------
+
+ def _raw(self, value) -> bool:
+ kind, sp, db = self.d["kind"], self.d["setpoint"], self.d.get("deadband", 0.0)
+ if kind == "DISCRETE":
+ return bool(value) == bool(sp)
+ if kind in ("HI", "HIHI"):
+ return value > (sp - db if self.condition else sp)
+ if kind in ("LO", "LOLO"):
+ return value < (sp + db if self.condition else sp)
+ raise ValueError(f"Unknown alarm kind {kind}")
+
+ def in_alarm_now(self) -> bool:
+ """Current value against the setpoint, ignoring deadband and delays."""
+ if self.value is None:
+ return False
+ was, self.condition = self.condition, False
+ try:
+ return self._raw(self.value)
+ finally:
+ self.condition = was
+
+ def update_condition(self, value) -> bool:
+ """Feed a new value. Returns the debounced condition."""
+ self.value = value
+ raw = self._raw(value)
+ if raw:
+ self._off_count = 0
+ if not self.condition:
+ self._on_count += 1
+ if self._on_count > self.d.get("on_delay", 0):
+ self.condition = True
+ self._on_count = 0
+ else:
+ self._on_count = 0
+ if self.condition:
+ self._off_count += 1
+ if self._off_count > self.d.get("off_delay", 0):
+ self.condition = False
+ self._off_count = 0
+ return self.condition
+
+ def to_dict(self) -> Dict[str, Any]:
+ d = self.d
+ pr = d["priority"]
+ return {
+ "tag": self.tag,
+ "meter_id": d["meter_id"],
+ "meter_name": self.meter_name,
+ "message": d["message"],
+ "kind": d["kind"],
+ "parameter": d["parameter"],
+ "priority": pr,
+ "priority_label": ALARM_PRIORITIES[pr]["label"],
+ "state": self.state.value,
+ "state_label": STATE_LABELS[self.state],
+ "condition_active": self.condition,
+ "value": _num(self.value),
+ "alarm_value": _num(self.alarm_value),
+ "setpoint": d["setpoint"],
+ "unit": d.get("unit", ""),
+ "activated_at": _iso(self.activated_at) if self.activated_at else None,
+ "ack_at": _iso(self.ack_at) if self.ack_at else None,
+ "ack_by": self.ack_by,
+ "rtn_at": _iso(self.rtn_at) if self.rtn_at else None,
+ "shelved_at": _iso(self.shelved_at) if self.shelved_at else None,
+ "shelved_until": _iso(self.shelved_until) if self.shelved_until else None,
+ "shelved_by": self.shelved_by,
+ "shelve_reason": self.shelve_reason,
+ "suppress_reason": self.suppress_reason,
+ "consequence": d.get("consequence"),
+ "response": d.get("response"),
+ }
+
+
+def _num(v):
+ if isinstance(v, bool):
+ return int(v)
+ return v
+
+
+class AlarmManager:
+ """Evaluates alarm points against readings and handles operator actions."""
+
+ def __init__(self, definitions: Iterable[Dict[str, Any]], meter_names: Dict[str, str]):
+ self.points: Dict[str, AlarmPoint] = {
+ d["tag"]: AlarmPoint(d, meter_names.get(d["meter_id"], d["meter_id"]))
+ for d in definitions
+ }
+ self._listeners: List[Callable[[Dict[str, Any]], None]] = []
+ self._recent_alarms: deque = deque()
+ self._lock = threading.RLock()
+
+ # Listeners --------------------------------------------------------------
+
+ def add_listener(self, fn: Callable[[Dict[str, Any]], None]):
+ self._listeners.append(fn)
+
+ def _emit(self, p: AlarmPoint, event: str, now: float, operator: str = "system",
+ note: Optional[str] = None):
+ if event == EV_ALARM:
+ self._recent_alarms.append(now)
+ record = {
+ "epoch": now,
+ "timestamp": _iso(now),
+ "tag": p.tag,
+ "meter_id": p.d["meter_id"],
+ "meter_name": p.meter_name,
+ "event": event,
+ "state": p.state.value,
+ "priority": p.d["priority"],
+ "value": _num(p.value),
+ "setpoint": p.d["setpoint"],
+ "unit": p.d.get("unit", ""),
+ "message": p.d["message"],
+ "operator": operator,
+ "note": note,
+ }
+ for fn in self._listeners:
+ fn(record)
+
+ # Scan evaluation --------------------------------------------------------
+
+ def evaluate(self, readings: Dict[str, Dict[str, Any]], now: float) -> bool:
+ """Evaluate every alarm point against one scan. Returns True if anything changed."""
+ changed = False
+ with self._lock:
+ for p in self.points.values():
+ before = p.state
+ self._evaluate_point(p, readings, now)
+ changed = changed or p.state != before
+ return changed
+
+ def _evaluate_point(self, p: AlarmPoint, readings, now: float):
+ reading = readings.get(p.d["meter_id"])
+ value = reading.get(p.d["parameter"]) if reading else None
+ if value is None:
+ return # bad or missing data: hold the last state
+ condition = p.update_condition(value)
+
+ # Time limited shelving
+ if p.state == AlarmState.SHELVED:
+ if p.shelved_until is not None and now >= p.shelved_until:
+ self._unshelve(p, now, "system", "Shelve time expired")
+ return
+
+ # Designed (state based) suppression
+ rule = p.d.get("suppress")
+ suppressed = False
+ if rule:
+ src = readings.get(rule["meter_id"])
+ if src is not None and src.get(rule["parameter"]) is not None:
+ suppressed = bool(src.get(rule["parameter"])) == bool(rule["equals"])
+ else:
+ suppressed = p.state == AlarmState.SUPPRESSED
+ first_scan = not p.evaluated
+ p.evaluated = True
+ if suppressed and p.state != AlarmState.SUPPRESSED:
+ p.state = AlarmState.SUPPRESSED
+ p.suppress_reason = rule.get("reason")
+ if not first_scan: # starting up suppressed is not an event
+ self._emit(p, EV_SUPPRESS, now, note=p.suppress_reason)
+ return
+ if p.state == AlarmState.SUPPRESSED:
+ if suppressed:
+ return
+ p.suppress_reason = None
+ self._return_to_service(p, now, EV_UNSUPPRESS, "system", "Suppression condition cleared")
+ return
+
+ # Core ISA 18.2 state model
+ if p.state == AlarmState.NORMAL and condition:
+ self._annunciate(p, now)
+ elif p.state == AlarmState.RTN_UNACK and condition:
+ self._annunciate(p, now)
+ elif p.state == AlarmState.UNACK and not condition:
+ p.state = AlarmState.RTN_UNACK
+ p.rtn_at = now
+ self._emit(p, EV_RTN, now)
+ elif p.state == AlarmState.ACKED and not condition:
+ p.state = AlarmState.NORMAL
+ p.rtn_at = now
+ self._emit(p, EV_RTN, now)
+
+ def _annunciate(self, p: AlarmPoint, now: float):
+ p.state = AlarmState.UNACK
+ p.activated_at = now
+ p.alarm_value = p.value
+ p.ack_at = p.ack_by = p.rtn_at = None
+ self._emit(p, EV_ALARM, now)
+
+ def _return_to_service(self, p: AlarmPoint, now: float, event: str, operator: str, note: str):
+ # Re-evaluate from the live value so a stale debounced condition from
+ # while the alarm was out of service does not annunciate.
+ p.condition = p.in_alarm_now()
+ p.state = AlarmState.NORMAL
+ self._emit(p, event, now, operator, note)
+ if p.condition:
+ self._annunciate(p, now)
+
+ # Operator actions -------------------------------------------------------
+
+ def _get(self, tag: str) -> AlarmPoint:
+ if tag not in self.points:
+ raise KeyError(tag)
+ return self.points[tag]
+
+ def acknowledge(self, tag: str, now: float, operator: str = "operator") -> Dict[str, Any]:
+ with self._lock:
+ p = self._get(tag)
+ if p.state == AlarmState.UNACK:
+ p.state = AlarmState.ACKED
+ elif p.state == AlarmState.RTN_UNACK:
+ p.state = AlarmState.NORMAL
+ else:
+ raise AlarmActionError(f"{tag} is {STATE_LABELS[p.state].lower()}, nothing to acknowledge")
+ p.ack_at, p.ack_by = now, operator
+ self._emit(p, EV_ACK, now, operator)
+ return p.to_dict()
+
+ def shelve(self, tag: str, minutes: float, now: float, operator: str = "operator",
+ reason: Optional[str] = None) -> Dict[str, Any]:
+ with self._lock:
+ p = self._get(tag)
+ if p.state in (AlarmState.SHELVED, AlarmState.SUPPRESSED):
+ raise AlarmActionError(f"{tag} is {STATE_LABELS[p.state].lower()}")
+ if not 0 < minutes <= SHELVE_MAX_MIN:
+ raise AlarmActionError(f"Shelve time must be between 1 and {SHELVE_MAX_MIN} minutes")
+ p.state = AlarmState.SHELVED
+ p.shelved_at = now
+ p.shelved_until = now + minutes * 60
+ p.shelved_by = operator
+ p.shelve_reason = (reason or "").strip() or "Shelved by operator"
+ self._emit(p, EV_SHELVE, now, operator, f"{p.shelve_reason} ({minutes:g} min)")
+ return p.to_dict()
+
+ def unshelve(self, tag: str, now: float, operator: str = "operator") -> Dict[str, Any]:
+ with self._lock:
+ p = self._get(tag)
+ if p.state != AlarmState.SHELVED:
+ raise AlarmActionError(f"{tag} is not shelved")
+ self._unshelve(p, now, operator, "Unshelved by operator")
+ return p.to_dict()
+
+ def _unshelve(self, p: AlarmPoint, now: float, operator: str, note: str):
+ p.shelved_at = p.shelved_until = p.shelved_by = p.shelve_reason = None
+ self._return_to_service(p, now, EV_UNSHELVE, operator, note)
+
+ # Views ------------------------------------------------------------------
+
+ def alarm_list(self) -> List[Dict[str, Any]]:
+ """Every alarm point that is not NORMAL, highest priority and newest first."""
+ with self._lock:
+ items = [p.to_dict() for p in self.points.values() if p.state != AlarmState.NORMAL]
+ items.sort(key=lambda a: (a["priority"], -(_ts(a["activated_at"]))))
+ return items
+
+ def definitions(self) -> List[Dict[str, Any]]:
+ out = []
+ for p in self.points.values():
+ d = {k: v for k, v in p.d.items() if k != "suppress"}
+ d["meter_name"] = p.meter_name
+ d["priority_label"] = ALARM_PRIORITIES[d["priority"]]["label"]
+ d["suppression"] = p.d["suppress"]["reason"] if p.d.get("suppress") else None
+ out.append(d)
+ return out
+
+ def alarm_rate(self, now: float) -> int:
+ with self._lock:
+ while self._recent_alarms and self._recent_alarms[0] < now - RATE_WINDOW_S:
+ self._recent_alarms.popleft()
+ return len(self._recent_alarms)
+
+ def summary(self, now: float) -> Dict[str, Any]:
+ with self._lock:
+ states = [p.state for p in self.points.values()]
+ annunciated = [p for p in self.points.values() if p.state in ANNUNCIATED]
+ unacked = [p for p in annunciated if p.state in NEEDS_ACK]
+ rate = self.alarm_rate(now)
+ top = min(unacked, key=lambda p: (p.d["priority"], -(p.activated_at or 0)), default=None)
+
+ if rate >= RATE_FLOOD:
+ rate_status = "Flood"
+ elif rate > RATE_MANAGEABLE:
+ rate_status = "Above target"
+ elif rate > RATE_ACCEPTABLE:
+ rate_status = "Manageable"
+ else:
+ rate_status = "Acceptable"
+
+ by_priority = {str(k): 0 for k in ALARM_PRIORITIES}
+ for p in annunciated:
+ by_priority[str(p.d["priority"])] += 1
+
+ return {
+ "annunciated": len(annunciated),
+ "in_alarm": states.count(AlarmState.UNACK) + states.count(AlarmState.ACKED),
+ "unacknowledged": len(unacked),
+ "acknowledged": states.count(AlarmState.ACKED),
+ "rtn_unack": states.count(AlarmState.RTN_UNACK),
+ "shelved": states.count(AlarmState.SHELVED),
+ "suppressed": states.count(AlarmState.SUPPRESSED),
+ "by_priority": by_priority,
+ "alarm_rate_10min": rate,
+ "rate_status": rate_status,
+ "highest_unacked": top.to_dict() if top else None,
+ "configured": len(states),
+ }
+
+
+def _ts(iso: Optional[str]) -> float:
+ if not iso:
+ return 0.0
+ return datetime.fromisoformat(iso).timestamp()
diff --git a/backend/collector.py b/backend/collector.py
index 7469420..2e95fc4 100644
--- a/backend/collector.py
+++ b/backend/collector.py
@@ -1,68 +1,265 @@
-"""Background data collection service."""
+"""Scan loop: read the data source, evaluate alarms, store history, push to clients."""
import asyncio
import json
import logging
-from datetime import datetime, timezone
-from typing import Set
+import random
+import time
+from typing import Any, Callable, Dict, List, Optional, Set
-from simulator import DeviceSimulator
-from database import insert_reading, prune_old_readings
-from alarm_engine import process_reading
-from config import POLL_INTERVAL
+import config
+import database
+from alarms import ANNUNCIATED, NEEDS_ACK, AlarmManager, AlarmState
+from simulator import SCENARIOS, FacilitySimulator
logger = logging.getLogger(__name__)
-_simulator = DeviceSimulator()
-_latest_readings: dict = {}
-_websocket_clients: Set = set()
+BACKFILL_STEP_S = 60.0
+BACKFILL_EVENT_STEP_S = 5.0 # finer steps while an event plays out
+SIM_OPERATOR = "operator (simulated)"
-def get_latest_readings_cache() -> dict:
- return _latest_readings
+class Collector:
+ """Owns the live state of the system and the set of WebSocket clients."""
+ def __init__(self, source: str = None, clock: Callable[[], float] = time.time):
+ self.source = (source or config.DATA_SOURCE)
+ self.clock = clock
+ self.started_at = clock()
+ self.meter_names = {m["id"]: m["name"] for m in config.METERS}
+ self.alarms = AlarmManager(config.ALARM_POINTS, self.meter_names)
+ self.alarms.add_listener(database.insert_alarm_event)
+ self.latest: Dict[str, Dict[str, Any]] = {}
+ self.clients: Set[Any] = set()
+ self.ready = False
+ self.scan_count = 0
+ self._last_history = 0.0
+ self._last_prune = 0.0
+ self._task: Optional[asyncio.Task] = None
-def register_ws_client(ws):
- _websocket_clients.add(ws)
+ self.sim: Optional[FacilitySimulator] = None
+ self.modbus = None
+ if self.source == "modbus":
+ from modbus_source import ModbusSource, load_config
+ self.modbus = ModbusSource(load_config(config.MODBUS_CONFIG))
+ else:
+ self.source = "simulator"
+ seed = int(config.SIM_SEED) if config.SIM_SEED else None
+ self.sim = FacilitySimulator(
+ seed=seed,
+ event_interval=(config.SIM_EVENT_MIN, config.SIM_EVENT_MAX),
+ first_event_delay=config.SIM_FIRST_EVENT,
+ )
+ @property
+ def simulated(self) -> bool:
+ return self.sim is not None
-def unregister_ws_client(ws):
- _websocket_clients.discard(ws)
+ # ------------------------------------------------------------------
+ # Startup
+ # ------------------------------------------------------------------
+ def initialize(self):
+ """Prepare storage. A simulated demo starts fresh with backfilled history."""
+ database.init_db(reset=self.simulated)
+ if self.simulated and config.BACKFILL_HOURS > 0:
+ t0 = time.perf_counter()
+ self.backfill(config.BACKFILL_HOURS, self.clock())
+ logger.info("Backfilled %.0f h of simulated history in %.2f s",
+ config.BACKFILL_HOURS, time.perf_counter() - t0)
-async def broadcast_readings():
- """Push latest readings to all connected WebSocket clients."""
- if not _websocket_clients or not _latest_readings:
- return
- payload = json.dumps({"type": "readings", "data": list(_latest_readings.values())})
- dead = set()
- for ws in _websocket_clients:
+ def backfill(self, hours: float, now: float):
+ """Run the simulator over the past few hours so trends and history are populated.
+
+ Events are spaced out the way a real site sees them, and a simulated
+ operator acknowledges alarms shortly after they annunciate.
+ """
+ sim = self.sim
+ live_interval, live_first = sim.event_interval, sim.first_event_delay
+ sim.event_interval, sim.first_event_delay = (1800.0, 5400.0), 600.0
+ rng = random.Random(sim.rng.random())
+
+ events: List[Dict[str, Any]] = []
+ mgr = AlarmManager(config.ALARM_POINTS, self.meter_names)
+ mgr.add_listener(events.append)
+ rows: List[Dict[str, Any]] = []
+
+ t = now - hours * 3600
+ quiet_from = now - 900 # no new events in the last 15 minutes
+ while t < now:
+ sim.auto_events = t < quiet_from
+ readings = sim.step(t)
+ rows.extend(readings.values())
+ mgr.evaluate(readings, t)
+ for p in mgr.points.values():
+ if p.state in NEEDS_ACK and p.activated_at is not None \
+ and t - p.activated_at >= rng.uniform(30, 150):
+ mgr.acknowledge(p.tag, t, SIM_OPERATOR)
+ t += BACKFILL_EVENT_STEP_S if sim.scenario else BACKFILL_STEP_S
+
+ database.insert_readings(rows)
+ database.insert_alarm_events(events)
+ sim.auto_events = True
+ sim.event_interval, sim.first_event_delay = live_interval, live_first
+ sim.schedule_event_at(now + live_first) # first live event soon after startup
+ self._last_history = now
+
+ async def start(self):
+ self.initialize()
+ await self.scan_once()
+ self.ready = True
+ self._task = asyncio.create_task(self._run())
+
+ async def stop(self):
+ if self._task:
+ self._task.cancel()
+ try:
+ await self._task
+ except asyncio.CancelledError:
+ pass
+ if self.modbus:
+ self.modbus.close()
+
+ # ------------------------------------------------------------------
+ # Scan loop
+ # ------------------------------------------------------------------
+
+ async def _run(self):
+ while True:
+ started = self.clock()
+ await asyncio.sleep(max(0.05, config.POLL_INTERVAL - (self.clock() - started)))
+ try:
+ await self.scan_once()
+ except Exception:
+ logger.exception("Scan failed")
+
+ async def scan_once(self, now: Optional[float] = None) -> Dict[str, Dict[str, Any]]:
+ now = self.clock() if now is None else now
+ if self.sim:
+ readings = self.sim.step(now)
+ else:
+ readings = await self.modbus.read_all(now)
+ self.latest.update(readings)
+ self.scan_count += 1
+
+ alarms_changed = self.alarms.evaluate(readings, now)
+
+ if now - self._last_history >= config.HISTORY_INTERVAL:
+ database.insert_readings(r for r in readings.values() if r.get("voltage") is not None)
+ self._last_history = now
+ if now - self._last_prune >= 3600:
+ database.prune(config.RETENTION_DAYS, now)
+ self._last_prune = now
+
+ await self.broadcast({"type": "readings", "data": {
+ "readings": list(readings.values()),
+ "system": self.system_status(now),
+ }})
+ if alarms_changed or self.scan_count % 5 == 0:
+ await self.broadcast_alarms(now)
+ return readings
+
+ # ------------------------------------------------------------------
+ # Views
+ # ------------------------------------------------------------------
+
+ def alarm_payload(self, now: Optional[float] = None) -> Dict[str, Any]:
+ now = self.clock() if now is None else now
+ return {"alarms": self.alarms.alarm_list(), "summary": self.alarms.summary(now)}
+
+ def system_status(self, now: Optional[float] = None) -> Dict[str, Any]:
+ now = self.clock() if now is None else now
+ r = self.latest
+ facility_kw = r.get("MSB-1", {}).get("power_kw") or 0.0
+ it_kw = sum((r.get(m["id"], {}).get("power_kw") or 0.0)
+ for m in config.METERS if m["type"] == "pdu")
+ ups_runtime = [r[u].get("battery_runtime_min") for u in ("UPS-A", "UPS-B")
+ if u in r and r[u].get("battery_runtime_min") is not None]
+
+ on_battery = any(r.get(u, {}).get("on_battery") for u in ("UPS-A", "UPS-B"))
+
+ in_alarm = [p for p in self.alarms.points.values()
+ if p.state in (AlarmState.UNACK, AlarmState.ACKED)]
+ priorities = {p.d["priority"] for p in in_alarm}
+ if 1 in priorities:
+ health = "critical"
+ elif 2 in priorities:
+ health = "warning"
+ elif priorities:
+ health = "caution"
+ else:
+ health = "normal"
+
+ online = sum(1 for v in r.values() if v.get("comm_ok", True))
+ return {
+ "health": health,
+ "data_source": self.source,
+ "simulated": self.simulated,
+ "facility_kw": round(facility_kw, 1),
+ "it_kw": round(it_kw, 1),
+ "pue": round(facility_kw / it_kw, 2) if facility_kw > 0 and it_kw > 0 and not on_battery else None,
+ "power_source": r.get("ATS-1", {}).get("position"),
+ "utility_status": r.get("UTIL-1", {}).get("status"),
+ "generator_status": r.get("GEN-1", {}).get("status"),
+ "ups_min_runtime_min": min(ups_runtime) if ups_runtime else None,
+ "annunciated_alarms": sum(1 for p in self.alarms.points.values() if p.state in ANNUNCIATED),
+ "meter_count": len(config.METERS),
+ "meters_online": online,
+ "scenario": self.sim.scenario_info(now) if self.sim else None,
+ "server_time": now,
+ }
+
+ def demo_info(self) -> Dict[str, Any]:
+ return {
+ "simulated": self.simulated,
+ "events_enabled": self.simulated and config.DEMO_EVENTS_ENABLED,
+ "scenarios": [{"name": k, **v} for k, v in SCENARIOS.items()] if self.simulated else [],
+ "poll_interval_s": config.POLL_INTERVAL,
+ "version": config.VERSION,
+ }
+
+ def snapshot(self) -> Dict[str, Any]:
+ now = self.clock()
+ return {
+ "meters": config.METERS,
+ "readings": list(self.latest.values()),
+ "system": self.system_status(now),
+ "demo": self.demo_info(),
+ **self.alarm_payload(now),
+ }
+
+ # ------------------------------------------------------------------
+ # Demo events
+ # ------------------------------------------------------------------
+
+ def trigger_event(self, name: str) -> Dict[str, Any]:
+ if not self.simulated or not config.DEMO_EVENTS_ENABLED:
+ raise PermissionError("Demo events are only available in simulation mode")
+ now = self.clock()
+ info = self.sim.trigger(name, now)
+ return info
+
+ # ------------------------------------------------------------------
+ # WebSocket clients
+ # ------------------------------------------------------------------
+
+ def register(self, ws):
+ self.clients.add(ws)
+
+ def unregister(self, ws):
+ self.clients.discard(ws)
+
+ async def _send(self, ws, text: str):
try:
- await ws.send_text(payload)
+ await asyncio.wait_for(ws.send_text(text), timeout=5)
except Exception:
- dead.add(ws)
- for ws in dead:
- _websocket_clients.discard(ws)
+ self.clients.discard(ws)
+ async def broadcast(self, message: Dict[str, Any]):
+ if not self.clients:
+ return
+ text = json.dumps(message, separators=(",", ":"))
+ await asyncio.gather(*(self._send(ws, text) for ws in list(self.clients)))
-async def collection_loop():
- """Main polling loop: read all meters, store, process alarms, broadcast."""
- prune_counter = 0
- while True:
- try:
- readings = _simulator.read_all()
- for meter_id, reading in readings.items():
- insert_reading(reading)
- process_reading(reading)
- _latest_readings[meter_id] = reading
-
- await broadcast_readings()
- prune_counter += 1
- if prune_counter >= 720: # every hour at 5s interval
- prune_old_readings(days=7)
- prune_counter = 0
-
- except Exception as exc:
- logger.error("Collection loop error: %s", exc)
-
- await asyncio.sleep(POLL_INTERVAL)
+ async def broadcast_alarms(self, now: Optional[float] = None):
+ await self.broadcast({"type": "alarms", "data": self.alarm_payload(now)})
diff --git a/backend/config.py b/backend/config.py
index 36cd81e..dbb2b36 100644
--- a/backend/config.py
+++ b/backend/config.py
@@ -1,126 +1,291 @@
-"""System configuration for Power Dashboard."""
+"""System configuration for Power Dashboard.
-# Polling interval in seconds
-POLL_INTERVAL = 5
+Runtime settings come from environment variables so the same code runs the
+public demo (simulated data) and a real site (Modbus TCP polling).
+"""
-# Database
-DB_PATH = "power_dashboard.db"
+import os
-# WebSocket broadcast interval (seconds)
-WS_INTERVAL = 2
+VERSION = "1.1.0"
+
+
+def _env_bool(name: str, default: bool) -> bool:
+ raw = os.getenv(name)
+ if raw is None:
+ return default
+ return raw.strip().lower() in ("1", "true", "yes", "on")
+
+
+def _env_float(name: str, default: float) -> float:
+ raw = os.getenv(name)
+ return float(raw) if raw not in (None, "") else default
+
+
+# ---------------------------------------------------------------------------
+# Runtime settings
+# ---------------------------------------------------------------------------
+
+# "simulator" (default, used by the public demo) or "modbus"
+DATA_SOURCE = os.getenv("DATA_SOURCE", "simulator").strip().lower()
+
+# Path to the Modbus device map (only used when DATA_SOURCE=modbus)
+MODBUS_CONFIG = os.getenv("MODBUS_CONFIG", "modbus_config.json")
+
+# Scan rate for the live feed and alarm evaluation (seconds)
+POLL_INTERVAL = _env_float("POLL_INTERVAL", 2.0)
+
+# How often a scan is written to the history table (seconds)
+HISTORY_INTERVAL = _env_float("HISTORY_INTERVAL", 10.0)
+
+# SQLite database file
+DB_PATH = os.getenv("DB_PATH", "power_dashboard.db")
+
+# Days of readings kept before pruning
+RETENTION_DAYS = int(_env_float("RETENTION_DAYS", 7))
+
+# Simulator: hours of history generated on startup so trends are never empty
+BACKFILL_HOURS = _env_float("BACKFILL_HOURS", 24)
+
+# Simulator: seconds between automatic demo events (min, max)
+SIM_EVENT_MIN = _env_float("SIM_EVENT_MIN", 150)
+SIM_EVENT_MAX = _env_float("SIM_EVENT_MAX", 360)
+
+# Simulator: delay before the first live event after startup (seconds)
+SIM_FIRST_EVENT = _env_float("SIM_FIRST_EVENT", 25)
+
+# Simulator: optional fixed random seed for repeatable runs
+SIM_SEED = os.getenv("SIM_SEED")
+
+# Allow visitors to trigger demo events from the UI (simulator only)
+DEMO_EVENTS_ENABLED = _env_bool("DEMO_EVENTS_ENABLED", True)
+
+# Comma separated list of allowed browser origins, "*" for any
+CORS_ORIGINS = [o.strip() for o in os.getenv("CORS_ORIGINS", "*").split(",") if o.strip()]
+
+# ISA 18.2 shelving is time limited. Durations offered to the operator (minutes)
+SHELVE_DURATIONS_MIN = [5, 15, 60]
+SHELVE_MAX_MIN = 60
+
+# Nominal system frequency
+NOMINAL_FREQUENCY = 60.0
+
+
+# ---------------------------------------------------------------------------
+# Facility topology
+#
+# A small 2N data center: utility service and standby generator feed an
+# automatic transfer switch, which feeds the main switchboard. Two UPS
+# systems (A side and B side) each feed two PDUs. Mechanical (cooling) load
+# is fed directly from the switchboard.
+# ---------------------------------------------------------------------------
-# Meter hierarchy definition
METERS = [
{
- "id": 1,
- "name": "Utility Feed",
- "description": "Main utility supply point",
- "type": "utility",
- "parent_id": None,
- "nominal_voltage": 480.0,
- "nominal_current": 800.0,
- "ct_ratio": 800,
- "pt_ratio": 1,
- },
- {
- "id": 2,
- "name": "Main Switchgear",
- "description": "Main distribution switchgear",
- "type": "switchgear",
- "parent_id": 1,
- "nominal_voltage": 480.0,
- "nominal_current": 750.0,
- "ct_ratio": 750,
- "pt_ratio": 1,
- },
- {
- "id": 3,
- "name": "Panel A",
- "description": "Distribution Panel A - IT Load",
- "type": "panel",
- "parent_id": 2,
- "nominal_voltage": 480.0,
- "nominal_current": 400.0,
- "ct_ratio": 400,
- "pt_ratio": 1,
- },
- {
- "id": 4,
- "name": "Panel B",
- "description": "Distribution Panel B - Cooling Load",
- "type": "panel",
- "parent_id": 2,
- "nominal_voltage": 480.0,
- "nominal_current": 350.0,
- "ct_ratio": 350,
- "pt_ratio": 1,
- },
- {
- "id": 5,
- "name": "Server Row A1",
- "description": "Server row PDU - Rack group A1",
- "type": "meter",
- "parent_id": 3,
- "nominal_voltage": 480.0,
- "nominal_current": 120.0,
- "ct_ratio": 120,
- "pt_ratio": 1,
- },
- {
- "id": 6,
- "name": "Server Row A2",
- "description": "Server row PDU - Rack group A2",
- "type": "meter",
- "parent_id": 3,
- "nominal_voltage": 480.0,
- "nominal_current": 120.0,
- "ct_ratio": 120,
- "pt_ratio": 1,
- },
- {
- "id": 7,
- "name": "CRAC Unit 1",
- "description": "Computer Room Air Conditioner 1",
- "type": "meter",
- "parent_id": 4,
- "nominal_voltage": 480.0,
- "nominal_current": 180.0,
- "ct_ratio": 180,
- "pt_ratio": 1,
- },
- {
- "id": 8,
- "name": "CRAC Unit 2",
- "description": "Computer Room Air Conditioner 2",
- "type": "meter",
- "parent_id": 4,
- "nominal_voltage": 480.0,
- "nominal_current": 170.0,
- "ct_ratio": 170,
- "pt_ratio": 1,
+ "id": "UTIL-1", "name": "Utility Service", "type": "utility", "parent_id": None,
+ "description": "Utility service entrance, 480 V secondary of the service transformer",
+ "nominal_voltage": 480.0, "rated_current": 1600.0, "rated_kva": 1330.0,
+ },
+ {
+ "id": "GEN-1", "name": "Standby Generator", "type": "generator", "parent_id": None,
+ "description": "750 kW diesel standby generator",
+ "nominal_voltage": 480.0, "rated_current": 1128.0, "rated_kva": 937.5,
+ },
+ {
+ "id": "ATS-1", "name": "Automatic Transfer Switch", "type": "ats", "parent_id": None,
+ "description": "Open transition ATS, utility normal / generator emergency",
+ "nominal_voltage": 480.0, "rated_current": 1600.0, "rated_kva": 1330.0,
+ },
+ {
+ "id": "MSB-1", "name": "Main Switchboard", "type": "switchgear", "parent_id": "ATS-1",
+ "description": "480 V main switchboard",
+ "nominal_voltage": 480.0, "rated_current": 1600.0, "rated_kva": 1330.0,
+ },
+ {
+ "id": "UPS-A", "name": "UPS A", "type": "ups", "parent_id": "MSB-1",
+ "description": "400 kVA double conversion UPS, A side",
+ "nominal_voltage": 480.0, "rated_current": 481.0, "rated_kva": 400.0,
+ },
+ {
+ "id": "UPS-B", "name": "UPS B", "type": "ups", "parent_id": "MSB-1",
+ "description": "400 kVA double conversion UPS, B side",
+ "nominal_voltage": 480.0, "rated_current": 481.0, "rated_kva": 400.0,
+ },
+ {
+ "id": "MECH-1", "name": "Mechanical Panel", "type": "panel", "parent_id": "MSB-1",
+ "description": "Cooling plant: CRAH units, pumps, condenser fans",
+ "nominal_voltage": 480.0, "rated_current": 400.0, "rated_kva": 332.0,
+ },
+ {
+ "id": "PDU-A1", "name": "PDU A1", "type": "pdu", "parent_id": "UPS-A",
+ "description": "150 kVA PDU, 480 to 208/120 V, server rows 1 and 2",
+ "nominal_voltage": 208.0, "rated_current": 416.0, "rated_kva": 150.0,
+ },
+ {
+ "id": "PDU-A2", "name": "PDU A2", "type": "pdu", "parent_id": "UPS-A",
+ "description": "150 kVA PDU, 480 to 208/120 V, server rows 3 and 4",
+ "nominal_voltage": 208.0, "rated_current": 416.0, "rated_kva": 150.0,
+ },
+ {
+ "id": "PDU-B1", "name": "PDU B1", "type": "pdu", "parent_id": "UPS-B",
+ "description": "150 kVA PDU, 480 to 208/120 V, server rows 1 and 2",
+ "nominal_voltage": 208.0, "rated_current": 416.0, "rated_kva": 150.0,
+ },
+ {
+ "id": "PDU-B2", "name": "PDU B2", "type": "pdu", "parent_id": "UPS-B",
+ "description": "150 kVA PDU, 480 to 208/120 V, server rows 3 and 4",
+ "nominal_voltage": 208.0, "rated_current": 416.0, "rated_kva": 150.0,
},
]
-# Alarm thresholds per meter type
-ALARM_THRESHOLDS = {
- "voltage_high_critical": 1.10, # 110% of nominal
- "voltage_high_warning": 1.05, # 105% of nominal
- "voltage_low_warning": 0.95, # 95% of nominal
- "voltage_low_critical": 0.90, # 90% of nominal
- "current_high_critical": 1.00, # 100% of nominal (full load)
- "current_high_warning": 0.85, # 85% of nominal
- "pf_low_warning": 0.85, # Power factor warning
- "pf_low_critical": 0.75, # Power factor critical
- "freq_high_warning": 60.5, # Hz
- "freq_high_critical": 61.0, # Hz
- "freq_low_warning": 59.5, # Hz
- "freq_low_critical": 59.0, # Hz
-}
+METERS_BY_ID = {m["id"]: m for m in METERS}
+
+
+# ---------------------------------------------------------------------------
+# Alarm priorities (ISA 18.2: priority reflects consequence severity and the
+# time available for the operator to respond)
+# ---------------------------------------------------------------------------
-# ISA 18.2 Priority labels
ALARM_PRIORITIES = {
- 1: "Critical",
- 2: "High",
- 3: "Medium",
- 4: "Low",
+ 1: {"label": "Critical", "response": "Respond immediately (under 5 min)"},
+ 2: {"label": "High", "response": "Respond promptly (under 15 min)"},
+ 3: {"label": "Medium", "response": "Respond within the shift (under 1 h)"},
+ 4: {"label": "Low", "response": "Respond when practical"},
}
+
+
+# ---------------------------------------------------------------------------
+# Master alarm database
+#
+# Each alarm was rationalized: one alarm per abnormal condition that needs an
+# operator response, with a documented consequence and corrective action.
+#
+# kind HI / HIHI / LO / LOLO compare an analog value to a setpoint.
+# DISCRETE alarms when the value is true.
+# deadband hysteresis in engineering units before the alarm can clear
+# on_delay consecutive scans in alarm before annunciating (anti-chatter)
+# off_delay consecutive scans normal before returning to normal
+# suppress designed (state based) suppression rule, see alarms.py
+# ---------------------------------------------------------------------------
+
+def _pdu_alarms(pdu: str):
+ name = METERS_BY_ID[pdu]["name"]
+ return [
+ {
+ "tag": f"{pdu}.LOAD.HI", "meter_id": pdu, "parameter": "load_pct", "kind": "HI",
+ "setpoint": 80.0, "deadband": 3.0, "priority": 3, "on_delay": 2, "off_delay": 2,
+ "unit": "%",
+ "message": f"{name} load high",
+ "consequence": "Reduced margin before the PDU main breaker trips.",
+ "response": "Check for new rack load or a failed partner PDU. Plan load rebalancing.",
+ },
+ {
+ "tag": f"{pdu}.LOAD.HIHI", "meter_id": pdu, "parameter": "load_pct", "kind": "HIHI",
+ "setpoint": 90.0, "deadband": 3.0, "priority": 2, "on_delay": 2, "off_delay": 2,
+ "unit": "%",
+ "message": f"{name} load very high",
+ "consequence": "PDU breaker trip drops every server on this side.",
+ "response": "Shed or move load now. Confirm the partner PDU can carry the racks.",
+ },
+ ]
+
+
+def _ups_alarms(ups: str):
+ name = METERS_BY_ID[ups]["name"]
+ return [
+ {
+ "tag": f"{ups}.ONBATT", "meter_id": ups, "parameter": "on_battery", "kind": "DISCRETE",
+ "setpoint": 1, "deadband": 0, "priority": 1, "on_delay": 0, "off_delay": 1,
+ "unit": "",
+ "message": f"{name} on battery",
+ "consequence": "Critical load is on limited battery runtime.",
+ "response": "Verify generator start and ATS transfer. Watch battery runtime.",
+ },
+ {
+ "tag": f"{ups}.LOAD.HI", "meter_id": ups, "parameter": "load_pct", "kind": "HI",
+ "setpoint": 50.0, "deadband": 2.0, "priority": 2, "on_delay": 2, "off_delay": 2,
+ "unit": "%",
+ "message": f"{name} load above 2N limit",
+ "consequence": "If the other side fails, this UPS cannot carry the full IT load.",
+ "response": "Rebalance IT load between A and B sides below 50 percent each.",
+ },
+ {
+ "tag": f"{ups}.BATT.LO", "meter_id": ups, "parameter": "battery_soc", "kind": "LO",
+ "setpoint": 60.0, "deadband": 5.0, "priority": 2, "on_delay": 1, "off_delay": 2,
+ "unit": "%",
+ "message": f"{name} battery charge low",
+ "consequence": "Reduced ride through time on the next utility event.",
+ "response": "Confirm the rectifier is charging. Avoid planned transfers until recharged.",
+ },
+ ]
+
+
+ALARM_POINTS = [
+ {
+ "tag": "UTIL-1.FAIL", "meter_id": "UTIL-1", "parameter": "source_failed", "kind": "DISCRETE",
+ "setpoint": 1, "deadband": 0, "priority": 2, "on_delay": 0, "off_delay": 1,
+ "unit": "",
+ "message": "Utility power failure",
+ "consequence": "Facility is running on UPS battery and standby generator.",
+ "response": "Confirm generator pickup. Notify the utility and facility manager.",
+ },
+ {
+ "tag": "UTIL-1.V.LO", "meter_id": "UTIL-1", "parameter": "voltage", "kind": "LO",
+ "setpoint": 432.0, "deadband": 4.8, "priority": 3, "on_delay": 0, "off_delay": 1,
+ "unit": "V",
+ "message": "Utility voltage low (sag)",
+ "consequence": "UPS may transfer to battery. Motors may drop out.",
+ "response": "Check UPS mode and mechanical equipment restart.",
+ "suppress": {"meter_id": "UTIL-1", "parameter": "source_failed", "equals": True,
+ "reason": "Utility failure alarm is active (cascade suppression)"},
+ },
+ {
+ "tag": "GEN-1.F.LO", "meter_id": "GEN-1", "parameter": "frequency", "kind": "LO",
+ "setpoint": 59.5, "deadband": 0.1, "priority": 3, "on_delay": 0, "off_delay": 1,
+ "unit": "Hz",
+ "message": "Generator frequency low",
+ "consequence": "UPS may stay on battery if frequency is out of tolerance.",
+ "response": "Check generator governor and block load step.",
+ "suppress": {"meter_id": "GEN-1", "parameter": "running", "equals": False,
+ "reason": "Generator not running (state based suppression)"},
+ },
+ {
+ "tag": "ATS-1.EMERG", "meter_id": "ATS-1", "parameter": "on_emergency", "kind": "DISCRETE",
+ "setpoint": 1, "deadband": 0, "priority": 3, "on_delay": 0, "off_delay": 1,
+ "unit": "",
+ "message": "ATS on emergency source",
+ "consequence": "Facility depends on generator fuel and runtime.",
+ "response": "Check fuel level and generator load. Track utility restoration.",
+ },
+ {
+ "tag": "MSB-1.LOAD.HI", "meter_id": "MSB-1", "parameter": "load_pct", "kind": "HI",
+ "setpoint": 80.0, "deadband": 3.0, "priority": 3, "on_delay": 3, "off_delay": 3,
+ "unit": "%",
+ "message": "Main switchboard load high",
+ "consequence": "Main breaker approaching continuous rating.",
+ "response": "Review facility load growth and defer non critical load.",
+ },
+ *_ups_alarms("UPS-A"),
+ *_ups_alarms("UPS-B"),
+ {
+ "tag": "MECH-1.PF.LO", "meter_id": "MECH-1", "parameter": "power_factor", "kind": "LO",
+ "setpoint": 0.85, "deadband": 0.01, "priority": 4, "on_delay": 2, "off_delay": 2,
+ "unit": "",
+ "message": "Mechanical panel power factor low",
+ "consequence": "Higher current and possible utility power factor penalty.",
+ "response": "Check the capacitor bank and VFD bypass status.",
+ "suppress": {"meter_id": "MECH-1", "parameter": "energized", "equals": False,
+ "reason": "Panel de-energized (state based suppression)"},
+ },
+ {
+ "tag": "MECH-1.LOAD.HI", "meter_id": "MECH-1", "parameter": "load_pct", "kind": "HI",
+ "setpoint": 85.0, "deadband": 3.0, "priority": 3, "on_delay": 3, "off_delay": 3,
+ "unit": "%",
+ "message": "Mechanical panel load high",
+ "consequence": "Cooling plant near feeder capacity.",
+ "response": "Check for stuck dampers or a failed CRAH unit forcing others to full speed.",
+ },
+ *_pdu_alarms("PDU-A1"),
+ *_pdu_alarms("PDU-A2"),
+ *_pdu_alarms("PDU-B1"),
+ *_pdu_alarms("PDU-B2"),
+]
diff --git a/backend/database.py b/backend/database.py
index 9d30137..cc67d2b 100644
--- a/backend/database.py
+++ b/backend/database.py
@@ -1,214 +1,158 @@
-"""Database layer for Power Dashboard using SQLite."""
+"""SQLite storage for meter history and the alarm event journal."""
import sqlite3
-import json
-from datetime import datetime, timezone
-from typing import List, Dict, Any, Optional
-from config import DB_PATH, METERS
-
-
-def get_connection() -> sqlite3.Connection:
- conn = sqlite3.connect(DB_PATH)
- conn.row_factory = sqlite3.Row
- conn.execute("PRAGMA journal_mode=WAL")
- conn.execute("PRAGMA synchronous=NORMAL")
- return conn
-
-
-def init_db():
- """Initialize database schema."""
- conn = get_connection()
- cur = conn.cursor()
-
- cur.executescript("""
- CREATE TABLE IF NOT EXISTS meters (
- id INTEGER PRIMARY KEY,
- name TEXT NOT NULL,
- description TEXT,
- type TEXT,
- parent_id INTEGER,
- nominal_voltage REAL,
- nominal_current REAL,
- ct_ratio INTEGER,
- pt_ratio INTEGER,
- created_at TEXT DEFAULT (datetime('now'))
- );
-
- CREATE TABLE IF NOT EXISTS readings (
- id INTEGER PRIMARY KEY AUTOINCREMENT,
- meter_id INTEGER NOT NULL,
- voltage REAL,
- current REAL,
- power_kw REAL,
- power_factor REAL,
- energy_kwh REAL,
- frequency REAL,
- demand_kw REAL,
- timestamp TEXT NOT NULL,
- FOREIGN KEY (meter_id) REFERENCES meters(id)
- );
-
- CREATE INDEX IF NOT EXISTS idx_readings_meter_ts
- ON readings(meter_id, timestamp DESC);
-
- CREATE TABLE IF NOT EXISTS alarms (
- id INTEGER PRIMARY KEY AUTOINCREMENT,
- meter_id INTEGER NOT NULL,
- parameter TEXT NOT NULL,
- condition TEXT NOT NULL,
- value REAL NOT NULL,
- threshold REAL NOT NULL,
- priority INTEGER NOT NULL,
- state TEXT NOT NULL DEFAULT 'active',
- acknowledged INTEGER NOT NULL DEFAULT 0,
- ack_time TEXT,
- ack_by TEXT,
- raised_at TEXT NOT NULL,
- cleared_at TEXT,
- FOREIGN KEY (meter_id) REFERENCES meters(id)
- );
-
- CREATE INDEX IF NOT EXISTS idx_alarms_state
- ON alarms(state, meter_id);
- """)
-
- # Seed meter definitions
- for m in METERS:
- cur.execute("""
- INSERT OR IGNORE INTO meters
- (id, name, description, type, parent_id, nominal_voltage, nominal_current, ct_ratio, pt_ratio)
- VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
- """, (m["id"], m["name"], m["description"], m["type"], m["parent_id"],
- m["nominal_voltage"], m["nominal_current"], m["ct_ratio"], m["pt_ratio"]))
-
- conn.commit()
- conn.close()
-
-
-def insert_reading(reading: Dict[str, Any]):
- conn = get_connection()
- conn.execute("""
- INSERT INTO readings (meter_id, voltage, current, power_kw, power_factor,
- energy_kwh, frequency, demand_kw, timestamp)
- VALUES (:meter_id, :voltage, :current, :power_kw, :power_factor,
- :energy_kwh, :frequency, :demand_kw, :timestamp)
- """, reading)
- conn.commit()
- conn.close()
-
-
-def get_latest_readings() -> List[Dict]:
- conn = get_connection()
- rows = conn.execute("""
- SELECT r.*, m.name AS meter_name, m.type, m.parent_id,
- m.nominal_voltage, m.nominal_current
- FROM readings r
- JOIN meters m ON m.id = r.meter_id
- WHERE r.id IN (
- SELECT MAX(id) FROM readings GROUP BY meter_id
+import threading
+import time
+from typing import Any, Dict, Iterable, List, Optional
+
+import config
+
+HISTORY_FIELDS = ("voltage", "current", "power_kw", "power_factor", "frequency",
+ "energy_kwh", "demand_kw", "load_pct")
+
+PERIODS = {"1h": 3600, "24h": 86400, "7d": 7 * 86400}
+HISTORY_POINTS = 360 # target points per trend after downsampling
+
+_lock = threading.Lock()
+_conn: Optional[sqlite3.Connection] = None
+
+
+def _db() -> sqlite3.Connection:
+ global _conn
+ if _conn is None:
+ _conn = sqlite3.connect(config.DB_PATH, check_same_thread=False)
+ _conn.row_factory = sqlite3.Row
+ if config.DB_PATH != ":memory:":
+ _conn.execute("PRAGMA journal_mode=WAL")
+ _conn.execute("PRAGMA synchronous=NORMAL")
+ return _conn
+
+
+def close():
+ global _conn
+ with _lock:
+ if _conn is not None:
+ _conn.close()
+ _conn = None
+
+
+def init_db(reset: bool = False):
+ """Create tables. reset=True clears stored data (used for a fresh demo)."""
+ with _lock:
+ db = _db()
+ if reset:
+ db.executescript("DROP TABLE IF EXISTS meter_readings; DROP TABLE IF EXISTS alarm_events;")
+ db.executescript("""
+ CREATE TABLE IF NOT EXISTS meter_readings (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ meter_id TEXT NOT NULL,
+ epoch REAL NOT NULL,
+ timestamp TEXT NOT NULL,
+ voltage REAL,
+ current REAL,
+ power_kw REAL,
+ power_factor REAL,
+ frequency REAL,
+ energy_kwh REAL,
+ demand_kw REAL,
+ load_pct REAL
+ );
+ CREATE INDEX IF NOT EXISTS idx_meter_readings_meter_epoch
+ ON meter_readings(meter_id, epoch);
+
+ CREATE TABLE IF NOT EXISTS alarm_events (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ epoch REAL NOT NULL,
+ timestamp TEXT NOT NULL,
+ tag TEXT NOT NULL,
+ meter_id TEXT NOT NULL,
+ meter_name TEXT,
+ event TEXT NOT NULL,
+ state TEXT NOT NULL,
+ priority INTEGER NOT NULL,
+ value REAL,
+ setpoint REAL,
+ unit TEXT,
+ message TEXT,
+ operator TEXT,
+ note TEXT
+ );
+ CREATE INDEX IF NOT EXISTS idx_alarm_events_epoch ON alarm_events(epoch);
+ """)
+ db.commit()
+
+
+def insert_readings(readings: Iterable[Dict[str, Any]]):
+ rows = [
+ (r["meter_id"], r["epoch"], r["timestamp"], *(r.get(f) for f in HISTORY_FIELDS))
+ for r in readings
+ ]
+ with _lock:
+ db = _db()
+ db.executemany(
+ f"INSERT INTO meter_readings (meter_id, epoch, timestamp, {', '.join(HISTORY_FIELDS)}) "
+ f"VALUES (?, ?, ?, {', '.join('?' * len(HISTORY_FIELDS))})",
+ rows,
)
- ORDER BY m.id
- """).fetchall()
- conn.close()
+ db.commit()
+
+
+def get_meter_history(meter_id: str, period: str = "1h", now: Optional[float] = None) -> List[Dict]:
+ """Return averaged history buckets for one meter, oldest first."""
+ span = PERIODS[period]
+ now = time.time() if now is None else now
+ bucket = max(span / HISTORY_POINTS, 1.0)
+ avgs = ", ".join(f"ROUND(AVG({f}), 3) AS {f}" for f in HISTORY_FIELDS)
+ with _lock:
+ rows = _db().execute(
+ f"""
+ SELECT CAST(epoch / ? AS INTEGER) * ? AS epoch, {avgs}
+ FROM meter_readings
+ WHERE meter_id = ? AND epoch >= ?
+ GROUP BY CAST(epoch / ? AS INTEGER)
+ ORDER BY epoch ASC
+ """,
+ (bucket, bucket, meter_id, now - span, bucket),
+ ).fetchall()
return [dict(r) for r in rows]
-def get_meter_history(meter_id: int, period: str = "1h") -> List[Dict]:
- period_map = {"1h": "1 hours", "24h": "24 hours", "7d": "7 days"}
- interval = period_map.get(period, "1 hours")
- conn = get_connection()
- rows = conn.execute(f"""
- SELECT * FROM readings
- WHERE meter_id = ?
- AND timestamp >= datetime('now', '-{interval}')
- ORDER BY timestamp ASC
- """, (meter_id,)).fetchall()
- conn.close()
- return [dict(r) for r in rows]
-
-
-def get_all_meters() -> List[Dict]:
- conn = get_connection()
- rows = conn.execute("SELECT * FROM meters ORDER BY id").fetchall()
- conn.close()
- return [dict(r) for r in rows]
+def insert_alarm_event(event: Dict[str, Any]):
+ insert_alarm_events([event])
-def upsert_alarm(meter_id: int, parameter: str, condition: str,
- value: float, threshold: float, priority: int) -> Optional[int]:
- """Insert new alarm if one does not already exist for this meter+parameter."""
- conn = get_connection()
- existing = conn.execute("""
- SELECT id FROM alarms
- WHERE meter_id=? AND parameter=? AND state='active'
- """, (meter_id, parameter)).fetchone()
-
- if existing:
- conn.close()
- return existing["id"]
-
- now = datetime.now(timezone.utc).isoformat()
- cur = conn.execute("""
- INSERT INTO alarms (meter_id, parameter, condition, value, threshold, priority, state, raised_at)
- VALUES (?, ?, ?, ?, ?, ?, 'active', ?)
- """, (meter_id, parameter, condition, value, threshold, priority, now))
- alarm_id = cur.lastrowid
- conn.commit()
- conn.close()
- return alarm_id
-
-
-def clear_alarm(meter_id: int, parameter: str):
- """Clear active alarm when condition resolves."""
- now = datetime.now(timezone.utc).isoformat()
- conn = get_connection()
- conn.execute("""
- UPDATE alarms SET state='cleared', cleared_at=?
- WHERE meter_id=? AND parameter=? AND state='active'
- """, (now, meter_id, parameter))
- conn.commit()
- conn.close()
-
-
-def get_active_alarms() -> List[Dict]:
- conn = get_connection()
- rows = conn.execute("""
- SELECT a.*, m.name AS meter_name
- FROM alarms a JOIN meters m ON m.id = a.meter_id
- WHERE a.state = 'active'
- ORDER BY a.priority ASC, a.raised_at DESC
- """).fetchall()
- conn.close()
- return [dict(r) for r in rows]
-
-
-def get_alarm_history(limit: int = 200) -> List[Dict]:
- conn = get_connection()
- rows = conn.execute("""
- SELECT a.*, m.name AS meter_name
- FROM alarms a JOIN meters m ON m.id = a.meter_id
- ORDER BY a.raised_at DESC LIMIT ?
- """, (limit,)).fetchall()
- conn.close()
+def insert_alarm_events(events: Iterable[Dict[str, Any]]):
+ cols = ("epoch", "timestamp", "tag", "meter_id", "meter_name", "event", "state",
+ "priority", "value", "setpoint", "unit", "message", "operator", "note")
+ rows = [tuple(e.get(c) for c in cols) for e in events]
+ if not rows:
+ return
+ with _lock:
+ db = _db()
+ db.executemany(
+ f"INSERT INTO alarm_events ({', '.join(cols)}) VALUES ({', '.join('?' * len(cols))})",
+ rows,
+ )
+ db.commit()
+
+
+def get_alarm_events(limit: int = 200, tag: Optional[str] = None) -> List[Dict]:
+ sql = "SELECT * FROM alarm_events"
+ args: list = []
+ if tag:
+ sql += " WHERE tag = ?"
+ args.append(tag)
+ sql += " ORDER BY epoch DESC, id DESC LIMIT ?"
+ args.append(limit)
+ with _lock:
+ rows = _db().execute(sql, args).fetchall()
return [dict(r) for r in rows]
-def acknowledge_alarm(alarm_id: int, ack_by: str = "operator"):
- now = datetime.now(timezone.utc).isoformat()
- conn = get_connection()
- conn.execute("""
- UPDATE alarms SET acknowledged=1, ack_time=?, ack_by=?
- WHERE id=?
- """, (now, ack_by, alarm_id))
- conn.commit()
- conn.close()
-
-
-def prune_old_readings(days: int = 7):
- """Remove readings older than specified days to manage DB size."""
- conn = get_connection()
- conn.execute("""
- DELETE FROM readings WHERE timestamp < datetime('now', ? || ' days')
- """, (f"-{days}",))
- conn.commit()
- conn.close()
+def prune(days: int, now: Optional[float] = None):
+ cutoff = (time.time() if now is None else now) - days * 86400
+ with _lock:
+ db = _db()
+ db.execute("DELETE FROM meter_readings WHERE epoch < ?", (cutoff,))
+ db.execute("DELETE FROM alarm_events WHERE epoch < ?", (cutoff,))
+ db.commit()
diff --git a/backend/main.py b/backend/main.py
index 4bfdf6c..f69b25b 100644
--- a/backend/main.py
+++ b/backend/main.py
@@ -1,142 +1,234 @@
"""FastAPI application entry point for Power Dashboard."""
-import asyncio
import json
import logging
from contextlib import asynccontextmanager
from typing import Optional
-from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException
+from fastapi import FastAPI, HTTPException, Request, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel, Field
-from database import (
- init_db, get_all_meters, get_latest_readings, get_meter_history,
- get_active_alarms, get_alarm_history, acknowledge_alarm
-)
-from collector import (
- collection_loop, register_ws_client, unregister_ws_client,
- get_latest_readings_cache, broadcast_readings
-)
-from config import METERS, WS_INTERVAL
+import config
+import database
+from alarms import AlarmActionError
+from collector import Collector
-logging.basicConfig(level=logging.INFO)
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
logger = logging.getLogger(__name__)
@asynccontextmanager
async def lifespan(app: FastAPI):
- init_db()
- task = asyncio.create_task(collection_loop())
- logger.info("Power Dashboard collection loop started")
+ collector = Collector()
+ app.state.collector = collector
+ await collector.start()
+ logger.info("Power Dashboard %s started (data source: %s)", config.VERSION, collector.source)
yield
- task.cancel()
+ await collector.stop()
+ database.close()
app = FastAPI(
title="Power Dashboard API",
- description="Real-time Electrical Power Monitoring System",
- version="1.0.0",
+ description="Real-time data center electrical power monitoring with ISA 18.2 alarm management",
+ version=config.VERSION,
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
- allow_origins=["*"],
+ allow_origins=config.CORS_ORIGINS,
allow_credentials=False,
allow_methods=["*"],
allow_headers=["*"],
)
-# REST Endpoints
+def _collector(request: Request) -> Collector:
+ return request.app.state.collector
+
+
+def _operator(name: Optional[str]) -> str:
+ name = (name or "").strip()[:40]
+ return name or "operator"
+
+
+class OperatorAction(BaseModel):
+ operator: Optional[str] = None
+
+
+class ShelveRequest(BaseModel):
+ minutes: float = Field(15, gt=0, le=config.SHELVE_MAX_MIN)
+ reason: Optional[str] = Field(None, max_length=200)
+ operator: Optional[str] = None
+
+
+# ---------------------------------------------------------------------------
+# Health and configuration
+# ---------------------------------------------------------------------------
+
+@app.get("/api/health")
+def health(request: Request):
+ c = _collector(request)
+ return {
+ "status": "ok" if c.ready else "starting",
+ "version": config.VERSION,
+ "data_source": c.source,
+ "uptime_s": round(c.clock() - c.started_at, 1),
+ "clients": len(c.clients),
+ }
+
+
+@app.get("/api/config")
+def get_config(request: Request):
+ c = _collector(request)
+ return {
+ "version": config.VERSION,
+ "data_source": c.source,
+ "poll_interval_s": config.POLL_INTERVAL,
+ "history_interval_s": config.HISTORY_INTERVAL,
+ "retention_days": config.RETENTION_DAYS,
+ "shelve_durations_min": config.SHELVE_DURATIONS_MIN,
+ "shelve_max_min": config.SHELVE_MAX_MIN,
+ "priorities": config.ALARM_PRIORITIES,
+ "meters": config.METERS,
+ }
+
+
+# ---------------------------------------------------------------------------
+# Meters
+# ---------------------------------------------------------------------------
@app.get("/api/meters")
-def list_meters():
- meters = get_all_meters()
- readings_cache = get_latest_readings_cache()
- result = []
- for m in meters:
- reading = readings_cache.get(m["id"], {})
- result.append({**m, "current_reading": reading})
- return result
+def list_meters(request: Request):
+ latest = _collector(request).latest
+ return [{**m, "current_reading": latest.get(m["id"], {})} for m in config.METERS]
@app.get("/api/meters/{meter_id}")
-def meter_detail(meter_id: int):
- meters = {m["id"]: m for m in get_all_meters()}
- if meter_id not in meters:
+def meter_detail(meter_id: str, request: Request):
+ if meter_id not in config.METERS_BY_ID:
raise HTTPException(status_code=404, detail="Meter not found")
- reading = get_latest_readings_cache().get(meter_id, {})
- return {**meters[meter_id], "current_reading": reading}
+ reading = _collector(request).latest.get(meter_id, {})
+ return {**config.METERS_BY_ID[meter_id], "current_reading": reading}
@app.get("/api/meters/{meter_id}/history")
-def meter_history(meter_id: int, period: str = "1h"):
- if period not in ("1h", "24h", "7d"):
+def meter_history(meter_id: str, period: str = "1h"):
+ if meter_id not in config.METERS_BY_ID:
+ raise HTTPException(status_code=404, detail="Meter not found")
+ if period not in database.PERIODS:
raise HTTPException(status_code=400, detail="period must be 1h, 24h, or 7d")
- return get_meter_history(meter_id, period)
+ return database.get_meter_history(meter_id, period)
+
+
+@app.get("/api/system/status")
+def system_status(request: Request):
+ return _collector(request).system_status()
+
+# ---------------------------------------------------------------------------
+# Alarms (ISA 18.2)
+# ---------------------------------------------------------------------------
@app.get("/api/alarms")
-def active_alarms():
- return get_active_alarms()
+def alarms(request: Request):
+ return _collector(request).alarm_payload()
@app.get("/api/alarms/history")
-def alarms_history(limit: int = 200):
- return get_alarm_history(limit)
+def alarm_history(limit: int = 200, tag: Optional[str] = None):
+ return database.get_alarm_events(min(max(limit, 1), 1000), tag)
-@app.post("/api/alarms/{alarm_id}/acknowledge")
-def ack_alarm(alarm_id: int, ack_by: str = "operator"):
- acknowledge_alarm(alarm_id, ack_by)
- return {"status": "acknowledged", "alarm_id": alarm_id}
+@app.get("/api/alarms/definitions")
+def alarm_definitions(request: Request):
+ return _collector(request).alarms.definitions()
-@app.get("/api/system/status")
-def system_status():
- cache = get_latest_readings_cache()
- active = get_active_alarms()
-
- total_power = sum(r.get("power_kw", 0) for r in cache.values())
- pf_values = [r["power_factor"] for r in cache.values() if r.get("power_factor")]
- avg_pf = round(sum(pf_values) / len(pf_values), 4) if pf_values else 0.0
-
- critical_count = sum(1 for a in active if a["priority"] == 1)
- high_count = sum(1 for a in active if a["priority"] == 2)
-
- if critical_count > 0:
- health = "critical"
- elif high_count > 0:
- health = "warning"
- elif len(active) > 0:
- health = "caution"
- else:
- health = "normal"
+async def _alarm_action(request: Request, fn, *args, **kwargs):
+ c = _collector(request)
+ try:
+ result = fn(*args, now=c.clock(), **kwargs)
+ except KeyError:
+ raise HTTPException(status_code=404, detail="Alarm tag not found")
+ except AlarmActionError as exc:
+ raise HTTPException(status_code=409, detail=str(exc))
+ await c.broadcast_alarms()
+ return result
- return {
- "health": health,
- "total_power_kw": round(total_power, 2),
- "avg_power_factor": avg_pf,
- "active_alarm_count": len(active),
- "critical_alarm_count": critical_count,
- "high_alarm_count": high_count,
- "meter_count": len(METERS),
- "meters_online": len(cache),
- }
+@app.post("/api/alarms/{tag}/acknowledge")
+async def acknowledge(tag: str, request: Request, body: Optional[OperatorAction] = None):
+ c = _collector(request)
+ return await _alarm_action(request, c.alarms.acknowledge, tag,
+ operator=_operator(body.operator if body else None))
-# WebSocket
+
+@app.post("/api/alarms/{tag}/shelve")
+async def shelve(tag: str, body: ShelveRequest, request: Request):
+ c = _collector(request)
+ return await _alarm_action(request, c.alarms.shelve, tag, body.minutes,
+ operator=_operator(body.operator), reason=body.reason)
+
+
+@app.post("/api/alarms/{tag}/unshelve")
+async def unshelve(tag: str, request: Request, body: Optional[OperatorAction] = None):
+ c = _collector(request)
+ return await _alarm_action(request, c.alarms.unshelve, tag,
+ operator=_operator(body.operator if body else None))
+
+
+# ---------------------------------------------------------------------------
+# Demo (simulation only)
+# ---------------------------------------------------------------------------
+
+@app.get("/api/demo")
+def demo(request: Request):
+ c = _collector(request)
+ return {**c.demo_info(), "active_scenario": c.sim.scenario_info(c.clock()) if c.sim else None}
+
+
+@app.post("/api/demo/events/{name}")
+async def trigger_event(name: str, request: Request):
+ c = _collector(request)
+ try:
+ info = c.trigger_event(name)
+ except PermissionError as exc:
+ raise HTTPException(status_code=403, detail=str(exc))
+ except KeyError:
+ raise HTTPException(status_code=404, detail="Unknown scenario")
+ except RuntimeError as exc:
+ raise HTTPException(status_code=409, detail=str(exc))
+ return info
+
+
+# ---------------------------------------------------------------------------
+# WebSocket live feed
+#
+# On connect the server sends a "snapshot" with everything the UI needs. After
+# that it pushes "readings" every scan and "alarms" whenever the alarm list
+# changes. Clients may send {"type": "ping"} and get {"type": "pong"} back.
+# ---------------------------------------------------------------------------
@app.websocket("/ws/live")
async def websocket_live(websocket: WebSocket):
+ c: Collector = websocket.app.state.collector
await websocket.accept()
- register_ws_client(websocket)
+ await websocket.send_text(json.dumps({"type": "snapshot", "data": c.snapshot()},
+ separators=(",", ":")))
+ c.register(websocket)
try:
while True:
- # Keep connection alive; data is pushed by broadcast_readings
- await asyncio.sleep(1)
+ text = await websocket.receive_text()
+ try:
+ msg = json.loads(text)
+ except ValueError:
+ continue
+ if isinstance(msg, dict) and msg.get("type") == "ping":
+ await websocket.send_text('{"type":"pong"}')
except WebSocketDisconnect:
pass
finally:
- unregister_ws_client(websocket)
+ c.unregister(websocket)
diff --git a/backend/modbus_config.example.json b/backend/modbus_config.example.json
new file mode 100644
index 0000000..4a3df61
--- /dev/null
+++ b/backend/modbus_config.example.json
@@ -0,0 +1,37 @@
+{
+ "defaults": {
+ "port": 502,
+ "timeout": 2.0,
+ "word_order": "big"
+ },
+ "devices": [
+ {
+ "meter_id": "UTIL-1",
+ "host": "meter-util.example.internal",
+ "unit_id": 1,
+ "registers": {
+ "voltage": { "address": 3000, "type": "float32", "function": "holding" },
+ "current": { "address": 3002, "type": "float32", "function": "holding" },
+ "power_kw": { "address": 3004, "type": "float32", "function": "holding" },
+ "power_factor": { "address": 3006, "type": "float32", "function": "holding" },
+ "frequency": { "address": 3008, "type": "float32", "function": "holding" },
+ "energy_kwh": { "address": 3010, "type": "uint32", "function": "holding", "scale": 0.1 },
+ "demand_kw": { "address": 3012, "type": "float32", "function": "holding" }
+ }
+ },
+ {
+ "meter_id": "UPS-A",
+ "host": "ups-a.example.internal",
+ "unit_id": 1,
+ "registers": {
+ "voltage": { "address": 100, "type": "uint16", "function": "input", "scale": 0.1 },
+ "current": { "address": 101, "type": "uint16", "function": "input", "scale": 0.1 },
+ "power_kw": { "address": 102, "type": "uint16", "function": "input", "scale": 0.1 },
+ "power_factor": { "address": 103, "type": "uint16", "function": "input", "scale": 0.001 },
+ "frequency": { "address": 104, "type": "uint16", "function": "input", "scale": 0.01 },
+ "battery_soc": { "address": 110, "type": "uint16", "function": "input" },
+ "on_battery": { "address": 0, "type": "bool", "function": "discrete" }
+ }
+ }
+ ]
+}
diff --git a/backend/modbus_source.py b/backend/modbus_source.py
new file mode 100644
index 0000000..2fb583c
--- /dev/null
+++ b/backend/modbus_source.py
@@ -0,0 +1,134 @@
+"""Modbus TCP data source for real power meters.
+
+Enabled with DATA_SOURCE=modbus. The device map is a JSON file (see
+modbus_config.example.json) that lists each meter's host, unit id, and the
+register for every value. Requires the optional dependency in
+requirements-modbus.txt (pymodbus).
+
+Readings come back in the same shape the simulator produces, so the alarm
+engine, history, and front end work unchanged. A meter that fails to answer
+is reported with status "Comm Fail" and empty values; alarms on that meter
+hold their last state until communication returns.
+"""
+
+import json
+import logging
+import struct
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional
+
+from config import METERS_BY_ID
+
+logger = logging.getLogger(__name__)
+
+REGISTER_COUNT = {"int16": 1, "uint16": 1, "int32": 2, "uint32": 2, "float32": 2, "bool": 1}
+NUMERIC_FIELDS = ("voltage", "current", "power_kw", "power_factor", "frequency",
+ "energy_kwh", "demand_kw", "load_pct")
+
+
+def decode(registers: List[int], data_type: str, word_order: str = "big") -> float:
+ """Decode raw 16 bit registers into a value."""
+ if data_type == "bool":
+ return bool(registers[0])
+ if data_type == "uint16":
+ return registers[0]
+ if data_type == "int16":
+ return struct.unpack(">h", struct.pack(">H", registers[0]))[0]
+ words = list(registers[:2])
+ if word_order == "little":
+ words.reverse()
+ raw = struct.pack(">HH", *words)
+ fmt = {"int32": ">i", "uint32": ">I", "float32": ">f"}[data_type]
+ return struct.unpack(fmt, raw)[0]
+
+
+def load_config(path: str) -> Dict[str, Any]:
+ with open(path, encoding="utf-8") as fh:
+ cfg = json.load(fh)
+ for dev in cfg.get("devices", []):
+ if dev["meter_id"] not in METERS_BY_ID:
+ raise ValueError(f"Modbus device references unknown meter {dev['meter_id']}")
+ return cfg
+
+
+class ModbusSource:
+ """Polls every configured device once per scan."""
+
+ def __init__(self, cfg: Dict[str, Any], client_factory=None):
+ self.cfg = cfg
+ self.defaults = cfg.get("defaults", {})
+ self._clients: Dict[str, Any] = {}
+ if client_factory is None:
+ from pymodbus.client import AsyncModbusTcpClient # optional dependency
+
+ def client_factory(host, port, timeout):
+ return AsyncModbusTcpClient(host, port=port, timeout=timeout)
+ self._factory = client_factory
+
+ async def _client(self, host: str, port: int):
+ key = f"{host}:{port}"
+ client = self._clients.get(key)
+ if client is None:
+ client = self._factory(host, port, self.defaults.get("timeout", 2.0))
+ self._clients[key] = client
+ if not client.connected:
+ await client.connect()
+ return client
+
+ async def _read_register(self, client, unit: int, spec: Dict[str, Any]):
+ function = spec.get("function", "holding")
+ address = spec["address"]
+ dtype = spec.get("type", "float32")
+ count = REGISTER_COUNT[dtype]
+ if function == "coil":
+ rr = await client.read_coils(address, count=1, slave=unit)
+ elif function == "discrete":
+ rr = await client.read_discrete_inputs(address, count=1, slave=unit)
+ elif function == "input":
+ rr = await client.read_input_registers(address, count=count, slave=unit)
+ else:
+ rr = await client.read_holding_registers(address, count=count, slave=unit)
+ if rr.isError():
+ raise IOError(f"Modbus error reading {address}: {rr}")
+ if function in ("coil", "discrete"):
+ return bool(rr.bits[0])
+ value = decode(rr.registers, dtype, spec.get("word_order", self.defaults.get("word_order", "big")))
+ if isinstance(value, bool):
+ return value
+ return value * spec.get("scale", 1.0)
+
+ async def read_device(self, dev: Dict[str, Any], now: float) -> Dict[str, Any]:
+ mid = dev["meter_id"]
+ reading: Dict[str, Any] = {
+ "meter_id": mid,
+ "timestamp": datetime.fromtimestamp(now, timezone.utc).isoformat(timespec="milliseconds"),
+ "epoch": round(now, 3),
+ **{f: None for f in NUMERIC_FIELDS},
+ }
+ try:
+ client = await self._client(dev["host"], dev.get("port", self.defaults.get("port", 502)))
+ for field, spec in dev["registers"].items():
+ reading[field] = await self._read_register(client, dev.get("unit_id", 1), spec)
+ except Exception as exc: # comm failure: report it, keep polling others
+ logger.warning("Modbus read failed for %s: %s", mid, exc)
+ return {**reading, **{f: None for f in NUMERIC_FIELDS}, "status": "Comm Fail", "comm_ok": False}
+
+ rated = METERS_BY_ID[mid]["rated_current"]
+ if reading["load_pct"] is None and reading["current"] is not None and rated:
+ reading["load_pct"] = round(reading["current"] / rated * 100.0, 1)
+ reading["status"] = dev.get("status_label", "Online")
+ reading["comm_ok"] = True
+ return reading
+
+ async def read_all(self, now: float) -> Dict[str, Dict[str, Any]]:
+ out = {}
+ for dev in self.cfg.get("devices", []):
+ out[dev["meter_id"]] = await self.read_device(dev, now)
+ return out
+
+ def close(self):
+ for client in self._clients.values():
+ try:
+ client.close()
+ except Exception:
+ pass
diff --git a/backend/requirements-dev.txt b/backend/requirements-dev.txt
new file mode 100644
index 0000000..b918a15
--- /dev/null
+++ b/backend/requirements-dev.txt
@@ -0,0 +1,3 @@
+-r requirements.txt
+pytest==8.3.3
+httpx==0.27.2
diff --git a/backend/requirements-modbus.txt b/backend/requirements-modbus.txt
new file mode 100644
index 0000000..5ae2f13
--- /dev/null
+++ b/backend/requirements-modbus.txt
@@ -0,0 +1 @@
+pymodbus==3.6.9
diff --git a/backend/requirements.txt b/backend/requirements.txt
index c83aa69..1eab1e7 100644
--- a/backend/requirements.txt
+++ b/backend/requirements.txt
@@ -1,2 +1,3 @@
fastapi==0.111.0
uvicorn[standard]==0.29.0
+pydantic==2.14.1
diff --git a/backend/simulator.py b/backend/simulator.py
index 063a367..86d1e26 100644
--- a/backend/simulator.py
+++ b/backend/simulator.py
@@ -1,129 +1,445 @@
-"""Modbus device simulator for Power Dashboard.
+"""Data center power system simulator.
-Generates realistic electrical meter readings using sine-wave modulation,
-Gaussian noise, and occasional fault injection to simulate real power meters.
+Models a small 2N data center so the public demo runs without hardware:
+
+ Utility ---+
+ +--> ATS --> Main Switchboard --+--> UPS A --> PDU A1, PDU A2
+ Generator -+ +--> UPS B --> PDU B1, PDU B2
+ +--> Mechanical panel
+
+Values are computed bottom up so they stay physically consistent: IT load at
+each PDU rolls up through transformer and UPS losses to the switchboard, and
+the switchboard load appears on whichever source the ATS has selected.
+
+Equipment behaves like the real thing: the generator cranks and warms up, the
+ATS waits for a stable source before transferring, UPS systems ride through on
+battery and recharge afterward, and cooling load drops out and restarts when
+the switchboard loses power.
+
+Scenarios inject abnormal conditions (utility outage, voltage sag, PDU
+overload, low power factor) so the alarm system has something to show. They
+fire automatically at random intervals and can be triggered on demand.
"""
import math
import random
-import time
from datetime import datetime, timezone
-from typing import Dict, Any
-
-from config import METERS
-
-
-class MeterSimulator:
- """Simulates a single power meter with realistic electrical values."""
-
- def __init__(self, meter_cfg: dict):
- self.meter_id = meter_cfg["id"]
- self.name = meter_cfg["name"]
- self.nominal_voltage = meter_cfg["nominal_voltage"]
- self.nominal_current = meter_cfg["nominal_current"]
-
- # Internal state
- self._energy_kwh = random.uniform(1000, 50000)
- self._phase_offset = random.uniform(0, 2 * math.pi)
- self._drift = 0.0
- self._fault_active = False
- self._fault_type = None
- self._fault_counter = 0
- self._last_demand_window = []
- self._start_time = time.time()
-
- # Load profile: base load as fraction of nominal
- self._base_load = random.uniform(0.45, 0.72)
-
- def _inject_fault(self):
- """Occasionally inject a fault condition."""
- if not self._fault_active and random.random() < 0.0015:
- self._fault_type = random.choice([
- "voltage_sag", "current_spike", "pf_drop", "freq_deviation"
- ])
- self._fault_active = True
- self._fault_counter = random.randint(3, 12)
-
- if self._fault_active:
- self._fault_counter -= 1
- if self._fault_counter <= 0:
- self._fault_active = False
- self._fault_type = None
-
- def read(self) -> Dict[str, Any]:
- """Return a simulated meter reading."""
- self._inject_fault()
-
- t = time.time() - self._start_time
-
- # Base voltage: 480V nominal with slow sine oscillation + noise
- v_swing = math.sin(t / 47 + self._phase_offset) * 3.5
- v_noise = random.gauss(0, 0.4)
- voltage = self.nominal_voltage + v_swing + v_noise
-
- # Current: follows a daily load profile approximated with a sine wave
- load_cycle = 0.82 + 0.18 * math.sin(t / 3600 * math.pi - math.pi / 2)
- i_base = self.nominal_current * self._base_load * load_cycle
- i_noise = random.gauss(0, i_base * 0.015)
- current = max(0.0, i_base + i_noise)
-
- # Power factor
- pf_base = 0.92 + math.sin(t / 120 + self._phase_offset) * 0.025
- pf = min(1.0, max(0.70, pf_base + random.gauss(0, 0.005)))
-
- # Frequency
- freq = 60.0 + math.sin(t / 23) * 0.08 + random.gauss(0, 0.03)
-
- # Apply faults
- if self._fault_active:
- ft = self._fault_type
- p = self._fault_counter / 12
- if ft == "voltage_sag":
- voltage *= (0.88 + 0.07 * p)
- elif ft == "current_spike":
- current *= (1.25 - 0.15 * p)
- elif ft == "pf_drop":
- pf *= (0.78 + 0.12 * p)
- elif ft == "freq_deviation":
- freq += (0.6 - 0.3 * p) * random.choice([-1, 1])
-
- # Power
- power_kw = (voltage * current * pf * math.sqrt(3)) / 1000.0
-
- # Demand (15-minute rolling average approximation)
- self._last_demand_window.append(power_kw)
- if len(self._last_demand_window) > 180:
- self._last_demand_window.pop(0)
- demand_kw = sum(self._last_demand_window) / len(self._last_demand_window)
-
- # Energy accumulation
- self._energy_kwh += power_kw * (5 / 3600)
+from typing import Any, Dict, Optional, Tuple
+
+from config import METERS, METERS_BY_ID, NOMINAL_FREQUENCY
+
+SQRT3 = math.sqrt(3)
+
+SCENARIOS = {
+ "utility_outage": {
+ "label": "Utility outage",
+ "description": "Utility power is lost. UPS systems carry the IT load on battery, the "
+ "generator starts, the ATS transfers, then retransfers when utility returns.",
+ },
+ "voltage_sag": {
+ "label": "Utility voltage sag",
+ "description": "A short utility voltage dip to about 85 percent. UPS systems ride "
+ "through on battery for a few seconds.",
+ },
+ "pdu_overload": {
+ "label": "PDU overload",
+ "description": "New servers push one PDU past its load alarm limits, then the load "
+ "is rebalanced.",
+ },
+ "pf_drop": {
+ "label": "Low power factor",
+ "description": "The cooling plant capacitor bank drops out and the mechanical panel "
+ "power factor falls below its alarm limit.",
+ },
+}
+
+# Base IT load per PDU (kW) and the cooling load ratio
+_PDU_BASE_KW = {"PDU-A1": 74.0, "PDU-A2": 66.0, "PDU-B1": 71.0, "PDU-B2": 63.0}
+_PDU_EFFICIENCY = 0.975
+_UPS_EFFICIENCY = 0.955
+_UPS_BATTERY_KWH = 50.0
+_UPS_CHARGE_KW = 10.0
+_HOUSE_KW = 18.0
+_MECH_RATIO = 0.48
+
+# Equipment timing (seconds)
+_GEN_START_DELAY = 2.0 # utility must be bad this long before engine start
+_GEN_CRANK_TIME = 6.0 # crank to rated voltage and frequency
+_ATS_GEN_STABLE = 4.0 # generator must be stable before transfer
+_ATS_RETRANSFER = 30.0 # utility must be stable before retransfer
+_GEN_COOLDOWN = 30.0 # unloaded run after retransfer
+_UPS_WALK_IN = 3.0 # rectifier walk in after input returns
+_MECH_RESTART = 20.0 # cooling equipment restart ramp
+
+
+def _iso(epoch: float) -> str:
+ return datetime.fromtimestamp(epoch, timezone.utc).isoformat(timespec="milliseconds")
+
+
+class FacilitySimulator:
+ """Stateful simulator. Call step(now) once per scan."""
+ def __init__(self, seed: Optional[int] = None, auto_events: bool = True,
+ event_interval: Tuple[float, float] = (150.0, 360.0),
+ first_event_delay: float = 25.0):
+ self.rng = random.Random(seed)
+ self.auto_events = auto_events
+ self.event_interval = event_interval
+ self.first_event_delay = first_event_delay
+
+ self._t_last: Optional[float] = None
+ self._next_event_at: Optional[float] = None
+ self.scenario: Optional[Dict[str, Any]] = None
+
+ # Slow random walks (Ornstein Uhlenbeck) keep values drifting naturally
+ self._walk: Dict[str, float] = {k: 0.0 for k in (
+ "util_v", "util_f", "gen_v", "mech", "mech_pf", *_PDU_BASE_KW.keys())}
+
+ # Meter registers
+ sizes = {"utility": 1.2e6, "generator": 4.0e3, "ats": 1.2e6, "switchgear": 1.2e6,
+ "ups": 4.5e5, "panel": 4.0e5, "pdu": 2.0e5}
+ self._energy = {m["id"]: sizes[m["type"]] * self.rng.uniform(0.6, 1.4) for m in METERS}
+ self._demand: Dict[str, Optional[float]] = {m["id"]: None for m in METERS}
+
+ # Equipment state
+ self.gen_state = "STANDBY"
+ self._gen_since = 0.0
+ self._gen_running_since: Optional[float] = None
+ self.ats_position = "UTILITY"
+ self._transfer_at: Optional[float] = None
+ self._util_bad_since: Optional[float] = None
+ self._util_good_since: Optional[float] = 0.0
+ self._source_ok_since: Optional[float] = 0.0
+ self._mech_restart_since: Optional[float] = None
+ self.ups = {
+ side: {"soc": 100.0, "on_battery": False, "input_ok_since": 0.0}
+ for side in ("UPS-A", "UPS-B")
+ }
+
+ # ------------------------------------------------------------------
+ # Scenarios
+ # ------------------------------------------------------------------
+
+ def trigger(self, name: str, now: float, **params) -> Dict[str, Any]:
+ """Start a scenario. Raises KeyError if unknown, RuntimeError if busy."""
+ if name not in SCENARIOS:
+ raise KeyError(name)
+ if self.scenario is not None:
+ raise RuntimeError(f"Scenario '{self.scenario['name']}' is still running")
+
+ rng = self.rng
+ if name == "utility_outage":
+ p = {"duration": params.get("duration", rng.uniform(90, 150))}
+ elif name == "voltage_sag":
+ p = {"duration": params.get("duration", rng.uniform(4, 8)),
+ "depth": params.get("depth", rng.uniform(0.84, 0.875))}
+ elif name == "pdu_overload":
+ ramp = params.get("ramp", 40.0)
+ hold = params.get("hold", rng.uniform(60, 90))
+ p = {"pdu": params.get("pdu", rng.choice(list(_PDU_BASE_KW))),
+ "target_pct": params.get("target_pct", rng.uniform(86, 95)),
+ "ramp": ramp, "hold": hold, "duration": 2 * ramp + hold}
+ else: # pf_drop
+ p = {"duration": params.get("duration", rng.uniform(70, 130)),
+ "pf": params.get("pf", rng.uniform(0.78, 0.82))}
+
+ self.scenario = {"name": name, "label": SCENARIOS[name]["label"],
+ "started_at": now, "params": p}
+ return self.scenario_info(now)
+
+ def scenario_info(self, now: float) -> Optional[Dict[str, Any]]:
+ if self.scenario is None:
+ return None
return {
- "meter_id": self.meter_id,
- "voltage": round(voltage, 2),
- "current": round(current, 2),
- "power_kw": round(power_kw, 2),
- "power_factor": round(pf, 4),
- "energy_kwh": round(self._energy_kwh, 2),
- "frequency": round(freq, 3),
- "demand_kw": round(demand_kw, 2),
- "timestamp": datetime.now(timezone.utc).isoformat(),
+ "name": self.scenario["name"],
+ "label": self.scenario["label"],
+ "started_at": _iso(self.scenario["started_at"]),
+ "elapsed_s": round(now - self.scenario["started_at"], 1),
}
+ def _scenario_elapsed(self, name: str, now: float) -> Optional[float]:
+ if self.scenario and self.scenario["name"] == name:
+ return now - self.scenario["started_at"]
+ return None
+
+ def _update_scenario(self, now: float):
+ sc = self.scenario
+ if sc is not None:
+ elapsed = now - sc["started_at"]
+ done = elapsed >= sc["params"]["duration"]
+ if sc["name"] == "utility_outage":
+ done = done and self.ats_position == "UTILITY" and self.gen_state == "STANDBY"
+ if done:
+ self.scenario = None
+ self._schedule_next(now)
+
+ if self._next_event_at is None:
+ self._next_event_at = now + self.first_event_delay
-class DeviceSimulator:
- """Manages all meter simulators."""
+ if self.auto_events and self.scenario is None and now >= self._next_event_at:
+ weights = {"voltage_sag": 0.3, "pdu_overload": 0.25, "pf_drop": 0.25,
+ "utility_outage": 0.2}
+ name = self.rng.choices(list(weights), weights=list(weights.values()))[0]
+ self.trigger(name, now)
- def __init__(self):
- self._meters: Dict[int, MeterSimulator] = {}
- for cfg in METERS:
- self._meters[cfg["id"]] = MeterSimulator(cfg)
+ def _schedule_next(self, now: float):
+ lo, hi = self.event_interval
+ self._next_event_at = now + self.rng.uniform(lo, hi)
- def read_all(self) -> Dict[int, Dict[str, Any]]:
- return {mid: m.read() for mid, m in self._meters.items()}
+ def schedule_event_at(self, at: float):
+ """Set the time of the next automatic event."""
+ self._next_event_at = at
- def read_meter(self, meter_id: int) -> Dict[str, Any]:
- if meter_id not in self._meters:
- raise ValueError(f"Meter {meter_id} not found")
- return self._meters[meter_id].read()
+ # ------------------------------------------------------------------
+ # Helpers
+ # ------------------------------------------------------------------
+
+ def _ou(self, key: str, dt: float, sigma: float, tau: float) -> float:
+ """Advance an Ornstein Uhlenbeck random walk with mean 0."""
+ decay = math.exp(-dt / tau)
+ x = self._walk[key] * decay + sigma * math.sqrt(1 - decay * decay) * self.rng.gauss(0, 1)
+ self._walk[key] = x
+ return x
+
+ def _noise(self, sigma: float) -> float:
+ return self.rng.gauss(0, sigma)
+
+ def _register(self, mid: str, kw: float, dt: float) -> Tuple[float, float]:
+ self._energy[mid] += max(kw, 0.0) * dt / 3600.0
+ d = self._demand[mid]
+ d = kw if d is None else d + (kw - d) * min(1.0, dt / 900.0) # 15 min demand
+ self._demand[mid] = d
+ return self._energy[mid], d
+
+ # ------------------------------------------------------------------
+ # Main step
+ # ------------------------------------------------------------------
+
+ def step(self, now: float) -> Dict[str, Dict[str, Any]]:
+ dt = 2.0 if self._t_last is None else min(max(now - self._t_last, 0.0), 3600.0)
+ self._t_last = now
+ self._update_scenario(now)
+
+ hour = (now % 86400) / 3600.0
+ it_profile = 1.0 + 0.04 * math.sin(2 * math.pi * (hour - 10) / 24)
+ temp_profile = 1.0 + 0.12 * math.sin(2 * math.pi * (hour - 9) / 24)
+
+ # --- Utility source ------------------------------------------------
+ outage_t = self._scenario_elapsed("utility_outage", now)
+ util_available = not (outage_t is not None and outage_t < self.scenario["params"]["duration"])
+ sag_t = self._scenario_elapsed("voltage_sag", now)
+ v_factor = 1.0
+ if sag_t is not None and sag_t < self.scenario["params"]["duration"]:
+ v_factor = self.scenario["params"]["depth"]
+
+ util_walk_v = self._ou("util_v", dt, 2.5, 600)
+ util_walk_f = self._ou("util_f", dt, 0.012, 120)
+ if util_available:
+ v_util = (482.0 + util_walk_v + self._noise(0.6)) * v_factor
+ f_util = NOMINAL_FREQUENCY + util_walk_f + self._noise(0.004)
+ else:
+ v_util, f_util = 0.0, 0.0
+ util_ok = v_util >= 0.80 * 480.0 # ATS normal source dropout
+
+ if util_ok:
+ self._util_bad_since = None
+ if self._util_good_since is None:
+ self._util_good_since = now
+ else:
+ self._util_good_since = None
+ if self._util_bad_since is None:
+ self._util_bad_since = now
+
+ # --- Generator and ATS controller -----------------------------------
+ start_signal = (self._util_bad_since is not None and now - self._util_bad_since >= _GEN_START_DELAY) \
+ or self.ats_position == "GENERATOR"
+
+ if self.gen_state == "STANDBY" and start_signal:
+ self.gen_state, self._gen_since = "CRANKING", now
+ elif self.gen_state == "CRANKING":
+ if not start_signal:
+ self.gen_state, self._gen_since = "STANDBY", now
+ elif now - self._gen_since >= _GEN_CRANK_TIME:
+ self.gen_state, self._gen_since = "RUNNING", now
+ self._gen_running_since = now
+ elif self.gen_state == "RUNNING" and not start_signal and self.ats_position == "UTILITY":
+ self.gen_state, self._gen_since = "COOLDOWN", now
+ elif self.gen_state == "COOLDOWN":
+ if start_signal:
+ self.gen_state, self._gen_since = "RUNNING", now
+ self._gen_running_since = now
+ elif now - self._gen_since >= _GEN_COOLDOWN:
+ self.gen_state, self._gen_since = "STANDBY", now
+ self._gen_running_since = None
+
+ gen_running = self.gen_state in ("RUNNING", "COOLDOWN")
+ if self.ats_position == "UTILITY" and not util_ok and self.gen_state == "RUNNING" \
+ and now - self._gen_running_since >= _ATS_GEN_STABLE:
+ self.ats_position, self._transfer_at = "GENERATOR", now
+ elif self.ats_position == "GENERATOR" and util_ok \
+ and now - self._util_good_since >= _ATS_RETRANSFER:
+ self.ats_position, self._transfer_at = "UTILITY", now
+
+ gen_walk_v = self._ou("gen_v", dt, 1.5, 60)
+ if gen_running:
+ v_gen = 480.0 + gen_walk_v + self._noise(0.8)
+ f_gen = NOMINAL_FREQUENCY + self._noise(0.04)
+ if self.ats_position == "GENERATOR" and self._transfer_at is not None:
+ since = now - self._transfer_at
+ if since < 6.0: # frequency dip from the block load step
+ f_gen -= 1.2 * math.exp(-since / 1.5)
+ elif self.gen_state == "CRANKING":
+ ramp = min(1.0, (now - self._gen_since) / _GEN_CRANK_TIME)
+ v_gen, f_gen = 480.0 * ramp, NOMINAL_FREQUENCY * ramp
+ else:
+ v_gen, f_gen = 0.0, 0.0
+
+ if self.ats_position == "UTILITY":
+ v_src, f_src = v_util, f_util
+ else:
+ v_src, f_src = v_gen, f_gen
+ src_live = v_src >= 0.5 * 480.0
+ if src_live:
+ if self._source_ok_since is None:
+ self._source_ok_since = now
+ else:
+ self._source_ok_since = None
+
+ # --- PDUs (IT load) --------------------------------------------------
+ readings: Dict[str, Dict[str, Any]] = {}
+ overload_t = self._scenario_elapsed("pdu_overload", now)
+ ups_out: Dict[str, Dict[str, float]] = {"UPS-A": {"kw": 0.0, "kvar": 0.0},
+ "UPS-B": {"kw": 0.0, "kvar": 0.0}}
+ it_total = 0.0
+ for pdu, base in _PDU_BASE_KW.items():
+ cfg = METERS_BY_ID[pdu]
+ walk = self._ou(pdu, dt, 0.025, 1800)
+ kw = base * it_profile * (1 + walk) * (1 + self._noise(0.006))
+ if overload_t is not None and self.scenario["params"]["pdu"] == pdu:
+ p = self.scenario["params"]
+ target_kw = cfg["rated_kva"] * p["target_pct"] / 100.0 * 0.98
+ shape = min(1.0, overload_t / p["ramp"],
+ max(0.0, (p["duration"] - overload_t) / p["ramp"]))
+ kw = kw * (1 - shape) + target_kw * shape
+ pf = min(0.995, 0.98 + self._noise(0.003))
+ kva = kw / pf
+ v = 208.0 * (1 - 0.012 * kva / cfg["rated_kva"]) + self._noise(0.25)
+ side = "UPS-A" if pdu.startswith("PDU-A") else "UPS-B"
+ kw_in = kw / _PDU_EFFICIENCY
+ ups_out[side]["kw"] += kw_in
+ ups_out[side]["kvar"] += kw_in * math.tan(math.acos(0.97))
+ it_total += kw
+ readings[pdu] = self._reading(pdu, now, dt, v, kva, kw, pf, f_src or NOMINAL_FREQUENCY,
+ status="Normal")
+
+ # --- UPS -------------------------------------------------------------
+ input_ok = 0.88 * 480.0 <= v_src <= 1.12 * 480.0 and 55.0 <= f_src <= 65.0
+ ups_in_kw = 0.0
+ for side, st in self.ups.items():
+ cfg = METERS_BY_ID[side]
+ if input_ok:
+ if st["input_ok_since"] is None:
+ st["input_ok_since"] = now
+ else:
+ st["input_ok_since"] = None
+ st["on_battery"] = st["input_ok_since"] is None or now - st["input_ok_since"] < _UPS_WALK_IN
+
+ out_kw = ups_out[side]["kw"]
+ out_kva = math.hypot(out_kw, ups_out[side]["kvar"])
+ if st["on_battery"]:
+ st["soc"] = max(0.0, st["soc"] - out_kw / 0.96 * dt / 3600.0 / _UPS_BATTERY_KWH * 100.0)
+ in_kw = 0.0
+ else:
+ charge = _UPS_CHARGE_KW if st["soc"] < 100.0 else 0.0
+ st["soc"] = min(100.0, st["soc"] + charge * dt / 3600.0 / _UPS_BATTERY_KWH * 100.0)
+ in_kw = out_kw / _UPS_EFFICIENCY + charge
+ ups_in_kw += in_kw
+ runtime = st["soc"] / 100.0 * _UPS_BATTERY_KWH / max(out_kw, 1.0) * 60.0
+ v_out = 480.0 + self._noise(0.4)
+ readings[side] = self._reading(
+ side, now, dt, v_out, out_kva, out_kw, out_kw / out_kva if out_kva else 1.0,
+ NOMINAL_FREQUENCY if st["on_battery"] else (f_src or NOMINAL_FREQUENCY),
+ status="On Battery" if st["on_battery"] else "Online",
+ on_battery=st["on_battery"], battery_soc=round(st["soc"], 1),
+ battery_runtime_min=round(runtime, 1), input_kw=round(in_kw, 1),
+ )
+
+ # --- Mechanical panel (not on UPS) -----------------------------------
+ cfg = METERS_BY_ID["MECH-1"]
+ mech_walk = self._ou("mech", dt, 0.04, 900)
+ pf_walk = self._ou("mech_pf", dt, 0.008, 300)
+ mech_pf = 0.90 + pf_walk + self._noise(0.002)
+ pf_t = self._scenario_elapsed("pf_drop", now)
+ if pf_t is not None:
+ p = self.scenario["params"]
+ shape = min(1.0, pf_t / 10.0, max(0.0, (p["duration"] - pf_t) / 10.0))
+ mech_pf = mech_pf * (1 - shape) + (p["pf"] + self._noise(0.004)) * shape
+ if src_live:
+ if self._mech_restart_since is not None:
+ ramp = min(1.0, (now - self._mech_restart_since) / _MECH_RESTART)
+ if ramp >= 1.0:
+ self._mech_restart_since = None
+ else:
+ ramp = 1.0
+ mech_kw = _MECH_RATIO * it_total * temp_profile * (1 + mech_walk) * ramp
+ mech_status = "Restarting" if ramp < 1.0 else "Normal"
+ house_kw = _HOUSE_KW
+ else:
+ self._mech_restart_since = now
+ mech_kw, house_kw, mech_status = 0.0, 0.0, "De-energized"
+ mech_kva = mech_kw / mech_pf
+ mech_v = v_src * (1 - 0.006 * mech_kva / cfg["rated_kva"]) if src_live else 0.0
+ readings["MECH-1"] = self._reading("MECH-1", now, dt, mech_v, mech_kva, mech_kw,
+ mech_pf if mech_kw > 0 else 0.0, f_src, status=mech_status,
+ energized=src_live)
+
+ # --- Main switchboard and ATS ----------------------------------------
+ total_kw = ups_in_kw + mech_kw + house_kw
+ total_kvar = ups_in_kw * math.tan(math.acos(0.99)) \
+ + mech_kw * math.tan(math.acos(max(mech_pf, 0.5))) \
+ + house_kw * math.tan(math.acos(0.95))
+ total_kva = math.hypot(total_kw, total_kvar)
+ msb_pf = total_kw / total_kva if total_kva else 0.0
+ msb_v = v_src * (1 - 0.004 * total_kva / 1330.0) if src_live else 0.0
+ readings["MSB-1"] = self._reading("MSB-1", now, dt, msb_v, total_kva, total_kw, msb_pf, f_src,
+ status="Energized" if src_live else "De-energized")
+ on_gen = self.ats_position == "GENERATOR"
+ readings["ATS-1"] = self._reading(
+ "ATS-1", now, dt, v_src, total_kva, total_kw, msb_pf, f_src,
+ status="On Generator" if on_gen else "On Utility",
+ position=self.ats_position, on_emergency=on_gen,
+ )
+
+ # --- Sources ---------------------------------------------------------
+ util_load = (total_kva, total_kw, msb_pf) if not on_gen else (0.0, 0.0, 0.0)
+ util_status = "Failed" if not util_available else ("Sag" if v_factor < 1.0 else "Normal")
+ readings["UTIL-1"] = self._reading("UTIL-1", now, dt, v_util, *util_load, f_util,
+ status=util_status, source_failed=not util_available)
+ gen_load = (total_kva, total_kw, msb_pf) if on_gen else (0.0, 0.0, 0.0)
+ readings["GEN-1"] = self._reading("GEN-1", now, dt, v_gen, *gen_load, f_gen,
+ status=self.gen_state.title(),
+ running=self.gen_state == "RUNNING",
+ gen_state=self.gen_state)
+
+ return {m["id"]: readings[m["id"]] for m in METERS}
+
+ def _reading(self, mid: str, now: float, dt: float, voltage: float, kva: float, kw: float,
+ pf: float, freq: float, status: str, **extra) -> Dict[str, Any]:
+ cfg = METERS_BY_ID[mid]
+ current = kva * 1000.0 / (SQRT3 * voltage) if voltage > 1.0 else 0.0
+ energy, demand = self._register(mid, kw, dt)
+ return {
+ "meter_id": mid,
+ "timestamp": _iso(now),
+ "epoch": round(now, 3),
+ "voltage": round(max(voltage, 0.0), 1),
+ "current": round(current, 1),
+ "power_kw": round(kw, 1),
+ "power_factor": round(pf, 3),
+ "frequency": round(max(freq, 0.0), 2),
+ "energy_kwh": round(energy, 1),
+ "demand_kw": round(demand, 1),
+ "load_pct": round(current / cfg["rated_current"] * 100.0, 1),
+ "status": status,
+ **extra,
+ }
diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py
new file mode 100644
index 0000000..261622d
--- /dev/null
+++ b/backend/tests/conftest.py
@@ -0,0 +1,15 @@
+import os
+import sys
+import tempfile
+
+# Settings must be in place before config.py is imported
+_tmp = tempfile.mkdtemp(prefix="power-dashboard-tests-")
+os.environ.setdefault("DB_PATH", os.path.join(_tmp, "test.db"))
+os.environ.setdefault("BACKFILL_HOURS", "1")
+os.environ.setdefault("POLL_INTERVAL", "0.2")
+os.environ.setdefault("HISTORY_INTERVAL", "0.2")
+os.environ.setdefault("SIM_SEED", "7")
+os.environ.setdefault("SIM_FIRST_EVENT", "100000")
+os.environ["DATA_SOURCE"] = "simulator"
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
diff --git a/backend/tests/test_alarms.py b/backend/tests/test_alarms.py
new file mode 100644
index 0000000..9ea1d58
--- /dev/null
+++ b/backend/tests/test_alarms.py
@@ -0,0 +1,274 @@
+import pytest
+
+import config
+from alarms import AlarmActionError, AlarmManager, AlarmState
+from simulator import FacilitySimulator
+
+T0 = 1_760_000_000.0
+
+HI = {
+ "tag": "PDU-1.LOAD.HI", "meter_id": "PDU-1", "parameter": "load_pct", "kind": "HI",
+ "setpoint": 80.0, "deadband": 3.0, "priority": 3, "on_delay": 0, "off_delay": 0,
+ "message": "PDU load high", "unit": "%",
+}
+LO = {
+ "tag": "UTIL.V.LO", "meter_id": "UTIL", "parameter": "voltage", "kind": "LO",
+ "setpoint": 432.0, "deadband": 5.0, "priority": 3, "on_delay": 0, "off_delay": 0,
+ "message": "Voltage low",
+ "suppress": {"meter_id": "UTIL", "parameter": "failed", "equals": True, "reason": "Utility failed"},
+}
+DISCRETE = {
+ "tag": "UPS.ONBATT", "meter_id": "UPS", "parameter": "on_battery", "kind": "DISCRETE",
+ "setpoint": 1, "deadband": 0, "priority": 1, "on_delay": 0, "off_delay": 0,
+ "message": "UPS on battery",
+}
+
+
+class Harness:
+ def __init__(self, *defs):
+ self.mgr = AlarmManager(defs, {})
+ self.events = []
+ self.mgr.add_listener(self.events.append)
+ self.t = T0
+
+ def scan(self, **values):
+ """values: meter_id -> dict of parameters"""
+ self.t += 2
+ return self.mgr.evaluate(values, self.t)
+
+ def load(self, pct):
+ return self.scan(**{"PDU-1": {"load_pct": pct}})
+
+ def state(self, tag="PDU-1.LOAD.HI"):
+ return self.mgr.points[tag].state
+
+ def kinds(self):
+ return [e["event"] for e in self.events]
+
+
+def test_alarm_ack_then_clear():
+ h = Harness(HI)
+ h.load(50)
+ assert h.state() == AlarmState.NORMAL
+ assert h.load(85) is True
+ assert h.state() == AlarmState.UNACK
+ h.mgr.acknowledge(HI["tag"], h.t, "jdoe")
+ assert h.state() == AlarmState.ACKED
+ assert h.mgr.points[HI["tag"]].ack_by == "jdoe"
+ h.load(70)
+ assert h.state() == AlarmState.NORMAL
+ assert h.kinds() == ["ALARM", "ACK", "RTN"]
+
+
+def test_return_to_normal_unacknowledged_then_ack():
+ h = Harness(HI)
+ h.load(85)
+ h.load(70)
+ assert h.state() == AlarmState.RTN_UNACK
+ item = h.mgr.alarm_list()[0]
+ assert item["state_label"] == "Returned to normal, unacknowledged"
+ h.mgr.acknowledge(HI["tag"], h.t)
+ assert h.state() == AlarmState.NORMAL
+ assert h.mgr.alarm_list() == []
+ assert h.kinds() == ["ALARM", "RTN", "ACK"]
+
+
+def test_rtn_unack_realarms():
+ h = Harness(HI)
+ h.load(85)
+ h.load(70)
+ h.load(88)
+ assert h.state() == AlarmState.UNACK
+ assert h.kinds() == ["ALARM", "RTN", "ALARM"]
+
+
+def test_ack_is_rejected_when_nothing_to_acknowledge():
+ h = Harness(HI)
+ h.load(50)
+ with pytest.raises(AlarmActionError):
+ h.mgr.acknowledge(HI["tag"], h.t)
+ h.load(85)
+ h.mgr.acknowledge(HI["tag"], h.t)
+ with pytest.raises(AlarmActionError):
+ h.mgr.acknowledge(HI["tag"], h.t)
+ with pytest.raises(KeyError):
+ h.mgr.acknowledge("NOPE", h.t)
+
+
+def test_deadband_prevents_chatter():
+ h = Harness(HI)
+ h.load(81)
+ h.mgr.acknowledge(HI["tag"], h.t)
+ for v in (79.5, 80.5, 78.0, 79.9, 77.5): # inside the 3% deadband
+ h.load(v)
+ assert h.state() == AlarmState.ACKED
+ h.load(76.9)
+ assert h.state() == AlarmState.NORMAL
+ assert h.kinds().count("ALARM") == 1
+
+
+def test_on_and_off_delay():
+ h = Harness({**HI, "on_delay": 2, "off_delay": 1})
+ h.load(85)
+ h.load(85)
+ assert h.state() == AlarmState.NORMAL
+ h.load(70) # a single normal scan resets the on delay
+ h.load(85)
+ h.load(85)
+ assert h.state() == AlarmState.NORMAL
+ h.load(85)
+ assert h.state() == AlarmState.UNACK
+ h.load(70)
+ assert h.state() == AlarmState.UNACK
+ h.load(70)
+ assert h.state() == AlarmState.RTN_UNACK
+
+
+def test_lo_and_discrete_kinds():
+ h = Harness(LO, DISCRETE)
+ h.scan(UTIL={"voltage": 480, "failed": False}, UPS={"on_battery": False})
+ assert h.state("UTIL.V.LO") == AlarmState.NORMAL
+ assert h.state("UPS.ONBATT") == AlarmState.NORMAL
+ h.scan(UTIL={"voltage": 410, "failed": False}, UPS={"on_battery": True})
+ assert h.state("UTIL.V.LO") == AlarmState.UNACK
+ assert h.state("UPS.ONBATT") == AlarmState.UNACK
+ h.scan(UTIL={"voltage": 434, "failed": False}, UPS={"on_battery": False})
+ assert h.state("UTIL.V.LO") == AlarmState.UNACK # within deadband
+ assert h.state("UPS.ONBATT") == AlarmState.RTN_UNACK
+
+
+def test_missing_value_holds_state():
+ h = Harness(HI)
+ h.load(85)
+ h.scan() # meter offline
+ h.scan(**{"PDU-1": {"load_pct": None}})
+ assert h.state() == AlarmState.UNACK
+ assert h.kinds() == ["ALARM"]
+
+
+def test_shelve_hides_alarm_and_expires():
+ h = Harness(HI)
+ h.load(85)
+ h.mgr.shelve(HI["tag"], 5, h.t, "jdoe", "Known overload during migration")
+ assert h.state() == AlarmState.SHELVED
+ p = h.mgr.points[HI["tag"]]
+ assert p.shelve_reason == "Known overload during migration"
+ summary = h.mgr.summary(h.t)
+ assert summary["shelved"] == 1 and summary["annunciated"] == 0
+
+ h.load(95)
+ assert h.state() == AlarmState.SHELVED # no annunciation while shelved
+ h.t += 5 * 60
+ h.load(95)
+ assert h.state() == AlarmState.UNACK # timer expired, condition still active
+ assert h.kinds() == ["ALARM", "SHELVE", "UNSHELVE", "ALARM"]
+ assert h.events[2]["note"] == "Shelve time expired"
+
+
+def test_unshelve_when_condition_cleared_returns_to_normal():
+ h = Harness(HI)
+ h.load(85)
+ h.mgr.shelve(HI["tag"], 15, h.t)
+ h.load(60)
+ h.mgr.unshelve(HI["tag"], h.t)
+ assert h.state() == AlarmState.NORMAL
+ with pytest.raises(AlarmActionError):
+ h.mgr.unshelve(HI["tag"], h.t)
+
+
+def test_shelve_limits():
+ h = Harness(HI)
+ h.load(85)
+ with pytest.raises(AlarmActionError):
+ h.mgr.shelve(HI["tag"], config.SHELVE_MAX_MIN + 1, h.t)
+ with pytest.raises(AlarmActionError):
+ h.mgr.shelve(HI["tag"], 0, h.t)
+ h.mgr.shelve(HI["tag"], 5, h.t)
+ with pytest.raises(AlarmActionError):
+ h.mgr.shelve(HI["tag"], 5, h.t)
+
+
+def test_designed_suppression():
+ h = Harness(LO)
+ h.scan(UTIL={"voltage": 480, "failed": False})
+ h.scan(UTIL={"voltage": 0, "failed": True})
+ assert h.state("UTIL.V.LO") == AlarmState.SUPPRESSED
+ assert h.mgr.points["UTIL.V.LO"].suppress_reason == "Utility failed"
+ assert h.kinds() == ["SUPPRESS"]
+ # Utility back at normal voltage: return to service without a stale alarm
+ h.scan(UTIL={"voltage": 481, "failed": False})
+ assert h.state("UTIL.V.LO") == AlarmState.NORMAL
+ assert h.kinds() == ["SUPPRESS", "UNSUPPRESS"]
+ # Utility back but low: return to service straight into alarm
+ h.scan(UTIL={"voltage": 0, "failed": True})
+ h.scan(UTIL={"voltage": 420, "failed": False})
+ assert h.state("UTIL.V.LO") == AlarmState.UNACK
+
+
+def test_starting_suppressed_is_silent():
+ h = Harness(LO)
+ h.scan(UTIL={"voltage": 0, "failed": True})
+ assert h.state("UTIL.V.LO") == AlarmState.SUPPRESSED
+ assert h.events == []
+
+
+def test_summary_rate_and_priorities():
+ h = Harness(HI, DISCRETE)
+ h.scan(**{"PDU-1": {"load_pct": 85}}, UPS={"on_battery": True})
+ s = h.mgr.summary(h.t)
+ assert s["annunciated"] == 2 and s["unacknowledged"] == 2
+ assert s["by_priority"]["1"] == 1 and s["by_priority"]["3"] == 1
+ assert s["highest_unacked"]["tag"] == "UPS.ONBATT"
+ assert s["alarm_rate_10min"] == 2 and s["rate_status"] == "Manageable"
+
+ # Chatter between the states to build up a flood
+ for _ in range(10):
+ h.scan(**{"PDU-1": {"load_pct": 60}}, UPS={"on_battery": False})
+ h.scan(**{"PDU-1": {"load_pct": 85}}, UPS={"on_battery": True})
+ assert h.mgr.summary(h.t)["rate_status"] == "Flood"
+ # Older than 10 minutes drops out of the rate window
+ assert h.mgr.summary(h.t + 601)["alarm_rate_10min"] == 0
+
+
+def test_alarm_list_orders_by_priority():
+ h = Harness(HI, DISCRETE)
+ h.scan(**{"PDU-1": {"load_pct": 85}}, UPS={"on_battery": True})
+ tags = [a["tag"] for a in h.mgr.alarm_list()]
+ assert tags == ["UPS.ONBATT", "PDU-1.LOAD.HI"]
+
+
+def test_configured_alarm_database_is_valid():
+ tags = [d["tag"] for d in config.ALARM_POINTS]
+ assert len(tags) == len(set(tags))
+ for d in config.ALARM_POINTS:
+ assert d["meter_id"] in config.METERS_BY_ID
+ assert d["priority"] in config.ALARM_PRIORITIES
+ assert d["kind"] in ("HI", "HIHI", "LO", "LOLO", "DISCRETE")
+ assert d["response"] and d["consequence"]
+
+
+def test_outage_through_real_alarm_database():
+ """The simulator and the configured alarms together tell the outage story."""
+ sim = FacilitySimulator(seed=5, auto_events=False)
+ names = {m["id"]: m["name"] for m in config.METERS}
+ mgr = AlarmManager(config.ALARM_POINTS, names)
+ events = []
+ mgr.add_listener(events.append)
+
+ t = T0
+ for _ in range(10):
+ mgr.evaluate(sim.step(t), t)
+ t += 2
+ assert events == [] # quiet plant, no standing alarms
+
+ sim.trigger("utility_outage", t, duration=40)
+ for _ in range(80):
+ mgr.evaluate(sim.step(t), t)
+ t += 2
+
+ alarmed = {e["tag"] for e in events if e["event"] == "ALARM"}
+ assert {"UTIL-1.FAIL", "UPS-A.ONBATT", "UPS-B.ONBATT", "ATS-1.EMERG"} <= alarmed
+ assert "UTIL-1.V.LO" not in alarmed # cascade suppressed by the failure alarm
+ assert "MECH-1.PF.LO" not in alarmed # no nuisance alarm from a de-energized panel
+ states = {a["tag"]: a["state"] for a in mgr.alarm_list()}
+ assert states.get("UPS-A.ONBATT") == "RTN_UNACK" # ran on battery, nobody acked
diff --git a/backend/tests/test_api_websocket.py b/backend/tests/test_api_websocket.py
new file mode 100644
index 0000000..5ee387e
--- /dev/null
+++ b/backend/tests/test_api_websocket.py
@@ -0,0 +1,131 @@
+import json
+import time
+
+import pytest
+from fastapi.testclient import TestClient
+
+from main import app
+
+
+@pytest.fixture(scope="module")
+def client():
+ with TestClient(app) as c:
+ yield c
+
+
+def receive_until(ws, predicate, limit=200):
+ """Read messages until one matches. Returns that message."""
+ for _ in range(limit):
+ msg = json.loads(ws.receive_text())
+ if predicate(msg):
+ return msg
+ raise AssertionError("expected message never arrived")
+
+
+def test_health(client):
+ body = client.get("/api/health").json()
+ assert body["status"] == "ok"
+ assert body["data_source"] == "simulator"
+ assert body["version"] == "1.1.0"
+
+
+def test_websocket_sends_snapshot_on_connect(client):
+ with client.websocket_connect("/ws/live") as ws:
+ msg = json.loads(ws.receive_text())
+ assert msg["type"] == "snapshot"
+ data = msg["data"]
+ assert len(data["meters"]) == 11
+ assert {r["meter_id"] for r in data["readings"]} == {m["id"] for m in data["meters"]}
+ assert data["demo"]["simulated"] is True
+ assert {s["name"] for s in data["demo"]["scenarios"]} >= {"utility_outage", "voltage_sag"}
+ assert "summary" in data and "alarms" in data
+ assert data["system"]["health"] in ("normal", "caution", "warning", "critical")
+
+
+def test_websocket_streams_readings(client):
+ with client.websocket_connect("/ws/live") as ws:
+ ws.receive_text() # snapshot
+ first = receive_until(ws, lambda m: m["type"] == "readings")
+ second = receive_until(ws, lambda m: m["type"] == "readings")
+ assert len(first["data"]["readings"]) == 11
+ assert second["data"]["readings"][0]["epoch"] > first["data"]["readings"][0]["epoch"]
+ assert first["data"]["system"]["facility_kw"] > 0
+
+
+def test_websocket_ping_pong(client):
+ with client.websocket_connect("/ws/live") as ws:
+ ws.receive_text()
+ ws.send_text(json.dumps({"type": "ping"}))
+ receive_until(ws, lambda m: m["type"] == "pong")
+
+
+def test_alarm_lifecycle_over_rest_and_websocket(client):
+ """Trigger an outage, see the alarm arrive live, acknowledge, shelve, unshelve."""
+ with client.websocket_connect("/ws/live") as ws:
+ ws.receive_text()
+ r = client.post("/api/demo/events/utility_outage")
+ assert r.status_code == 200 and r.json()["name"] == "utility_outage"
+ assert client.post("/api/demo/events/voltage_sag").status_code == 409
+
+ def has_state(tag, state):
+ def check(m):
+ return m["type"] == "alarms" and any(
+ a["tag"] == tag and a["state"] == state for a in m["data"]["alarms"])
+ return check
+
+ msg = receive_until(ws, has_state("UPS-A.ONBATT", "UNACK"))
+ summary = msg["data"]["summary"]
+ assert summary["highest_unacked"]["priority"] == 1
+
+ r = client.post("/api/alarms/UPS-A.ONBATT/acknowledge", json={"operator": "tester"})
+ assert r.status_code == 200
+ assert r.json()["state"] in ("ACKED", "NORMAL")
+ receive_until(ws, lambda m: m["type"] == "alarms" and all(
+ a["state"] != "UNACK" for a in m["data"]["alarms"] if a["tag"] == "UPS-A.ONBATT"))
+
+ r = client.post("/api/alarms/UTIL-1.FAIL/shelve",
+ json={"minutes": 5, "reason": "Utility crew on site", "operator": "tester"})
+ assert r.status_code == 200 and r.json()["state"] == "SHELVED"
+ receive_until(ws, has_state("UTIL-1.FAIL", "SHELVED"))
+ assert client.post("/api/alarms/UTIL-1.FAIL/unshelve").json()["state"] != "SHELVED"
+
+ history = client.get("/api/alarms/history?limit=500").json()
+ kinds = {(e["tag"], e["event"]) for e in history}
+ assert ("UPS-A.ONBATT", "ALARM") in kinds
+ assert ("UPS-A.ONBATT", "ACK") in kinds
+ assert ("UTIL-1.FAIL", "SHELVE") in kinds
+ assert ("UTIL-1.FAIL", "UNSHELVE") in kinds
+ ack = next(e for e in history if e["tag"] == "UPS-A.ONBATT" and e["event"] == "ACK")
+ assert ack["operator"] == "tester"
+
+
+def test_alarm_action_errors(client):
+ assert client.post("/api/alarms/NOPE/acknowledge").status_code == 404
+ assert client.post("/api/alarms/MSB-1.LOAD.HI/acknowledge").status_code == 409
+ assert client.post("/api/alarms/MSB-1.LOAD.HI/shelve", json={"minutes": 999}).status_code == 422
+ assert client.post("/api/alarms/MSB-1.LOAD.HI/unshelve").status_code == 409
+
+
+def test_meters_and_history(client):
+ meters = client.get("/api/meters").json()
+ assert len(meters) == 11
+ by_id = {m["id"]: m for m in meters}
+ # PDUs sit behind the UPS, so they stay energized even if an outage is in progress
+ assert by_id["PDU-A1"]["current_reading"]["voltage"] > 200
+ assert client.get("/api/meters/UPS-A").json()["type"] == "ups"
+ assert client.get("/api/meters/NOPE").status_code == 404
+ hist = client.get("/api/meters/MSB-1/history?period=1h").json()
+ assert len(hist) > 30 # backfilled on startup
+ epochs = [h["epoch"] for h in hist]
+ assert epochs == sorted(epochs)
+ assert time.time() - epochs[-1] < 120
+ assert client.get("/api/meters/MSB-1/history?period=2y").status_code == 400
+
+
+def test_definitions_and_config(client):
+ defs = client.get("/api/alarms/definitions").json()
+ assert any(d["suppression"] for d in defs)
+ cfg = client.get("/api/config").json()
+ assert cfg["data_source"] == "simulator"
+ assert cfg["shelve_max_min"] == 60
+ assert client.post("/api/demo/events/nope").status_code == 404
diff --git a/backend/tests/test_modbus.py b/backend/tests/test_modbus.py
new file mode 100644
index 0000000..3fd1b1a
--- /dev/null
+++ b/backend/tests/test_modbus.py
@@ -0,0 +1,71 @@
+import asyncio
+import struct
+
+from modbus_source import ModbusSource, decode
+
+
+def float_regs(value):
+ hi, lo = struct.unpack(">HH", struct.pack(">f", value))
+ return [hi, lo]
+
+
+def test_decode_types():
+ assert decode([0xFFFF], "int16") == -1
+ assert decode([0xFFFF], "uint16") == 65535
+ assert abs(decode(float_regs(480.5), "float32") - 480.5) < 1e-4
+ assert abs(decode(list(reversed(float_regs(59.98))), "float32", "little") - 59.98) < 1e-4
+ assert decode([0x0001, 0x0000], "uint32") == 65536
+ assert decode([1], "bool") is True
+
+
+class FakeResult:
+ def __init__(self, registers=None, bits=None, error=False):
+ self.registers, self.bits, self._error = registers, bits, error
+
+ def isError(self):
+ return self._error
+
+
+class FakeClient:
+ def __init__(self, fail=False):
+ self.connected = False
+ self.fail = fail
+
+ async def connect(self):
+ self.connected = True
+
+ async def read_holding_registers(self, address, count, slave):
+ if self.fail:
+ return FakeResult(error=True)
+ return FakeResult(registers={3000: float_regs(481.0), 3002: float_regs(800.0)}[address])
+
+ async def read_discrete_inputs(self, address, count, slave):
+ return FakeResult(bits=[True])
+
+ def close(self):
+ pass
+
+
+def make_source(fail=False):
+ cfg = {"devices": [{
+ "meter_id": "UTIL-1", "host": "meter.example.internal", "unit_id": 1,
+ "registers": {
+ "voltage": {"address": 3000, "type": "float32"},
+ "current": {"address": 3002, "type": "float32"},
+ },
+ }]}
+ return ModbusSource(cfg, client_factory=lambda host, port, timeout: FakeClient(fail))
+
+
+def test_modbus_reading_shape():
+ out = asyncio.run(make_source().read_all(1_760_000_000.0))
+ r = out["UTIL-1"]
+ assert r["voltage"] == 481.0 and r["current"] == 800.0
+ assert r["load_pct"] == 50.0 # 800 A on a 1600 A service
+ assert r["comm_ok"] is True and r["status"] == "Online"
+
+
+def test_modbus_comm_failure_is_reported():
+ r = asyncio.run(make_source(fail=True).read_all(1_760_000_000.0))["UTIL-1"]
+ assert r["comm_ok"] is False and r["status"] == "Comm Fail"
+ assert r["voltage"] is None
diff --git a/backend/tests/test_simulator.py b/backend/tests/test_simulator.py
new file mode 100644
index 0000000..ccc8c9b
--- /dev/null
+++ b/backend/tests/test_simulator.py
@@ -0,0 +1,164 @@
+import pytest
+
+from config import METERS, METERS_BY_ID
+from simulator import SCENARIOS, FacilitySimulator
+
+T0 = 1_760_000_000.0
+FIELDS = ("voltage", "current", "power_kw", "power_factor", "frequency",
+ "energy_kwh", "demand_kw", "load_pct", "status", "timestamp", "epoch")
+
+
+def run(sim, start, seconds, step=1.0):
+ """Step the simulator and return the list of scans."""
+ out, t = [], start
+ while t < start + seconds:
+ out.append(sim.step(t))
+ t += step
+ return out
+
+
+def quiet_sim(seed=1):
+ return FacilitySimulator(seed=seed, auto_events=False)
+
+
+def test_every_meter_reports_all_fields():
+ r = quiet_sim().step(T0)
+ assert set(r) == {m["id"] for m in METERS}
+ for mid, reading in r.items():
+ assert reading["meter_id"] == mid
+ for f in FIELDS:
+ assert f in reading, f"{mid} missing {f}"
+
+
+def test_normal_operation_values_are_believable():
+ sim = quiet_sim()
+ for scan in run(sim, T0, 600, step=2):
+ for mid, r in scan.items():
+ cfg = METERS_BY_ID[mid]
+ if cfg["type"] == "generator":
+ assert r["status"] == "Standby" and r["voltage"] == 0
+ continue
+ assert abs(r["voltage"] - cfg["nominal_voltage"]) / cfg["nominal_voltage"] < 0.03
+ assert 59.9 < r["frequency"] < 60.1
+ assert 5 < r["load_pct"] < 75
+ assert 0.8 < r["power_factor"] <= 1.0
+ assert scan["ATS-1"]["position"] == "UTILITY"
+ assert not scan["UPS-A"]["on_battery"] and not scan["UPS-B"]["on_battery"]
+
+
+def test_power_balances_through_the_hierarchy():
+ r = quiet_sim().step(T0)
+ pdu_a = r["PDU-A1"]["power_kw"] + r["PDU-A2"]["power_kw"]
+ # UPS output covers its PDUs plus transformer losses
+ assert pdu_a < r["UPS-A"]["power_kw"] < pdu_a * 1.05
+ # Switchboard carries both UPS inputs, cooling, and house load
+ floor = r["UPS-A"]["input_kw"] + r["UPS-B"]["input_kw"] + r["MECH-1"]["power_kw"]
+ assert floor < r["MSB-1"]["power_kw"] < floor + 25
+ # On utility, the ATS and the utility meter see the switchboard load
+ assert r["UTIL-1"]["power_kw"] == r["ATS-1"]["power_kw"] == r["MSB-1"]["power_kw"]
+ # A plausible PUE for a small facility
+ it = sum(r[p]["power_kw"] for p in ("PDU-A1", "PDU-A2", "PDU-B1", "PDU-B2"))
+ assert 1.3 < r["MSB-1"]["power_kw"] / it < 2.0
+
+
+def test_values_move_and_energy_accumulates():
+ scans = run(quiet_sim(), T0, 300, step=2)
+ kw = {s["MSB-1"]["power_kw"] for s in scans}
+ assert len(kw) > 20
+ energy = [s["UTIL-1"]["energy_kwh"] for s in scans]
+ assert energy == sorted(energy) and energy[-1] > energy[0]
+
+
+def test_same_seed_is_repeatable():
+ a = run(quiet_sim(seed=42), T0, 60, step=2)
+ b = run(quiet_sim(seed=42), T0, 60, step=2)
+ assert a == b
+
+
+def test_utility_outage_sequence():
+ sim = quiet_sim()
+ sim.step(T0)
+ sim.trigger("utility_outage", T0 + 1, duration=60)
+ scans = run(sim, T0 + 1, 200, step=1)
+
+ def first(pred):
+ return next(i for i, s in enumerate(scans) if pred(s))
+
+ lost = first(lambda s: s["UTIL-1"]["source_failed"])
+ on_batt = first(lambda s: s["UPS-A"]["on_battery"] and s["UPS-B"]["on_battery"])
+ cranking = first(lambda s: s["GEN-1"]["gen_state"] == "CRANKING")
+ running = first(lambda s: s["GEN-1"]["gen_state"] == "RUNNING")
+ on_gen = first(lambda s: s["ATS-1"]["position"] == "GENERATOR")
+ ups_back = first(lambda s: s["ATS-1"]["position"] == "GENERATOR" and not s["UPS-A"]["on_battery"])
+ assert lost == on_batt == 0
+ assert 0 < cranking < running < on_gen < ups_back
+
+ # While on battery the critical IT load is still served
+ assert scans[on_batt]["PDU-A1"]["power_kw"] > 50
+ assert scans[on_batt]["MSB-1"]["power_kw"] == 0
+ # The block load step dips generator frequency
+ assert scans[on_gen]["GEN-1"]["frequency"] < 59.5
+ # On generator, the generator meter carries the facility
+ assert scans[ups_back + 5]["GEN-1"]["power_kw"] > 300
+ assert scans[ups_back + 5]["UTIL-1"]["power_kw"] == 0
+
+ back = first(lambda s: not s["UTIL-1"]["source_failed"])
+ retransfer = first(lambda s: s["ATS-1"]["position"] == "UTILITY" and scans.index(s) > on_gen)
+ assert retransfer - back >= 29 # retransfer delay
+ assert scans[-1]["GEN-1"]["gen_state"] == "STANDBY"
+ assert sim.scenario is None
+ assert scans[-1]["UPS-A"]["battery_soc"] < 100 # still recharging
+
+
+def test_voltage_sag_rides_through_on_battery_without_generator():
+ sim = quiet_sim()
+ sim.step(T0)
+ sim.trigger("voltage_sag", T0 + 1, duration=5, depth=0.85)
+ scans = run(sim, T0 + 1, 20)
+ assert min(s["UTIL-1"]["voltage"] for s in scans) < 432
+ assert any(s["UPS-A"]["on_battery"] for s in scans)
+ assert all(s["GEN-1"]["gen_state"] == "STANDBY" for s in scans)
+ assert not scans[-1]["UPS-A"]["on_battery"]
+
+
+def test_pdu_overload_crosses_alarm_limits():
+ sim = quiet_sim()
+ sim.step(T0)
+ sim.trigger("pdu_overload", T0 + 1, pdu="PDU-B2", target_pct=93, hold=60)
+ scans = run(sim, T0 + 1, 160, step=2)
+ peak = max(s["PDU-B2"]["load_pct"] for s in scans)
+ assert peak > 90
+ assert max(s["PDU-A1"]["load_pct"] for s in scans) < 75
+ assert scans[-1]["PDU-B2"]["load_pct"] < 75
+
+
+def test_power_factor_drop():
+ sim = quiet_sim()
+ sim.step(T0)
+ sim.trigger("pf_drop", T0 + 1, duration=60, pf=0.79)
+ scans = run(sim, T0 + 1, 80, step=2)
+ assert min(s["MECH-1"]["power_factor"] for s in scans) < 0.82
+ assert scans[-1]["MECH-1"]["power_factor"] > 0.85
+
+
+def test_trigger_validation():
+ sim = quiet_sim()
+ sim.step(T0)
+ with pytest.raises(KeyError):
+ sim.trigger("meteor_strike", T0)
+ sim.trigger("pf_drop", T0)
+ with pytest.raises(RuntimeError):
+ sim.trigger("voltage_sag", T0 + 1)
+
+
+def test_automatic_events_fire_and_finish():
+ sim = FacilitySimulator(seed=3, auto_events=True, event_interval=(20, 30), first_event_delay=5)
+ seen = set()
+ t = T0
+ for _ in range(1500):
+ sim.step(t)
+ if sim.scenario:
+ seen.add(sim.scenario["name"])
+ t += 2
+ assert seen <= set(SCENARIOS)
+ assert len(seen) >= 3
diff --git a/frontend/.env.example b/frontend/.env.example
index 5934e2e..d42e601 100644
--- a/frontend/.env.example
+++ b/frontend/.env.example
@@ -1 +1,6 @@
-VITE_API_URL=http://localhost:8000
+# Base URL of the backend API, read at build time.
+# Optional: a production build without it uses the live demo API
+# (https://power-dashboard-api.onrender.com). The dev server without it
+# proxies /api and /ws to http://localhost:8000.
+# VITE_API_URL=https://your-api.example.com
+VITE_API_URL=
diff --git a/frontend/index.html b/frontend/index.html
index 67b1b2f..b43a095 100644
--- a/frontend/index.html
+++ b/frontend/index.html
@@ -2,14 +2,25 @@
-
- Demo events are disabled on this server.
+ } + + const fire = async (name) => { + setBusy(name) + setError(null) + try { + await api.triggerEvent(name) + onDone?.() + } catch (e) { + setError(e.message) + } finally { + setBusy(null) + } + } + + return ( +{error}
} ++ A 2N data center of about 450 kW: a 480 V utility service and a 750 kW standby generator + feed an automatic transfer switch, then the main switchboard. Two 400 kVA UPS systems + (A side and B side) each feed two 150 kVA PDUs at 208 V, and the cooling plant runs from + its own panel. Values roll up from the IT load through transformer and UPS losses, so + the numbers add up the way a real site does. +
++ Events also happen on their own every few minutes. The event rate is faster than a real + site so there is always something to look at. +
++ Every alarm in the master alarm database is rationalized with a priority, a consequence, + and an operator response. Alarms follow the ISA 18.2 state model, use deadbands and + delays to prevent chatter, can be shelved for a limited time with a logged reason, and + some are suppressed by design (for example, generator alarms while the generator is off). + Every transition is written to the alarm journal. +
+
+ The same backend polls Modbus TCP meters when started with DATA_SOURCE=modbus and
+ a register map. The alarm engine, history, and this interface work the same either way.
+
Python, FastAPI, WebSocket, SQLite, React, Vite. Hosted on Render.
+ ++ This portfolio demo is hosted on a free tier that sleeps when nobody is using it. + The first visit after a quiet period takes about 30 to 60 seconds while the server + starts. After that, everything updates live every 2 seconds. +
+ )} + + +This is taking longer than usual. The server may still be starting.
+ +{error}
} +ISA 18.2 shelving is time limited. The alarm returns automatically when the timer runs out.
+ + ) +} + +function AlarmRow({ alarm, now }) { + const [open, setOpen] = useState(false) + const [shelving, setShelving] = useState(false) + const [error, setError] = useState(null) + const needsAck = NEEDS_ACK.includes(alarm.state) + + const act = useCallback(async (fn) => { + setError(null) + try { await fn() } catch (e) { setError(e.message) } + }, []) + + return ( +{error}
} + {open && ( +{empty}
Loading journal...
| Time | Event | Alarm | Priority | Value | State after | By |
|---|---|---|---|---|---|---|
| {dateTimeOf(e.timestamp)} | +{EVENT_LABELS[e.event] || e.event} | +
+ {e.message}
+ {e.tag}{e.note ? `, ${e.note}` : ''}
+ |
+ {e.value == null ? '--' : alarmValue({ unit: e.unit, kind: e.unit === '' && e.setpoint === 1 ? 'DISCRETE' : 'HI' }, e.value)} | +{STATES[e.state]?.short || e.state} | +{e.operator} | +|
| No journal entries |
Loading alarm database...
+ The master alarm database. Each alarm was rationalized: it has one cause, a priority based on + consequence and time to respond, and a documented operator response. Deadbands and delays + keep noisy signals from chattering. +
+| Tag | Alarm | Limit | Deadband | Delay on/off | Priority | Operator response |
|---|---|---|---|---|---|---|
| {d.tag} | +
+ {d.message}
+ {d.suppression && Suppressed when: {d.suppression} }
+ |
+ {setpointText(d)} | +{d.kind === 'DISCRETE' ? '--' : `${d.deadband} ${d.unit}`} | +{d.on_delay} / {d.off_delay} scans | +{d.response} | +
ISA 18.2 aligned alarm management system
+Alarm management following ANSI/ISA-18.2
+| Timestamp | -Meter | -Condition | -Value | -Threshold | -Priority | -Action | -
|---|---|---|---|---|---|---|
| - {tab === 'active' ? 'No active alarms' : 'No alarm history'} - |