SystemSimulationApp/app/system_xml.py

from __future__ import annotations

from collections.abc import Mapping
from dataclasses import dataclass, replace
from functools import lru_cache
from math import isfinite
from pathlib import Path
from typing import Literal

from lxml import etree

from app.simulation.core.ports import PortDefinition
from app.simulation.registry import (
    COMPONENT_MODEL_REGISTRY,
    ParameterSpec,
)


ValidationLayer = Literal["xml", "schema", "semantic"]
ValidationSeverity = Literal["error", "warning"]
SYSTEM_XML_MAX_BYTES = 5 * 1024 * 1024
SYSTEM_XML_V2_SCHEMA_PATH = (
    Path(__file__).resolve().parent.parent / "schemas" / "system-simulation-v2.xsd"
)
SUPPORTED_SOLVER_METHODS = {"RK45", "RK23", "DOP853", "Radau", "BDF", "LSODA"}
AMESIM_PARAMETER_ENCODING_VERSION = "1"
MECMAS21_LEGACY_BINARY_PARAMETERS = ("useFriction", "strib")


@dataclass(frozen=True)
class ValidationIssue:
    layer: ValidationLayer
    code: str
    message: str
    severity: ValidationSeverity = "error"
    path: str | None = None
    line: int | None = None

    def as_dict(self) -> dict[str, object]:
        result: dict[str, object] = {
            "severity": self.severity,
            "layer": self.layer,
            "code": self.code,
            "message": self.message,
        }
        if self.path is not None:
            result["path"] = self.path
        if self.line is not None:
            result["line"] = self.line
        return result


@dataclass(frozen=True)
class SystemXmlSimulation:
    t_start: float
    t_stop: float
    step: float
    max_step: float
    method: str
    line: int | None = None


@dataclass(frozen=True)
class SystemXmlPort:
    name: str
    kind: str
    domain: str
    nominal_role: str
    positive_flow_direction: str | None
    side: str
    line: int | None = None

    def as_project_data(self) -> dict[str, object]:
        data: dict[str, object] = {
            "name": self.name,
            "kind": self.kind,
            "domain": self.domain,
            "nominalRole": self.nominal_role,
            "side": self.side,
        }
        if self.positive_flow_direction is not None:
            data["positiveFlowDirection"] = self.positive_flow_direction
        return data


@dataclass(frozen=True)
class SystemXmlParameter:
    name: str
    value: float
    line: int | None = None


@dataclass(frozen=True)
class SystemXmlComponent:
    id: str
    name: str
    model_type: str
    component_type: str
    x: float
    y: float
    rotation: int
    mirrored: bool
    ports: tuple[SystemXmlPort, ...]
    parameters: tuple[SystemXmlParameter, ...]
    line: int | None = None

    @property
    def port_by_name(self) -> dict[str, SystemXmlPort]:
        return {port.name: port for port in self.ports}


@dataclass(frozen=True)
class SystemXmlEndpoint:
    component: str
    port: str
    role: str | None
    line: int | None = None

    @property
    def key(self) -> tuple[str, str]:
        return self.component, self.port


@dataclass(frozen=True)
class SystemXmlConnection:
    id: str
    kind: str
    domain: str
    endpoints: tuple[SystemXmlEndpoint, SystemXmlEndpoint]
    line: int | None = None

    @property
    def undirected_key(self) -> tuple[tuple[str, str], tuple[str, str]]:
        first, second = sorted(endpoint.key for endpoint in self.endpoints)
        return first, second


@dataclass(frozen=True)
class SystemXmlDocument:
    name: str
    schema_version: str
    unit_system: str
    medium_reference_version: str | None
    amesim_parameter_encoding_version: str | None
    simulation: SystemXmlSimulation
    components: tuple[SystemXmlComponent, ...]
    connections: tuple[SystemXmlConnection, ...]

    def summary(self) -> dict[str, object]:
        result: dict[str, object] = {
            "name": self.name,
            "schemaVersion": self.schema_version,
            "unitSystem": self.unit_system,
            "componentCount": len(self.components),
            "connectionCount": len(self.connections),
        }
        if self.medium_reference_version is not None:
            result["mediumReferenceVersion"] = self.medium_reference_version
        if self.amesim_parameter_encoding_version is not None:
            result["amesimParameterEncodingVersion"] = (
                self.amesim_parameter_encoding_version
            )
        return result

    def as_project_data(self) -> dict[str, object]:
        edges = []
        for connection in self.connections:
            first, second = connection.endpoints
            if connection.kind == "signal":
                by_role = {endpoint.role: endpoint for endpoint in connection.endpoints}
                first = by_role.get("source", first)
                second = by_role.get("target", second)
            edges.append(
                {
                    "id": connection.id,
                    "source": first.component,
                    "sourceHandle": first.port,
                    "target": second.component,
                    "targetHandle": second.port,
                }
            )

        return {
            "name": self.name,
            "mediumReferenceVersion": 1,
            "amesimParameterEncodingVersion": 1,
            "nodes": [
                {
                    "id": component.id,
                    "type": "simulationComponent",
                    "position": {"x": component.x, "y": component.y},
                    "data": {
                        "label": component.name,
                        "componentType": component.component_type,
                        "modelType": component.model_type,
                        "ports": [port.as_project_data() for port in component.ports],
                        "parameters": {
                            parameter.name: parameter.value
                            for parameter in component.parameters
                        },
                        "rotation": component.rotation,
                        "mirrored": component.mirrored,
                    },
                }
                for component in self.components
            ],
            "edges": edges,
            "simulation": {
                "t_start": self.simulation.t_start,
                "t_stop": self.simulation.t_stop,
                "step": self.simulation.step,
                "max_step": self.simulation.max_step,
                "method": self.simulation.method,
            },
        }


