103 lines
3.1 KiB
Python
103 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from collections import Counter
|
|
from dataclasses import dataclass
|
|
|
|
from app.simulation.reporting.amesim_results import AmesimResults
|
|
from app.simulation.reporting.test_mql_variables import (
|
|
TestMqlVariableCatalog,
|
|
build_test_mql_variable_catalog,
|
|
)
|
|
from app.simulation.examples.test_mql.system import CONNECTION_SPECS
|
|
|
|
|
|
TEST_MQL_PNEUMATIC_LINE_SUBMODELS = ("PNL0001", "PNL0002", "PNL0003", "PNL00R")
|
|
_LINE_PATTERN_RE = re.compile(r"\(([^()]+)\)\s*$")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TestMqlLineConnection:
|
|
index: int
|
|
alias: str
|
|
submodel: str
|
|
pattern: str
|
|
source_component: str
|
|
source_port: str
|
|
target_component: str
|
|
target_port: str
|
|
label: str
|
|
data_paths: tuple[str, ...]
|
|
signal_names: tuple[str, ...]
|
|
|
|
@property
|
|
def has_compliance(self) -> bool:
|
|
return "C" in self.pattern
|
|
|
|
@property
|
|
def has_resistance(self) -> bool:
|
|
return "R" in self.pattern
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TestMqlLineAssembly:
|
|
lines: tuple[TestMqlLineConnection, ...]
|
|
|
|
@property
|
|
def line_count(self) -> int:
|
|
return len(self.lines)
|
|
|
|
def by_alias(self, alias: str) -> TestMqlLineConnection:
|
|
for line in self.lines:
|
|
if line.alias == alias:
|
|
return line
|
|
raise KeyError(alias)
|
|
|
|
def by_submodel(self, submodel: str) -> tuple[TestMqlLineConnection, ...]:
|
|
return tuple(line for line in self.lines if line.submodel == submodel)
|
|
|
|
def counts_by_submodel(self) -> dict[str, int]:
|
|
return dict(Counter(line.submodel for line in self.lines))
|
|
|
|
def aliases(self) -> tuple[str, ...]:
|
|
return tuple(line.alias for line in self.lines)
|
|
|
|
|
|
def build_test_mql_line_assembly(
|
|
amesim_results: AmesimResults,
|
|
variable_catalog: TestMqlVariableCatalog | None = None,
|
|
) -> TestMqlLineAssembly:
|
|
variable_catalog = variable_catalog or build_test_mql_variable_catalog(amesim_results)
|
|
lines = []
|
|
for spec in CONNECTION_SPECS:
|
|
submodel = str(spec["submodel"])
|
|
if submodel not in TEST_MQL_PNEUMATIC_LINE_SUBMODELS:
|
|
continue
|
|
data_paths = variable_catalog.data_paths_for_owner(str(spec["alias"]))
|
|
signal_names = tuple(path.rsplit("@", 1)[0] for path in data_paths)
|
|
lines.append(
|
|
TestMqlLineConnection(
|
|
index=int(spec["index"]),
|
|
alias=str(spec["alias"]),
|
|
submodel=submodel,
|
|
pattern=_line_pattern(str(spec["label"]), submodel),
|
|
source_component=str(spec["source_component"]),
|
|
source_port=str(spec["source_port"]),
|
|
target_component=str(spec["target_component"]),
|
|
target_port=str(spec["target_port"]),
|
|
label=str(spec["label"]),
|
|
data_paths=data_paths,
|
|
signal_names=signal_names,
|
|
)
|
|
)
|
|
return TestMqlLineAssembly(lines=tuple(lines))
|
|
|
|
|
|
def _line_pattern(label: str, submodel: str) -> str:
|
|
match = _LINE_PATTERN_RE.search(label)
|
|
if match is not None:
|
|
return match.group(1)
|
|
if submodel == "PNL00R":
|
|
return "R"
|
|
return submodel
|