SystemSimulationApp/app/simulation/core/ports.py

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Literal


PortKind = Literal["physical", "signal"]
PortNominalRole = Literal["inlet", "outlet", "bidirectional", "input", "output"]
ActualFlowDirection = Literal["in", "out", "stagnant"]
VariableRole = Literal["effort", "flow", "stream", "signal"]
ConnectionRule = Literal["equal", "sumToZero", "streamMix", "directed"]


@dataclass(frozen=True)
class PortVariableDefinition:
    name: str
    role: VariableRole
    connection_rule: ConnectionRule
    label: str = field(default="", compare=False)
    quantity: str = field(default="", compare=False)
    unit: str = field(default="", compare=False)
    result_visible: bool = field(default=True, compare=False)
    order: int = field(default=0, compare=False)

    def as_interface_dict(self) -> dict[str, object]:
        return {
            "name": self.name,
            "role": self.role,
            "connectionRule": self.connection_rule,
            "label": self.label or self.name,
            "quantity": self.quantity or self.name,
            "unit": self.unit,
            "resultVisible": self.result_visible,
            "order": self.order,
        }


@dataclass(frozen=True)
class PortDefinition:
    """Stable connector contract shared by components, XML, and the compiler."""

    name: str
    kind: PortKind
    domain: str
    nominal_role: PortNominalRole
    positive_flow_direction: Literal["intoComponent"] | None = None
    variables: tuple[PortVariableDefinition, ...] = ()

    @classmethod
    def pneumatic(
        cls,
        name: str,
        *,
        nominal_role: Literal["inlet", "outlet", "bidirectional"] = "bidirectional",
    ) -> PortDefinition:
        return cls(
            name=name,
            kind="physical",
            domain="pneumatic",
            nominal_role=nominal_role,
            positive_flow_direction="intoComponent",
            variables=(
                PortVariableDefinition(
                    "p",
                    "effort",
                    "equal",
                    label="压力",
                    quantity="pressure",
                    unit="Pa",
                    order=10,
                ),
                PortVariableDefinition(
                    "m_flow",
                    "flow",
                    "sumToZero",
                    label="质量流量",
                    quantity="mass_flow",
                    unit="kg/s",
                    order=20,
                ),
                PortVariableDefinition(
                    "h_outflow",
                    "stream",
                    "streamMix",
                    label="流出比焓",
                    quantity="specific_enthalpy",
                    unit="J/kg",
                    order=30,
                ),
                PortVariableDefinition(
                    "volume",
                    "signal",
                    "directed",
                    label="外部容积",
                    quantity="volume",
                    unit="m3",
                    result_visible=False,
                    order=40,
                ),
                PortVariableDefinition(
                    "volume_flow",
                    "signal",
                    "directed",
                    label="外部容积变化率",
                    quantity="volume_flow",
                    unit="m3/s",
                    result_visible=False,
                    order=50,
                ),
            ),
        )

    @classmethod
    def mechanical_translational(
        cls,
        name: str,
        *,
        nominal_role: Literal["inlet", "outlet", "bidirectional"] = "bidirectional",
    ) -> PortDefinition:
        return cls(
            name=name,
            kind="physical",
            domain="mechanical",
            nominal_role=nominal_role,
            positive_flow_direction="intoComponent",
            variables=(
                PortVariableDefinition(
                    "x",
                    "effort",
                    "equal",
                    label="位移",
                    quantity="length",
                    unit="m",
                    order=10,
                ),
                PortVariableDefinition(
                    "v",
                    "effort",
                    "equal",
                    label="速度",
                    quantity="velocity",
                    unit="m/s",
                    order=20,
                ),
                PortVariableDefinition(
                    "f",
                    "flow",
                    "sumToZero",
                    label="力",
                    quantity="force",
                    unit="N",
                    order=30,
                ),
            ),
        )

    @classmethod
    def signal(
        cls,
        name: str,
        *,
        nominal_role: Literal["input", "output"],
        domain: str = "signal",
    ) -> PortDefinition:
        return cls(
            name=name,
            kind="signal",
            domain=domain,
            nominal_role=nominal_role,
            variables=(
                PortVariableDefinition(
                    "signal",
                    "signal",
                    "directed",
                    label="信号值",
                    quantity="dimensionless",
                    unit="",
                    order=10,
                ),
            ),
        )

    def as_interface_dict(self) -> dict[str, object]:
        return {
            "name": self.name,
            "kind": self.kind,
            "domain": self.domain,
            "nominalRole": self.nominal_role,
            "positiveFlowDirection": self.positive_flow_direction,
            "variables": [variable.as_interface_dict() for variable in self.variables],
        }


@dataclass
class PortState:
    """Python-side analogue of a Modelica fluid port."""

    p: float = 0.0
    m_flow: float = 0.0
    h_outflow: float = 0.0
    volume: float = 0.0
    volume_flow: float = 0.0
    signal: float = 0.0
    x: float = 0.0
    v: float = 0.0
    f: float = 0.0
    definition: PortDefinition | None = field(default=None, repr=False, compare=False)

    @classmethod
    def pneumatic(
        cls,
        name: str,
        *,
        nominal_role: Literal["inlet", "outlet", "bidirectional"] = "bidirectional",
    ) -> PortState:
        return cls(definition=PortDefinition.pneumatic(name, nominal_role=nominal_role))

    @property
    def inflow_rate(self) -> float:
        return max(self.m_flow, 0.0)

    @property
    def outflow_rate(self) -> float:
        return max(-self.m_flow, 0.0)

    def actual_direction(self, tolerance: float = 1e-12) -> ActualFlowDirection:
        if self.m_flow > tolerance:
            return "in"
        if self.m_flow < -tolerance:
            return "out"
        return "stagnant"