845 lines
29 KiB
Python
845 lines
29 KiB
Python
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,
|
|
)
|