alro65/AR-VMS-Seaman
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

sprint-4: Runtime server base — tag_store + historian + alarm engine + API FastAPI

Browse Source 
Arquitectura asincronica completa para correr 24/7 a bordo del buque.

vmssailor/runtime/server/tag_store.py
- TagStore in-memory con pub/sub asyncio.Queue
- register_tag/register_many con valores iniciales
- TagValue dataclass: value + quality + timestamp + raw_value
- subscribe()/unsubscribe() para fan-out
- stats() con breakdown por Quality

vmssailor/runtime/server/historian.py
- Historian DuckDB embebido (in-memory o archivo)
- Reader loop suscrito al tag_store + buffer + flush periodico (1s)
- query(tag_id, since, until, limit) para series temporales
- Soporta valores numericos y boolean separadamente

vmssailor/runtime/server/alarm_engine.py
- Suscriptor al tag_store que evalua AlarmConfig por update
- Operators: >, >=, <, <=, ==, !=
- Hysteresis correcta: aplica al SALIR de alarma, no a entrar
- Delay configurable (persistencia minima antes de disparar)
- Estados: ACTIVE -> ACK (con user) -> CLEARED
- ack(alarm_id, user) reconoce sin clear

vmssailor/runtime/server/drivers.py
- SimulatorDriver: produce valores sinteticos creibles por UnitSI
- Tick configurable (default 0.5s)
- Respeta range_normal_min/max del tag para mantenerse en rango
- Permite probar UI/API sin hardware ni Modbus real

vmssailor/runtime/server/runtime_app.py
- RuntimeApp dataclass ensambla todos los servicios
- build_runtime(project) construye listo para correr
- start()/stop() async lifecycle ordenado

vmssailor/runtime/server/api.py
- FastAPI app con lifespan que arranca/detiene el runtime
- GET /health, /project
- GET /tags, /tags/{id}, /tags/{id}/history
- GET /alarms, POST /alarms/{id}/ack
- WebSocket /ws/realtime con snapshot inicial + push + heartbeat

runtime_server_main.py
- Entry point con argparse: --vmsproj, --host, --port, --db
- Sin --vmsproj usa proyecto demo Sprint 0 (genera simulator vivo)
- Lanza con uvicorn

Tests (tests/runtime/, 16 nuevos, total 142/142):
- test_tag_store: register, update, subscribe, unsubscribe, stats
- test_historian: roundtrip query, stats
- test_alarm_engine: fire when below, hysteresis clears, ack
- test_api: health, project, tags listing, history, alarms via httpx.ASGITransport

Para correr el servidor en vivo:
    uv run python runtime_server_main.py --verbose
Luego en otro shell:
    curl http://127.0.0.1:8765/health
    curl http://127.0.0.1:8765/tags | jq .

Dependencias agregadas:
- fastapi >=0.110
- uvicorn[standard] >=0.27
- websockets >=12.0
- duckdb >=0.10
- pymodbus >=3.5 (Sprint 5)
- python-can >=4.3 (Sprint 5)
- httpx >=0.27 (testing + cliente HTTP)