@dataclass(frozen=True)
class SystemXmlValidationReport:
    document: SystemXmlDocument | None
    issues: tuple[ValidationIssue, ...]

    @property
    def valid(self) -> bool:
        return self.document is not None and not any(
            issue.severity == "error" for issue in self.issues
        )

    def as_dict(self) -> dict[str, object]:
        errors = sum(issue.severity == "error" for issue in self.issues)
        warnings = sum(issue.severity == "warning" for issue in self.issues)
        result: dict[str, object] = {
            "valid": self.valid,
            "errorCount": errors,
            "warningCount": warnings,
            "issues": [issue.as_dict() for issue in self.issues],
        }
        if self.document is not None:
            result["system"] = self.document.summary()
        return result


def validate_system_xml_document(
    source: bytes | str,
) -> SystemXmlValidationReport:
    xml_bytes = source.encode("utf-8") if isinstance(source, str) else source
    if not xml_bytes.strip():
        return _failed_report("xml", "XML_EMPTY", "The XML document is empty.")
    if len(xml_bytes) > SYSTEM_XML_MAX_BYTES:
        return _failed_report(
            "xml",
            "XML_TOO_LARGE",
            f"The XML document exceeds {SYSTEM_XML_MAX_BYTES} bytes.",
        )

    parser = etree.XMLParser(
        resolve_entities=False,
        no_network=True,
        load_dtd=False,
        recover=False,
        huge_tree=False,
    )
    try:
        root = etree.fromstring(xml_bytes, parser=parser)
    except etree.XMLSyntaxError as exc:
        line, _ = exc.position
        return _failed_report(
            "xml",
            "XML_SYNTAX_ERROR",
            str(exc).split(", line", maxsplit=1)[0],
            line=line,
        )

    if root.getroottree().docinfo.doctype:
        return _failed_report(
            "xml",
            "XML_DTD_NOT_ALLOWED",
            "DTD and entity declarations are not allowed.",
            line=root.sourceline,
        )

    schema = _system_xml_v2_schema()
    if not schema.validate(root):
        issues = tuple(
            ValidationIssue(
                layer="schema",
                code="XSD_VALIDATION_ERROR",
                message=entry.message.strip(),
                path=entry.path or None,
                line=entry.line or None,
            )
            for entry in schema.error_log
        )
        return SystemXmlValidationReport(document=None, issues=issues)

    document = _parse_validated_root(root)
    document, compatibility_issues = _normalize_legacy_amesim_gas_references(
        document
    )
    document, property_model_issues = _normalize_amesim_gas_property_models(
        document
    )
    document, parameter_encoding_issues = (
        _normalize_legacy_amesim_parameter_encoding(document)
    )
    document = replace(
        document,
        medium_reference_version="1",
        amesim_parameter_encoding_version=AMESIM_PARAMETER_ENCODING_VERSION,
    )
    issues = tuple(
        [
            *compatibility_issues,
            *property_model_issues,
            *parameter_encoding_issues,
            *_semantic_issues(document),
        ]
    )
    return SystemXmlValidationReport(document=document, issues=issues)


@lru_cache(maxsize=1)
def _system_xml_v2_schema() -> etree.XMLSchema:
    schema_document = etree.parse(str(SYSTEM_XML_V2_SCHEMA_PATH))
    return etree.XMLSchema(schema_document)


def _failed_report(
    layer: ValidationLayer,
    code: str,
    message: str,
    *,
    line: int | None = None,
) -> SystemXmlValidationReport:
    return SystemXmlValidationReport(
        document=None,
        issues=(ValidationIssue(layer=layer, code=code, message=message, line=line),),
    )


def _parse_validated_root(root: etree._Element) -> SystemXmlDocument:
    simulation_element = root.find("Simulation")
    components_element = root.find("Components")
    connections_element = root.find("Connections")
    assert simulation_element is not None
    assert components_element is not None
    assert connections_element is not None

    simulation = SystemXmlSimulation(
        t_start=float(simulation_element.get("tStart")),
        t_stop=float(simulation_element.get("tStop")),
        step=float(simulation_element.get("step")),
        max_step=float(simulation_element.get("maxStep")),
        method=str(simulation_element.get("method")),
        line=simulation_element.sourceline,
    )
    components = tuple(
        _parse_component(component) for component in components_element.findall("Component")
    )
    connections = tuple(
        _parse_connection(connection)
        for connection in connections_element.findall("Connection")
    )
    return SystemXmlDocument(
        name=str(root.get("name")),
        schema_version=str(root.get("schemaVersion")),
        unit_system=str(root.get("unitSystem")),
        medium_reference_version=root.get("mediumReferenceVersion"),
        amesim_parameter_encoding_version=root.get(
            "amesimParameterEncodingVersion"
        ),
        simulation=simulation,
        components=components,
        connections=connections,
    )


