"""Project — agregado raíz del modelo de datos. Un Project es el conjunto completo de configuración para UN buque de UN cliente. Se persiste como archivo único `.vmsproj` (SQLite portable) y se compila a `.vmspack` (ZIP firmado) para distribuir al Runtime del buque. """ from __future__ import annotations from datetime import UTC, datetime from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator from vmssailor.core.card import Topology from vmssailor.core.enums import SystemId from vmssailor.core.equipment import Equipment from vmssailor.core.permissive import PermissiveRule from vmssailor.core.tag import Tag from vmssailor.core.vessel import Vessel from vmssailor.version import __version__ def _now_utc() -> datetime: return datetime.now(UTC) class Project(BaseModel): """Configuración completa de un buque de un cliente.""" model_config = ConfigDict(extra="forbid") id: str = Field( ..., min_length=1, max_length=128, pattern=r"^[a-z0-9][a-z0-9_-]*$", description="ID estable snake_case-kebab, ej: 'm_y_aurora_sunseeker_76'.", ) name: str = Field(..., min_length=1, max_length=256, description='Nombre humano: "M/Y Aurora".') customer: str = Field(default="", max_length=256) notes: str = Field(default="", max_length=4096) vessel: Vessel systems_enabled: list[SystemId] = Field( default_factory=list, description="Sistemas habilitados — definen el menú lateral del Runtime.", ) equipment: list[Equipment] = Field(default_factory=list) tags: list[Tag] = Field(default_factory=list) topology: Topology = Field(default_factory=Topology) permissive_rules: list[PermissiveRule] = Field(default_factory=list) created_at: datetime = Field(default_factory=_now_utc) updated_at: datetime = Field(default_factory=_now_utc) vmssailor_version: str = Field(default=__version__, max_length=32) # ---- Validadores ---------------------------------------------------- @field_validator("systems_enabled") @classmethod def _systems_unique(cls, v: list[SystemId]) -> list[SystemId]: if len(v) != len(set(v)): raise ValueError("systems_enabled no debe contener duplicados.") return v @field_validator("equipment") @classmethod def _equipment_unique_ids(cls, v: list[Equipment]) -> list[Equipment]: ids = [e.id for e in v] if len(ids) != len(set(ids)): raise ValueError("Equipment IDs deben ser únicos en un Project.") prefixes = [e.tag_prefix for e in v] if len(prefixes) != len(set(prefixes)): raise ValueError("Equipment tag_prefix deben ser únicos en un Project.") return v @field_validator("tags") @classmethod def _tags_unique_ids(cls, v: list[Tag]) -> list[Tag]: ids = [t.id for t in v] if len(ids) != len(set(ids)): raise ValueError("Tag IDs deben ser únicos en un Project.") return v @field_validator("permissive_rules") @classmethod def _rules_unique_ids(cls, v: list[PermissiveRule]) -> list[PermissiveRule]: ids = [r.id for r in v] if len(ids) != len(set(ids)): raise ValueError("PermissiveRule IDs deben ser únicos en un Project.") return v @model_validator(mode="after") def _equipment_systems_must_be_enabled(self) -> Project: enabled = set(self.systems_enabled) for eq in self.equipment: if eq.system_id not in enabled: raise ValueError( f"Equipment '{eq.id}' pertenece a sistema " f"'{eq.system_id.value}' que no está en systems_enabled." ) return self @model_validator(mode="after") def _tags_reference_existing_equipment(self) -> Project: eq_ids = {e.id for e in self.equipment} for t in self.tags: if t.equipment_id is not None and t.equipment_id not in eq_ids: raise ValueError( f"Tag '{t.id}' referencia equipment_id='{t.equipment_id}' " "que no existe en este Project." ) return self @model_validator(mode="after") def _tag_bindings_reference_existing_cards(self) -> Project: card_ids = {c.id for c in self.topology.cards} for t in self.tags: if t.physical_binding is None: continue if t.physical_binding.card_id not in card_ids: raise ValueError( f"Tag '{t.id}' tiene physical_binding.card_id=" f"'{t.physical_binding.card_id}' que no existe en topology." ) return self @model_validator(mode="after") def _decks_referenced_by_equipment_exist(self) -> Project: deck_ids = {d.id for d in self.vessel.decks} for eq in self.equipment: if eq.deck_id is not None and eq.deck_id not in deck_ids: raise ValueError( f"Equipment '{eq.id}' referencia deck_id='{eq.deck_id}' " "que no existe en vessel.decks." ) return self @model_validator(mode="after") def _permissive_conditions_reference_existing_tags(self) -> Project: tag_ids = {t.id for t in self.tags} for rule in self.permissive_rules: for cond in rule.conditions: if cond.tag_ref not in tag_ids: raise ValueError( f"PermissiveRule '{rule.id}' tiene Condition.tag_ref=" f"'{cond.tag_ref}' que no existe en tags." ) return self # ---- Conveniencias -------------------------------------------------- def equipment_by_id(self, equipment_id: str) -> Equipment | None: for e in self.equipment: if e.id == equipment_id: return e return None def tag_by_id(self, tag_id: str) -> Tag | None: for t in self.tags: if t.id == tag_id: return t return None def tags_for_equipment(self, equipment_id: str) -> list[Tag]: return [t for t in self.tags if t.equipment_id == equipment_id] def tags_for_system(self, system_id: SystemId) -> list[Tag]: eq_ids = {e.id for e in self.equipment if e.system_id == system_id} return [t for t in self.tags if t.equipment_id in eq_ids] def touch(self) -> None: """Actualiza `updated_at`. Llamar antes de cada `save`.""" self.updated_at = _now_utc() def stats(self) -> dict[str, int]: """Resumen numérico para debug y para el panel resumen del Studio.""" return { "systems": len(self.systems_enabled), "equipment": len(self.equipment), "tags": len(self.tags), "tags_with_alarms": sum(1 for t in self.tags if t.alarms), "tags_controllable": sum(1 for t in self.tags if t.controllable), "buses": len(self.topology.buses), "cards": len(self.topology.cards), "permissive_rules": len(self.permissive_rules), "permissive_conditions": sum(len(r.conditions) for r in self.permissive_rules), }