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

sprint-0: fundaciones VMS-Sailor

Browse Source 
Sprint 0 completo del producto VMS-Sailor (Vessel Management System
integrado para buques 30-40m). Brief de referencia en
VMS_Sailor_v2_Parte_*.md (intacto).

Core (vmssailor.core, 95.17% coverage, 99 tests verde):
- ShipCoord: sistema naval x_pp/y_cl/z_bl frozen
- Vessel, Deck, Bulkhead
- Equipment, EquipmentModel, Sensor, EquipmentSpec
- Tag, AlarmConfig, TagBinding, Scaling
- CardInstance, Bus, Topology con validacion 21 puntos I/O AR-NMEA-IO-v1.0
- Alarm, PermissiveRule, Condition
- Project agregado raiz con validacion cross-entity
- Persistencia portable .vmsproj (SQLite) con roundtrip verificable

Biblioteca curada seed (vmssailor.library):
- systems_catalog.json completo (catalogo maestro Parte 1 sec 7)
- 2 vessels: Sunseeker 76, Ferretti 850
- 2 motores: MTU 12V 2000 M96, Volvo D13-900
- 1 genset: Northern Lights M65C13
- yacht_motor_planeo.yaml (reglas heuristicas)
- TODO marcado data_source=seed_estimate - requiere validacion datasheets

Tools:
- vms-validate-library: CLI valida biblioteca completa
- vms-generate-test-project: CLI demo + verificacion roundtrip persistencia

Design System + 8 mockups HTML estaticos:
- docs/design_system.md (paleta Deep Ocean, gradientes, typography, motion)
- docs/brand/ (logo + variantes SVG)
- docs/mockups/splash, studio_main, runtime_overview,
  runtime_mimic_fuel (P&ID animado), runtime_alarms, runtime_trim (panel
  estrella con horizonte artificial), mobile_overview, mobile_trim
- docs/mockups/index.html (galeria)

Firmware (Sprint 12+ implementacion):
- firmware/ar_nmea_io_v1/src/config/pinout.h con macros GPIO

Decisiones autonomas documentadas en docs/decisions_sprint0.md.

Stack: Python 3.11 + uv + Pydantic v2 + SQLite stdlib + hatchling +
pytest 9 + ruff + mypy. Sin PySide6, FastAPI, Flutter ni firmware
funcional (entran en sprints siguientes).