def _parse_component(element: etree._Element) -> SystemXmlComponent:
    ports = tuple(
        SystemXmlPort(
            name=str(port.get("name")),
            kind=str(port.get("kind")),
            domain=str(port.get("domain")),
            nominal_role=str(port.get("nominalRole")),
            positive_flow_direction=port.get("positiveFlowDirection"),
            side=str(port.get("side")),
            line=port.sourceline,
        )
        for port in element.findall("Port")
    )
    parameters = tuple(
        SystemXmlParameter(
            name=str(parameter.get("name")),
            value=float(parameter.get("value")),
            line=parameter.sourceline,
        )
        for parameter in element.findall("Parameter")
    )
    return SystemXmlComponent(
        id=str(element.get("id")),
        name=str(element.get("name")),
        model_type=str(element.get("type")),
        component_type=str(element.get("componentType")),
        x=float(element.get("x")),
        y=float(element.get("y")),
        rotation=int(element.get("rotation", "0")),
        mirrored=element.get("mirrored", "false") in {"true", "1"},
        ports=ports,
        parameters=parameters,
        line=element.sourceline,
    )


def _normalize_legacy_amesim_parameter_encoding(
    document: SystemXmlDocument,
) -> tuple[SystemXmlDocument, tuple[ValidationIssue, ...]]:
    """Upgrade legacy app-local MECMAS21 0/1 codes to AMESim 1/2 codes."""

    if document.amesim_parameter_encoding_version is not None:
        return document, ()

    normalized_components: list[SystemXmlComponent] = []
    issues: list[ValidationIssue] = []
    for component in document.components:
        if component.model_type != "amesim_mecmas21":
            normalized_components.append(component)
            continue

        changed_parameters: list[str] = []
        normalized_parameters: list[SystemXmlParameter] = []
        parameter_names: set[str] = set()
        for parameter in component.parameters:
            parameter_names.add(parameter.name)
            value = parameter.value
            if parameter.name in MECMAS21_LEGACY_BINARY_PARAMETERS:
                if value == 0.0:
                    value = 1.0
                    changed_parameters.append(parameter.name)
                elif value == 1.0:
                    value = 2.0
                    changed_parameters.append(parameter.name)
            normalized_parameters.append(replace(parameter, value=value))

        for name in MECMAS21_LEGACY_BINARY_PARAMETERS:
            if name not in parameter_names:
                normalized_parameters.append(
                    SystemXmlParameter(name=name, value=2.0, line=component.line)
                )
                changed_parameters.append(name)

        normalized_components.append(
            replace(component, parameters=tuple(normalized_parameters))
        )
        if changed_parameters:
            issues.append(
                ValidationIssue(
                    layer="semantic",
                    code="AMESIM_PARAMETER_ENCODING_MIGRATED",
                    message=(
                        f"Component '{component.id}' used legacy MECMAS21 "
                        "option semantics; normalized "
                        + ", ".join(changed_parameters)
                        + " to canonical AMESim 1/2 codes."
                    ),
                    severity="warning",
                    path=f"/System/Components/Component[@id='{component.id}']",
                    line=component.line,
                )
            )

    return (
        replace(
            document,
            components=tuple(normalized_components),
            amesim_parameter_encoding_version=AMESIM_PARAMETER_ENCODING_VERSION,
        ),
        tuple(issues),
    )


