356 lines
12 KiB
Python
356 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
from math import floor
|
|
|
|
from app.simulation.core.base import AlgebraicComponent
|
|
from app.simulation.core.catalog import (
|
|
ComponentDisplaySpec,
|
|
ParameterGroupDisplaySpec,
|
|
PortDisplaySpec,
|
|
)
|
|
from app.simulation.core.metadata import (
|
|
ParameterCondition,
|
|
ParameterDefinition,
|
|
ParameterOption,
|
|
ResultVariableDefinition,
|
|
)
|
|
from app.simulation.core.medium import IdealGasMedium
|
|
from app.simulation.core.ports import PortDefinition
|
|
|
|
|
|
def _ud00_stage_parameters(index: int) -> tuple[ParameterDefinition, ...]:
|
|
visible_when = (
|
|
()
|
|
if index == 1
|
|
else (
|
|
ParameterCondition(
|
|
"nstages",
|
|
tuple(float(stage_count) for stage_count in range(index, 9)),
|
|
),
|
|
)
|
|
)
|
|
return (
|
|
ParameterDefinition(
|
|
f"start{index}",
|
|
0.0 if index == 1 else 1.0,
|
|
label=f"第 {index} 段起点",
|
|
quantity="dimensionless",
|
|
unit="",
|
|
description=f"第 {index} 段开始时的输出值。",
|
|
visible_when=visible_when,
|
|
),
|
|
ParameterDefinition(
|
|
f"end{index}",
|
|
1.0,
|
|
label=f"第 {index} 段终点",
|
|
quantity="dimensionless",
|
|
unit="",
|
|
description=f"第 {index} 段结束时的输出值。",
|
|
visible_when=visible_when,
|
|
),
|
|
ParameterDefinition(
|
|
f"t{index}",
|
|
1.0 if index == 1 else 0.0,
|
|
label=f"第 {index} 段时长",
|
|
quantity="time",
|
|
unit="s",
|
|
minimum=0.0,
|
|
description=f"第 {index} 段的持续时间。",
|
|
visible_when=visible_when,
|
|
),
|
|
)
|
|
|
|
|
|
_UD00_STAGE_PARAMETERS = tuple(
|
|
parameter
|
|
for stage_index in range(1, 9)
|
|
for parameter in _ud00_stage_parameters(stage_index)
|
|
)
|
|
|
|
|
|
class AmesimStep0(AlgebraicComponent):
|
|
"""AMESim STEP0 scalar step signal source."""
|
|
|
|
MODEL_TYPE = "amesim_step0"
|
|
MODEL_VERSION = "0.1.0"
|
|
PORTS = (PortDefinition.signal("out", nominal_role="output"),)
|
|
PARAMETERS = (
|
|
ParameterDefinition("initial", 0.0, label="初始值", quantity="dimensionless", unit=""),
|
|
ParameterDefinition("final", 1.0, label="阶跃后值", quantity="dimensionless", unit=""),
|
|
ParameterDefinition("time", 0.0, label="阶跃时间", quantity="time", unit="s"),
|
|
)
|
|
RESULT_VARIABLES = (
|
|
ResultVariableDefinition("y", "输出", "dimensionless", "", "signal", 10),
|
|
)
|
|
DISPLAY = ComponentDisplaySpec(
|
|
label="STEP0 阶跃信号",
|
|
library_id="amesim",
|
|
category_id="signals",
|
|
symbol="amesim_step0",
|
|
ports=(PortDisplaySpec("out", "right", order=10),),
|
|
order=10,
|
|
)
|
|
|
|
def __init__(
|
|
self,
|
|
name: str,
|
|
medium: IdealGasMedium,
|
|
*,
|
|
initial: float = 0.0,
|
|
final: float = 1.0,
|
|
time: float = 0.0,
|
|
) -> None:
|
|
super().__init__(name=name)
|
|
self.set_parameter_values({"initial": initial, "final": final, "time": time})
|
|
self.initial = float(initial)
|
|
self.final = float(final)
|
|
self.time = float(time)
|
|
self.out = self.register_declared_port("out")
|
|
self.out.signal = self.output_at(0.0)
|
|
|
|
@classmethod
|
|
def create(
|
|
cls,
|
|
*,
|
|
name: str,
|
|
medium: IdealGasMedium,
|
|
parameters: Mapping[str, float],
|
|
) -> "AmesimStep0":
|
|
return cls(
|
|
name=name,
|
|
medium=medium,
|
|
initial=parameters["initial"],
|
|
final=parameters["final"],
|
|
time=parameters["time"],
|
|
)
|
|
|
|
def output_at(self, time: float) -> float:
|
|
return self.final if time >= self.time else self.initial
|
|
|
|
def signal_output_values(self, time: float) -> dict[str, float]:
|
|
return {"out": self.output_at(time)}
|
|
|
|
def signal_event_times(
|
|
self,
|
|
start_time: float,
|
|
stop_time: float,
|
|
) -> tuple[float, ...]:
|
|
"""Expose the exact STEP0 switch time as an integration split point."""
|
|
|
|
return (self.time,) if start_time < self.time < stop_time else ()
|
|
|
|
def component_result_values(self) -> Mapping[str, float]:
|
|
return {"y": self.out.signal}
|
|
|
|
|
|
class AmesimUd00(AlgebraicComponent):
|
|
"""AMESim UD00 piecewise-linear scalar signal source."""
|
|
|
|
MODEL_TYPE = "amesim_ud00"
|
|
MODEL_VERSION = "0.2.0"
|
|
PORTS = (PortDefinition.signal("out", nominal_role="output"),)
|
|
PARAMETERS = (
|
|
ParameterDefinition(
|
|
"tstart",
|
|
0.0,
|
|
label="启动时间",
|
|
quantity="time",
|
|
unit="s",
|
|
description="分段信号开始输出第一段之前的等待时间。",
|
|
),
|
|
*_UD00_STAGE_PARAMETERS,
|
|
ParameterDefinition(
|
|
"nstages",
|
|
1.0,
|
|
label="段数",
|
|
quantity="dimensionless",
|
|
unit="",
|
|
minimum=1.0,
|
|
maximum=8.0,
|
|
editor="choice",
|
|
options=tuple(
|
|
ParameterOption(float(stage_count), str(stage_count))
|
|
for stage_count in range(1, 9)
|
|
),
|
|
description="参与输出计算的有效线性分段数量。",
|
|
),
|
|
ParameterDefinition(
|
|
"iscyclic",
|
|
0.0,
|
|
label="循环",
|
|
quantity="dimensionless",
|
|
unit="",
|
|
minimum=0.0,
|
|
maximum=1.0,
|
|
editor="choice",
|
|
options=(
|
|
ParameterOption(0.0, "否"),
|
|
ParameterOption(1.0, "是"),
|
|
),
|
|
description="当前公共协议编码：0 表示单次输出，1 表示循环输出。",
|
|
),
|
|
)
|
|
RESULT_VARIABLES = (
|
|
ResultVariableDefinition("y", "输出", "dimensionless", "", "signal", 10),
|
|
)
|
|
DISPLAY = ComponentDisplaySpec(
|
|
label="UD00 分段线性信号",
|
|
library_id="amesim",
|
|
category_id="signals",
|
|
symbol="amesim_ud00",
|
|
ports=(PortDisplaySpec("out", "right", order=10),),
|
|
order=20,
|
|
parameter_groups=(
|
|
ParameterGroupDisplaySpec(
|
|
id="stages",
|
|
label="分段参数",
|
|
parameters=tuple(
|
|
parameter.name for parameter in _UD00_STAGE_PARAMETERS
|
|
),
|
|
order=10,
|
|
),
|
|
),
|
|
)
|
|
|
|
def __init__(
|
|
self,
|
|
name: str,
|
|
medium: IdealGasMedium,
|
|
*,
|
|
tstart: float = 0.0,
|
|
starts: tuple[float, ...] = (0.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0),
|
|
ends: tuple[float, ...] = (1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0),
|
|
durations: tuple[float, ...] = (1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0),
|
|
nstages: int = 1,
|
|
iscyclic: bool = False,
|
|
) -> None:
|
|
super().__init__(name=name)
|
|
if len(starts) != 8 or len(ends) != 8 or len(durations) != 8:
|
|
raise ValueError("UD00 requires exactly eight start, end, and duration values.")
|
|
if nstages < 1 or nstages > 8:
|
|
raise ValueError("UD00 nstages must be between 1 and 8.")
|
|
self.tstart = float(tstart)
|
|
self.starts = tuple(float(value) for value in starts)
|
|
self.ends = tuple(float(value) for value in ends)
|
|
self.durations = tuple(float(value) for value in durations)
|
|
self.nstages = int(nstages)
|
|
self.iscyclic = bool(iscyclic)
|
|
values: dict[str, float] = {"tstart": self.tstart, "nstages": float(self.nstages), "iscyclic": float(int(self.iscyclic))}
|
|
for index in range(1, 9):
|
|
values[f"start{index}"] = self.starts[index - 1]
|
|
values[f"end{index}"] = self.ends[index - 1]
|
|
values[f"t{index}"] = self.durations[index - 1]
|
|
self.set_parameter_values(values)
|
|
self.out = self.register_declared_port("out")
|
|
self.out.signal = self.output_at(0.0)
|
|
|
|
@classmethod
|
|
def create(
|
|
cls,
|
|
*,
|
|
name: str,
|
|
medium: IdealGasMedium,
|
|
parameters: Mapping[str, float],
|
|
) -> "AmesimUd00":
|
|
nstages = parameters["nstages"]
|
|
iscyclic = parameters["iscyclic"]
|
|
definitions = {definition.name: definition for definition in cls.PARAMETERS}
|
|
for parameter_name, value in (
|
|
("nstages", nstages),
|
|
("iscyclic", iscyclic),
|
|
):
|
|
numeric_value = float(value)
|
|
if not numeric_value.is_integer():
|
|
raise ValueError(f"UD00 {parameter_name} must be an integer.")
|
|
message = definitions[parameter_name].validation_message(numeric_value)
|
|
if message is not None:
|
|
raise ValueError(f"UD00 {parameter_name} {message}.")
|
|
return cls(
|
|
name=name,
|
|
medium=medium,
|
|
tstart=parameters["tstart"],
|
|
starts=tuple(parameters[f"start{index}"] for index in range(1, 9)),
|
|
ends=tuple(parameters[f"end{index}"] for index in range(1, 9)),
|
|
durations=tuple(parameters[f"t{index}"] for index in range(1, 9)),
|
|
nstages=int(nstages),
|
|
iscyclic=bool(int(iscyclic)),
|
|
)
|
|
|
|
def output_at(self, time: float) -> float:
|
|
elapsed = max(float(time) - self.tstart, 0.0)
|
|
active_durations = self.durations[: self.nstages]
|
|
total_duration = sum(active_durations)
|
|
if self.iscyclic and total_duration > 0.0:
|
|
elapsed = elapsed % total_duration
|
|
|
|
stage_start_time = 0.0
|
|
for index, duration in enumerate(active_durations):
|
|
stage_end_time = stage_start_time + duration
|
|
if elapsed < stage_end_time or index == self.nstages - 1:
|
|
if duration <= 0.0:
|
|
return self.ends[index]
|
|
fraction = (elapsed - stage_start_time) / duration
|
|
return self.starts[index] + fraction * (self.ends[index] - self.starts[index])
|
|
stage_start_time = stage_end_time
|
|
return self.ends[self.nstages - 1]
|
|
|
|
def signal_output_values(self, time: float) -> dict[str, float]:
|
|
return {"out": self.output_at(time)}
|
|
|
|
def signal_event_times(
|
|
self,
|
|
start_time: float,
|
|
stop_time: float,
|
|
) -> tuple[float, ...]:
|
|
"""Return UD00 start, stage, and repeated cycle boundaries.
|
|
|
|
The final non-cyclic stage is intentionally not given an end event:
|
|
``output_at`` continues that stage's slope after its configured duration.
|
|
"""
|
|
|
|
if stop_time <= start_time:
|
|
return ()
|
|
|
|
active_durations = self.durations[: self.nstages]
|
|
stage_offsets = [0.0]
|
|
elapsed = 0.0
|
|
for duration in active_durations[:-1]:
|
|
elapsed += duration
|
|
stage_offsets.append(elapsed)
|
|
|
|
if not self.iscyclic:
|
|
return tuple(
|
|
sorted(
|
|
{
|
|
event_time
|
|
for offset in stage_offsets
|
|
if start_time
|
|
< (event_time := self.tstart + offset)
|
|
< stop_time
|
|
}
|
|
)
|
|
)
|
|
|
|
cycle_duration = sum(active_durations)
|
|
if cycle_duration <= 0.0:
|
|
return ()
|
|
|
|
events: set[float] = set()
|
|
for offset in stage_offsets:
|
|
first_boundary = self.tstart + offset
|
|
cycle_index = max(
|
|
0,
|
|
floor((start_time - first_boundary) / cycle_duration) + 1,
|
|
)
|
|
event_time = first_boundary + cycle_index * cycle_duration
|
|
while event_time < stop_time:
|
|
if event_time > start_time:
|
|
events.add(event_time)
|
|
cycle_index += 1
|
|
event_time = first_boundary + cycle_index * cycle_duration
|
|
return tuple(sorted(events))
|
|
|
|
def component_result_values(self) -> Mapping[str, float]:
|
|
return {"y": self.out.signal}
|