142/142 pytest verde, ruff clean.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Alvaro Enrique Romero Donado
2026-05-17 20:03:19 -04:00
parent fbce1ecb42
commit 36dda85259
16 changed files with 1579 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
vmssailor/runtime/__init__.py
+9 -3
View File
@@ -1,5 +1,11 @@
"""vmssailor.runtime — Servidor 24/7 y cliente desktop a bordo (Sprint 4+).
"""vmssailor.runtime — Servidor 24/7 y cliente desktop a bordo.
En Sprint 0 está vacío. Sprint 4 trae el servicio Windows + drivers Modbus +
historian + alarm engine + API. Ver `VMS_Sailor_v2_Parte_03_Runtime.md`.
Sprint 4: servidor con tag_store + historian + alarm engine + API FastAPI +
simulator driver. Sprint 5: NMEA 2000 + Log Book. Sprint 6: cliente desktop.
Entry point: `runtime_server_main.py` en la raíz.
"""
from vmssailor.runtime.server.runtime_app import RuntimeApp, build_runtime
__all__ = ["RuntimeApp", "build_runtime"]
vmssailor/runtime/server/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Runtime servidor (servicio Windows 24/7 en el PC industrial del buque)."""
vmssailor/runtime/server/alarm_engine.py
+184
View File
@@ -0,0 +1,184 @@
"""Alarm engine — evalúa AlarmConfig contra valores del tag_store.
Para cada update de tag, recorre sus AlarmConfig y produce instancias
`Alarm` en estados ACTIVE/ACK/CLEARED.
Sprint 4: lógica básica. Sprint 8: escalación + permissives integration.
"""
from __future__ import annotations
import asyncio
import logging
from collections.abc import Callable
from contextlib import suppress
from datetime import UTC, datetime
from vmssailor.core.alarm import Alarm
from vmssailor.core.enums import AlarmState
from vmssailor.core.tag import AlarmConfig, Tag
from vmssailor.runtime.server.tag_store import TagStore, TagValue
from vmssailor.shared.ids import make_alarm_instance_id
logger = logging.getLogger(__name__)
def _evaluate(value: float, alarm: AlarmConfig) -> bool:
"""¿La condición de alarma está activa según el valor?"""
op = alarm.operator
t = alarm.threshold
if op == ">":
return value > t
if op == "<":
return value < t
if op == ">=":
return value >= t
if op == "<=":
return value <= t
if op == "==":
return value == t
if op == "!=":
return value != t
return False
def _evaluate_with_hysteresis(
value: float, alarm: AlarmConfig, currently_active: bool
) -> bool:
"""Aplica histéresis para no salir/entrar de alarma con jitter."""
base = _evaluate(value, alarm)
if alarm.hysteresis == 0:
return base
# Si actualmente activo: requiere cruzar threshold ± hysteresis para salir
if currently_active:
h = alarm.hysteresis
op = alarm.operator
if op in (">", ">="):
# Para salir, value debe ser < threshold - h
return value > (alarm.threshold - h)
if op in ("<", "<="):
return value < (alarm.threshold + h)
return base
class AlarmEngine:
"""Consume del tag_store y emite eventos de alarma."""
def __init__(
self,
tag_store: TagStore,
on_alarm_event: Callable[[Alarm], None] | None = None,
) -> None:
self._tag_store = tag_store
self._on_event = on_alarm_event or (lambda _a: None)
self._active_alarms: dict[str, Alarm] = {}
# tracking del primer momento que la condicion se vuelve verdadera
# para soportar `delay_seconds`
self._pending: dict[str, datetime] = {}
self._stop = False
self._task: asyncio.Task | None = None
# ----- Lifecycle ---------------------------------------------------
async def start(self) -> None:
self._sub_q = self._tag_store.subscribe(maxsize=4096)
self._task = asyncio.create_task(self._loop())
async def stop(self) -> None:
self._stop = True
if self._task is not None:
self._task.cancel()
with suppress(asyncio.CancelledError):
await self._task
self._tag_store.unsubscribe(self._sub_q)
async def _loop(self) -> None:
try:
while not self._stop:
tv = await self._sub_q.get()
await self._evaluate_tag(tv)
except asyncio.CancelledError:
pass
# ----- Evaluation --------------------------------------------------
async def _evaluate_tag(self, tv: TagValue) -> None:
if not isinstance(tv.value, (int, float)) or isinstance(tv.value, bool):
return
tag = self._tag_store.get_tag(tv.tag_id)
if tag is None or not tag.alarms:
return
for a in tag.alarms:
await self._evaluate_alarm(tag, a, float(tv.value), tv.timestamp)
async def _evaluate_alarm(
self,
tag: Tag,
config: AlarmConfig,
value: float,
ts: datetime,
) -> None:
key = f"{tag.id}|{config.id}"
currently_active = key in self._active_alarms
condition = _evaluate_with_hysteresis(value, config, currently_active)
if condition and not currently_active:
# Check delay: si delay_seconds > 0 requiere persistencia previa
if config.delay_seconds > 0:
first_seen = self._pending.get(key)
if first_seen is None:
self._pending[key] = ts
return
if (ts - first_seen).total_seconds() < config.delay_seconds:
return
# Disparar
self._pending.pop(key, None)
alarm = Alarm(
id=make_alarm_instance_id(tag.id, config.id, ts.timestamp()),
tag_id=tag.id,
alarm_config_id=config.id,
priority=config.priority,
state=AlarmState.ACTIVE,
timestamp_active=ts,
message=config.message or f"{tag.id} {config.operator} {config.threshold}",
value_at_trigger=value,
)
self._active_alarms[key] = alarm
self._on_event(alarm)
logger.info("ALARM ACTIVE %s = %s (%s)", tag.id, value, config.priority.value)
return
if not condition and currently_active:
# Clear
self._pending.pop(key, None)
alarm = self._active_alarms.pop(key)
cleared = alarm.model_copy(
update={"state": AlarmState.CLEARED, "timestamp_cleared": ts}
)
self._on_event(cleared)
logger.info("ALARM CLEARED %s", tag.id)
return
if not condition:
self._pending.pop(key, None)
def ack(self, alarm_id: str, user: str) -> Alarm | None:
"""Reconoce una alarma activa por su id."""
for key, a in list(self._active_alarms.items()):
if a.id == alarm_id:
ts = datetime.now(UTC)
acked = a.model_copy(
update={
"state": AlarmState.ACK,
"timestamp_ack": ts,
"acknowledged_by": user,
}
)
self._active_alarms[key] = acked
self._on_event(acked)
return acked
return None
def active_alarms(self) -> list[Alarm]:
return list(self._active_alarms.values())
vmssailor/runtime/server/api.py
+171
View File
@@ -0,0 +1,171 @@
"""API FastAPI del Runtime server.
- WebSocket: `/ws/realtime` — pub/sub de tag updates + alarm events
- REST: `/tags`, `/tags/{id}/history`, `/alarms`, `/health`, `/project`
"""
from __future__ import annotations
import asyncio
import logging
from contextlib import asynccontextmanager, suppress
from datetime import datetime
from typing import Any
from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect
from vmssailor.runtime.server.runtime_app import RuntimeApp
from vmssailor.version import __version__
logger = logging.getLogger(__name__)
def create_app(runtime: RuntimeApp) -> FastAPI:
"""Construye la FastAPI app y monta endpoints."""
@asynccontextmanager
async def lifespan(_app: FastAPI):
await runtime.start()
try:
yield
finally:
await runtime.stop()
app = FastAPI(
title="VMS-Sailor Runtime API",
version=__version__,
description="On-board Runtime API: tags, history, alarms, control.",
lifespan=lifespan,
)
app.state.runtime = runtime
# ----- Health ------------------------------------------------------
@app.get("/health")
def health() -> dict[str, Any]:
return {
"status": "ok",
"vmssailor_version": __version__,
"tag_store": runtime.tag_store.stats(),
"historian": runtime.historian.stats(),
"active_alarms": len(runtime.alarm_engine.active_alarms()),
"project": {
"id": runtime.project.id,
"name": runtime.project.name,
"vessel": runtime.project.vessel.name,
},
}
# ----- Project ----------------------------------------------------
@app.get("/project")
def project() -> dict[str, Any]:
p = runtime.project
return {
"id": p.id,
"name": p.name,
"customer": p.customer,
"vessel": {
"id": p.vessel.id,
"name": p.vessel.name,
"type": p.vessel.type.value,
"loa_m": p.vessel.length_overall_m,
"beam_m": p.vessel.beam_max_m,
"draft_m": p.vessel.draft_m,
},
"systems_enabled": [s.value for s in p.systems_enabled],
"stats": p.stats(),
}
# ----- Tags --------------------------------------------------------
@app.get("/tags")
def list_tags() -> list[dict[str, Any]]:
out: list[dict[str, Any]] = []
for tag_id, tag in runtime.tag_store.all_tags().items():
tv = runtime.tag_store.get(tag_id)
out.append(
{
"id": tag_id,
"description": tag.description,
"unit_si": tag.unit_si.value,
"controllable": tag.controllable,
"value": tv.value if tv else None,
"quality": tv.quality.value if tv else None,
"timestamp": tv.timestamp.isoformat() if tv else None,
"range_normal_min": tag.range_normal_min,
"range_normal_max": tag.range_normal_max,
}
)
return out
@app.get("/tags/{tag_id}")
def get_tag(tag_id: str) -> dict[str, Any]:
tag = runtime.tag_store.get_tag(tag_id)
tv = runtime.tag_store.get(tag_id)
if tag is None or tv is None:
raise HTTPException(status_code=404, detail=f"Tag '{tag_id}' no encontrado.")
return {
"id": tag_id,
"description": tag.description,
"unit_si": tag.unit_si.value,
"value": tv.value,
"quality": tv.quality.value,
"timestamp": tv.timestamp.isoformat(),
"alarms": [a.model_dump(mode="json") for a in tag.alarms],
}
@app.get("/tags/{tag_id}/history")
def history(
tag_id: str,
since: str | None = None,
until: str | None = None,
limit: int = 1000,
) -> list[dict[str, Any]]:
if tag_id not in runtime.tag_store:
raise HTTPException(status_code=404, detail=f"Tag '{tag_id}' no encontrado.")
since_dt = datetime.fromisoformat(since) if since else None
until_dt = datetime.fromisoformat(until) if until else None
return runtime.historian.query(tag_id, since=since_dt, until=until_dt, limit=limit)
# ----- Alarms ------------------------------------------------------
@app.get("/alarms")
def alarms(state: str | None = None) -> list[dict[str, Any]]:
active = runtime.alarm_engine.active_alarms()
if state == "active":
return [a.model_dump(mode="json") for a in active]
all_events = runtime.alarm_events
return [a.model_dump(mode="json") for a in all_events]
@app.post("/alarms/{alarm_id}/ack")
def ack_alarm(alarm_id: str, user: str = "anonymous") -> dict[str, Any]:
acked = runtime.alarm_engine.ack(alarm_id, user=user)
if acked is None:
raise HTTPException(status_code=404, detail=f"Alarma '{alarm_id}' no activa.")
return acked.model_dump(mode="json")
# ----- WebSocket ---------------------------------------------------
@app.websocket("/ws/realtime")
async def ws_realtime(ws: WebSocket) -> None:
await ws.accept()
q = runtime.tag_store.subscribe(maxsize=512)
# Push current snapshot first
for tv in runtime.tag_store.all_values().values():
await ws.send_json(tv.to_event_dict())
try:
while True:
try:
tv = await asyncio.wait_for(q.get(), timeout=30.0)
await ws.send_json(tv.to_event_dict())
except TimeoutError:
await ws.send_json({"type": "heartbeat"})
except WebSocketDisconnect:
pass
finally:
runtime.tag_store.unsubscribe(q)
with suppress(Exception):
await ws.close()
return app
vmssailor/runtime/server/drivers.py
+114
View File
@@ -0,0 +1,114 @@
"""Drivers de protocolo (Sprint 4: Simulator).
Sprint 5 traerá los drivers reales:
- modbus_rtu.py (pymodbus, serial COM)
- nmea2000.py (python-can)
- j1939.py
- card_discovery.py
Por ahora, `SimulatorDriver` produce valores sintéticos creíbles para los
tags registrados. Sirve para:
- Probar el alarm engine + historian sin hardware
- Probar el cliente desktop y mobile contra datos vivos
- Tests de integración
"""
from __future__ import annotations
import asyncio
import logging
import math
import random
from collections.abc import Callable
from contextlib import suppress
from vmssailor.core.enums import Quality, UnitSI
from vmssailor.runtime.server.tag_store import TagStore
logger = logging.getLogger(__name__)
class SimulatorDriver:
"""Genera valores sintéticos para todos los tags numéricos del store.
Estrategia por unidad:
- rpm: oscila entre 1200 y 1800 ± ruido
- bar: estable cerca del medio del rango con jitter
- C (temperatura): se calienta hacia 80 °C con jitter
- %: 30-70 % oscilación
- V: 27-28 V (24V system) o 13-14 V (12V system)
- bool: random toggle ocasional
Tags pueden tener `range_normal_min/max`; respeta esos rangos.
"""
def __init__(self, tag_store: TagStore, tick_period_s: float = 0.5) -> None:
self._store = tag_store
self._tick_s = tick_period_s
self._t0 = 0.0
self._stop = False
self._task: asyncio.Task | None = None
self._on_tick: Callable[[], None] | None = None
async def start(self) -> None:
self._task = asyncio.create_task(self._loop())
async def stop(self) -> None:
self._stop = True
if self._task is not None:
self._task.cancel()
with suppress(asyncio.CancelledError):
await self._task
async def _loop(self) -> None:
try:
while not self._stop:
await self._tick()
await asyncio.sleep(self._tick_s)
except asyncio.CancelledError:
pass
async def _tick(self) -> None:
self._t0 += self._tick_s
for tag_id, tag in self._store.all_tags().items():
value = self._synthesize(tag_id, tag.unit_si, tag)
if value is None:
continue
await self._store.update(tag_id, value, quality=Quality.GOOD)
if self._on_tick is not None:
self._on_tick()
def _synthesize(self, tag_id: str, unit: UnitSI, tag) -> float | bool | None:
t = self._t0
if unit == UnitSI.RPM:
base = 1500
return base + 250 * math.sin(t / 30 + hash(tag_id) % 100) + random.uniform(-15, 15)
if unit == UnitSI.BAR:
mid = (
(tag.range_normal_min + tag.range_normal_max) / 2
if tag.range_normal_min is not None and tag.range_normal_max is not None
else 4.5
)
return mid + 0.2 * math.sin(t / 5 + hash(tag_id) % 50) + random.uniform(-0.05, 0.05)
if unit == UnitSI.DEGREE_CELSIUS:
target = 82.0
return target + 8 * math.sin(t / 60 + hash(tag_id) % 30) + random.uniform(-0.5, 0.5)
if unit == UnitSI.PERCENT:
return 50 + 25 * math.sin(t / 20 + hash(tag_id) % 20) + random.uniform(-2, 2)
if unit == UnitSI.VOLT:
mid = (
(tag.range_normal_min + tag.range_normal_max) / 2
if tag.range_normal_min is not None and tag.range_normal_max is not None
else 27.5
)
return mid + random.uniform(-0.2, 0.2)
if unit == UnitSI.HERTZ:
return 50.0 + random.uniform(-0.1, 0.1)
if unit == UnitSI.AMPERE:
return 50 + 30 * math.sin(t / 25 + hash(tag_id) % 40)
if unit == UnitSI.HOUR:
return 1200 + t / 3600 # accumulating engine hours
if unit == UnitSI.BOOL:
# 99% del tiempo False, eventualmente True
return random.random() < 0.02
return None
vmssailor/runtime/server/historian.py
+165
View File
@@ -0,0 +1,165 @@
"""Historian DuckDB embebido.
Almacena series temporales de los tags con retención configurable.
Para Sprint 4 implementamos:
- Inserción batch (cada 1s) desde el subscriber del tag_store
- Query simple por tag_id + rango temporal
Sprint 5+: downsampling automático, compresión.
"""
from __future__ import annotations
import asyncio
import logging
from contextlib import suppress
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any
import duckdb
from vmssailor.runtime.server.tag_store import TagStore, TagValue
logger = logging.getLogger(__name__)
_SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS samples (
tag_id VARCHAR NOT NULL,
timestamp TIMESTAMP NOT NULL,
value DOUBLE,
bool_value BOOLEAN,
quality VARCHAR NOT NULL DEFAULT 'good'
);
CREATE INDEX IF NOT EXISTS idx_samples_tag_ts ON samples(tag_id, timestamp);
"""
class Historian:
"""DuckDB-backed historian. Apenas un wrapper async-friendly."""
def __init__(self, db_path: Path | str | None = None) -> None:
if db_path is None:
db_path = ":memory:"
self._db_path = str(db_path)
self._conn = duckdb.connect(self._db_path)
self._conn.execute(_SCHEMA_SQL)
self._buffer: list[tuple] = []
self._buffer_lock = asyncio.Lock()
self._flush_task: asyncio.Task | None = None
self._stopping = False
# ----- Lifecycle ---------------------------------------------------
async def start(self, tag_store: TagStore) -> None:
"""Comienza a suscribirse al tag store y a flush periodico."""
self._tag_store = tag_store
self._subscriber_q = tag_store.subscribe(maxsize=8192)
self._reader_task = asyncio.create_task(self._reader_loop())
self._flush_task = asyncio.create_task(self._flush_loop())
async def stop(self) -> None:
self._stopping = True
for t in (getattr(self, "_reader_task", None), self._flush_task):
if t is not None:
t.cancel()
with suppress(asyncio.CancelledError):
await t
if hasattr(self, "_subscriber_q"):
self._tag_store.unsubscribe(self._subscriber_q)
await self._flush(force=True)
self._conn.close()
# ----- Loops -------------------------------------------------------
async def _reader_loop(self) -> None:
while not self._stopping:
try:
tv: TagValue = await self._subscriber_q.get()
except asyncio.CancelledError:
break
await self._enqueue(tv)
async def _flush_loop(self) -> None:
try:
while not self._stopping:
await asyncio.sleep(1.0)
await self._flush()
except asyncio.CancelledError:
pass
async def _enqueue(self, tv: TagValue) -> None:
async with self._buffer_lock:
value_num: float | None
value_bool: bool | None
if isinstance(tv.value, bool):
value_num = None
value_bool = tv.value
elif isinstance(tv.value, (int, float)):
value_num = float(tv.value)
value_bool = None
else:
value_num = None
value_bool = None
self._buffer.append(
(tv.tag_id, tv.timestamp, value_num, value_bool, tv.quality.value)
)
async def _flush(self, force: bool = False) -> None:
async with self._buffer_lock:
if not self._buffer:
return
batch = self._buffer
self._buffer = []
try:
# DuckDB executemany es síncrono pero rápido
self._conn.executemany(
"INSERT INTO samples (tag_id, timestamp, value, bool_value, quality) "
"VALUES (?, ?, ?, ?, ?)",
batch,
)
except Exception:
logger.exception("Failed to flush %d samples", len(batch))
# ----- Queries -----------------------------------------------------
def query(
self,
tag_id: str,
*,
since: datetime | None = None,
until: datetime | None = None,
limit: int | None = None,
) -> list[dict[str, Any]]:
"""Consulta histórica de un tag."""
if since is None:
since = datetime.now(UTC) - timedelta(hours=1)
if until is None:
until = datetime.now(UTC)
sql = (
"SELECT tag_id, timestamp, value, bool_value, quality "
"FROM samples WHERE tag_id = ? AND timestamp >= ? AND timestamp <= ? "
"ORDER BY timestamp"
)
if limit:
sql += f" LIMIT {int(limit)}"
rows = self._conn.execute(sql, [tag_id, since, until]).fetchall()
result: list[dict[str, Any]] = []
for r in rows:
value = r[2] if r[2] is not None else r[3]
result.append(
{
"tag_id": r[0],
"timestamp": r[1].isoformat(),
"value": value,
"quality": r[4],
}
)
return result
def stats(self) -> dict[str, Any]:
total = self._conn.execute("SELECT COUNT(*) FROM samples").fetchone()[0]
tags = self._conn.execute("SELECT COUNT(DISTINCT tag_id) FROM samples").fetchone()[0]
return {"total_samples": int(total), "tags": int(tags), "db_path": self._db_path}
vmssailor/runtime/server/runtime_app.py
+73
View File
@@ -0,0 +1,73 @@
"""Orquestador del Runtime server: ensambla tag_store + historian + alarm engine + drivers.
Sprint 4: simulator-only. Sprint 5: Modbus + NMEA 2000 reales.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from pathlib import Path
from vmssailor.core.alarm import Alarm
from vmssailor.core.project import Project
from vmssailor.runtime.server.alarm_engine import AlarmEngine
from vmssailor.runtime.server.drivers import SimulatorDriver
from vmssailor.runtime.server.historian import Historian
from vmssailor.runtime.server.tag_store import TagStore
logger = logging.getLogger(__name__)
@dataclass(slots=True)
class RuntimeApp:
"""Conjunto de servicios del Runtime. Se construye con `build_runtime()`."""
project: Project
tag_store: TagStore
historian: Historian
alarm_engine: AlarmEngine
simulator: SimulatorDriver
alarm_events: list[Alarm] = field(default_factory=list)
async def start(self) -> None:
await self.historian.start(self.tag_store)
await self.alarm_engine.start()
await self.simulator.start()
logger.info("RuntimeApp started — %d tags, simulator @ 0.5s", len(self.tag_store))
async def stop(self) -> None:
await self.simulator.stop()
await self.alarm_engine.stop()
await self.historian.stop()
logger.info("RuntimeApp stopped")
def build_runtime(
project: Project,
*,
historian_db: Path | str | None = None,
simulator_tick_s: float = 0.5,
) -> RuntimeApp:
"""Construye un Runtime listo para usar (no arrancado todavía)."""
tag_store = TagStore()
tag_store.register_many(project.tags)
historian = Historian(historian_db)
alarm_events: list[Alarm] = []
def _on_alarm_event(alarm: Alarm) -> None:
alarm_events.append(alarm)
alarm_engine = AlarmEngine(tag_store, on_alarm_event=_on_alarm_event)
simulator = SimulatorDriver(tag_store, tick_period_s=simulator_tick_s)
return RuntimeApp(
project=project,
tag_store=tag_store,
historian=historian,
alarm_engine=alarm_engine,
simulator=simulator,
alarm_events=alarm_events,
)
vmssailor/runtime/server/tag_store.py
+183
View File
@@ -0,0 +1,183 @@
"""Tag store en memoria con pub/sub asíncrono.
Esta es la "fuente de verdad" del estado del buque en tiempo real.
Cada Tag está registrado por su `id` y tiene:
- Valor actual + calidad + timestamp
- Lista de subscribers asyncio.Queue para fan-out a:
- Alarm engine
- Historian
- WebSocket clients
- Authority/permissive engines
"""
from __future__ import annotations
import asyncio
import logging
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any
from vmssailor.core.enums import Quality
from vmssailor.core.tag import Tag
logger = logging.getLogger(__name__)
@dataclass(slots=True)
class TagValue:
"""Snapshot del valor actual de un tag."""
tag_id: str
value: float | bool | str | None
quality: Quality
timestamp: datetime
raw_value: float | None = None # valor crudo pre-escalado, útil para debug
def to_event_dict(self) -> dict[str, Any]:
"""Formato para enviar por WebSocket / alarmas / historian."""
return {
"type": "tag_update",
"tag_id": self.tag_id,
"value": self.value,
"quality": self.quality.value,
"timestamp": self.timestamp.isoformat(),
}
@dataclass(slots=True)
class _TagEntry:
tag: Tag
value: TagValue
last_quality_change: datetime = field(default_factory=lambda: datetime.now(UTC))
class TagStore:
"""Estado de runtime en memoria con pub/sub asíncrono.
Uso:
store = TagStore()
store.register_tag(tag_config)
await store.update("ME_PORT.OIL_PRESS", 4.8)
async for ev in store.subscribe():
...
"""
def __init__(self) -> None:
self._entries: dict[str, _TagEntry] = {}
self._subscribers: set[asyncio.Queue[TagValue]] = set()
self._lock = asyncio.Lock()
# ----- Registro -----------------------------------------------------
def register_tag(self, tag: Tag, initial_value: Any = None) -> None:
"""Registra un tag con valor inicial.
Si ya existe, mantiene el valor actual.
"""
if tag.id in self._entries:
return
ts = datetime.now(UTC)
self._entries[tag.id] = _TagEntry(
tag=tag,
value=TagValue(
tag_id=tag.id,
value=initial_value,
quality=Quality.UNCERTAIN if initial_value is None else Quality.GOOD,
timestamp=ts,
),
)
def register_many(self, tags: list[Tag]) -> None:
for t in tags:
self.register_tag(t)
# ----- Lectura ------------------------------------------------------
def get(self, tag_id: str) -> TagValue | None:
entry = self._entries.get(tag_id)
return entry.value if entry else None
def get_tag(self, tag_id: str) -> Tag | None:
entry = self._entries.get(tag_id)
return entry.tag if entry else None
def all_values(self) -> dict[str, TagValue]:
return {tid: e.value for tid, e in self._entries.items()}
def all_tags(self) -> dict[str, Tag]:
return {tid: e.tag for tid, e in self._entries.items()}
def __len__(self) -> int:
return len(self._entries)
def __contains__(self, tag_id: str) -> bool:
return tag_id in self._entries
# ----- Escritura ----------------------------------------------------
async def update(
self,
tag_id: str,
value: float | bool | str,
*,
quality: Quality = Quality.GOOD,
raw_value: float | None = None,
timestamp: datetime | None = None,
) -> TagValue:
"""Actualiza el valor de un tag y notifica a todos los subscribers."""
entry = self._entries.get(tag_id)
if entry is None:
raise KeyError(f"Tag '{tag_id}' no está registrado en el tag_store.")
ts = timestamp or datetime.now(UTC)
new_value = TagValue(
tag_id=tag_id,
value=value,
quality=quality,
timestamp=ts,
raw_value=raw_value,
)
async with self._lock:
old = entry.value
entry.value = new_value
if old.quality != quality:
entry.last_quality_change = ts
# Notificar a todos los subscribers (no-blocking)
for q in list(self._subscribers):
try:
q.put_nowait(new_value)
except asyncio.QueueFull:
logger.warning("Subscriber queue full, dropping update for %s", tag_id)
return new_value
# ----- Pub/Sub ------------------------------------------------------
def subscribe(self, maxsize: int = 1024) -> asyncio.Queue[TagValue]:
"""Devuelve una Queue donde van los updates.
El subscriber DEBE llamar `unsubscribe(queue)` al terminar para
liberar la referencia.
"""
q: asyncio.Queue[TagValue] = asyncio.Queue(maxsize=maxsize)
self._subscribers.add(q)
return q
def unsubscribe(self, queue: asyncio.Queue[TagValue]) -> None:
self._subscribers.discard(queue)
# ----- Stats --------------------------------------------------------
def stats(self) -> dict[str, Any]:
from collections import Counter
quality_counts: Counter[str] = Counter()
for e in self._entries.values():
quality_counts[e.value.quality.value] += 1
return {
"total_tags": len(self._entries),
"subscribers": len(self._subscribers),
"by_quality": dict(quality_counts),
}