def _normalize_legacy_amesim_gas_references(
    document: SystemXmlDocument,
) -> tuple[SystemXmlDocument, list[ValidationIssue]]:
    """Safely normalize the pre-v1 implicit-air ``gi`` convention.

    System XML v2 files without ``mediumReferenceVersion`` predate explicit
    project medium definitions.  Such a document can only be migrated when
    every component type is known and no component declares the
    ``amesimGasMediumDefinition`` catalog role.  Parameter discovery is driven
    by the catalog editor contract instead of hard-coded AMESim model names.
    """

    if document.medium_reference_version is not None:
        return document, []

    specs = [
        COMPONENT_MODEL_REGISTRY.get(component.model_type)
        for component in document.components
    ]
    if any(spec is None for spec in specs):
        return document, []
    if any(
        spec is not None
        and spec.display.role == "amesimGasMediumDefinition"
        for spec in specs
    ):
        return document, []

    issues: list[ValidationIssue] = []
    normalized_components: list[SystemXmlComponent] = []
    changed = False
    for component, spec in zip(document.components, specs, strict=True):
        assert spec is not None
        reference_parameters = {
            parameter.name: parameter
            for parameter in spec.parameters
            if parameter.editor == "amesimGasReference"
            and parameter.default == 0.0
        }
        if not reference_parameters:
            normalized_components.append(component)
            continue

        parameters_by_name = {
            parameter.name: parameter for parameter in component.parameters
        }
        normalized_parameters = list(component.parameters)
        for name, definition in reference_parameters.items():
            existing = parameters_by_name.get(name)
            path = (
                "/System/Components/"
                f"Component[@id='{component.id}']/Parameter[@name='{name}']"
            )
            if existing is None:
                normalized_parameters.append(
                    SystemXmlParameter(
                        name=name,
                        value=definition.default,
                        line=component.line,
                    )
                )
                issues.append(
                    _semantic_issue(
                        "AMESIM_GI_DEFAULT_INJECTED",
                        f"Legacy component {component.id} omitted {name}; "
                        "the implicit ideal-air reference 0 was applied.",
                        path,
                        component.line,
                        severity="warning",
                    )
                )
                changed = True
                continue
            if existing.value != 1.0:
                continue
            normalized_parameters = [
                replace(parameter, value=0.0)
                if parameter is existing
                else parameter
                for parameter in normalized_parameters
            ]
            issues.append(
                _semantic_issue(
                    "AMESIM_GI_LEGACY_AIR_MAPPED",
                    f"Legacy component {component.id} used {name}=1 without "
                    "a medium-definition component; it was mapped to the "
                    "implicit ideal-air reference 0.",
                    path,
                    existing.line,
                    severity="warning",
                )
            )
            changed = True
        normalized_components.append(
            replace(component, parameters=tuple(normalized_parameters))
        )

    if not changed:
        return document, issues
    return replace(document, components=tuple(normalized_components)), issues


def _normalize_amesim_gas_property_models(
    document: SystemXmlDocument,
) -> tuple[SystemXmlDocument, list[ValidationIssue]]:
    """Add the catalog default for pre-selector medium definitions.

    Medium-definition nodes created before the property-model selector existed
    only contain ``gi``.  The selector is catalog driven, so its default can be
    injected without hard-coding a component type or a calculation method.
    """

    issues: list[ValidationIssue] = []
    normalized_components: list[SystemXmlComponent] = []
    changed = False

    for component in document.components:
        spec = COMPONENT_MODEL_REGISTRY.get(component.model_type)
        if spec is None or spec.display.role != "amesimGasMediumDefinition":
            normalized_components.append(component)
            continue

        existing_names = {parameter.name for parameter in component.parameters}
        normalized_parameters = list(component.parameters)
        for definition in spec.parameters:
            if (
                definition.editor != "amesimGasPropertyModel"
                or definition.name in existing_names
            ):
                continue
            normalized_parameters.append(
                SystemXmlParameter(
                    name=definition.name,
                    value=definition.default,
                    line=component.line,
                )
            )
            issues.append(
                _semantic_issue(
                    "AMESIM_GAS_PROPERTY_MODEL_DEFAULTED",
                    f"Medium definition {component.id} omitted "
                    f"{definition.name}; catalog default {definition.default:g} "
                    "was applied.",
                    "/System/Components/"
                    f"Component[@id='{component.id}']/"
                    f"Parameter[@name='{definition.name}']",
                    component.line,
                    severity="warning",
                )
            )
            changed = True

        normalized_components.append(
            replace(component, parameters=tuple(normalized_parameters))
        )

    if not changed:
        return document, issues
    return replace(document, components=tuple(normalized_components)), issues


def _parse_connection(element: etree._Element) -> SystemXmlConnection:
    endpoints = tuple(
        SystemXmlEndpoint(
            component=str(endpoint.get("component")),
            port=str(endpoint.get("port")),
            role=endpoint.get("role"),
            line=endpoint.sourceline,
        )
        for endpoint in element.findall("Endpoint")
    )
    assert len(endpoints) == 2
    return SystemXmlConnection(
        id=str(element.get("id")),
        kind=str(element.get("kind")),
        domain=str(element.get("domain")),
        endpoints=(endpoints[0], endpoints[1]),
        line=element.sourceline,
    )


def _semantic_issues(document: SystemXmlDocument) -> list[ValidationIssue]:
    issues: list[ValidationIssue] = []
    _validate_system_and_simulation(document, issues)
    component_by_id = _validate_components(document, issues)
    _validate_connections(document, component_by_id, issues)
    _validate_amesim_medium_references(document, component_by_id, issues)
    return issues


