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"} @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 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 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, "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 = replace(document, medium_reference_version="1") issues = tuple( [ *compatibility_issues, *property_model_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"), 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_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, )