977 lines
33 KiB
Python
977 lines
33 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.performance import profile_phase
|
|
from app.simulation.registry import COMPONENT_MODEL_REGISTRY, ParameterSpec
|
|
from app.simulation.config import SolveIVPConfig
|
|
from app.simulation.sampling import (
|
|
SimulationSampleTimeError,
|
|
simulation_sample_times,
|
|
)
|
|
|
|
|
|
ValidationLayer = Literal["xml", "schema", "semantic"]
|
|
ValidationSeverity = Literal["error", "warning"]
|
|
SYSTEM_XML_MAX_BYTES = 5 * 1024 * 1024
|
|
SYSTEM_XML_V3_SCHEMA_PATH = (
|
|
Path(__file__).resolve().parent.parent / "schemas" / "system-simulation-v3.xsd"
|
|
)
|
|
SUPPORTED_SOLVER_METHODS = {"RK45", "BDF"}
|
|
|
|
|
|
@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
|
|
sample_step: float
|
|
max_step: float
|
|
method: str
|
|
line: int | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SystemXmlParameter:
|
|
name: str
|
|
value: float
|
|
line: int | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SystemXmlComponent:
|
|
id: str
|
|
model_type: str
|
|
model_version: str
|
|
parameters: tuple[SystemXmlParameter, ...]
|
|
line: int | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SystemXmlEndpoint:
|
|
component: str
|
|
port: str
|
|
line: int | None = None
|
|
|
|
@property
|
|
def key(self) -> tuple[str, str]:
|
|
return self.component, self.port
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SystemXmlConnection:
|
|
id: 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 | None
|
|
schema_version: str
|
|
unit_system: str
|
|
simulation: SystemXmlSimulation
|
|
components: tuple[SystemXmlComponent, ...]
|
|
connections: tuple[SystemXmlConnection, ...]
|
|
|
|
def summary(self) -> dict[str, object]:
|
|
result: dict[str, object] = {
|
|
"schemaVersion": self.schema_version,
|
|
"unitSystem": self.unit_system,
|
|
"componentCount": len(self.components),
|
|
"connectionCount": len(self.connections),
|
|
}
|
|
if self.name is not None:
|
|
result["name"] = self.name
|
|
return result
|
|
|
|
def as_model_data(self) -> dict[str, object]:
|
|
"""Return the v3 execution-model contract without editor-only fields."""
|
|
|
|
return {
|
|
"name": self.name or "untitled",
|
|
"simulation": {
|
|
"t_start": self.simulation.t_start,
|
|
"t_stop": self.simulation.t_stop,
|
|
"sample_step": self.simulation.sample_step,
|
|
"max_step": self.simulation.max_step,
|
|
"method": self.simulation.method,
|
|
},
|
|
"components": [
|
|
{
|
|
"id": component.id,
|
|
"model_type": component.model_type,
|
|
"model_version": component.model_version,
|
|
"parameters": {
|
|
parameter.name: parameter.value
|
|
for parameter in component.parameters
|
|
},
|
|
}
|
|
for component in self.components
|
|
],
|
|
"connections": [
|
|
{
|
|
"id": connection.id,
|
|
"endpoints": [
|
|
{
|
|
"component": endpoint.component,
|
|
"port": endpoint.port,
|
|
}
|
|
for endpoint in connection.endpoints
|
|
],
|
|
}
|
|
for connection in self.connections
|
|
],
|
|
}
|
|
|
|
def as_dict(self) -> dict[str, object]:
|
|
return {
|
|
"schemaVersion": self.schema_version,
|
|
"unitSystem": self.unit_system,
|
|
**self.as_model_data(),
|
|
}
|
|
|
|
|
|
@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
|
|
|
|
|
|
@profile_phase("simulation.xml_validation")
|
|
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_v3_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)
|
|
return SystemXmlValidationReport(
|
|
document=document,
|
|
issues=tuple(_semantic_issues(document)),
|
|
)
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def _system_xml_v3_schema() -> etree.XMLSchema:
|
|
schema_document = etree.parse(str(SYSTEM_XML_V3_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")),
|
|
sample_step=float(simulation_element.get("sampleStep")),
|
|
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, index)
|
|
for index, connection in enumerate(
|
|
connections_element.findall("Connection"), start=1
|
|
)
|
|
)
|
|
return SystemXmlDocument(
|
|
name=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:
|
|
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")),
|
|
model_type=str(element.get("type")),
|
|
model_version=str(element.get("modelVersion")),
|
|
parameters=parameters,
|
|
line=element.sourceline,
|
|
)
|
|
|
|
|
|
def _parse_connection(
|
|
element: etree._Element,
|
|
index: int,
|
|
) -> SystemXmlConnection:
|
|
endpoints = tuple(
|
|
SystemXmlEndpoint(
|
|
component=str(endpoint.get("component")),
|
|
port=str(endpoint.get("port")),
|
|
line=endpoint.sourceline,
|
|
)
|
|
for endpoint in element.findall("Endpoint")
|
|
)
|
|
assert len(endpoints) == 2
|
|
return SystemXmlConnection(
|
|
id=element.get("id") or f"connection_{index}",
|
|
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.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,
|
|
"sampleStep": simulation.sample_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,
|
|
)
|
|
)
|
|
elif isfinite(simulation.sample_step) and simulation.sample_step > 0.0:
|
|
try:
|
|
simulation_sample_times(
|
|
SolveIVPConfig(
|
|
t_start=simulation.t_start,
|
|
t_stop=simulation.t_stop,
|
|
),
|
|
simulation.sample_step,
|
|
)
|
|
except SimulationSampleTimeError as exc:
|
|
issues.append(
|
|
_semantic_issue(
|
|
exc.code,
|
|
str(exc),
|
|
(
|
|
"/System/Simulation"
|
|
if exc.code == "SIMULATION_TIME_SPAN_NOT_FINITE"
|
|
else "/System/Simulation/@sampleStep"
|
|
),
|
|
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] = {}
|
|
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
|
|
|
|
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.model_version != spec.model_version:
|
|
issues.append(
|
|
_semantic_issue(
|
|
"COMPONENT_MODEL_VERSION_MISMATCH",
|
|
f"Component {component.id} declares modelVersion "
|
|
f"'{component.model_version}', expected '{spec.model_version}'.",
|
|
f"{path}/@modelVersion",
|
|
component.line,
|
|
)
|
|
)
|
|
_validate_component_parameters(
|
|
component,
|
|
spec.parameter_by_name,
|
|
path,
|
|
issues,
|
|
)
|
|
return component_by_id
|
|
|
|
|
|
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 _registered_port(
|
|
component: SystemXmlComponent | None,
|
|
port_name: str,
|
|
) -> PortDefinition | None:
|
|
if component is None:
|
|
return None
|
|
spec = COMPONENT_MODEL_REGISTRY.get(component.model_type)
|
|
if spec is None:
|
|
return None
|
|
return next((port for port in spec.ports if port.name == port_name), None)
|
|
|
|
def _active_registered_ports(
|
|
component: SystemXmlComponent,
|
|
) -> tuple[PortDefinition, ...]:
|
|
spec = COMPONENT_MODEL_REGISTRY.get(component.model_type)
|
|
if spec is None:
|
|
return ()
|
|
values = {
|
|
parameter.name: parameter.value
|
|
for parameter in component.parameters
|
|
}
|
|
try:
|
|
return spec.active_ports(values)
|
|
except (KeyError, ValueError):
|
|
return ()
|
|
|
|
|
|
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] = {}
|
|
driven_signal_inputs: 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].component
|
|
== connection.endpoints[1].component
|
|
):
|
|
issues.append(
|
|
_semantic_issue(
|
|
"CONNECTION_SELF_REFERENCE",
|
|
f"Connection {connection.id} connects component "
|
|
f"{connection.endpoints[0].component} to itself.",
|
|
path,
|
|
connection.line,
|
|
)
|
|
)
|
|
|
|
resolved_endpoints: list[tuple[SystemXmlEndpoint, PortDefinition]] = []
|
|
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 "
|
|
f"{endpoint.component}.",
|
|
endpoint_path,
|
|
endpoint.line,
|
|
)
|
|
)
|
|
continue
|
|
port = _registered_port(component, endpoint.port)
|
|
if port is None:
|
|
issues.append(
|
|
_semantic_issue(
|
|
"ENDPOINT_PORT_UNKNOWN",
|
|
f"Connection {connection.id} references unknown port "
|
|
f"{endpoint.component}.{endpoint.port}.",
|
|
endpoint_path,
|
|
endpoint.line,
|
|
)
|
|
)
|
|
continue
|
|
if port not in _active_registered_ports(component):
|
|
issues.append(
|
|
_semantic_issue(
|
|
"ENDPOINT_PORT_INACTIVE",
|
|
f"Connection {connection.id} references inactive port "
|
|
f"{endpoint.component}.{endpoint.port} for its parameters.",
|
|
endpoint_path,
|
|
endpoint.line,
|
|
)
|
|
)
|
|
continue
|
|
resolved_endpoints.append((endpoint, port))
|
|
referenced_ports.add(endpoint.key)
|
|
if port.kind == "physical":
|
|
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 "
|
|
f"already used by connection {previous}; use a Tee for branching.",
|
|
endpoint_path,
|
|
endpoint.line,
|
|
)
|
|
)
|
|
else:
|
|
occupied_physical_ports[endpoint.key] = connection.id
|
|
elif port.nominal_role == "input":
|
|
previous = driven_signal_inputs.get(endpoint.key)
|
|
if previous is not None:
|
|
issues.append(
|
|
_semantic_issue(
|
|
"SIGNAL_INPUT_MULTIPLE_DRIVERS",
|
|
f"Signal input {endpoint.component}.{endpoint.port} is "
|
|
f"already driven by connection {previous}.",
|
|
endpoint_path,
|
|
endpoint.line,
|
|
)
|
|
)
|
|
else:
|
|
driven_signal_inputs[endpoint.key] = connection.id
|
|
|
|
if len(resolved_endpoints) != 2:
|
|
continue
|
|
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,
|
|
)
|
|
)
|
|
continue
|
|
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,
|
|
)
|
|
)
|
|
continue
|
|
if first_port.variables != second_port.variables:
|
|
issues.append(
|
|
_semantic_issue(
|
|
"CONNECTION_VARIABLE_CONTRACT_MISMATCH",
|
|
f"Connection {connection.id} joins incompatible port variable contracts.",
|
|
path,
|
|
connection.line,
|
|
)
|
|
)
|
|
if first_port.kind == "signal" and {
|
|
first_port.nominal_role,
|
|
second_port.nominal_role,
|
|
} != {"input", "output"}:
|
|
issues.append(
|
|
_semantic_issue(
|
|
"SIGNAL_PORT_ROLES_INVALID",
|
|
f"Signal connection {connection.id} must join one output and one input.",
|
|
path,
|
|
connection.line,
|
|
)
|
|
)
|
|
|
|
for component in document.components:
|
|
spec = COMPONENT_MODEL_REGISTRY.get(component.model_type)
|
|
if spec is None:
|
|
continue
|
|
for port in _active_registered_ports(component):
|
|
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}']",
|
|
component.line,
|
|
severity="warning",
|
|
)
|
|
)
|
|
|
|
|
|
def _validate_amesim_medium_references(
|
|
document: SystemXmlDocument,
|
|
component_by_id: Mapping[str, SystemXmlComponent],
|
|
issues: list[ValidationIssue],
|
|
) -> None:
|
|
"""Validate project-scoped AMESim gas slots in the canonical v3 contract."""
|
|
|
|
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:
|
|
first_endpoint, second_endpoint = connection.endpoints
|
|
first_component = component_by_id.get(first_endpoint.component)
|
|
second_component = component_by_id.get(second_endpoint.component)
|
|
first_port = _registered_port(first_component, first_endpoint.port)
|
|
second_port = _registered_port(second_component, second_endpoint.port)
|
|
if (
|
|
first_component is None
|
|
or second_component is None
|
|
or 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 _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,
|
|
)
|