def _validate_system_and_simulation(
    document: SystemXmlDocument,
    issues: list[ValidationIssue],
) -> None:
    if not document.name.strip():
        issues.append(_semantic_issue("SYSTEM_NAME_EMPTY", "System name cannot be blank.", "/System"))
    if not document.components:
        issues.append(
            _semantic_issue(
                "SYSTEM_HAS_NO_COMPONENTS",
                "The system must contain at least one component.",
                "/System/Components",
            )
        )

    simulation = document.simulation
    values = {
        "tStart": simulation.t_start,
        "tStop": simulation.t_stop,
        "step": simulation.step,
        "maxStep": simulation.max_step,
    }
    for name, value in values.items():
        if not isfinite(value):
            issues.append(
                _semantic_issue(
                    "SIMULATION_VALUE_NOT_FINITE",
                    f"Simulation value {name} must be finite.",
                    f"/System/Simulation/@{name}",
                    simulation.line,
                )
            )
    if isfinite(simulation.t_start) and isfinite(simulation.t_stop):
        if simulation.t_stop <= simulation.t_start:
            issues.append(
                _semantic_issue(
                    "SIMULATION_TIME_RANGE_INVALID",
                    "Simulation tStop must be greater than tStart.",
                    "/System/Simulation",
                    simulation.line,
                )
            )
    for name, value in {
        "step": simulation.step,
        "maxStep": simulation.max_step,
    }.items():
        if isfinite(value) and value <= 0.0:
            issues.append(
                _semantic_issue(
                    "SIMULATION_STEP_INVALID",
                    f"Simulation value {name} must be greater than zero.",
                    f"/System/Simulation/@{name}",
                    simulation.line,
                )
            )
    if simulation.method not in SUPPORTED_SOLVER_METHODS:
        issues.append(
            _semantic_issue(
                "SIMULATION_METHOD_UNSUPPORTED",
                f"Unsupported solver method: {simulation.method}.",
                "/System/Simulation/@method",
                simulation.line,
            )
        )


def _validate_components(
    document: SystemXmlDocument,
    issues: list[ValidationIssue],
) -> dict[str, SystemXmlComponent]:
    component_by_id: dict[str, SystemXmlComponent] = {}
    names: dict[str, str] = {}
    for index, component in enumerate(document.components, start=1):
        path = f"/System/Components/Component[{index}]"
        if component.id in component_by_id:
            issues.append(
                _semantic_issue(
                    "COMPONENT_ID_DUPLICATE",
                    f"Duplicate component id: {component.id}.",
                    path,
                    component.line,
                )
            )
        else:
            component_by_id[component.id] = component

        if component.name in names:
            issues.append(
                _semantic_issue(
                    "COMPONENT_NAME_DUPLICATE",
                    f"Duplicate component name: {component.name}.",
                    path,
                    component.line,
                )
            )
        else:
            names[component.name] = component.id

        spec = COMPONENT_MODEL_REGISTRY.get(component.model_type)
        if spec is None:
            issues.append(
                _semantic_issue(
                    "COMPONENT_TYPE_UNSUPPORTED",
                    f"Unsupported component model type: {component.model_type}.",
                    f"{path}/@type",
                    component.line,
                )
            )
            continue
        if component.component_type != component.model_type:
            issues.append(
                _semantic_issue(
                    "COMPONENT_TYPE_MISMATCH",
                    f"componentType '{component.component_type}' does not match model type '{component.model_type}'.",
                    f"{path}/@componentType",
                    component.line,
                )
            )
        if not isfinite(component.x) or not isfinite(component.y):
            issues.append(
                _semantic_issue(
                    "COMPONENT_POSITION_NOT_FINITE",
                    f"Component {component.id} position must be finite.",
                    path,
                    component.line,
                )
            )
        _validate_component_ports(component, spec.ports, path, issues)
        _validate_component_parameters(component, spec.parameter_by_name, path, issues)
    return component_by_id


def _validate_component_ports(
    component: SystemXmlComponent,
    expected_ports: tuple[PortDefinition, ...],
    component_path: str,
    issues: list[ValidationIssue],
) -> None:
    actual_by_name: dict[str, SystemXmlPort] = {}
    for port_index, port in enumerate(component.ports, start=1):
        path = f"{component_path}/Port[{port_index}]"
        if port.name in actual_by_name:
            issues.append(
                _semantic_issue(
                    "PORT_NAME_DUPLICATE",
                    f"Component {component.id} contains duplicate port {port.name}.",
                    path,
                    port.line,
                )
            )
        else:
            actual_by_name[port.name] = port

    expected_by_name = {port.name: port for port in expected_ports}
    for name in sorted(set(expected_by_name) - set(actual_by_name)):
        issues.append(
            _semantic_issue(
                "PORT_REQUIRED_MISSING",
                f"Component {component.id} is missing registered port {name}.",
                component_path,
                component.line,
            )
        )
    for name in sorted(set(actual_by_name) - set(expected_by_name)):
        port = actual_by_name[name]
        issues.append(
            _semantic_issue(
                "PORT_UNSUPPORTED",
                f"Component {component.id} contains unsupported port {name}.",
                component_path,
                port.line,
            )
        )

    for name in sorted(set(actual_by_name) & set(expected_by_name)):
        actual = actual_by_name[name]
        expected = expected_by_name[name]
        path = f"{component_path}/Port[@name='{name}']"
        if actual.kind != expected.kind or actual.domain != expected.domain:
            issues.append(
                _semantic_issue(
                    "PORT_INTERFACE_MISMATCH",
                    f"Port {component.id}.{name} has an incompatible kind or domain.",
                    path,
                    actual.line,
                )
            )
        if actual.nominal_role != expected.nominal_role:
            issues.append(
                _semantic_issue(
                    "PORT_NOMINAL_ROLE_MISMATCH",
                    f"Port {component.id}.{name} has nominalRole '{actual.nominal_role}', expected '{expected.nominal_role}'.",
                    path,
                    actual.line,
                )
            )
        if actual.kind == "physical" and (
            actual.positive_flow_direction != expected.positive_flow_direction
        ):
            issues.append(
                _semantic_issue(
                    "PORT_FLOW_SIGN_MISMATCH",
                    f"Port {component.id}.{name} must use positiveFlowDirection='intoComponent'.",
                    path,
                    actual.line,
                )
            )


