"""Motor de reglas heurísticas — Studio Sprint 2. Lee `vmssailor/library/rules/*.yaml` y, dado el contexto del wizard (tipo de buque, subtipo, eslora, sistemas habilitados), produce una lista de `EquipmentProposal` que el wizard muestra al integrador. El integrador acepta/edita/rechaza cada propuesta antes de pasar al paso 6 (refinamiento manual). """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any from vmssailor.core.coords import ShipCoord from vmssailor.core.enums import SystemId, VesselSubtype, VesselType from vmssailor.library import load_library @dataclass(slots=True) class EquipmentProposal: """Una propuesta concreta de equipo para el integrador. El integrador puede: - aceptarla tal cual → se materializa en un Equipment del proyecto - editarla (modelo distinto, ubicación distinta, prefix diferente) - rechazarla """ system_id: SystemId model_ref: str """ID de EquipmentModel de la biblioteca.""" tag_prefix: str """Prefijo de tags sugerido (ME_PORT, GEN_1, BILGE_FWD, etc.).""" display_name: str """Nombre humano para el integrador.""" location_x_pp: float location_y_cl: float location_z_bl: float rationale: str = "" """Por qué la regla propone este equipo.""" accepted: bool = True """El integrador puede desmarcar para rechazar.""" def to_ship_coord(self) -> ShipCoord: return ShipCoord( x_pp=self.location_x_pp, y_cl=self.location_y_cl, z_bl=self.location_z_bl ) @dataclass(slots=True) class RuleContext: """Input que el wizard pasa al rule engine.""" vessel_type: VesselType vessel_subtype: VesselSubtype length_overall_m: float beam_max_m: float draft_m: float systems_enabled: list[SystemId] = field(default_factory=list) class RuleEngine: """Carga reglas YAML y produce propuestas para un contexto dado.""" def __init__(self, rules: dict[str, dict[str, Any]] | None = None) -> None: if rules is None: try: rules = load_library().rules except Exception: rules = {} self._rules = rules def applicable_rules(self, ctx: RuleContext) -> list[str]: """Lista de rule_ids cuyo `meta.applies_to` matchea el contexto.""" out: list[str] = [] for rid, data in self._rules.items(): applies = (data.get("meta") or {}).get("applies_to") or {} v_types = applies.get("vessel_types") or [] v_subs = applies.get("vessel_subtypes") or [] loa_range = applies.get("length_overall_m") or {} if v_types and ctx.vessel_type.value not in v_types: continue if v_subs and ctx.vessel_subtype.value not in v_subs: continue lo = loa_range.get("min") hi = loa_range.get("max") if lo is not None and ctx.length_overall_m < float(lo): continue if hi is not None and ctx.length_overall_m > float(hi): continue out.append(rid) return out def propose(self, ctx: RuleContext) -> list[EquipmentProposal]: """Genera la lista completa de propuestas para un contexto.""" proposals: list[EquipmentProposal] = [] for rid in self.applicable_rules(ctx): proposals.extend(self._propose_from_rule(rid, ctx)) return proposals # ----- internals -------------------------------------------------- def _propose_from_rule( self, rule_id: str, ctx: RuleContext ) -> list[EquipmentProposal]: rule = self._rules.get(rule_id) or {} sys_proposals = rule.get("equipment_proposals") or {} out: list[EquipmentProposal] = [] for sys_str, block in sys_proposals.items(): if not isinstance(block, dict): continue try: sys_id = SystemId(sys_str) except ValueError: continue if sys_id not in ctx.systems_enabled: continue count = int(block.get("count", 1)) candidates = block.get("candidates") or [] # Pick the first candidate whose `when` matches ctx chosen = self._pick_candidate(candidates, ctx) if not chosen: continue model_ref = chosen.get("model_ref", "") rationale = chosen.get("rationale", "") tag_template = block.get("tag_prefix_template", sys_str.upper()) sides = block.get("sides") or [] location_tpl = block.get("location_template") or {} for idx in range(count): if sides and idx < len(sides): side = sides[idx] prefix = tag_template.replace("{side}", side) side_key = side.lower() loc = location_tpl.get(side_key) or location_tpl.get("default") or {} else: prefix = tag_template.replace("{idx}", str(idx + 1)) loc = location_tpl.get("default") or {} x_pp = self._resolve_x(loc, ctx) y_cl = float(loc.get("y_cl", 0.0)) z_bl = float(loc.get("z_bl", 1.0)) out.append( EquipmentProposal( system_id=sys_id, model_ref=model_ref, tag_prefix=prefix, display_name=f"{prefix} · {model_ref}", location_x_pp=x_pp, location_y_cl=y_cl, location_z_bl=z_bl, rationale=rationale, ) ) return out def _pick_candidate( self, candidates: list[dict[str, Any]], ctx: RuleContext ) -> dict[str, Any] | None: for c in candidates: when = c.get("when") or {} loa = when.get("length_overall_m") or {} lo = loa.get("min") hi = loa.get("max") if lo is not None and ctx.length_overall_m < float(lo): continue if hi is not None and ctx.length_overall_m > float(hi): continue return c # Si no matchea ninguno con `when`, devolver el primero sin condiciones for c in candidates: if not c.get("when"): return c return candidates[0] if candidates else None def _resolve_x(self, loc: dict[str, Any], ctx: RuleContext) -> float: if "x_pp" in loc: return float(loc["x_pp"]) if "x_pp_pct" in loc: return float(loc["x_pp_pct"]) * ctx.length_overall_m return ctx.length_overall_m * 0.30 def apply_rules(ctx: RuleContext) -> list[EquipmentProposal]: """Conveniencia: aplica todas las reglas globales al contexto.""" return RuleEngine().propose(ctx)