1249 lines
44 KiB
Python
1249 lines
44 KiB
Python
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,
|
|
)
|