def _validate_component_parameters(
    component: SystemXmlComponent,
    expected_parameters: Mapping[str, ParameterSpec],
    component_path: str,
    issues: list[ValidationIssue],
) -> None:
    actual_by_name: dict[str, SystemXmlParameter] = {}
    for parameter_index, parameter in enumerate(component.parameters, start=1):
        path = f"{component_path}/Parameter[{parameter_index}]"
        if parameter.name in actual_by_name:
            issues.append(
                _semantic_issue(
                    "PARAMETER_NAME_DUPLICATE",
                    f"Component {component.id} contains duplicate parameter {parameter.name}.",
                    path,
                    parameter.line,
                )
            )
        else:
            actual_by_name[parameter.name] = parameter

    for name in sorted(set(expected_parameters) - set(actual_by_name)):
        issues.append(
            _semantic_issue(
                "PARAMETER_REQUIRED_MISSING",
                f"Component {component.id} is missing required parameter {name}.",
                component_path,
                component.line,
            )
        )
    for name in sorted(set(actual_by_name) - set(expected_parameters)):
        parameter = actual_by_name[name]
        issues.append(
            _semantic_issue(
                "PARAMETER_UNSUPPORTED",
                f"Component {component.id} contains unsupported parameter {name}.",
                component_path,
                parameter.line,
            )
        )
    for name in sorted(set(actual_by_name) & set(expected_parameters)):
        parameter = actual_by_name[name]
        message = expected_parameters[name].validation_message(parameter.value)
        if message is not None:
            issues.append(
                _semantic_issue(
                    "PARAMETER_VALUE_INVALID",
                    f"Parameter {component.id}.{name} {message}.",
                    f"{component_path}/Parameter[@name='{name}']",
                    parameter.line,
                )
            )


def _validate_amesim_medium_references(
    document: SystemXmlDocument,
    component_by_id: Mapping[str, SystemXmlComponent],
    issues: list[ValidationIssue],
) -> None:
    """Validate project-scoped AMESim gas slots before model compilation."""

    component_paths = {
        id(component): f"/System/Components/Component[{index}]"
        for index, component in enumerate(document.components, start=1)
    }
    definitions_by_index: dict[int, SystemXmlComponent] = {}

    for component in document.components:
        spec = COMPONENT_MODEL_REGISTRY.get(component.model_type)
        if spec is None or spec.display.role != "amesimGasMediumDefinition":
            continue
        component_path = component_paths[id(component)]
        parameters = [
            parameter for parameter in component.parameters if parameter.name == "gi"
        ]
        if len(parameters) != 1:
            if not parameters:
                issues.append(
                    _semantic_issue(
                        "AMESIM_GAS_MEDIUM_INDEX_MISSING",
                        f"AMESim gas medium definition {component.id} must "
                        "declare exactly one gi parameter.",
                        component_path,
                        component.line,
                    )
                )
            continue

        parameter = parameters[0]
        parameter_path = f"{component_path}/Parameter[@name='gi']"
        gas_index = _integer_index(parameter.value, minimum=1, maximum=99)
        if gas_index is None:
            issues.append(
                _semantic_issue(
                    "AMESIM_GAS_MEDIUM_INDEX_INVALID",
                    f"AMESim gas medium definition {component.id} gi must be "
                    "an integer between 1 and 99.",
                    parameter_path,
                    parameter.line,
                )
            )
            continue

        existing = definitions_by_index.get(gas_index)
        if existing is not None:
            issues.append(
                _semantic_issue(
                    "AMESIM_GAS_MEDIUM_INDEX_DUPLICATE",
                    f"AMESim gas medium index gi={gas_index} is defined by "
                    f"both {existing.id} and {component.id}.",
                    parameter_path,
                    parameter.line,
                )
            )
            continue
        definitions_by_index[gas_index] = component

    resolved_references: dict[
        str,
        list[tuple[str, int, SystemXmlParameter]],
    ] = {}
    for component in document.components:
        spec = COMPONENT_MODEL_REGISTRY.get(component.model_type)
        if spec is None:
            continue
        component_path = component_paths[id(component)]
        for definition in spec.parameters:
            if definition.editor != "amesimGasReference":
                continue
            parameters = [
                parameter
                for parameter in component.parameters
                if parameter.name == definition.name
            ]
            if len(parameters) != 1:
                continue

            parameter = parameters[0]
            parameter_path = (
                f"{component_path}/Parameter[@name='{definition.name}']"
            )
            gas_index = _integer_index(parameter.value, minimum=0, maximum=99)
            if gas_index is None:
                issues.append(
                    _semantic_issue(
                        "AMESIM_GAS_REFERENCE_INDEX_INVALID",
                        f"AMESim gas reference {component.id}.{definition.name} "
                        "must be an integer between 0 and 99.",
                        parameter_path,
                        parameter.line,
                    )
                )
                continue
            if gas_index > 0 and gas_index not in definitions_by_index:
                issues.append(
                    _semantic_issue(
                        "AMESIM_GAS_REFERENCE_UNDEFINED",
                        f"AMESim gas reference {component.id}.{definition.name} "
                        f"uses gi={gas_index}, but no gas medium definition "
                        "declares that index.",
                        parameter_path,
                        parameter.line,
                    )
                )
                continue
            resolved_references.setdefault(component.id, []).append(
                (definition.name, gas_index, parameter)
            )

    _validate_connected_amesim_gas_references(
        document,
        component_by_id,
        resolved_references,
        issues,
    )


