1056 lines
38 KiB
Python
1056 lines
38 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping, Sequence
|
|
from dataclasses import dataclass
|
|
from importlib import import_module
|
|
from math import isfinite
|
|
import re
|
|
from typing import cast
|
|
|
|
from app.simulation.core.base import Component
|
|
from app.simulation.core.catalog import (
|
|
ComponentCategorySpec,
|
|
ComponentDisplaySpec,
|
|
ComponentLibrarySpec,
|
|
ParameterGroupDisplaySpec,
|
|
PortDisplaySpec,
|
|
)
|
|
from app.simulation.core.metadata import (
|
|
SI_UNIT_BY_QUANTITY,
|
|
ParameterCondition,
|
|
ParameterDefinition,
|
|
ParameterOption,
|
|
ResultVariableDefinition,
|
|
)
|
|
from app.simulation.core.medium import GasMedium, IdealGasMedium
|
|
from app.simulation.core.ports import PortDefinition, PortVariableDefinition
|
|
from app.simulation.core.port_computation import PortComputation
|
|
|
|
|
|
ParameterSpec = ParameterDefinition
|
|
|
|
ENABLED_COMPONENT_LIBRARIES = (
|
|
"app.simulation.components.experimental.library:LIBRARY",
|
|
"app.simulation.components.amesim.library:LIBRARY",
|
|
)
|
|
|
|
_MACHINE_ID_PATTERN = re.compile(r"[a-z][a-z0-9_]*")
|
|
_MEMBER_ID_PATTERN = re.compile(r"[A-Za-z][A-Za-z0-9_]*")
|
|
_SEMANTIC_VERSION_PATTERN = re.compile(r"\d+\.\d+\.\d+")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ComponentModelSpec:
|
|
"""Validated registry entry backed by one component implementation class."""
|
|
|
|
component_class: type[Component]
|
|
library: ComponentLibrarySpec
|
|
|
|
@property
|
|
def model_type(self) -> str:
|
|
return cast(str, self.component_class.MODEL_TYPE)
|
|
|
|
@property
|
|
def model_version(self) -> str:
|
|
return cast(str, self.component_class.MODEL_VERSION)
|
|
|
|
@property
|
|
def ports(self) -> tuple[PortDefinition, ...]:
|
|
return self.component_class.PORTS
|
|
|
|
@property
|
|
def parameters(self) -> tuple[ParameterDefinition, ...]:
|
|
return self.component_class.PARAMETERS
|
|
|
|
@property
|
|
def result_variables(self) -> tuple[ResultVariableDefinition, ...]:
|
|
return self.component_class.RESULT_VARIABLES
|
|
|
|
@property
|
|
def display(self) -> ComponentDisplaySpec:
|
|
return cast(ComponentDisplaySpec, self.component_class.DISPLAY)
|
|
|
|
@property
|
|
def parameter_by_name(self) -> dict[str, ParameterDefinition]:
|
|
return {parameter.name: parameter for parameter in self.parameters}
|
|
|
|
def active_ports(
|
|
self,
|
|
values: Mapping[str, float],
|
|
) -> tuple[PortDefinition, ...]:
|
|
resolved = {
|
|
parameter.name: values.get(parameter.name, parameter.default)
|
|
for parameter in self.parameters
|
|
}
|
|
return self.component_class.active_port_definitions_for_parameters(resolved)
|
|
|
|
def as_catalog_dict(self) -> dict[str, object]:
|
|
category = self.library.category_by_id[self.display.category_id]
|
|
display_ports = self.display.port_by_name
|
|
ports: list[dict[str, object]] = []
|
|
for port in self.ports:
|
|
payload = port.as_interface_dict()
|
|
payload["side"] = display_ports[port.name].side
|
|
payload["order"] = display_ports[port.name].order
|
|
ports.append(payload)
|
|
ports.sort(key=lambda item: (int(item["order"]), str(item["name"])))
|
|
|
|
payload: dict[str, object] = {
|
|
"type": self.model_type,
|
|
"modelType": self.model_type,
|
|
"modelVersion": self.model_version,
|
|
"label": self.display.label,
|
|
"symbol": self.display.symbol,
|
|
"order": self.display.order,
|
|
"category": category.as_catalog_dict(),
|
|
"ports": ports,
|
|
"parameters": [
|
|
parameter.as_interface_dict() for parameter in self.parameters
|
|
],
|
|
}
|
|
if self.display.role is not None:
|
|
payload["role"] = self.display.role
|
|
if self.display.parameter_groups:
|
|
payload["parameterGroups"] = [
|
|
group.as_catalog_dict()
|
|
for group in sorted(
|
|
self.display.parameter_groups,
|
|
key=lambda item: (item.order, item.id),
|
|
)
|
|
]
|
|
return payload
|
|
|
|
def create(
|
|
self,
|
|
name: str,
|
|
medium: GasMedium,
|
|
values: Mapping[str, float],
|
|
) -> Component:
|
|
unknown = sorted(set(values) - set(self.parameter_by_name))
|
|
if unknown:
|
|
raise ValueError(
|
|
f"Component '{name}' contains unsupported parameters: "
|
|
+ ", ".join(unknown)
|
|
+ "."
|
|
)
|
|
|
|
resolved = {
|
|
parameter.name: values.get(parameter.name, parameter.default)
|
|
for parameter in self.parameters
|
|
}
|
|
for parameter in self.parameters:
|
|
message = parameter.validation_message(resolved[parameter.name])
|
|
if message is not None:
|
|
raise ValueError(
|
|
f"Parameter '{parameter.name}' on component '{name}' {message}."
|
|
)
|
|
|
|
component = self.component_class.create(
|
|
name=name,
|
|
medium=medium,
|
|
parameters=resolved,
|
|
)
|
|
self._validate_instance(component, resolved)
|
|
return component
|
|
|
|
def _validate_instance(
|
|
self,
|
|
component: Component,
|
|
resolved_parameters: Mapping[str, float],
|
|
) -> None:
|
|
if not isinstance(component, self.component_class):
|
|
raise ValueError(
|
|
f"Component model '{self.model_type}' create() returned "
|
|
f"{type(component).__name__}, expected {self.component_class.__name__}."
|
|
)
|
|
if component.model_type != self.model_type:
|
|
raise ValueError(
|
|
f"Component implementation '{self.model_type}' created instance "
|
|
f"with model type '{component.model_type}'."
|
|
)
|
|
if component.port_definitions != self.ports:
|
|
raise ValueError(
|
|
f"Component implementation '{self.model_type}' does not match "
|
|
"its declared ports."
|
|
)
|
|
if component.parameter_values != dict(resolved_parameters):
|
|
raise ValueError(
|
|
f"Component implementation '{self.model_type}' did not preserve "
|
|
"its normalized parameters."
|
|
)
|
|
|
|
|
|
def _load_reference(reference: str) -> object:
|
|
try:
|
|
module_name, attribute_name = reference.rsplit(":", maxsplit=1)
|
|
except ValueError as exc:
|
|
raise ValueError(
|
|
f"Component reference '{reference}' must use 'module:attribute'."
|
|
) from exc
|
|
if not module_name or not attribute_name:
|
|
raise ValueError(
|
|
f"Component reference '{reference}' must use 'module:attribute'."
|
|
)
|
|
|
|
try:
|
|
module = import_module(module_name)
|
|
except Exception as exc:
|
|
raise ValueError(
|
|
f"Could not import component module '{module_name}' from '{reference}'."
|
|
) from exc
|
|
try:
|
|
return getattr(module, attribute_name)
|
|
except AttributeError as exc:
|
|
raise ValueError(
|
|
f"Component module '{module_name}' has no attribute '{attribute_name}'."
|
|
) from exc
|
|
|
|
|
|
def _validate_machine_id(value: object, *, field: str) -> str:
|
|
if not isinstance(value, str) or _MACHINE_ID_PATTERN.fullmatch(value) is None:
|
|
raise ValueError(
|
|
f"{field} must start with a lowercase letter and contain only "
|
|
"lowercase letters, digits, and underscores."
|
|
)
|
|
return value
|
|
|
|
|
|
def _validate_member_id(value: object, *, field: str) -> str:
|
|
if not isinstance(value, str) or _MEMBER_ID_PATTERN.fullmatch(value) is None:
|
|
raise ValueError(
|
|
f"{field} must start with a letter and contain only letters, "
|
|
"digits, and underscores."
|
|
)
|
|
return value
|
|
|
|
|
|
def _validate_version(value: object, *, field: str) -> str:
|
|
if (
|
|
not isinstance(value, str)
|
|
or _SEMANTIC_VERSION_PATTERN.fullmatch(value) is None
|
|
):
|
|
raise ValueError(f"{field} must use 'major.minor.patch'.")
|
|
return value
|
|
|
|
|
|
def _validate_label(value: object, *, field: str) -> str:
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise ValueError(f"{field} must not be empty.")
|
|
return value
|
|
|
|
|
|
def _validate_order(value: object, *, field: str) -> int:
|
|
if type(value) is not int:
|
|
raise ValueError(f"{field} must be an integer.")
|
|
return value
|
|
|
|
|
|
def _validate_quantity_unit(
|
|
quantity: object,
|
|
unit: object,
|
|
*,
|
|
field: str,
|
|
) -> None:
|
|
if not isinstance(quantity, str) or not quantity:
|
|
raise ValueError(f"{field} quantity must not be empty.")
|
|
if not isinstance(unit, str):
|
|
raise ValueError(f"{field} unit must be a string.")
|
|
try:
|
|
expected_unit = SI_UNIT_BY_QUANTITY[quantity]
|
|
except KeyError as exc:
|
|
raise ValueError(
|
|
f"{field} uses unsupported quantity '{quantity}'."
|
|
) from exc
|
|
if unit != expected_unit:
|
|
raise ValueError(
|
|
f"{field} uses unit '{unit}' for quantity '{quantity}', "
|
|
f"expected SI unit '{expected_unit}'."
|
|
)
|
|
|
|
|
|
def _validate_unique_names(
|
|
names: Sequence[str],
|
|
*,
|
|
field: str,
|
|
) -> None:
|
|
duplicates = sorted({name for name in names if names.count(name) > 1})
|
|
if duplicates:
|
|
raise ValueError(f"{field} contains duplicate names: {', '.join(duplicates)}.")
|
|
|
|
|
|
def _validate_category(
|
|
category: ComponentCategorySpec,
|
|
*,
|
|
library_id: str,
|
|
) -> None:
|
|
if not isinstance(category, ComponentCategorySpec):
|
|
raise ValueError(
|
|
f"Library '{library_id}' categories must use ComponentCategorySpec."
|
|
)
|
|
_validate_machine_id(
|
|
category.id,
|
|
field=f"Library '{library_id}' category id",
|
|
)
|
|
_validate_label(
|
|
category.label,
|
|
field=f"Library '{library_id}' category '{category.id}' label",
|
|
)
|
|
_validate_order(
|
|
category.order,
|
|
field=f"Library '{library_id}' category '{category.id}' order",
|
|
)
|
|
|
|
|
|
def _validate_parameter_groups(
|
|
parameter_groups: object,
|
|
*,
|
|
model_type: str,
|
|
parameter_names: Sequence[str],
|
|
) -> None:
|
|
field = f"Component '{model_type}' display parameter groups"
|
|
if not isinstance(parameter_groups, tuple):
|
|
raise ValueError(f"{field} must be a tuple.")
|
|
|
|
group_ids: list[str] = []
|
|
grouped_parameter_names: list[str] = []
|
|
known_parameter_names = set(parameter_names)
|
|
for group in parameter_groups:
|
|
if not isinstance(group, ParameterGroupDisplaySpec):
|
|
raise ValueError(
|
|
f"{field} must use ParameterGroupDisplaySpec."
|
|
)
|
|
group_id = _validate_machine_id(
|
|
group.id,
|
|
field=f"{field} id",
|
|
)
|
|
_validate_label(
|
|
group.label,
|
|
field=f"{field} '{group_id}' label",
|
|
)
|
|
_validate_order(
|
|
group.order,
|
|
field=f"{field} '{group_id}' order",
|
|
)
|
|
if type(group.default_expanded) is not bool:
|
|
raise ValueError(
|
|
f"{field} '{group_id}' default_expanded must be a boolean."
|
|
)
|
|
if not isinstance(group.parameters, tuple) or not group.parameters:
|
|
raise ValueError(
|
|
f"{field} '{group_id}' parameters must use a non-empty tuple."
|
|
)
|
|
for parameter_name in group.parameters:
|
|
_validate_member_id(
|
|
parameter_name,
|
|
field=f"{field} '{group_id}' parameter name",
|
|
)
|
|
_validate_unique_names(
|
|
list(group.parameters),
|
|
field=f"{field} '{group_id}' parameters",
|
|
)
|
|
unknown_parameters = sorted(
|
|
set(group.parameters) - known_parameter_names
|
|
)
|
|
if unknown_parameters:
|
|
raise ValueError(
|
|
f"{field} '{group_id}' references unknown parameters: "
|
|
+ ", ".join(unknown_parameters)
|
|
+ "."
|
|
)
|
|
group_ids.append(group_id)
|
|
grouped_parameter_names.extend(group.parameters)
|
|
|
|
_validate_unique_names(group_ids, field=field)
|
|
duplicate_memberships = sorted(
|
|
{
|
|
name
|
|
for name in grouped_parameter_names
|
|
if grouped_parameter_names.count(name) > 1
|
|
}
|
|
)
|
|
if duplicate_memberships:
|
|
raise ValueError(
|
|
f"{field} assign parameters to multiple groups: "
|
|
+ ", ".join(duplicate_memberships)
|
|
+ "."
|
|
)
|
|
|
|
|
|
def validate_component_library(library: ComponentLibrarySpec) -> None:
|
|
if not isinstance(library, ComponentLibrarySpec):
|
|
raise ValueError("Enabled component libraries must use ComponentLibrarySpec.")
|
|
library_id = _validate_machine_id(library.id, field="Component library id")
|
|
_validate_label(library.label, field=f"Library '{library_id}' label")
|
|
_validate_version(library.version, field=f"Library '{library_id}' version")
|
|
_validate_order(library.order, field=f"Library '{library_id}' order")
|
|
if type(library.temporary) is not bool:
|
|
raise ValueError(f"Library '{library_id}' temporary must be a boolean.")
|
|
if (
|
|
not isinstance(library.source_package, str)
|
|
or not library.source_package.strip()
|
|
):
|
|
raise ValueError(f"Library '{library_id}' source package must not be empty.")
|
|
if not library.categories:
|
|
raise ValueError(f"Library '{library_id}' must declare at least one category.")
|
|
if not library.models:
|
|
raise ValueError(f"Library '{library_id}' must declare at least one model.")
|
|
|
|
for category in library.categories:
|
|
_validate_category(category, library_id=library_id)
|
|
category_ids = [category.id for category in library.categories]
|
|
_validate_unique_names(
|
|
category_ids,
|
|
field=f"Library '{library_id}' categories",
|
|
)
|
|
|
|
_validate_unique_names(
|
|
list(library.models),
|
|
field=f"Library '{library_id}' model references",
|
|
)
|
|
for reference in library.models:
|
|
if not isinstance(reference, str) or ":" not in reference:
|
|
raise ValueError(
|
|
f"Library '{library_id}' contains invalid model reference "
|
|
f"'{reference}'."
|
|
)
|
|
module_name = reference.rsplit(":", maxsplit=1)[0]
|
|
if not (
|
|
module_name == library.source_package
|
|
or module_name.startswith(f"{library.source_package}.")
|
|
):
|
|
raise ValueError(
|
|
f"Library '{library_id}' model '{reference}' is outside source "
|
|
f"package '{library.source_package}'."
|
|
)
|
|
|
|
|
|
def _validate_port_variable(
|
|
variable: PortVariableDefinition,
|
|
*,
|
|
model_type: str,
|
|
port_name: str,
|
|
) -> None:
|
|
field = f"Component '{model_type}' port '{port_name}' variable"
|
|
if not isinstance(variable, PortVariableDefinition):
|
|
raise ValueError(f"{field} must use PortVariableDefinition.")
|
|
_validate_member_id(variable.name, field=f"{field} name")
|
|
_validate_label(variable.label or variable.name, field=f"{field} label")
|
|
_validate_quantity_unit(
|
|
variable.quantity or variable.name,
|
|
variable.unit,
|
|
field=f"{field} '{variable.name}'",
|
|
)
|
|
_validate_order(variable.order, field=f"{field} '{variable.name}' order")
|
|
if variable.role not in {"effort", "flow", "stream", "signal"}:
|
|
raise ValueError(f"{field} '{variable.name}' has invalid role.")
|
|
if variable.connection_rule not in {
|
|
"equal",
|
|
"sumToZero",
|
|
"streamMix",
|
|
"directed",
|
|
}:
|
|
raise ValueError(
|
|
f"{field} '{variable.name}' has invalid connection rule."
|
|
)
|
|
if type(variable.result_visible) is not bool:
|
|
raise ValueError(
|
|
f"{field} '{variable.name}' result_visible must be a boolean."
|
|
)
|
|
|
|
|
|
def _validate_port(port: PortDefinition, *, model_type: str) -> None:
|
|
field = f"Component '{model_type}' port"
|
|
if not isinstance(port, PortDefinition):
|
|
raise ValueError(f"{field} declarations must use PortDefinition.")
|
|
_validate_machine_id(port.name, field=f"{field} name")
|
|
_validate_machine_id(port.domain, field=f"{field} '{port.name}' domain")
|
|
if port.kind not in {"physical", "signal"}:
|
|
raise ValueError(f"{field} '{port.name}' has invalid kind.")
|
|
if port.nominal_role not in {
|
|
"inlet",
|
|
"outlet",
|
|
"bidirectional",
|
|
"input",
|
|
"output",
|
|
}:
|
|
raise ValueError(f"{field} '{port.name}' has invalid nominal role.")
|
|
if port.kind == "physical" and port.positive_flow_direction != "intoComponent":
|
|
raise ValueError(
|
|
f"{field} '{port.name}' must use positive flow into the component."
|
|
)
|
|
if port.kind == "physical" and port.nominal_role in {"input", "output"}:
|
|
raise ValueError(
|
|
f"{field} '{port.name}' uses a signal-only nominal role."
|
|
)
|
|
if port.kind == "signal" and port.nominal_role not in {"input", "output"}:
|
|
raise ValueError(
|
|
f"{field} '{port.name}' must use input or output as its nominal role."
|
|
)
|
|
|
|
variable_names = [variable.name for variable in port.variables]
|
|
_validate_unique_names(
|
|
variable_names,
|
|
field=f"{field} '{port.name}' variables",
|
|
)
|
|
for variable in port.variables:
|
|
_validate_port_variable(
|
|
variable,
|
|
model_type=model_type,
|
|
port_name=port.name,
|
|
)
|
|
|
|
|
|
def _validate_parameter(
|
|
parameter: ParameterDefinition,
|
|
*,
|
|
model_type: str,
|
|
) -> None:
|
|
field = f"Component '{model_type}' parameter"
|
|
if not isinstance(parameter, ParameterDefinition):
|
|
raise ValueError(f"{field} declarations must use ParameterDefinition.")
|
|
_validate_member_id(parameter.name, field=f"{field} name")
|
|
_validate_label(parameter.label, field=f"{field} '{parameter.name}' label")
|
|
if not isinstance(parameter.description, str):
|
|
raise ValueError(
|
|
f"{field} '{parameter.name}' description must be a string."
|
|
)
|
|
if parameter.description and not parameter.description.strip():
|
|
raise ValueError(
|
|
f"{field} '{parameter.name}' description must not be blank."
|
|
)
|
|
_validate_quantity_unit(
|
|
parameter.quantity,
|
|
parameter.unit,
|
|
field=f"{field} '{parameter.name}'",
|
|
)
|
|
if not isinstance(parameter.default, (int, float)) or not isfinite(
|
|
parameter.default
|
|
):
|
|
raise ValueError(f"{field} '{parameter.name}' default must be finite.")
|
|
for boundary_name, boundary in (
|
|
("minimum", parameter.minimum),
|
|
("maximum", parameter.maximum),
|
|
):
|
|
if boundary is not None and (
|
|
not isinstance(boundary, (int, float)) or not isfinite(boundary)
|
|
):
|
|
raise ValueError(
|
|
f"{field} '{parameter.name}' {boundary_name} must be finite."
|
|
)
|
|
if type(parameter.minimum_exclusive) is not bool:
|
|
raise ValueError(
|
|
f"{field} '{parameter.name}' minimum_exclusive must be a boolean."
|
|
)
|
|
if parameter.editor not in {
|
|
None,
|
|
"amesimGasReference",
|
|
"amesimGasPropertyModel",
|
|
"choice",
|
|
}:
|
|
raise ValueError(
|
|
f"{field} '{parameter.name}' uses unsupported editor "
|
|
f"'{parameter.editor}'."
|
|
)
|
|
if not isinstance(parameter.options, tuple):
|
|
raise ValueError(
|
|
f"{field} '{parameter.name}' options must use a tuple."
|
|
)
|
|
if parameter.editor == "amesimGasPropertyModel" and not parameter.options:
|
|
raise ValueError(
|
|
f"{field} '{parameter.name}' property-model editor must declare options."
|
|
)
|
|
if parameter.editor == "choice" and not parameter.options:
|
|
raise ValueError(
|
|
f"{field} '{parameter.name}' choice editor must declare options."
|
|
)
|
|
if parameter.options and parameter.editor not in {
|
|
"amesimGasPropertyModel",
|
|
"choice",
|
|
}:
|
|
raise ValueError(
|
|
f"{field} '{parameter.name}' options require the "
|
|
"'choice' or 'amesimGasPropertyModel' editor."
|
|
)
|
|
option_values: list[float] = []
|
|
for option in parameter.options:
|
|
if not isinstance(option, ParameterOption):
|
|
raise ValueError(
|
|
f"{field} '{parameter.name}' options must use ParameterOption."
|
|
)
|
|
if (
|
|
isinstance(option.value, bool)
|
|
or not isinstance(option.value, (int, float))
|
|
or not isfinite(option.value)
|
|
):
|
|
raise ValueError(
|
|
f"{field} '{parameter.name}' option values must be finite numbers."
|
|
)
|
|
_validate_label(
|
|
option.label,
|
|
field=(
|
|
f"{field} '{parameter.name}' option "
|
|
f"'{option.value:g}' label"
|
|
),
|
|
)
|
|
numeric_option = float(option.value)
|
|
if (
|
|
parameter.editor == "amesimGasPropertyModel"
|
|
and not numeric_option.is_integer()
|
|
):
|
|
raise ValueError(
|
|
f"{field} '{parameter.name}' property-model option values "
|
|
"must be integers."
|
|
)
|
|
option_message = parameter.validation_message(numeric_option)
|
|
if option_message is not None:
|
|
raise ValueError(
|
|
f"{field} '{parameter.name}' option {numeric_option:g} "
|
|
f"{option_message}."
|
|
)
|
|
option_values.append(numeric_option)
|
|
duplicate_option_values = sorted(
|
|
{value for value in option_values if option_values.count(value) > 1}
|
|
)
|
|
if duplicate_option_values:
|
|
duplicates = ", ".join(f"{value:g}" for value in duplicate_option_values)
|
|
raise ValueError(
|
|
f"{field} '{parameter.name}' contains duplicate option values: "
|
|
f"{duplicates}."
|
|
)
|
|
if not isinstance(parameter.visible_when, tuple):
|
|
raise ValueError(
|
|
f"{field} '{parameter.name}' visible_when must use a tuple."
|
|
)
|
|
for condition in parameter.visible_when:
|
|
if not isinstance(condition, ParameterCondition):
|
|
raise ValueError(
|
|
f"{field} '{parameter.name}' visible_when entries must use "
|
|
"ParameterCondition."
|
|
)
|
|
_validate_member_id(
|
|
condition.parameter,
|
|
field=(
|
|
f"{field} '{parameter.name}' visibility condition parameter"
|
|
),
|
|
)
|
|
if not isinstance(condition.values, tuple) or not condition.values:
|
|
raise ValueError(
|
|
f"{field} '{parameter.name}' visibility condition values "
|
|
"must use a non-empty tuple."
|
|
)
|
|
numeric_condition_values: list[float] = []
|
|
for value in condition.values:
|
|
if (
|
|
isinstance(value, bool)
|
|
or not isinstance(value, (int, float))
|
|
or not isfinite(value)
|
|
):
|
|
raise ValueError(
|
|
f"{field} '{parameter.name}' visibility condition values "
|
|
"must be finite numbers."
|
|
)
|
|
numeric_condition_values.append(float(value))
|
|
duplicate_condition_values = sorted(
|
|
{
|
|
value
|
|
for value in numeric_condition_values
|
|
if numeric_condition_values.count(value) > 1
|
|
}
|
|
)
|
|
if duplicate_condition_values:
|
|
duplicates = ", ".join(
|
|
f"{value:g}" for value in duplicate_condition_values
|
|
)
|
|
raise ValueError(
|
|
f"{field} '{parameter.name}' visibility condition for "
|
|
f"'{condition.parameter}' contains duplicate values: "
|
|
f"{duplicates}."
|
|
)
|
|
if (
|
|
parameter.minimum is not None
|
|
and parameter.maximum is not None
|
|
and parameter.minimum > parameter.maximum
|
|
):
|
|
raise ValueError(
|
|
f"{field} '{parameter.name}' minimum exceeds its maximum."
|
|
)
|
|
message = parameter.validation_message(parameter.default)
|
|
if message is not None:
|
|
raise ValueError(
|
|
f"{field} '{parameter.name}' default value {message}."
|
|
)
|
|
|
|
|
|
def _validate_parameter_visibility(
|
|
parameters: tuple[ParameterDefinition, ...],
|
|
*,
|
|
model_type: str,
|
|
) -> None:
|
|
"""Validate cross-parameter references used by catalog visibility rules."""
|
|
|
|
parameter_by_name = {parameter.name: parameter for parameter in parameters}
|
|
dependencies: dict[str, tuple[str, ...]] = {}
|
|
|
|
for parameter in parameters:
|
|
controllers: list[str] = []
|
|
for condition in parameter.visible_when:
|
|
controller_name = condition.parameter
|
|
controller = parameter_by_name.get(controller_name)
|
|
if controller is None:
|
|
raise ValueError(
|
|
f"Component '{model_type}' parameter '{parameter.name}' "
|
|
"visibility condition references unknown parameter "
|
|
f"'{controller_name}'."
|
|
)
|
|
if controller_name == parameter.name:
|
|
raise ValueError(
|
|
f"Component '{model_type}' parameter '{parameter.name}' "
|
|
"visibility condition cannot reference itself."
|
|
)
|
|
if controller_name in controllers:
|
|
raise ValueError(
|
|
f"Component '{model_type}' parameter '{parameter.name}' "
|
|
"contains duplicate visibility controller "
|
|
f"'{controller_name}'."
|
|
)
|
|
if not controller.options:
|
|
raise ValueError(
|
|
f"Component '{model_type}' parameter '{parameter.name}' "
|
|
f"visibility controller '{controller_name}' must declare "
|
|
"options."
|
|
)
|
|
controller_values = {
|
|
float(option.value) for option in controller.options
|
|
}
|
|
unsupported_values = sorted(
|
|
{
|
|
float(value)
|
|
for value in condition.values
|
|
if float(value) not in controller_values
|
|
}
|
|
)
|
|
if unsupported_values:
|
|
values = ", ".join(f"{value:g}" for value in unsupported_values)
|
|
raise ValueError(
|
|
f"Component '{model_type}' parameter '{parameter.name}' "
|
|
f"visibility condition values for '{controller_name}' "
|
|
f"must belong to its options; unsupported: {values}."
|
|
)
|
|
controllers.append(controller_name)
|
|
dependencies[parameter.name] = tuple(controllers)
|
|
|
|
states: dict[str, int] = {}
|
|
stack: list[str] = []
|
|
|
|
def visit(parameter_name: str) -> None:
|
|
state = states.get(parameter_name, 0)
|
|
if state == 2:
|
|
return
|
|
if state == 1:
|
|
cycle_start = stack.index(parameter_name)
|
|
cycle = stack[cycle_start:] + [parameter_name]
|
|
raise ValueError(
|
|
f"Component '{model_type}' parameter visibility dependencies "
|
|
f"contain a cycle: {' -> '.join(cycle)}."
|
|
)
|
|
states[parameter_name] = 1
|
|
stack.append(parameter_name)
|
|
for controller_name in dependencies[parameter_name]:
|
|
visit(controller_name)
|
|
stack.pop()
|
|
states[parameter_name] = 2
|
|
|
|
for parameter in parameters:
|
|
visit(parameter.name)
|
|
|
|
|
|
def _validate_result_variable(
|
|
variable: ResultVariableDefinition,
|
|
*,
|
|
model_type: str,
|
|
) -> None:
|
|
field = f"Component '{model_type}' result variable"
|
|
if not isinstance(variable, ResultVariableDefinition):
|
|
raise ValueError(
|
|
f"{field} declarations must use ResultVariableDefinition."
|
|
)
|
|
_validate_member_id(variable.name, field=f"{field} name")
|
|
_validate_label(variable.label, field=f"{field} '{variable.name}' label")
|
|
_validate_quantity_unit(
|
|
variable.quantity,
|
|
variable.unit,
|
|
field=f"{field} '{variable.name}'",
|
|
)
|
|
_validate_machine_id(
|
|
variable.category,
|
|
field=f"{field} '{variable.name}' category",
|
|
)
|
|
_validate_order(variable.order, field=f"{field} '{variable.name}' order")
|
|
if type(variable.visible) is not bool:
|
|
raise ValueError(
|
|
f"{field} '{variable.name}' visible must be a boolean."
|
|
)
|
|
|
|
|
|
def validate_component_model_class(
|
|
component_class: type[Component],
|
|
*,
|
|
library: ComponentLibrarySpec,
|
|
) -> None:
|
|
if not isinstance(component_class, type) or not issubclass(
|
|
component_class,
|
|
Component,
|
|
):
|
|
raise ValueError(
|
|
f"Library '{library.id}' model entries must be Component subclasses."
|
|
)
|
|
|
|
required_declarations = (
|
|
"MODEL_TYPE",
|
|
"MODEL_VERSION",
|
|
"PORTS",
|
|
"PARAMETERS",
|
|
"RESULT_VARIABLES",
|
|
"DISPLAY",
|
|
"create",
|
|
)
|
|
missing = [
|
|
name for name in required_declarations if name not in component_class.__dict__
|
|
]
|
|
if missing:
|
|
raise ValueError(
|
|
f"Component class '{component_class.__name__}' must declare: "
|
|
+ ", ".join(missing)
|
|
+ "."
|
|
)
|
|
if not isinstance(component_class.__dict__["create"], classmethod):
|
|
raise ValueError(
|
|
f"Component class '{component_class.__name__}' create must be a classmethod."
|
|
)
|
|
|
|
model_type = _validate_machine_id(
|
|
component_class.MODEL_TYPE,
|
|
field=f"Component class '{component_class.__name__}' model type",
|
|
)
|
|
_validate_version(
|
|
component_class.MODEL_VERSION,
|
|
field=f"Component '{model_type}' model version",
|
|
)
|
|
|
|
display = component_class.DISPLAY
|
|
if not isinstance(display, ComponentDisplaySpec):
|
|
raise ValueError(
|
|
f"Component '{model_type}' DISPLAY must use ComponentDisplaySpec."
|
|
)
|
|
_validate_label(display.label, field=f"Component '{model_type}' display label")
|
|
_validate_machine_id(
|
|
display.symbol,
|
|
field=f"Component '{model_type}' display symbol",
|
|
)
|
|
_validate_order(display.order, field=f"Component '{model_type}' display order")
|
|
if display.role not in {None, "amesimGasMediumDefinition"}:
|
|
raise ValueError(
|
|
f"Component '{model_type}' DISPLAY uses unsupported role "
|
|
f"'{display.role}'."
|
|
)
|
|
if not isinstance(display.ports, tuple):
|
|
raise ValueError(f"Component '{model_type}' DISPLAY ports must be a tuple.")
|
|
if display.library_id != library.id:
|
|
raise ValueError(
|
|
f"Component '{model_type}' references library '{display.library_id}', "
|
|
f"expected '{library.id}'."
|
|
)
|
|
if display.category_id not in library.category_by_id:
|
|
raise ValueError(
|
|
f"Component '{model_type}' references unknown category "
|
|
f"'{display.category_id}' in library '{library.id}'."
|
|
)
|
|
|
|
ports = component_class.PORTS
|
|
parameters = component_class.PARAMETERS
|
|
result_variables = component_class.RESULT_VARIABLES
|
|
if not isinstance(ports, tuple):
|
|
raise ValueError(f"Component '{model_type}' PORTS must be a tuple.")
|
|
if not isinstance(parameters, tuple):
|
|
raise ValueError(f"Component '{model_type}' PARAMETERS must be a tuple.")
|
|
if not isinstance(result_variables, tuple):
|
|
raise ValueError(
|
|
f"Component '{model_type}' RESULT_VARIABLES must be a tuple."
|
|
)
|
|
|
|
port_by_name = {port.name: port for port in ports if isinstance(port, PortDefinition)}
|
|
for port in ports:
|
|
_validate_port(port, model_type=model_type)
|
|
contract = port.computation
|
|
if contract is not None:
|
|
if port.domain != 'pneumatic' or not isinstance(contract, PortComputation):
|
|
raise ValueError(f"Component '{model_type}' has an invalid pneumatic computation contract.")
|
|
if contract.reference_port:
|
|
reference = port_by_name.get(contract.reference_port)
|
|
if (reference is None or reference.computation is None
|
|
or not {'p', 'T'}.issubset(reference.computation.inputs)):
|
|
raise ValueError(f"Component '{model_type}.{port.name}' must alias a pressure/temperature input port.")
|
|
for parameter in parameters:
|
|
_validate_parameter(parameter, model_type=model_type)
|
|
for variable in result_variables:
|
|
_validate_result_variable(variable, model_type=model_type)
|
|
for port in display.ports:
|
|
if not isinstance(port, PortDisplaySpec):
|
|
raise ValueError(
|
|
f"Component '{model_type}' display ports must use PortDisplaySpec."
|
|
)
|
|
_validate_machine_id(
|
|
port.name,
|
|
field=f"Component '{model_type}' display port name",
|
|
)
|
|
if port.side not in {"left", "right"}:
|
|
raise ValueError(
|
|
f"Component '{model_type}' display port '{port.name}' "
|
|
"must use side 'left' or 'right'."
|
|
)
|
|
_validate_order(
|
|
port.order,
|
|
field=f"Component '{model_type}' display port '{port.name}' order",
|
|
)
|
|
|
|
port_names = [port.name for port in ports]
|
|
parameter_names = [parameter.name for parameter in parameters]
|
|
result_names = [variable.name for variable in result_variables]
|
|
display_port_names = [port.name for port in display.ports]
|
|
_validate_unique_names(port_names, field=f"Component '{model_type}' ports")
|
|
_validate_unique_names(
|
|
parameter_names,
|
|
field=f"Component '{model_type}' parameters",
|
|
)
|
|
_validate_parameter_groups(
|
|
display.parameter_groups,
|
|
model_type=model_type,
|
|
parameter_names=parameter_names,
|
|
)
|
|
_validate_parameter_visibility(parameters, model_type=model_type)
|
|
_validate_unique_names(
|
|
result_names,
|
|
field=f"Component '{model_type}' result variables",
|
|
)
|
|
_validate_unique_names(
|
|
display_port_names,
|
|
field=f"Component '{model_type}' display ports",
|
|
)
|
|
if set(port_names) != set(display_port_names):
|
|
raise ValueError(
|
|
f"Component '{model_type}' display ports must exactly match "
|
|
"its physical port declarations."
|
|
)
|
|
|
|
|
|
def discover_component_registries(
|
|
library_references: Sequence[str] = ENABLED_COMPONENT_LIBRARIES,
|
|
) -> tuple[
|
|
dict[str, ComponentLibrarySpec],
|
|
dict[str, ComponentModelSpec],
|
|
]:
|
|
libraries: dict[str, ComponentLibrarySpec] = {}
|
|
models: dict[str, ComponentModelSpec] = {}
|
|
|
|
for library_reference in library_references:
|
|
library = _load_reference(library_reference)
|
|
if not isinstance(library, ComponentLibrarySpec):
|
|
raise ValueError(
|
|
f"Enabled library '{library_reference}' must reference "
|
|
"ComponentLibrarySpec."
|
|
)
|
|
validate_component_library(library)
|
|
if library.id in libraries:
|
|
raise ValueError(f"Duplicate component library id: {library.id}.")
|
|
libraries[library.id] = library
|
|
|
|
for model_reference in library.models:
|
|
component_class = _load_reference(model_reference)
|
|
if not isinstance(component_class, type) or not issubclass(
|
|
component_class,
|
|
Component,
|
|
):
|
|
raise ValueError(
|
|
f"Library '{library.id}' model '{model_reference}' is not "
|
|
"a Component subclass."
|
|
)
|
|
validate_component_model_class(component_class, library=library)
|
|
model_type = cast(str, component_class.MODEL_TYPE)
|
|
if model_type in models:
|
|
existing = models[model_type].component_class
|
|
raise ValueError(
|
|
f"Duplicate component model type '{model_type}' from "
|
|
f"{existing.__module__}.{existing.__name__} and "
|
|
f"{component_class.__module__}.{component_class.__name__}."
|
|
)
|
|
spec = ComponentModelSpec(
|
|
component_class=component_class,
|
|
library=library,
|
|
)
|
|
try:
|
|
spec.create(
|
|
f"__catalog_validation_{model_type}",
|
|
IdealGasMedium(),
|
|
{},
|
|
)
|
|
except Exception as exc:
|
|
raise ValueError(
|
|
f"Component model '{model_type}' cannot be created with "
|
|
"its declared defaults."
|
|
) from exc
|
|
models[model_type] = spec
|
|
|
|
if not libraries:
|
|
raise ValueError("At least one component library must be enabled.")
|
|
return libraries, models
|
|
|
|
|
|
COMPONENT_LIBRARY_REGISTRY, COMPONENT_MODEL_REGISTRY = (
|
|
discover_component_registries()
|
|
)
|
|
|
|
|
|
def validate_component_registries() -> None:
|
|
"""Re-run public registry invariants for tests and startup diagnostics."""
|
|
|
|
discovered_libraries, discovered_models = discover_component_registries()
|
|
if tuple(discovered_libraries) != tuple(COMPONENT_LIBRARY_REGISTRY):
|
|
raise ValueError("Component library registry differs from discovery output.")
|
|
if tuple(discovered_models) != tuple(COMPONENT_MODEL_REGISTRY):
|
|
raise ValueError("Component model registry differs from discovery output.")
|
|
|
|
|
|
def build_component_catalog() -> dict[str, object]:
|
|
components_by_library: dict[str, list[ComponentModelSpec]] = {
|
|
library_id: [] for library_id in COMPONENT_LIBRARY_REGISTRY
|
|
}
|
|
for component in COMPONENT_MODEL_REGISTRY.values():
|
|
components_by_library[component.library.id].append(component)
|
|
|
|
libraries: list[dict[str, object]] = []
|
|
for library in sorted(
|
|
COMPONENT_LIBRARY_REGISTRY.values(),
|
|
key=lambda item: (item.order, item.id),
|
|
):
|
|
payload = library.as_catalog_dict()
|
|
payload["components"] = [
|
|
component.as_catalog_dict()
|
|
for component in sorted(
|
|
components_by_library[library.id],
|
|
key=lambda item: (item.display.order, item.model_type),
|
|
)
|
|
]
|
|
libraries.append(payload)
|
|
|
|
return {
|
|
"schemaVersion": 1,
|
|
"libraries": libraries,
|
|
}
|
|
|
|
|
|
def get_component_model_spec(model_type: str) -> ComponentModelSpec:
|
|
try:
|
|
return COMPONENT_MODEL_REGISTRY[model_type]
|
|
except KeyError as exc:
|
|
raise ValueError(f"Unsupported model type: {model_type}.") from exc
|