from __future__ import annotations from collections.abc import Mapping from dataclasses import dataclass 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 simulation: SystemXmlSimulation components: tuple[SystemXmlComponent, ...] connections: tuple[SystemXmlConnection, ...] def summary(self) -> dict[str, object]: return { "name": self.name, "schemaVersion": self.schema_version, "unitSystem": self.unit_system, "componentCount": len(self.components), "connectionCount": len(self.connections), } 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, "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) issues = tuple(_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")), 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 _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) 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_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, )