def _integer_index(value: float, *, minimum: int, maximum: int) -> int | None:
    if not isfinite(value) or value != int(value):
        return None
    index = int(value)
    if not minimum <= index <= maximum:
        return None
    return index


def _validate_connected_amesim_gas_references(
    document: SystemXmlDocument,
    component_by_id: Mapping[str, SystemXmlComponent],
    resolved_references: Mapping[
        str,
        list[tuple[str, int, SystemXmlParameter]],
    ],
    issues: list[ValidationIssue],
) -> None:
    if len(component_by_id) != len(document.components):
        return

    parents = {component_id: component_id for component_id in component_by_id}

    def find(component_id: str) -> str:
        parent = parents[component_id]
        while parent != parents[parent]:
            parent = parents[parent]
        while component_id != parent:
            next_component = parents[component_id]
            parents[component_id] = parent
            component_id = next_component
        return parent

    def union(left: str, right: str) -> None:
        left_root = find(left)
        right_root = find(right)
        if left_root != right_root:
            parents[right_root] = left_root

    for connection in document.connections:
        if connection.kind != "physical" or connection.domain != "pneumatic":
            continue
        first_endpoint, second_endpoint = connection.endpoints
        first_component = component_by_id.get(first_endpoint.component)
        second_component = component_by_id.get(second_endpoint.component)
        if first_component is None or second_component is None:
            continue
        first_port = first_component.port_by_name.get(first_endpoint.port)
        second_port = second_component.port_by_name.get(second_endpoint.port)
        if (
            first_port is None
            or second_port is None
            or first_port.kind != "physical"
            or second_port.kind != "physical"
            or first_port.domain != "pneumatic"
            or second_port.domain != "pneumatic"
        ):
            continue
        union(first_component.id, second_component.id)

    references_by_root: dict[
        str,
        dict[int, list[tuple[str, str, SystemXmlParameter]]],
    ] = {}
    for component_id, references in resolved_references.items():
        if component_id not in parents:
            continue
        by_index = references_by_root.setdefault(find(component_id), {})
        for parameter_name, gas_index, parameter in references:
            by_index.setdefault(gas_index, []).append(
                (component_id, parameter_name, parameter)
            )

    for references_by_index in references_by_root.values():
        if len(references_by_index) <= 1:
            continue
        details = ", ".join(
            f"gi={gas_index} "
            f"({', '.join(sorted(component_id for component_id, _, _ in references))})"
            for gas_index, references in sorted(references_by_index.items())
        )
        first_reference = next(
            reference
            for references in references_by_index.values()
            for reference in references
        )
        component_id, parameter_name, parameter = first_reference
        issues.append(
            _semantic_issue(
                "AMESIM_GAS_REFERENCE_CONFLICT",
                "Connected pneumatic circuit contains conflicting AMESim "
                f"gas references: {details}.",
                "/System/Components/"
                f"Component[@id='{component_id}']/"
                f"Parameter[@name='{parameter_name}']",
                parameter.line,
            )
        )


