diff --git a/CHANGELOG.md b/CHANGELOG.md index 85794f7..bbd44de 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,17 @@ Versionado: una entrada por **entrega académica** del semestre (no SemVer estri ## [Unreleased] +### Added — H4 fase 3: modo simulación (RF-12, 2026-05-21) +- Nuevo módulo [`application/simulacion.py`](core-python/src/sentinel_dispatch/application/simulacion.py) con value object `ReporteSimulacion` (dataclass frozen+slots) que agrega: `incidentes_procesados`, tupla de `ResultadoDespacho`, porcentajes por motivo (`pct_optimo`, `pct_penalizado`, `pct_suboptimo_rn02`, `pct_saturacion`), ETA media y ETA p95. +- Función `simular(incidentes, flota_ficticia, grafo, *, repositorio_eventos=None, factor_hora=1.0, factor_sirena=1.0) → ReporteSimulacion`. **Semántica v1**: sin evolución temporal entre incidentes (cada uno ve la flota inicial). Determinístico: el resultado depende sólo de los inputs. +- Persistencia **opt-in**: por default NO escribe al log canónico (modo simulación ≠ operativo). Si se provee `repositorio_eventos`, se persisten eventos `despacho_creado` con `despacho_id` prefijado `SD-SIM-` para distinguirlos de despachos operativos. +- Nuevo subcomando CLI [`interfaces/cli/simular_cmd.py`](core-python/src/sentinel_dispatch/interfaces/cli/simular_cmd.py): `sentinel simular --flota --incidentes --graph --out [--persistir-en]`. El reporte JSON de salida incluye campo `"modo": "simulacion"` como marca explícita. +- Tests: **7 nuevos** verdes — Normal (2): resultados+métricas, ausencia de evolución temporal · Borde (2): lista vacía, flota vacía=100% saturación · ReglasNegocio (2): default no escribe, con repo escribe N eventos al archivo separado · Métricas (1): pcts suman 100.0. Suite total **256/256** verde; cobertura global **90.60 %**. + +### Changed — H4 fase 3 +- `docs/quality/trazabilidad.md`: **RF-12 marcado ✅** apuntando a `application/simulacion.py` + `interfaces/cli/simular_cmd.py`. +- `interfaces/cli/app.py`: registro del subcomando `simular`. + ### Added — H4 fase 2: exportador CSV/JSON (RF-11, 2026-05-21) - Nuevo adapter [`adapters/exportador.py`](core-python/src/sentinel_dispatch/adapters/exportador.py): funciones puras `exportar_a_csv(eventos, path)` y `exportar_a_json(eventos, path)`. CSV con flatten de `payload_*` (e.g. `payload_costo_total`) y encoding `utf-8-sig` (BOM para que Excel español abra correctamente). JSON como array indentado sin BOM. Helper `_aplanar_dict(d, prefijo)` aplana dicts recursivamente; listas (e.g. `ruta`) se serializan como JSON string en una sola celda. - Nuevo subcomando CLI [`interfaces/cli/export_cmd.py`](core-python/src/sentinel_dispatch/interfaces/cli/export_cmd.py): `sentinel export --formato {csv,json} --in eventos.jsonl --out reporte.{csv,json}`. Enum `FormatoExport(csv|json)` para validación de argumento. Exit 0 en éxito, 2 si `--in` no existe o el JSONL es corrupto. diff --git a/core-python/src/sentinel_dispatch/application/simulacion.py b/core-python/src/sentinel_dispatch/application/simulacion.py new file mode 100644 index 0000000..ab88cb8 --- /dev/null +++ b/core-python/src/sentinel_dispatch/application/simulacion.py @@ -0,0 +1,197 @@ +"""Modo simulación (RF-12) — ejecuta el cálculo de despacho sobre flota ficticia. + +El SRS sec. RF-12 exige un modo que "ejecuta el cálculo completo sobre +un estado de flota ficticio sin afectar el estado operativo real". Este +módulo implementa la semántica más simple compatible con esa exigencia: + +- **Sin evolución temporal entre incidentes**: cada incidente ve la flota + ficticia en su estado inicial. Equivalente a correr ``run-dataset`` N + veces sobre el mismo grafo + flota. Esta interpretación literal del + SRS es suficiente para v1 (academic). Para v2 con reloj virtual y + liberación de unidades por ``eta_segundos`` se necesitaría un ADR + nuevo y framework de event-driven simulation. + +- **Persistencia opt-in**: por default NO escribe al log canónico + (modo simulación ≠ modo operativo). Si el caller provee + ``repositorio_eventos``, se escribe ahí (típicamente un archivo + separado tipo ``eventos_sim.jsonl`` para no contaminar el log real). + +- **Reporte agregado**: el output incluye los resultados crudos + + métricas porcentuales por motivo y ETA media/p95 para defensa + académica del módulo. +""" + +from __future__ import annotations + +import math +import statistics +from dataclasses import dataclass +from datetime import UTC, datetime +from typing import TYPE_CHECKING + +from sentinel_dispatch.application.despachar_ambulancia import despachar +from sentinel_dispatch.application.serializacion import serializar_resultado_despacho +from sentinel_dispatch.application.tipos import MotivoDespacho, ResultadoDespacho +from sentinel_dispatch.ports.repositorio_eventos import EventoLog, TipoEvento + +if TYPE_CHECKING: + from collections.abc import Sequence + + from sentinel_dispatch.adapters.repositorio_jsonl import JsonlRepositorioEventos + from sentinel_dispatch.domain.dispatch.tipos import Incidente, Unidad + from sentinel_dispatch.domain.routing.grafo_vial import GrafoVial + + +@dataclass(frozen=True, slots=True) +class ReporteSimulacion: + """Resumen inmutable de una corrida de simulación. + + Attributes: + incidentes_procesados: número de incidentes ejecutados. + resultados: tupla con un :class:`ResultadoDespacho` por incidente, + en el orden en que se procesaron. + pct_optimo: porcentaje de despachos en motivo ``OPTIMO`` (0-100). + pct_penalizado: porcentaje en motivo ``PENALIZADO``. + pct_suboptimo_rn02: porcentaje en motivo ``SUBOPTIMO_RN02``. + pct_saturacion: porcentaje en motivo ``SATURACION``. + eta_media_s: ETA promedio sobre los despachos no saturados. + ``None`` si todos los incidentes fueron saturación. + eta_p95_s: percentil 95 de las ETAs sobre los despachos no + saturados. ``None`` si N < 1 o todos saturación. + """ + + incidentes_procesados: int + resultados: tuple[ResultadoDespacho, ...] + pct_optimo: float + pct_penalizado: float + pct_suboptimo_rn02: float + pct_saturacion: float + eta_media_s: float | None + eta_p95_s: float | None + + +def _calcular_metricas( + resultados: Sequence[ResultadoDespacho], +) -> tuple[float, float, float, float, float | None, float | None]: + """Devuelve `(pct_optimo, pct_penalizado, pct_suboptimo_rn02, pct_saturacion, eta_media, eta_p95)`.""" + n = len(resultados) + if n == 0: + return 0.0, 0.0, 0.0, 0.0, None, None + + conteos = dict.fromkeys(MotivoDespacho, 0) + for r in resultados: + conteos[r.motivo] += 1 + + pct = {motivo: 100.0 * c / n for motivo, c in conteos.items()} + + etas = [ + r.costo_elegida.t_viaje_s + for r in resultados + if r.motivo is not MotivoDespacho.SATURACION + and r.costo_elegida is not None + and math.isfinite(r.costo_elegida.t_viaje_s) + ] + if etas: + eta_media: float | None = statistics.fmean(etas) + # p95 con interpolación lineal estándar; statistics no expone percentile, + # uso ordered + índice clamp por simplicidad (N pequeño, exacto). + ordenado = sorted(etas) + idx_p95 = max(0, math.ceil(0.95 * len(ordenado)) - 1) + eta_p95: float | None = ordenado[idx_p95] + else: + eta_media = None + eta_p95 = None + + return ( + pct[MotivoDespacho.OPTIMO], + pct[MotivoDespacho.PENALIZADO], + pct[MotivoDespacho.SUBOPTIMO_RN02], + pct[MotivoDespacho.SATURACION], + eta_media, + eta_p95, + ) + + +def _persistir_evento( + repo: JsonlRepositorioEventos, + resultado: ResultadoDespacho, + incidente: Incidente, +) -> None: + """Persiste un evento ``despacho_creado`` por resultado (ADR-0018).""" + despacho_id = ( + f"SD-SIM-{incidente.timestamp_iso[:10].replace('-', '')}-{incidente.id.replace('I-', '')}" + ) + evento = EventoLog( + evento_id=repo.generar_evento_id(base_ts=datetime.now(UTC)), + timestamp_iso=incidente.timestamp_iso, + tipo=TipoEvento.DESPACHO_CREADO, + despacho_id=despacho_id, + incidente_id=incidente.id, + payload=serializar_resultado_despacho(resultado), + ) + repo.append(evento) + + +def simular( + incidentes: Sequence[Incidente], + flota_ficticia: Sequence[Unidad], + grafo: GrafoVial, + *, + repositorio_eventos: JsonlRepositorioEventos | None = None, + factor_hora: float = 1.0, + factor_sirena: float = 1.0, +) -> ReporteSimulacion: + """Ejecuta el cálculo de despacho sobre cada incidente sin tocar el sistema real. + + Args: + incidentes: secuencia de incidentes a procesar (deberían ya tener + la categoría MPDS calculada). + flota_ficticia: secuencia de unidades **ficticias** a usar como + estado inicial. Es responsabilidad del caller verificar que + esta flota no se cruza con la flota operativa real (por + convención v1, basta con ``id`` distintos). + grafo: :class:`GrafoVial` ya cargado (usualmente la región + real — la "ficción" está en la flota, no en el mapa). + repositorio_eventos: opcional. Si se provee, cada despacho + persiste un evento ``despacho_creado`` ahí. Default ``None`` + = no persiste; **la simulación no afecta el log canónico + del modo operativo** (cumple "sin afectar el estado real"). + factor_hora: multiplicador de tráfico horario para el A*. + factor_sirena: multiplicador de sirena. + + Returns: + :class:`ReporteSimulacion` con los resultados + métricas. + + Notas semánticas (decisiones v1): + - **Sin estado evolutivo entre incidentes**: cada incidente ve + la flota ficticia tal como llegó al inicio. Si una unidad fue + "despachada" al I-01, sigue ``DISPONIBLE`` al evaluar I-02. + Equivalente a paralelizar conceptualmente. + - **Determinismo**: el resultado depende sólo de los inputs, + no del wall-clock. Apto para tests y para reproducir corridas. + """ + resultados: list[ResultadoDespacho] = [] + for incidente in incidentes: + resultado = despachar( + incidente, + flota_ficticia, + grafo, + factor_hora=factor_hora, + factor_sirena=factor_sirena, + ) + resultados.append(resultado) + if repositorio_eventos is not None: + _persistir_evento(repositorio_eventos, resultado, incidente) + + pct_opt, pct_pen, pct_sub, pct_sat, eta_media, eta_p95 = _calcular_metricas(resultados) + + return ReporteSimulacion( + incidentes_procesados=len(resultados), + resultados=tuple(resultados), + pct_optimo=pct_opt, + pct_penalizado=pct_pen, + pct_suboptimo_rn02=pct_sub, + pct_saturacion=pct_sat, + eta_media_s=eta_media, + eta_p95_s=eta_p95, + ) diff --git a/core-python/src/sentinel_dispatch/interfaces/cli/app.py b/core-python/src/sentinel_dispatch/interfaces/cli/app.py index 10dc516..d51e60f 100644 --- a/core-python/src/sentinel_dispatch/interfaces/cli/app.py +++ b/core-python/src/sentinel_dispatch/interfaces/cli/app.py @@ -13,7 +13,12 @@ import typer -from sentinel_dispatch.interfaces.cli import export_cmd, run_dataset_cmd, triaje_cmd +from sentinel_dispatch.interfaces.cli import ( + export_cmd, + run_dataset_cmd, + simular_cmd, + triaje_cmd, +) app = typer.Typer( name="sentinel", @@ -25,6 +30,7 @@ app.add_typer(triaje_cmd.app, name="triaje") app.command("run-dataset")(run_dataset_cmd.run_dataset) app.command("export")(export_cmd.export) +app.command("simular")(simular_cmd.simular_cmd) if __name__ == "__main__": # pragma: no cover diff --git a/core-python/src/sentinel_dispatch/interfaces/cli/simular_cmd.py b/core-python/src/sentinel_dispatch/interfaces/cli/simular_cmd.py new file mode 100644 index 0000000..7d7092d --- /dev/null +++ b/core-python/src/sentinel_dispatch/interfaces/cli/simular_cmd.py @@ -0,0 +1,187 @@ +"""Subcomando ``sentinel simular`` — modo simulación con flota ficticia (RF-12). + +Equivale a ``run-dataset`` pero con dos garantías adicionales: + +1. **Etiqueta semántica explícita**: el comando se llama ``simular``, + marcando claramente al operador que esto NO es ejecución operativa. +2. **Persistencia opt-in a archivo separado**: si se usa ``--persistir-en``, + los eventos van a un archivo distinto del log canónico operativo + (sugerencia: ``data/eventos_sim.jsonl``), preservando "sin afectar + el estado operativo real" (SRS RF-12). + +Uso:: + + sentinel simular --flota data/ficticias.json --incidentes data/scenario.json \\ + --graph data/graphs/coquimbo.graphml --out reporte.json + + sentinel simular --flota ... --incidentes ... --persistir-en data/eventos_sim.jsonl +""" + +from __future__ import annotations + +import json +from pathlib import Path # noqa: TC003 — Typer inspecciona el tipo en runtime. +from typing import Annotated, Any, cast + +import typer + +from sentinel_dispatch.adapters.grafo_osmnx import OsmnxGrafoVial, cargar_grafo_iv_region +from sentinel_dispatch.adapters.repositorio_jsonl import JsonlRepositorioEventos +from sentinel_dispatch.application.serializacion import serializar_resultado_despacho +from sentinel_dispatch.application.simulacion import ReporteSimulacion, simular +from sentinel_dispatch.domain.dispatch.tipos import ( + EstadoUnidad, + Incidente, + TipoUnidad, + Unidad, +) +from sentinel_dispatch.domain.triaje.tipos import CategoriaMPDS + + +def _validar_path(path: Path, etiqueta: str) -> None: + if not path.exists(): + typer.secho( + f"Error: archivo de {etiqueta} no encontrado: {path}", + fg=typer.colors.RED, + err=True, + ) + raise typer.Exit(code=2) + + +def _cargar_json_lista(path: Path, etiqueta: str) -> list[dict[str, Any]]: + try: + return cast("list[dict[str, Any]]", json.loads(path.read_text(encoding="utf-8"))) + except json.JSONDecodeError as exc: + typer.secho( + f"Error: {etiqueta} JSON inválido — {exc.msg} (línea {exc.lineno}).", + fg=typer.colors.RED, + err=True, + ) + raise typer.Exit(code=2) from exc + + +def _unidad_desde_dict(data: dict[str, Any]) -> Unidad: + return Unidad( + id=data["id"], + patente=data["patente"], + tipo=TipoUnidad(data["tipo"]), + base_nombre=data["base_nombre"], + base_lat=float(data["base_lat"]), + base_lon=float(data["base_lon"]), + estado=EstadoUnidad(data["estado"]), + ) + + +def _incidente_desde_dict(data: dict[str, Any]) -> Incidente: + categoria = CategoriaMPDS(data["ground_truth"]["categoria_mpds"]) + return Incidente( + id=data["id"], + lat=float(data["lat"]), + lon=float(data["lon"]), + categoria_mpds=categoria, + timestamp_iso=data["timestamp"], + ) + + +def _serializar_reporte(reporte: ReporteSimulacion) -> dict[str, Any]: + """Convierte un :class:`ReporteSimulacion` a dict serializable.""" + return { + "modo": "simulacion", + "incidentes_procesados": reporte.incidentes_procesados, + "metricas": { + "pct_optimo": reporte.pct_optimo, + "pct_penalizado": reporte.pct_penalizado, + "pct_suboptimo_rn02": reporte.pct_suboptimo_rn02, + "pct_saturacion": reporte.pct_saturacion, + "eta_media_s": reporte.eta_media_s, + "eta_p95_s": reporte.eta_p95_s, + }, + "resultados": [serializar_resultado_despacho(r) for r in reporte.resultados], + } + + +def simular_cmd( + incidentes_path: Annotated[ + Path, + typer.Option( + "--incidentes", + help="Path al JSON con los incidentes a simular.", + ), + ], + flota_path: Annotated[ + Path, + typer.Option( + "--flota", + help="Path al JSON con la flota ficticia.", + ), + ], + graph_path: Annotated[ + Path, + typer.Option( + "--graph", + help="Path al GraphML del grafo vial.", + ), + ], + out_path: Annotated[ + Path, + typer.Option( + "--out", + help="Path al JSON de reporte de simulación a crear.", + ), + ], + persistir_en: Annotated[ + Path | None, + typer.Option( + "--persistir-en", + help=( + "(Opcional) Path a un eventos.jsonl SEPARADO para persistir " + "los eventos de la simulación. NO debe coincidir con el log " + "operativo real (RF-12 §'sin afectar el estado operativo')." + ), + ), + ] = None, +) -> None: + """Ejecuta el modo simulación (RF-12): cálculo completo sobre flota ficticia. + + Por default no escribe al log canónico. Si se provee ``--persistir-en``, + los eventos se persisten en ese archivo separado. + + Exit codes: + + - **0** si la simulación termina sin error. + - **2** si algún archivo de entrada no existe o es JSON inválido. + """ + _validar_path(incidentes_path, etiqueta="incidentes") + _validar_path(flota_path, etiqueta="flota") + _validar_path(graph_path, etiqueta="grafo") + + incidentes_raw = _cargar_json_lista(incidentes_path, etiqueta="incidentes") + flota_raw = _cargar_json_lista(flota_path, etiqueta="flota") + + grafo_nx = cargar_grafo_iv_region(ruta_cache=graph_path) + grafo = OsmnxGrafoVial(grafo=grafo_nx) + + incidentes = [_incidente_desde_dict(raw) for raw in incidentes_raw] + flota_ficticia = [_unidad_desde_dict(raw) for raw in flota_raw] + + repositorio = JsonlRepositorioEventos(persistir_en) if persistir_en is not None else None + + reporte = simular( + incidentes, + flota_ficticia, + grafo, + repositorio_eventos=repositorio, + ) + + out_path.parent.mkdir(parents=True, exist_ok=True) + out_path.write_text( + json.dumps(_serializar_reporte(reporte), ensure_ascii=False, indent=2) + "\n", + encoding="utf-8", + ) + + sufijo = f" · eventos en {persistir_en}" if persistir_en is not None else "" + typer.echo( + f"Simulación completa: {reporte.incidentes_procesados} incidente(s). " + f"Reporte en {out_path}{sufijo}." + ) + raise typer.Exit(code=0) diff --git a/core-python/tests/unit/application/test_simulacion.py b/core-python/tests/unit/application/test_simulacion.py new file mode 100644 index 0000000..00b5f00 --- /dev/null +++ b/core-python/tests/unit/application/test_simulacion.py @@ -0,0 +1,256 @@ +"""UT del modo simulación (RF-12).""" + +from __future__ import annotations + +import math +from typing import TYPE_CHECKING + +import pytest + +from sentinel_dispatch.adapters.repositorio_jsonl import JsonlRepositorioEventos +from sentinel_dispatch.application.simulacion import simular +from sentinel_dispatch.application.tipos import MotivoDespacho +from sentinel_dispatch.domain.dispatch.tipos import ( + EstadoUnidad, + Incidente, + TipoUnidad, + Unidad, +) +from sentinel_dispatch.domain.routing.tipos import Arista, NoRutaDisponibleError +from sentinel_dispatch.domain.triaje.tipos import CategoriaMPDS + +if TYPE_CHECKING: + from collections.abc import Iterable + from pathlib import Path + + +# --------------------------------------------------------------------------- +# FakeGrafo + monkeypatch del A* — semántica idéntica a test_despacho +# --------------------------------------------------------------------------- + + +class FakeGrafo: + def __init__( + self, + nodos_por_unidad: dict[str, int], + nodo_incidente: int = 99, + ) -> None: + self._nodos = nodos_por_unidad + self._nodo_inc = nodo_incidente + self._coords = dict.fromkeys([*nodos_por_unidad.values(), nodo_incidente], (-29.95, -71.34)) + + def vecinos(self, nodo: int) -> Iterable[Arista]: + return [] + + def coordenadas(self, nodo: int) -> tuple[float, float]: + return self._coords[nodo] + + def nodo_mas_cercano(self, lat: float, lon: float) -> int: + # Snap del incidente fijo al nodo_incidente; bases por id de unidad + # detectado por proximidad a (lat, lon) provistos en la fixture. + # Como las coords son arbitrarias, retorno el nodo del incidente + # por default y los nodos de unidad cuando esa coord coincide. + for uid, n in self._nodos.items(): + base_lat, base_lon = -29.0 - (hash(uid) % 100) * 0.001, -71.0 + if abs(lat - base_lat) < 1e-6 and abs(lon - base_lon) < 1e-6: + return n + return self._nodo_inc + + def distancia_snap_m(self, lat: float, lon: float, nodo: int) -> float: + return 0.0 + + +@pytest.fixture +def tiempos_por_unidad() -> dict[str, float]: + """Mapeo unidad.id → t_viaje_s sintético. Modificable por tests.""" + return {"U01": 180.0, "U02": 240.0} + + +@pytest.fixture +def fake_grafo_y_tiempos( + monkeypatch: pytest.MonkeyPatch, tiempos_por_unidad: dict[str, float] +) -> dict[str, float]: + """Sustituye `a_estrella` por un fake que lee de `tiempos_por_unidad`.""" + + nodos_por_unidad = {"U01": 1, "U02": 2} + + def fake_a_estrella( + grafo: object, + origen: int, + destino: int, + *, + factor_hora: float, + factor_sirena: float, + ) -> tuple[float, list[int]]: + for uid, n in nodos_por_unidad.items(): + if n == origen: + t = tiempos_por_unidad.get(uid) + if t is None or not math.isfinite(t): + raise NoRutaDisponibleError(origen, destino) + return t, [origen, destino] + raise NoRutaDisponibleError(origen, destino) + + import sentinel_dispatch.application.despachar_ambulancia as _da + + monkeypatch.setattr(_da, "a_estrella", fake_a_estrella) + return tiempos_por_unidad + + +def _unidad( + uid: str = "U01", + tipo: TipoUnidad = TipoUnidad.AVANZADA, + estado: EstadoUnidad = EstadoUnidad.DISPONIBLE, +) -> Unidad: + return Unidad( + id=uid, + patente=f"PAT-{uid}", + tipo=tipo, + base_nombre=f"Base {uid}", + base_lat=-29.0 - (hash(uid) % 100) * 0.001, + base_lon=-71.0, + estado=estado, + ) + + +def _incidente(iid: str, categoria: CategoriaMPDS = CategoriaMPDS.ALPHA) -> Incidente: + return Incidente( + id=iid, + lat=-29.95, + lon=-71.34, + categoria_mpds=categoria, + timestamp_iso="2026-05-21T12:00:00.000Z", + ) + + +# --------------------------------------------------------------------------- +# Normal +# --------------------------------------------------------------------------- + + +class TestNormal: + def test_2_incidentes_2_resultados_y_metricas_correctas( + self, fake_grafo_y_tiempos: dict[str, float] + ) -> None: + grafo = FakeGrafo(nodos_por_unidad={"U01": 1, "U02": 2}) + reporte = simular( + incidentes=[_incidente("I-01"), _incidente("I-02")], + flota_ficticia=[_unidad("U01")], + grafo=grafo, + ) + assert reporte.incidentes_procesados == 2 + assert len(reporte.resultados) == 2 + assert reporte.pct_optimo == 100.0 + assert reporte.pct_saturacion == 0.0 + assert reporte.eta_media_s is not None + assert reporte.eta_media_s == pytest.approx(180.0) + assert reporte.eta_p95_s == pytest.approx(180.0) + + def test_sin_estado_evolutivo_entre_incidentes( + self, fake_grafo_y_tiempos: dict[str, float] + ) -> None: + """Cada incidente ve la flota inicial: la misma unidad puede ser elegida en cada uno.""" + grafo = FakeGrafo(nodos_por_unidad={"U01": 1}) + reporte = simular( + incidentes=[_incidente("I-01"), _incidente("I-02"), _incidente("I-03")], + flota_ficticia=[_unidad("U01")], + grafo=grafo, + ) + ganadores = {r.elegida.id for r in reporte.resultados if r.elegida is not None} + assert ganadores == {"U01"} # mismo ganador, no consumida + + +# --------------------------------------------------------------------------- +# Borde +# --------------------------------------------------------------------------- + + +class TestBorde: + def test_lista_incidentes_vacia(self, fake_grafo_y_tiempos: dict[str, float]) -> None: + grafo = FakeGrafo(nodos_por_unidad={"U01": 1}) + reporte = simular( + incidentes=[], + flota_ficticia=[_unidad("U01")], + grafo=grafo, + ) + assert reporte.incidentes_procesados == 0 + assert reporte.resultados == () + assert reporte.eta_media_s is None + assert reporte.eta_p95_s is None + assert reporte.pct_optimo == 0.0 + + def test_flota_vacia_todos_saturacion(self, fake_grafo_y_tiempos: dict[str, float]) -> None: + grafo = FakeGrafo(nodos_por_unidad={}) + reporte = simular( + incidentes=[_incidente("I-01"), _incidente("I-02")], + flota_ficticia=[], + grafo=grafo, + ) + assert reporte.pct_saturacion == 100.0 + assert all(r.motivo is MotivoDespacho.SATURACION for r in reporte.resultados) + assert reporte.eta_media_s is None + + +# --------------------------------------------------------------------------- +# Reglas — "sin afectar el estado operativo real" +# --------------------------------------------------------------------------- + + +class TestSinAfectarEstadoReal: + def test_default_no_escribe_a_repositorio( + self, fake_grafo_y_tiempos: dict[str, float], tmp_path: Path + ) -> None: + """Sin `repositorio_eventos`, no se crea ningún archivo de log.""" + grafo = FakeGrafo(nodos_por_unidad={"U01": 1}) + log_no_creado = tmp_path / "log_no_creado.jsonl" + simular( + incidentes=[_incidente("I-01")], + flota_ficticia=[_unidad("U01")], + grafo=grafo, + ) + assert not log_no_creado.exists() + + def test_con_repositorio_provisto_escribe_n_eventos_a_archivo_separado( + self, fake_grafo_y_tiempos: dict[str, float], tmp_path: Path + ) -> None: + """Con repositorio, escribe 1 evento por incidente.""" + grafo = FakeGrafo(nodos_por_unidad={"U01": 1}) + eventos_sim_path = tmp_path / "eventos_sim.jsonl" + repo = JsonlRepositorioEventos(eventos_sim_path) + simular( + incidentes=[_incidente("I-01"), _incidente("I-02"), _incidente("I-03")], + flota_ficticia=[_unidad("U01")], + grafo=grafo, + repositorio_eventos=repo, + ) + eventos = list(repo.leer_todos()) + assert len(eventos) == 3 + # Convención: los despacho_id de simulación tienen prefijo SD-SIM- + assert all( + e.despacho_id is not None and e.despacho_id.startswith("SD-SIM-") for e in eventos + ) + + +# --------------------------------------------------------------------------- +# Métricas — porcentajes por motivo +# --------------------------------------------------------------------------- + + +class TestMetricas: + def test_pcts_suman_100_para_n_positivo(self, fake_grafo_y_tiempos: dict[str, float]) -> None: + grafo = FakeGrafo(nodos_por_unidad={"U01": 1, "U02": 2}) + # Mix Echo + Básica → suboptimo_rn02 + reporte = simular( + incidentes=[ + _incidente("I-01", CategoriaMPDS.ALPHA), # → OPTIMO + _incidente("I-02", CategoriaMPDS.ECHO), # con sólo Básica → SUBOPTIMO_RN02 + ], + flota_ficticia=[_unidad("U02", tipo=TipoUnidad.BASICA)], + grafo=grafo, + ) + suma = ( + reporte.pct_optimo + + reporte.pct_penalizado + + reporte.pct_suboptimo_rn02 + + reporte.pct_saturacion + ) + assert suma == pytest.approx(100.0) diff --git a/docs/quality/trazabilidad.md b/docs/quality/trazabilidad.md index 47493f1..44e06e4 100644 --- a/docs/quality/trazabilidad.md +++ b/docs/quality/trazabilidad.md @@ -40,7 +40,7 @@ La matriz cubre los **doce Requisitos Funcionales** (RF-01..RF-12), las **diez R | **RF-09** Panel de unidades en tiempo real | `interfaces/api` + UI HTMX (F5 diferido) | _Endpoint `/unidades/estado` (pendiente)_ | Verificación funcional durante FTR-02 | Estado actualizado refleja transiciones Disponible↔EnRuta↔EnEscena↔Taller con coordenadas | ⛔ post-H5 (ADR-0004 deferred) | | **RF-10** Detección de saturación y candidatas a re-dirección | `application/` → [`saturacion.py`](../../core-python/src/sentinel_dispatch/application/saturacion.py) | `detectar_saturacion(flota, progreso_por_unidad)` → `EstadoSaturacion(saturada, candidatas_redireccion)`; candidatas EnRuta ordenadas por `(progreso_pct asc, unidad.id lex asc)`; default conservador `progreso=0.0` para EnRuta sin progreso provisto | [CP-10](../SRS.md#213-casos-de-prueba) flota saturada | Sistema reporta saturación cuando ninguna unidad está en `DISPONIBLE`; lista candidatas EnRuta para re-dirección manual del operador | ✅ H3 fase 3 | | **RF-11** Exportación de logs CSV/JSON | `adapters/exportador.py` + `interfaces/cli/export_cmd.py` | `exportar_a_csv(eventos, path)` y `exportar_a_json(eventos, path)`; CLI `sentinel export --formato {csv,json} --in EVENTOS.jsonl --out PATH`. CSV con flatten de `payload_*` y encoding `utf-8-sig` (BOM Excel); JSON como array indentado sin BOM | `test_exportador.py` (14 UT — flatten, CSV/JSON Normal/Borde, end-to-end CLI con archivo inexistente o corrupto) | Logs derivados con union de columnas para payloads heterogéneos; el log canónico JSONL no se modifica (RN-03 preservado, este subcomando solo lee) | ✅ | -| **RF-12** Modo simulación sobre flota ficticia | `application/` → `simulacion.py` | `simular(flota_ficticia, incidente)` (pendiente) | _Test funcional (pendiente)_ | Cálculo completo se ejecuta sin afectar el estado operativo real; resultado claramente marcado como simulación | 🟡 H4 | +| **RF-12** Modo simulación sobre flota ficticia | `application/simulacion.py` + `interfaces/cli/simular_cmd.py` | `simular(incidentes, flota_ficticia, grafo, *, repositorio_eventos=None) → ReporteSimulacion`. Sin evolución temporal entre incidentes (cada uno ve la flota inicial). Persistencia **opt-in** vía `repositorio_eventos` para no contaminar el log operativo. CLI: `sentinel simular --flota --incidentes --graph --out [--persistir-en]` | `test_simulacion.py` (7 UT — Normal/Borde/RN/Métricas: resultados+métricas correctas, sin evolución temporal, lista vacía, flota vacía=100% saturación, default no escribe a repo, con repo escribe N eventos con prefijo `SD-SIM-`, pcts suman 100) | Reporte agregado: lista de `ResultadoDespacho` + pct por motivo + ETA media/p95. El reporte JSON queda etiquetado `"modo": "simulacion"` para distinguir del modo operativo | ✅ | ## 3. Reglas de Negocio