Criterio de aceptacion Sprint 0: cumplido.
- uv sync: OK
- pytest: 99/99 verde
- cov vmssailor.core: 95.17% (objetivo >=80%)
- ruff: clean
- vms-validate-library: OK
- vms-generate-test-project: INTEGRIDAD OK

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Alvaro Enrique Romero Donado
2026-05-17 07:26:06 -04:00
commit deb04c9315
96 changed files with 15335 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/__init__.py
View File
tests/conftest.py
+145
View File
@@ -0,0 +1,145 @@
"""Fixtures comunes para los tests de vmssailor.core."""
from __future__ import annotations
from pathlib import Path
import pytest
from vmssailor.core import (
AlarmConfig,
AlarmPriority,
Bus,
BusRole,
CardInstance,
ChannelType,
Equipment,
Project,
Protocol,
Scaling,
ShipCoord,
SignalType,
SystemId,
Tag,
TagBinding,
Topology,
UnitSI,
Vessel,
VesselSubtype,
VesselType,
)
from vmssailor.core.vessel import Deck
@pytest.fixture()
def repo_root() -> Path:
return Path(__file__).resolve().parents[1]
@pytest.fixture()
def sample_ship_coord() -> ShipCoord:
return ShipCoord(x_pp=10.0, y_cl=-1.5, z_bl=2.0)
@pytest.fixture()
def minimal_vessel() -> Vessel:
return Vessel(
id="test_vessel",
name="Test Vessel",
type=VesselType.YACHT_MOTOR,
subtype=VesselSubtype.PLANING,
length_overall_m=24.0,
beam_max_m=5.5,
draft_m=1.8,
decks=[
Deck(id="main", name="Main", z_bl_bottom=2.0, z_bl_top=4.5),
],
)
@pytest.fixture()
def sample_card() -> CardInstance:
return CardInstance(
id="card_001",
slot_number=1,
bus_id="bus_main",
bus_role=BusRole.MODBUS_SLAVE,
modbus_address=1,
physical_location="SM panel A",
)
@pytest.fixture()
def sample_bus() -> Bus:
return Bus(
id="bus_main",
name="Bus principal",
protocol=Protocol.MODBUS_RTU,
physical_port="COM3",
)
@pytest.fixture()
def sample_topology(sample_bus: Bus, sample_card: CardInstance) -> Topology:
return Topology(buses=[sample_bus], cards=[sample_card])
@pytest.fixture()
def sample_equipment() -> Equipment:
return Equipment(
id="eq_me_port",
model_ref="mtu_12v_2000_m96",
tag_prefix="ME_PORT",
display_name="Motor babor",
location=ShipCoord(x_pp=6.0, y_cl=-0.9, z_bl=1.2),
deck_id="main",
system_id=SystemId.MAIN_ENGINE,
)
@pytest.fixture()
def sample_tag(sample_equipment: Equipment) -> Tag:
return Tag(
id=f"{sample_equipment.tag_prefix}.OIL_PRESS",
equipment_id=sample_equipment.id,
description="Presión aceite",
unit_si=UnitSI.BAR,
range_normal_min=3.5,
range_normal_max=6.5,
alarms=[
AlarmConfig(
id=f"{sample_equipment.tag_prefix}.OIL_PRESS.LOW",
threshold=1.5,
operator="<",
priority=AlarmPriority.EMERGENCY,
hysteresis=0.2,
)
],
protocol=Protocol.MODBUS_RTU,
physical_binding=TagBinding(
card_id="card_001",
channel_type=ChannelType.AI,
channel_number=1,
signal_type=SignalType.SIG_4_20_MA,
scaling=Scaling(raw_min=4.0, raw_max=20.0, eng_min=0.0, eng_max=10.0),
),
)
@pytest.fixture()
def sample_project(
minimal_vessel: Vessel,
sample_equipment: Equipment,
sample_tag: Tag,
sample_topology: Topology,
) -> Project:
return Project(
id="test_project",
name="Test project",
customer="Pytest",
vessel=minimal_vessel,
systems_enabled=[SystemId.MAIN_ENGINE],
equipment=[sample_equipment],
tags=[sample_tag],
topology=sample_topology,
)
tests/core/__init__.py
View File
tests/core/persistence/__init__.py
View File
tests/core/persistence/test_roundtrip.py
+90
View File
@@ -0,0 +1,90 @@
"""Test crítico Sprint 0: roundtrip de persistencia .vmsproj.
Criterio de aceptación de la Parte 6: `load_project(save_project(p))` produce
un Project equivalente campo por campo.
"""
from __future__ import annotations
import sqlite3
from pathlib import Path
import pytest
from vmssailor.core import Project
from vmssailor.core.persistence import load_project, save_project
from vmssailor.core.persistence.vmsproj_reader import VmsProjError
from vmssailor.tools.generate_test_project import build_demo_project
def _projects_equivalent(a: Project, b: Project) -> bool:
return a.model_dump(mode="json", exclude={"updated_at"}) == b.model_dump(
mode="json", exclude={"updated_at"}
)
def test_roundtrip_minimal(sample_project: Project, tmp_path: Path) -> None:
out = tmp_path / "minimal.vmsproj"
save_project(sample_project, out)
assert out.exists()
loaded = load_project(out)
assert _projects_equivalent(sample_project, loaded)
def test_roundtrip_full_demo(tmp_path: Path) -> None:
project = build_demo_project()
out = tmp_path / "demo.vmsproj"
save_project(project, out)
loaded = load_project(out)
assert _projects_equivalent(project, loaded), (
"Roundtrip falló — el proyecto releído no coincide con el original."
)
# Stats deben coincidir 1:1
assert project.stats() == loaded.stats()
def test_vmsproj_is_real_sqlite(sample_project: Project, tmp_path: Path) -> None:
out = tmp_path / "x.vmsproj"
save_project(sample_project, out)
# Cualquier herramienta SQLite debe abrirlo
conn = sqlite3.connect(str(out))
try:
version = conn.execute("SELECT MAX(version) FROM schema_version").fetchone()[0]
assert version == 1
# Meta esperada
meta_rows = dict(conn.execute("SELECT key, value FROM meta"))
assert meta_rows["file_format"] == "vmsproj"
finally:
conn.close()
def test_load_nonexistent_raises(tmp_path: Path) -> None:
with pytest.raises(VmsProjError):
load_project(tmp_path / "does_not_exist.vmsproj")
def test_load_invalid_db_raises(tmp_path: Path) -> None:
bad = tmp_path / "bad.vmsproj"
bad.write_bytes(b"not a sqlite database at all")
with pytest.raises(Exception): # noqa: B017 -- sqlite3 levanta varias clases distintas
load_project(bad)
def test_save_overwrites_existing(sample_project: Project, tmp_path: Path) -> None:
out = tmp_path / "ow.vmsproj"
save_project(sample_project, out)
first_size = out.stat().st_size
# Modifico el proyecto y vuelvo a guardar
sample_project.notes = "Cambiado en memoria"
save_project(sample_project, out)
loaded = load_project(out)
assert loaded.notes == "Cambiado en memoria"
# El archivo sigue siendo válido
assert out.stat().st_size > 0
_ = first_size # no relevante para el assert principal
def test_save_creates_parent_dirs(sample_project: Project, tmp_path: Path) -> None:
out = tmp_path / "deep" / "nested" / "out.vmsproj"
save_project(sample_project, out)
assert out.exists()
tests/core/test_alarm.py
+82
View File
@@ -0,0 +1,82 @@
"""Tests de Alarm (instancia activa)."""
from __future__ import annotations
from datetime import UTC, datetime, timedelta
import pytest
from pydantic import ValidationError
from vmssailor.core import Alarm
from vmssailor.core.enums import AlarmPriority, AlarmState
def _now() -> datetime:
return datetime.now(UTC)
def test_alarm_active_basic() -> None:
a = Alarm(
id="alm_1",
tag_id="ME_PORT.OIL_PRESS",
alarm_config_id="ME_PORT.OIL_PRESS.LOW",
priority=AlarmPriority.EMERGENCY,
state=AlarmState.ACTIVE,
timestamp_active=_now(),
message="Oil pressure low",
)
assert a.state == AlarmState.ACTIVE
assert a.timestamp_ack is None
def test_alarm_ack_requires_timestamp_ack() -> None:
with pytest.raises(ValidationError):
Alarm(
id="a",
tag_id="t",
alarm_config_id="t.low",
priority=AlarmPriority.HIGH,
state=AlarmState.ACK,
timestamp_active=_now(),
# timestamp_ack faltante
)
def test_alarm_cleared_requires_timestamp_cleared() -> None:
with pytest.raises(ValidationError):
Alarm(
id="a",
tag_id="t",
alarm_config_id="t.low",
priority=AlarmPriority.HIGH,
state=AlarmState.CLEARED,
timestamp_active=_now(),
)
def test_alarm_ack_timestamp_must_be_after_active() -> None:
t0 = _now()
with pytest.raises(ValidationError):
Alarm(
id="a",
tag_id="t",
alarm_config_id="t.low",
priority=AlarmPriority.HIGH,
state=AlarmState.ACK,
timestamp_active=t0,
timestamp_ack=t0 - timedelta(seconds=1),
acknowledged_by="op1",
)
def test_alarm_acknowledged_by_without_ts_inconsistent() -> None:
with pytest.raises(ValidationError):
Alarm(
id="a",
tag_id="t",
alarm_config_id="t.low",
priority=AlarmPriority.HIGH,
state=AlarmState.ACTIVE,
timestamp_active=_now(),
acknowledged_by="op1",
)
tests/core/test_card.py
+86
View File
@@ -0,0 +1,86 @@
"""Tests de Bus, CardInstance, Topology."""
from __future__ import annotations
import pytest
from pydantic import ValidationError
from vmssailor.core import (
Bus,
BusRole,
CardInstance,
Protocol,
Topology,
)
def test_bus_modbus_rtu_ok() -> None:
Bus(id="b", name="b", protocol=Protocol.MODBUS_RTU, physical_port="COM3")
def test_bus_invalid_protocol_rejected() -> None:
with pytest.raises(ValidationError):
Bus(id="b", name="b", protocol=Protocol.J1939, physical_port="COM3")
def test_card_slave_requires_modbus_addr() -> None:
with pytest.raises(ValidationError):
CardInstance(
id="c", slot_number=1, bus_id="bm", bus_role=BusRole.MODBUS_SLAVE
)
def test_card_n2k_node_no_modbus_addr() -> None:
CardInstance(
id="c", slot_number=1, bus_id="bm", bus_role=BusRole.NMEA2000_NODE
)
with pytest.raises(ValidationError):
CardInstance(
id="c", slot_number=1, bus_id="bm",
bus_role=BusRole.NMEA2000_NODE, modbus_address=5,
)
def test_topology_unique_slots_per_bus() -> None:
b = Bus(id="bm", name="bm", protocol=Protocol.MODBUS_RTU, physical_port="COM3")
c1 = CardInstance(
id="c1", slot_number=1, bus_id="bm",
bus_role=BusRole.MODBUS_SLAVE, modbus_address=1,
)
c2 = CardInstance(
id="c2", slot_number=1, bus_id="bm",
bus_role=BusRole.MODBUS_SLAVE, modbus_address=2,
)
with pytest.raises(ValidationError):
Topology(buses=[b], cards=[c1, c2])
def test_topology_unique_modbus_addresses_per_bus() -> None:
b = Bus(id="bm", name="bm", protocol=Protocol.MODBUS_RTU, physical_port="COM3")
c1 = CardInstance(
id="c1", slot_number=1, bus_id="bm",
bus_role=BusRole.MODBUS_SLAVE, modbus_address=5,
)
c2 = CardInstance(
id="c2", slot_number=2, bus_id="bm",
bus_role=BusRole.MODBUS_SLAVE, modbus_address=5,
)
with pytest.raises(ValidationError):
Topology(buses=[b], cards=[c1, c2])
def test_topology_card_references_existing_bus() -> None:
b = Bus(id="bm", name="bm", protocol=Protocol.MODBUS_RTU, physical_port="COM3")
c = CardInstance(
id="c", slot_number=1, bus_id="other_bus",
bus_role=BusRole.MODBUS_SLAVE, modbus_address=1,
)
with pytest.raises(ValidationError):
Topology(buses=[b], cards=[c])
def test_topology_helpers(sample_topology: Topology) -> None:
assert sample_topology.bus_by_id("bus_main") is not None
assert sample_topology.bus_by_id("nope") is None
assert sample_topology.card_by_id("card_001") is not None
assert sample_topology.card_by_id("nope") is None
tests/core/test_coords.py
+69
View File
@@ -0,0 +1,69 @@
"""Tests del sistema de coordenadas naval."""
from __future__ import annotations
import pytest
from pydantic import ValidationError
from vmssailor.core import ShipCoord
def test_construct_valid() -> None:
c = ShipCoord(x_pp=10.0, y_cl=-1.5, z_bl=2.0)
assert c.x_pp == 10.0
assert c.y_cl == -1.5
assert c.z_bl == 2.0
def test_frozen() -> None:
c = ShipCoord(x_pp=10.0, y_cl=0.0, z_bl=1.0)
with pytest.raises(ValidationError):
c.x_pp = 99.0 # type: ignore[misc]
def test_as_tuple() -> None:
c = ShipCoord(x_pp=1.0, y_cl=2.0, z_bl=3.0)
assert c.as_tuple() == (1.0, 2.0, 3.0)
def test_is_starboard_port_centerline() -> None:
starboard = ShipCoord(x_pp=5.0, y_cl=1.0, z_bl=0.0)
port = ShipCoord(x_pp=5.0, y_cl=-1.0, z_bl=0.0)
centerline = ShipCoord(x_pp=5.0, y_cl=0.0, z_bl=0.0)
assert starboard.is_starboard()
assert not starboard.is_port()
assert not starboard.is_centerline()
assert port.is_port()
assert not port.is_starboard()
assert centerline.is_centerline()
assert not centerline.is_starboard()
assert not centerline.is_port()
def test_distance_to() -> None:
a = ShipCoord(x_pp=0.0, y_cl=0.0, z_bl=0.0)
b = ShipCoord(x_pp=3.0, y_cl=4.0, z_bl=0.0)
assert a.distance_to(b) == pytest.approx(5.0)
def test_out_of_range_rejected() -> None:
with pytest.raises(ValidationError):
ShipCoord(x_pp=999.0, y_cl=0.0, z_bl=0.0)
with pytest.raises(ValidationError):
ShipCoord(x_pp=10.0, y_cl=999.0, z_bl=0.0)
with pytest.raises(ValidationError):
ShipCoord(x_pp=10.0, y_cl=0.0, z_bl=-999.0)
def test_extra_fields_forbidden() -> None:
with pytest.raises(ValidationError):
ShipCoord(x_pp=1.0, y_cl=0.0, z_bl=0.0, foo="bar") # type: ignore[call-arg]
def test_str_representation() -> None:
c = ShipCoord(x_pp=10.0, y_cl=-1.5, z_bl=2.0)
s = str(c)
assert "10.00" in s
assert "-1.50" in s
assert "+2.00" in s
assert "[m]" in s
tests/core/test_enums.py
+53
View File
@@ -0,0 +1,53 @@
"""Smoke tests para los enums del core."""
from __future__ import annotations
from vmssailor.core.enums import (
AlarmPriority,
BusRole,
ChannelType,
ControlMode,
SystemId,
UnitSI,
VesselType,
)
def test_systemid_main_engine_value() -> None:
assert SystemId.MAIN_ENGINE.value == "main_engine"
def test_channel_type_values() -> None:
assert ChannelType.AI.value == "ai"
assert ChannelType.DI.value == "di"
assert ChannelType.DO.value == "do"
assert ChannelType.RPM.value == "rpm"
def test_alarm_priority_complete_set() -> None:
values = {p.value for p in AlarmPriority}
assert values == {"emergency", "high", "low", "info"}
def test_busrole_includes_dual_and_bridge() -> None:
assert BusRole.DUAL.value == "dual"
assert BusRole.BRIDGE.value == "bridge"
def test_controlmode_default_states_present() -> None:
assert ControlMode.MONITOR.value == "monitor"
assert ControlMode.MANUAL.value == "manual"
assert ControlMode.AUTO.value == "auto"
assert ControlMode.FUTURE.value == "future"
def test_vesseltype_complete() -> None:
values = {v.value for v in VesselType}
assert {"yacht_motor", "yacht_sail", "fishing", "patrol", "ferry", "offshore_support"} <= values
def test_unit_si_si_only() -> None:
# No deben aparecer unidades imperiales en el enum interno.
forbidden = {"ft", "psi", "F", "kt"}
values = {u.value.lower() for u in UnitSI}
assert not (forbidden & values), f"UnitSI debe ser solo SI. Encontrado: {forbidden & values}"
tests/core/test_equipment.py
+88
View File
@@ -0,0 +1,88 @@
"""Tests de Equipment, EquipmentModel, Sensor."""
from __future__ import annotations
import pytest
from pydantic import ValidationError
from vmssailor.core import (
EquipmentCategory,
EquipmentModel,
EquipmentSpec,
Sensor,
SystemId,
UnitSI,
)
def test_sensor_basic() -> None:
s = Sensor(
id="oil_press",
name="Oil Pressure",
unit_si=UnitSI.BAR,
range_normal_min=3.0,
range_normal_max=5.5,
)
assert s.unit_si == UnitSI.BAR
def test_sensor_range_max_below_min_rejected() -> None:
with pytest.raises(ValidationError):
Sensor(
id="x",
name="X",
unit_si=UnitSI.BAR,
range_normal_min=5.0,
range_normal_max=2.0,
)
def test_equipment_model_with_sensors() -> None:
em = EquipmentModel(
id="mtu_test",
manufacturer="MTU",
model_name="Test",
category=EquipmentCategory.ENGINE_MAIN,
typical_systems=[SystemId.MAIN_ENGINE],
specs=EquipmentSpec(power_kw=1432, rpm_nominal=2450),
default_sensors=[
Sensor(id="rpm", name="RPM", unit_si=UnitSI.RPM),
Sensor(id="oil_press", name="Oil P", unit_si=UnitSI.BAR),
],
)
assert em.specs.power_kw == 1432
assert len(em.default_sensors) == 2
def test_equipment_model_duplicate_sensor_ids_rejected() -> None:
with pytest.raises(ValidationError):
EquipmentModel(
id="x",
manufacturer="X",
model_name="X",
category=EquipmentCategory.ENGINE_MAIN,
default_sensors=[
Sensor(id="rpm", name="RPM"),
Sensor(id="rpm", name="RPM dup"),
],
)
def test_equipment_instance(sample_equipment) -> None:
assert sample_equipment.tag_prefix == "ME_PORT"
assert sample_equipment.system_id == SystemId.MAIN_ENGINE
assert sample_equipment.installed
def test_equipment_invalid_tag_prefix_rejected() -> None:
with pytest.raises(ValidationError):
from vmssailor.core import Equipment, ShipCoord
Equipment(
id="x",
model_ref="m",
tag_prefix="lowercase_bad", # debe ser mayúsculas
display_name="X",
location=ShipCoord(x_pp=1.0, y_cl=0.0, z_bl=0.0),
system_id=SystemId.MAIN_ENGINE,
)
tests/core/test_permissive.py
+49
View File
@@ -0,0 +1,49 @@
"""Tests de PermissiveRule y Condition."""
from __future__ import annotations
import pytest
from pydantic import ValidationError
from vmssailor.core import PermissiveRule
from vmssailor.core.permissive import Condition
def test_condition_basic() -> None:
c = Condition(
tag_ref="ME_PORT.OIL_PRESS",
operator=">",
threshold=0.3,
severity="fail",
)
assert c.severity == "fail"
def test_condition_between_requires_both_thresholds() -> None:
with pytest.raises(ValidationError):
Condition(
tag_ref="x",
operator="between",
threshold_low=1.0,
# threshold_high faltante
)
def test_permissive_rule_empty_conditions_rejected() -> None:
with pytest.raises(ValidationError):
PermissiveRule(id="r", action_id="START", conditions=[])
def test_permissive_rule_with_multiple_conditions() -> None:
r = PermissiveRule(
id="rule_start",
action_id="START_ME_PORT",
conditions=[
Condition(tag_ref="ME_PORT.OIL_PRESS", operator=">", threshold=0.3),
Condition(tag_ref="ME_PORT.COOLANT_TEMP", operator=">", threshold=5.0),
Condition(tag_ref="ME_PORT.ESTOP_ACTIVE", operator="is_false"),
],
on_fail_message="No es seguro arrancar.",
)
assert len(r.conditions) == 3
assert r.conditions[2].operator == "is_false"
tests/core/test_project.py
+194
View File
@@ -0,0 +1,194 @@
"""Tests del agregado raíz Project."""
from __future__ import annotations
import pytest
from pydantic import ValidationError
from vmssailor.core import (
Bus,
BusRole,
CardInstance,
Equipment,
PermissiveRule,
Project,
Protocol,
ShipCoord,
SystemId,
Tag,
Topology,
UnitSI,
Vessel,
)
from vmssailor.core.permissive import Condition
def test_project_stats(sample_project: Project) -> None:
stats = sample_project.stats()
assert stats["systems"] == 1
assert stats["equipment"] == 1
assert stats["tags"] == 1
assert stats["tags_with_alarms"] == 1
assert stats["buses"] == 1
assert stats["cards"] == 1
def test_project_equipment_requires_enabled_system(
minimal_vessel: Vessel, sample_topology: Topology
) -> None:
eq = Equipment(
id="eq",
model_ref="m",
tag_prefix="X",
display_name="x",
location=ShipCoord(x_pp=5.0, y_cl=0.0, z_bl=1.0),
system_id=SystemId.WATERMAKER, # NO en systems_enabled
)
with pytest.raises(ValidationError):
Project(
id="p",
name="p",
vessel=minimal_vessel,
systems_enabled=[SystemId.MAIN_ENGINE],
equipment=[eq],
topology=sample_topology,
)
def test_project_tag_equipment_id_must_exist(
minimal_vessel: Vessel, sample_topology: Topology, sample_equipment: Equipment
) -> None:
with pytest.raises(ValidationError):
Project(
id="p",
name="p",
vessel=minimal_vessel,
systems_enabled=[SystemId.MAIN_ENGINE],
equipment=[sample_equipment],
tags=[
Tag(
id="GHOST.X",
equipment_id="ghost_eq", # no existe
unit_si=UnitSI.BAR,
protocol=Protocol.MODBUS_RTU,
address=1,
)
],
topology=sample_topology,
)
def test_project_tag_binding_must_reference_existing_card(
minimal_vessel: Vessel,
) -> None:
from vmssailor.core import ChannelType, SignalType, TagBinding
b = Bus(id="bm", name="bm", protocol=Protocol.MODBUS_RTU, physical_port="COM3")
c = CardInstance(
id="card_001", slot_number=1, bus_id="bm",
bus_role=BusRole.MODBUS_SLAVE, modbus_address=1,
)
topo = Topology(buses=[b], cards=[c])
eq = Equipment(
id="eq",
model_ref="m",
tag_prefix="X",
display_name="x",
location=ShipCoord(x_pp=5.0, y_cl=0.0, z_bl=1.0),
system_id=SystemId.MAIN_ENGINE,
)
bad_tag = Tag(
id="X.PRESS",
equipment_id="eq",
unit_si=UnitSI.BAR,
protocol=Protocol.MODBUS_RTU,
physical_binding=TagBinding(
card_id="ghost_card",
channel_type=ChannelType.AI,
channel_number=1,
signal_type=SignalType.SIG_4_20_MA,
),
)
with pytest.raises(ValidationError):
Project(
id="p",
name="p",
vessel=minimal_vessel,
systems_enabled=[SystemId.MAIN_ENGINE],
equipment=[eq],
tags=[bad_tag],
topology=topo,
)
def test_project_permissive_condition_must_reference_existing_tag(
sample_project: Project,
) -> None:
bad_rule = PermissiveRule(
id="bad",
action_id="DO_X",
conditions=[Condition(tag_ref="GHOST.TAG", operator=">", threshold=1.0)],
)
with pytest.raises(ValidationError):
Project(
id=sample_project.id,
name=sample_project.name,
vessel=sample_project.vessel,
systems_enabled=sample_project.systems_enabled,
equipment=sample_project.equipment,
tags=sample_project.tags,
topology=sample_project.topology,
permissive_rules=[bad_rule],
)
def test_project_helpers(sample_project: Project) -> None:
eq = sample_project.equipment_by_id("eq_me_port")
assert eq is not None
assert eq.tag_prefix == "ME_PORT"
tag = sample_project.tag_by_id("ME_PORT.OIL_PRESS")
assert tag is not None
tags_eq = sample_project.tags_for_equipment("eq_me_port")
assert len(tags_eq) == 1
tags_sys = sample_project.tags_for_system(SystemId.MAIN_ENGINE)
assert len(tags_sys) == 1
def test_project_touch_updates_timestamp(sample_project: Project) -> None:
original = sample_project.updated_at
sample_project.touch()
assert sample_project.updated_at >= original
def test_project_equipment_unique_tag_prefix(minimal_vessel: Vessel) -> None:
e1 = Equipment(
id="e1", model_ref="m", tag_prefix="ME",
display_name="A", location=ShipCoord(x_pp=1, y_cl=0, z_bl=0),
system_id=SystemId.MAIN_ENGINE,
)
e2 = Equipment(
id="e2", model_ref="m", tag_prefix="ME", # mismo prefix
display_name="B", location=ShipCoord(x_pp=2, y_cl=0, z_bl=0),
system_id=SystemId.MAIN_ENGINE,
)
with pytest.raises(ValidationError):
Project(
id="p", name="p", vessel=minimal_vessel,
systems_enabled=[SystemId.MAIN_ENGINE],
equipment=[e1, e2],
)
def test_project_equipment_deck_must_exist(minimal_vessel: Vessel) -> None:
eq = Equipment(
id="eq", model_ref="m", tag_prefix="X",
display_name="x", location=ShipCoord(x_pp=1, y_cl=0, z_bl=0),
deck_id="non_existent_deck",
system_id=SystemId.MAIN_ENGINE,
)
with pytest.raises(ValidationError):
Project(
id="p", name="p", vessel=minimal_vessel,
systems_enabled=[SystemId.MAIN_ENGINE],
equipment=[eq],
)
tests/core/test_tag.py
+141
View File
@@ -0,0 +1,141 @@
"""Tests de Tag, AlarmConfig, TagBinding, Scaling."""
from __future__ import annotations
import pytest
from pydantic import ValidationError
from vmssailor.core import (
AlarmConfig,
AlarmPriority,
ChannelType,
ControlMode,
Protocol,
Scaling,
SignalType,
Tag,
TagBinding,
UnitSI,
)
def test_scaling_apply_and_invert() -> None:
s = Scaling(raw_min=4.0, raw_max=20.0, eng_min=0.0, eng_max=10.0)
assert s.apply(4.0) == pytest.approx(0.0)
assert s.apply(12.0) == pytest.approx(5.0)
assert s.apply(20.0) == pytest.approx(10.0)
assert s.invert(5.0) == pytest.approx(12.0)
def test_scaling_equal_raw_rejected() -> None:
with pytest.raises(ValidationError):
Scaling(raw_min=4.0, raw_max=4.0, eng_min=0.0, eng_max=10.0)
def test_scaling_equal_eng_rejected() -> None:
with pytest.raises(ValidationError):
Scaling(raw_min=4.0, raw_max=20.0, eng_min=5.0, eng_max=5.0)
def test_tag_binding_channel_within_capacity() -> None:
TagBinding(
card_id="card_001",
channel_type=ChannelType.AI,
channel_number=4,
signal_type=SignalType.SIG_4_20_MA,
)
def test_tag_binding_channel_exceeds_capacity_rejected() -> None:
with pytest.raises(ValidationError):
TagBinding(
card_id="card_001",
channel_type=ChannelType.AI,
channel_number=5, # cap = 4
signal_type=SignalType.SIG_4_20_MA,
)
with pytest.raises(ValidationError):
TagBinding(
card_id="card_001",
channel_type=ChannelType.RPM,
channel_number=2, # cap = 1
signal_type=SignalType.PULSE_MAGNETIC_PICKUP,
)
def test_alarm_config_basic() -> None:
a = AlarmConfig(
id="x.low",
threshold=1.5,
operator="<",
priority=AlarmPriority.EMERGENCY,
hysteresis=0.2,
)
assert a.priority == AlarmPriority.EMERGENCY
def test_alarm_config_invalid_operator() -> None:
with pytest.raises(ValidationError):
AlarmConfig(id="x", threshold=1.0, operator="??", priority=AlarmPriority.HIGH)
def test_tag_controllable_must_not_be_monitor() -> None:
with pytest.raises(ValidationError):
Tag(
id="X.START",
unit_si=UnitSI.BOOL,
controllable=True,
control_mode=ControlMode.MONITOR, # inconsistente
protocol=Protocol.MODBUS_RTU,
address=1,
)
def test_tag_non_controllable_must_be_monitor() -> None:
with pytest.raises(ValidationError):
Tag(
id="X.PRESS",
unit_si=UnitSI.BAR,
controllable=False,
control_mode=ControlMode.MANUAL,
protocol=Protocol.MODBUS_RTU,
address=1,
)
def test_tag_internal_no_address() -> None:
Tag(
id="VIRT.STATE",
unit_si=UnitSI.NONE,
protocol=Protocol.INTERNAL,
address=None,
)
with pytest.raises(ValidationError):
Tag(
id="VIRT.STATE",
unit_si=UnitSI.NONE,
protocol=Protocol.INTERNAL,
address=42, # INTERNAL no debe tener address
)
def test_tag_modbus_requires_address_or_binding() -> None:
with pytest.raises(ValidationError):
Tag(
id="X.OIL",
unit_si=UnitSI.BAR,
protocol=Protocol.MODBUS_RTU,
address=None,
physical_binding=None,
)
def test_tag_id_pattern() -> None:
with pytest.raises(ValidationError):
Tag(id="lowercase.bad", unit_si=UnitSI.NONE, protocol=Protocol.INTERNAL)
def test_tag_alarms_unique(sample_tag: Tag) -> None:
assert len(sample_tag.alarms) == 1
a = sample_tag.alarms[0]
assert a.priority == AlarmPriority.EMERGENCY
tests/core/test_validation.py
+131
View File
@@ -0,0 +1,131 @@
"""Tests del validador cross-entity."""
from __future__ import annotations
from vmssailor.core import (
Bus,
BusRole,
CardInstance,
ChannelType,
Equipment,
Project,
Protocol,
Scaling,
ShipCoord,
SignalType,
SystemId,
Tag,
TagBinding,
Topology,
UnitSI,
Vessel,
VesselSubtype,
VesselType,
)
from vmssailor.core.validation import Severity, validate_project
def _make_vessel() -> Vessel:
return Vessel(
id="v",
name="V",
type=VesselType.YACHT_MOTOR,
subtype=VesselSubtype.PLANING,
length_overall_m=20.0,
beam_max_m=5.0,
draft_m=1.5,
)
def _basic_topology() -> Topology:
b = Bus(id="bm", name="bm", protocol=Protocol.MODBUS_RTU, physical_port="COM3")
c = CardInstance(
id="c1", slot_number=1, bus_id="bm",
bus_role=BusRole.MODBUS_SLAVE, modbus_address=1,
)
return Topology(buses=[b], cards=[c])
def test_orphan_system_warning() -> None:
p = Project(
id="p", name="p", vessel=_make_vessel(),
systems_enabled=[SystemId.MAIN_ENGINE, SystemId.WATERMAKER],
equipment=[],
topology=_basic_topology(),
)
r = validate_project(p)
assert r.ok()
assert any(i.code == "SYSTEM_WITHOUT_EQUIPMENT" for i in r.warnings)
def test_card_capacity_high_info() -> None:
topo = _basic_topology()
eq = Equipment(
id="eq", model_ref="m", tag_prefix="X",
display_name="x", location=ShipCoord(x_pp=5, y_cl=0, z_bl=1),
system_id=SystemId.MAIN_ENGINE,
)
# 4 AI = 100% (at-limit -> warning)
tags = []
for i in range(4):
tags.append(
Tag(
id=f"X.AI{i + 1}",
equipment_id="eq",
unit_si=UnitSI.BAR,
protocol=Protocol.MODBUS_RTU,
physical_binding=TagBinding(
card_id="c1",
channel_type=ChannelType.AI,
channel_number=i + 1,
signal_type=SignalType.SIG_4_20_MA,
scaling=Scaling(raw_min=4, raw_max=20, eng_min=0, eng_max=10),
),
)
)
p = Project(
id="p", name="p", vessel=_make_vessel(),
systems_enabled=[SystemId.MAIN_ENGINE],
equipment=[eq],
tags=tags,
topology=topo,
)
r = validate_project(p)
codes = {i.code for i in r.issues}
assert "CARD_CAPACITY_AT_LIMIT" in codes
def test_equipment_out_of_hull_warning() -> None:
eq = Equipment(
id="eq", model_ref="m", tag_prefix="X",
display_name="x",
location=ShipCoord(x_pp=100.0, y_cl=0.0, z_bl=1.0), # fuera del buque
system_id=SystemId.MAIN_ENGINE,
)
p = Project(
id="p", name="p", vessel=_make_vessel(),
systems_enabled=[SystemId.MAIN_ENGINE],
equipment=[eq],
topology=_basic_topology(),
)
r = validate_project(p)
codes = {i.code for i in r.issues}
assert "EQUIPMENT_OUT_OF_HULL" in codes
def test_validation_report_format() -> None:
p = Project(
id="p", name="p", vessel=_make_vessel(),
systems_enabled=[],
equipment=[],
topology=_basic_topology(),
)
r = validate_project(p)
assert "OK" in r.format() or "Total" in r.format()
def test_severity_values() -> None:
assert Severity.ERROR.value == "error"
assert Severity.WARNING.value == "warning"
assert Severity.INFO.value == "info"
tests/core/test_vessel.py
+79
View File
@@ -0,0 +1,79 @@
"""Tests de Vessel, Deck, Bulkhead."""
from __future__ import annotations
import pytest
from pydantic import ValidationError
from vmssailor.core import Vessel, VesselSubtype, VesselType
from vmssailor.core.vessel import Bulkhead, Deck
def test_deck_height() -> None:
d = Deck(id="main", name="Main", z_bl_bottom=2.0, z_bl_top=4.5)
assert d.height() == pytest.approx(2.5)
def test_deck_polygon_empty_ok() -> None:
Deck(id="main", name="Main", z_bl_bottom=2.0, z_bl_top=4.5)
def test_deck_polygon_too_few_vertices_rejected() -> None:
with pytest.raises(ValidationError):
Deck(
id="main",
name="Main",
z_bl_bottom=2.0,
z_bl_top=4.5,
polygon_xy=[(0.0, 0.0), (1.0, 0.0)],
)
def test_minimal_vessel(minimal_vessel: Vessel) -> None:
assert minimal_vessel.length_overall_m == 24.0
assert minimal_vessel.type == VesselType.YACHT_MOTOR
assert minimal_vessel.subtype == VesselSubtype.PLANING
assert len(minimal_vessel.decks) == 1
def test_vessel_duplicate_deck_ids_rejected() -> None:
with pytest.raises(ValidationError):
Vessel(
id="x",
name="X",
type=VesselType.YACHT_MOTOR,
subtype=VesselSubtype.PLANING,
length_overall_m=20.0,
beam_max_m=5.0,
draft_m=1.5,
decks=[
Deck(id="main", name="A", z_bl_bottom=0.0, z_bl_top=2.0),
Deck(id="main", name="B", z_bl_bottom=2.0, z_bl_top=4.0),
],
)
def test_vessel_position_helpers(minimal_vessel: Vessel) -> None:
bow = minimal_vessel.position_at_bow()
assert bow.x_pp == minimal_vessel.length_overall_m
origin = minimal_vessel.position_at_origin()
assert origin.x_pp == 0.0
def test_vessel_invalid_length_rejected() -> None:
with pytest.raises(ValidationError):
Vessel(
id="x",
name="X",
type=VesselType.YACHT_MOTOR,
subtype=VesselSubtype.PLANING,
length_overall_m=-1.0,
beam_max_m=5.0,
draft_m=1.5,
)
def test_bulkhead_basic() -> None:
b = Bulkhead(id="col", name="Collision", x_pp=21.0)
assert b.x_pp == 21.0
assert b.description == ""
tests/library/__init__.py
View File
tests/library/test_loader.py
+39
View File
@@ -0,0 +1,39 @@
"""Tests del library loader."""
from __future__ import annotations
import json
from pathlib import Path
from vmssailor.library.loader import load_library
def test_loader_returns_result_with_format() -> None:
r = load_library()
s = r.format()
assert "Library load summary" in s
assert "Vessels" in s
assert "EquipmentModels" in s
def test_loader_handles_corrupt_json(tmp_path: Path) -> None:
# Construimos una biblioteca falsa con un JSON malo
fake_lib = tmp_path / "lib"
(fake_lib / "vessels").mkdir(parents=True)
(fake_lib / "vessels" / "broken.json").write_text("{ not valid json", encoding="utf-8")
(fake_lib / "equipment").mkdir(parents=True)
(fake_lib / "rules").mkdir(parents=True)
# systems_catalog.json mínimo válido
(fake_lib / "systems_catalog.json").write_text(
json.dumps({"_meta": {"version": 1}, "categories": []}), encoding="utf-8"
)
r = load_library(fake_lib)
assert not r.ok()
assert any("broken.json" in i.path for i in r.errors)
def test_loader_handles_missing_systems_catalog(tmp_path: Path) -> None:
fake_lib = tmp_path / "lib"
fake_lib.mkdir()
r = load_library(fake_lib)
assert any("systems_catalog.json" in i.path for i in r.errors)
tests/library/test_seed_integrity.py
+80
View File
@@ -0,0 +1,80 @@
"""Tests de integridad de la biblioteca seed."""
from __future__ import annotations
from vmssailor.core.enums import EquipmentCategory, SystemId
from vmssailor.library import load_library, load_systems_catalog
def test_systems_catalog_loads() -> None:
cat = load_systems_catalog()
assert "categories" in cat
assert len(cat["categories"]) >= 10 # 11 categorías mínimo
def test_systems_catalog_covers_all_enum_values() -> None:
cat = load_systems_catalog()
json_ids = set()
for c in cat["categories"]:
for s in c["systems"]:
json_ids.add(s["id"])
enum_ids = {s.value for s in SystemId}
missing_in_json = enum_ids - json_ids
missing_in_enum = json_ids - enum_ids
assert not missing_in_json, f"SystemId enum tiene valores no en catálogo: {missing_in_json}"
assert not missing_in_enum, f"Catálogo tiene IDs no en SystemId enum: {missing_in_enum}"
def test_library_loads_without_errors() -> None:
result = load_library()
assert result.ok(), f"Biblioteca con errores: {result.errors}"
def test_library_has_minimum_seed() -> None:
result = load_library()
vessel_ids = {v.id for v in result.vessels}
assert "sunseeker_76" in vessel_ids
assert "ferretti_850" in vessel_ids
eq_ids = {e.id for e in result.equipment_models}
assert "mtu_12v_2000_m96" in eq_ids
assert "volvo_d13_900hp" in eq_ids
assert "northern_lights_m65c13" in eq_ids
def test_seed_engines_are_engine_main() -> None:
result = load_library()
for em in result.equipment_models:
if em.id.startswith("mtu_") or em.id.startswith("volvo_"):
assert em.category == EquipmentCategory.ENGINE_MAIN, em.id
def test_seed_genset_category() -> None:
result = load_library()
nl = next(e for e in result.equipment_models if e.id == "northern_lights_m65c13")
assert nl.category == EquipmentCategory.GENSET
def test_rules_yacht_motor_planeo_present() -> None:
result = load_library()
assert "yacht_motor_planeo" in result.rules
def test_rules_reference_existing_equipment_models() -> None:
result = load_library()
# Filtrar warnings que no son del cross-ref
cross_ref_warns = [
w
for w in result.warnings
if "no existe en equipment_models" in w.message
]
assert not cross_ref_warns, f"Rules con model_ref roto: {cross_ref_warns}"
def test_seed_marked_as_estimate() -> None:
"""Todos los archivos seed deben venir marcados como seed_estimate."""
result = load_library()
for v in result.vessels:
assert v.data_source == "seed_estimate", v.id
for em in result.equipment_models:
assert em.data_source == "seed_estimate", em.id
tests/shared/__init__.py
View File
tests/shared/test_ids.py
+51
View File
@@ -0,0 +1,51 @@
"""Tests de generación de IDs."""
from __future__ import annotations
import pytest
from vmssailor.shared.ids import (
make_alarm_config_id,
make_alarm_instance_id,
make_project_id,
make_tag_id,
new_uuid,
)
def test_make_tag_id_basic() -> None:
assert make_tag_id("ME_PORT", "oil_press") == "ME_PORT.OIL_PRESS"
def test_make_tag_id_rejects_lowercase_prefix() -> None:
with pytest.raises(ValueError):
make_tag_id("me_port", "oil_press")
def test_make_tag_id_rejects_uppercase_sensor() -> None:
with pytest.raises(ValueError):
make_tag_id("ME_PORT", "OIL_PRESS")
def test_alarm_config_id() -> None:
cid = make_alarm_config_id("ME_PORT.OIL_PRESS", "low")
assert cid == "ME_PORT.OIL_PRESS.LOW"
def test_alarm_instance_id_deterministic() -> None:
a = make_alarm_instance_id("ME_PORT.OIL_PRESS", "ME_PORT.OIL_PRESS.LOW", 1234567890.0)
assert a.startswith("alm.ME_PORT.OIL_PRESS.LOW.")
def test_new_uuid_unique() -> None:
a = new_uuid()
b = new_uuid()
assert a != b
assert len(a) == 36 # UUID canonical
def test_project_id_normalization() -> None:
pid = make_project_id("Acme Yachts S.A.", "Sunseeker_76")
# Espacios y puntuación se normalizan
assert pid == "acme_yachts_s_a___sunseeker_76" or pid.startswith("acme_yachts_s_a")
assert "__" in pid # separador customer__vessel