def _validate_connections(
    document: SystemXmlDocument,
    component_by_id: dict[str, SystemXmlComponent],
    issues: list[ValidationIssue],
) -> None:
    connection_ids: set[str] = set()
    connection_keys: set[tuple[tuple[str, str], tuple[str, str]]] = set()
    occupied_physical_ports: dict[tuple[str, str], str] = {}
    referenced_ports: set[tuple[str, str]] = set()

    for index, connection in enumerate(document.connections, start=1):
        path = f"/System/Connections/Connection[{index}]"
        if connection.id in connection_ids:
            issues.append(
                _semantic_issue(
                    "CONNECTION_ID_DUPLICATE",
                    f"Duplicate connection id: {connection.id}.",
                    path,
                    connection.line,
                )
            )
        connection_ids.add(connection.id)

        if connection.undirected_key in connection_keys:
            issues.append(
                _semantic_issue(
                    "CONNECTION_DUPLICATE",
                    f"Connection {connection.id} duplicates an existing endpoint pair.",
                    path,
                    connection.line,
                )
            )
        connection_keys.add(connection.undirected_key)

        if connection.endpoints[0].key == connection.endpoints[1].key:
            issues.append(
                _semantic_issue(
                    "CONNECTION_SELF_REFERENCE",
                    f"Connection {connection.id} connects an endpoint to itself.",
                    path,
                    connection.line,
                )
            )

        resolved_endpoints: list[tuple[SystemXmlEndpoint, SystemXmlPort]] = []
        for endpoint_index, endpoint in enumerate(connection.endpoints, start=1):
            endpoint_path = f"{path}/Endpoint[{endpoint_index}]"
            component = component_by_id.get(endpoint.component)
            if component is None:
                issues.append(
                    _semantic_issue(
                        "ENDPOINT_COMPONENT_UNKNOWN",
                        f"Connection {connection.id} references unknown component {endpoint.component}.",
                        endpoint_path,
                        endpoint.line,
                    )
                )
                continue
            port = component.port_by_name.get(endpoint.port)
            if port is None:
                issues.append(
                    _semantic_issue(
                        "ENDPOINT_PORT_UNKNOWN",
                        f"Connection {connection.id} references unknown port {endpoint.component}.{endpoint.port}.",
                        endpoint_path,
                        endpoint.line,
                    )
                )
                continue
            resolved_endpoints.append((endpoint, port))
            referenced_ports.add(endpoint.key)
            if port.kind != connection.kind or port.domain != connection.domain:
                issues.append(
                    _semantic_issue(
                        "CONNECTION_INTERFACE_MISMATCH",
                        f"Connection {connection.id} kind/domain does not match {endpoint.component}.{endpoint.port}.",
                        endpoint_path,
                        endpoint.line,
                    )
                )
            if connection.kind == "physical":
                if endpoint.role is not None:
                    issues.append(
                        _semantic_issue(
                            "PHYSICAL_ENDPOINT_HAS_ROLE",
                            f"Physical endpoint {endpoint.component}.{endpoint.port} must not declare a source/target role.",
                            endpoint_path,
                            endpoint.line,
                        )
                    )
                previous = occupied_physical_ports.get(endpoint.key)
                if previous is not None:
                    issues.append(
                        _semantic_issue(
                            "PHYSICAL_PORT_ALREADY_CONNECTED",
                            f"Physical port {endpoint.component}.{endpoint.port} is already used by connection {previous}; use a Tee for branching.",
                            endpoint_path,
                            endpoint.line,
                        )
                    )
                else:
                    occupied_physical_ports[endpoint.key] = connection.id

        if len(resolved_endpoints) == 2:
            first_port = resolved_endpoints[0][1]
            second_port = resolved_endpoints[1][1]
            if first_port.kind != second_port.kind:
                issues.append(
                    _semantic_issue(
                        "CONNECTION_MIXES_PORT_KINDS",
                        f"Connection {connection.id} mixes physical and signal ports.",
                        path,
                        connection.line,
                    )
                )
            if first_port.domain != second_port.domain:
                issues.append(
                    _semantic_issue(
                        "CONNECTION_DOMAIN_MISMATCH",
                        f"Connection {connection.id} connects different physical domains.",
                        path,
                        connection.line,
                    )
                )

        if connection.kind == "signal":
            roles = {endpoint.role for endpoint in connection.endpoints}
            if roles != {"source", "target"}:
                issues.append(
                    _semantic_issue(
                        "SIGNAL_ENDPOINT_ROLES_INVALID",
                        f"Signal connection {connection.id} must contain source and target roles.",
                        path,
                        connection.line,
                    )
                )
            for endpoint, port in resolved_endpoints:
                expected_role = "source" if port.nominal_role == "output" else "target"
                if endpoint.role != expected_role:
                    issues.append(
                        _semantic_issue(
                            "SIGNAL_DIRECTION_MISMATCH",
                            f"Signal endpoint {endpoint.component}.{endpoint.port} has role '{endpoint.role}', expected '{expected_role}'.",
                            path,
                            endpoint.line,
                        )
                    )

    for component in document.components:
        for port in component.ports:
            if (component.id, port.name) not in referenced_ports:
                issues.append(
                    _semantic_issue(
                        "PORT_UNCONNECTED",
                        f"Port {component.id}.{port.name} is not connected.",
                        f"/System/Components/Component[@id='{component.id}']/Port[@name='{port.name}']",
                        port.line,
                        severity="warning",
                    )
                )


def _semantic_issue(
    code: str,
    message: str,
    path: str,
    line: int | None = None,
    *,
    severity: ValidationSeverity = "error",
) -> ValidationIssue:
    return ValidationIssue(
        layer="semantic",
        code=code,
        message=message,
        severity=severity,
        path=path,
        line=line,
    )