上传PythonModels文件

PythonModels/core/__init__.py

@@ -0,0 +1,2 @@
+"""Core abstractions for the Python system model."""
+

PythonModels/core/base.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+
+class Component(ABC):
+    def __init__(self, name: str) -> None:
+        self.name = name
+
+
+class DynamicComponent(Component):
+    state_size = 2
+
+    @staticmethod
+    def actual_stream_enthalpy(
+        port_m_flow: float,
+        connected_h: float,
+        internal_h: float,
+    ) -> float:
+        """Approximate `actualStream(port.h_outflow)` for a mixed control volume port."""
+
+        return connected_h if port_m_flow > 0.0 else internal_h
+
+    def connection_inlet_enthalpy(
+        self,
+        port_m_flow: float,
+        connected_h: float,
+        internal_h: float,
+    ) -> float:
+        """Resolve the enthalpy convected into this control volume through one port."""
+
+        return self.actual_stream_enthalpy(
+            port_m_flow=port_m_flow,
+            connected_h=connected_h,
+            internal_h=internal_h,
+        )
+
+    @abstractmethod
+    def get_state_vector(self) -> list[float]:
+        raise NotImplementedError
+
+    @abstractmethod
+    def set_state_vector(self, values: list[float]) -> None:
+        raise NotImplementedError
+
+
+class AlgebraicComponent(Component):
+    """Stateless element described by algebraic constraints only."""

PythonModels/core/medium.py

@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class ThermodynamicProperties:
+    p: float
+    T: float
+    rho: float
+    u: float
+    h: float
+
+
+@dataclass(frozen=True)
+class IdealGasMedium:
+    """Temperature-dependent ideal-gas air approximation.
+
+    This is still not a strict clone of `Modelica.Media.Air.SimpleAir`.
+    The small linear `cp(T)` term is kept configurable for calibration, but the
+    current default is calibrated against the committed Testmodel baseline and
+    therefore falls back to the constant-heat-capacity limit.
+    """
+
+    name: str = "SimpleAirApprox"
+    R_gas: float = 287.0
+    cp_ref: float = 1005.0
+    T_ref: float = 300.0
+    cp_slope: float = 0.0
+
+    @property
+    def cv(self) -> float:
+        return self.cv_at_temperature(self.T_ref)
+
+    @property
+    def gamma(self) -> float:
+        return self.cp_at_temperature(self.T_ref) / self.cv
+
+    def cp_at_temperature(self, T: float) -> float:
+        return self.cp_ref + self.cp_slope * (T - self.T_ref)
+
+    def cv_at_temperature(self, T: float) -> float:
+        return self.cp_at_temperature(T) - self.R_gas
+
+    def density(self, p: float, T: float) -> float:
+        return p / (self.R_gas * T)
+
+    def specific_internal_energy(self, T: float) -> float:
+        delta_T = T - self.T_ref
+        return (
+            self.cv * self.T_ref
+            + self.cv * delta_T
+            + 0.5 * self.cp_slope * delta_T * delta_T
+        )
+
+    def specific_enthalpy(self, T: float) -> float:
+        delta_T = T - self.T_ref
+        return (
+            self.cp_ref * self.T_ref
+            + self.cp_ref * delta_T
+            + 0.5 * self.cp_slope * delta_T * delta_T
+        )
+
+    def temperature_from_internal_energy(self, u: float) -> float:
+        reference_internal_energy = self.cv * self.T_ref
+        delta_u = u - reference_internal_energy
+
+        if abs(self.cp_slope) <= 1e-15:
+            return self.T_ref + delta_u / self.cv
+
+        a = 0.5 * self.cp_slope
+        b = self.cv
+        c = -delta_u
+        discriminant = max(b * b - 4.0 * a * c, 0.0)
+        positive_root = (-b + discriminant**0.5) / (2.0 * a)
+        negative_root = (-b - discriminant**0.5) / (2.0 * a)
+        delta_T = positive_root if abs(positive_root) <= abs(negative_root) else negative_root
+        return self.T_ref + delta_T
+
+    def temperature_from_mass_internal_energy(self, m: float, U: float) -> float:
+        if m <= 0.0:
+            raise ValueError("Mass must stay positive when recovering temperature.")
+        return self.temperature_from_internal_energy(U / m)
+
+    def pressure(self, m: float, T: float, V: float) -> float:
+        if V <= 0.0:
+            raise ValueError("Volume must stay positive.")
+        return m * self.R_gas * T / V
+
+    def properties_from_mU(self, m: float, U: float, V: float) -> ThermodynamicProperties:
+        T = self.temperature_from_mass_internal_energy(m, U)
+        p = self.pressure(m, T, V)
+        rho = m / V
+        u = U / m
+        h = self.specific_enthalpy(T)
+        return ThermodynamicProperties(p=p, T=T, rho=rho, u=u, h=h)

PythonModels/core/network.py

@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from PythonModels.core.base import Component, DynamicComponent
+
+
+@dataclass(frozen=True)
+class Connection:
+    source_component: str
+    source_port: str
+    target_component: str
+    target_port: str
+
+
+class SimulationNetwork:
+    """Container for components, topology, and state-vector bookkeeping."""
+
+    def __init__(self, name: str) -> None:
+        self.name = name
+        self.components: dict[str, Component] = {}
+        self.connections: list[Connection] = []
+
+    def add_component(self, component: Component) -> None:
+        if component.name in self.components:
+            raise ValueError(f"Duplicate component name: {component.name}")
+        self.components[component.name] = component
+
+    def connect(
+        self,
+        source_component: str,
+        source_port: str,
+        target_component: str,
+        target_port: str,
+    ) -> None:
+        self.connections.append(
+            Connection(
+                source_component=source_component,
+                source_port=source_port,
+                target_component=target_component,
+                target_port=target_port,
+            )
+        )
+
+    def dynamic_components(self) -> list[DynamicComponent]:
+        return [
+            component
+            for component in self.components.values()
+            if isinstance(component, DynamicComponent)
+        ]
+
+    def initial_state_vector(self) -> list[float]:
+        values: list[float] = []
+        for component in self.dynamic_components():
+            values.extend(component.get_state_vector())
+        return values
+
+    def apply_state_vector(self, values: list[float]) -> None:
+        cursor = 0
+        for component in self.dynamic_components():
+            next_cursor = cursor + component.state_size
+            component.set_state_vector(values[cursor:next_cursor])
+            cursor = next_cursor
+        if cursor != len(values):
+            raise ValueError("State vector length does not match dynamic components.")
+
+    def summary(self) -> str:
+        lines = [f"Network: {self.name}", "Components:"]
+        for name, component in self.components.items():
+            lines.append(f"  - {name}: {component.__class__.__name__}")
+        lines.append("Connections:")
+        for conn in self.connections:
+            lines.append(
+                f"  - {conn.source_component}.{conn.source_port}"
+                f" -> {conn.target_component}.{conn.target_port}"
+            )
+        return "\n".join(lines)
+

PythonModels/core/ports.py

@@ -0,0 +1,13 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass
+class PortState:
+    """Python-side analogue of a Modelica fluid port."""
+
+    p: float = 0.0
+    m_flow: float = 0.0
+    h_outflow: float = 0.0
+

PythonModels/core/solver.py

@@ -0,0 +1,102 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Callable
+
+
+@dataclass(frozen=True)
+class SolveIVPConfig:
+    t_start: float = 0.0
+    t_stop: float = 20.0
+    method: str = "BDF"
+    rtol: float = 1e-6
+    atol: float = 1e-8
+    max_step: float = 1e-3
+
+
+@dataclass(frozen=True)
+class ODESolution:
+    t: list[float]
+    y: list[list[float]]
+    success: bool
+    message: str
+
+
+def _vector_add(a: list[float], b: list[float], scale: float = 1.0) -> list[float]:
+    return [x + scale * y for x, y in zip(a, b)]
+
+
+def _runge_kutta_4(
+    rhs: Callable[[float, list[float]], list[float]],
+    initial_state: list[float],
+    config: SolveIVPConfig,
+    t_eval: list[float] | None,
+) -> ODESolution:
+    if t_eval is None:
+        point_count = max(
+            2,
+            int((config.t_stop - config.t_start) / max(config.max_step, 1e-6)) + 1,
+        )
+        step = (config.t_stop - config.t_start) / (point_count - 1)
+        t_eval = [config.t_start + index * step for index in range(point_count)]
+
+    state = list(initial_state)
+    states = [[value] for value in state]
+    times = [float(t_eval[0])]
+    current_time = float(t_eval[0])
+
+    for target_time in t_eval[1:]:
+        while current_time < target_time - 1e-15:
+            dt = min(config.max_step, target_time - current_time)
+            k1 = rhs(current_time, state)
+            k2 = rhs(current_time + 0.5 * dt, _vector_add(state, k1, 0.5 * dt))
+            k3 = rhs(current_time + 0.5 * dt, _vector_add(state, k2, 0.5 * dt))
+            k4 = rhs(current_time + dt, _vector_add(state, k3, dt))
+            state = [
+                value + (dt / 6.0) * (a + 2.0 * b + 2.0 * c + d)
+                for value, a, b, c, d in zip(state, k1, k2, k3, k4)
+            ]
+            current_time += dt
+
+        times.append(float(target_time))
+        for index, value in enumerate(state):
+            states[index].append(value)
+
+    return ODESolution(
+        t=times,
+        y=states,
+        success=True,
+        message="Integrated with built-in RK4 fallback because SciPy is unavailable.",
+    )
+
+
+def integrate_ode(
+    rhs: Callable[[float, list[float]], list[float]],
+    initial_state: list[float],
+    config: SolveIVPConfig,
+    t_eval: list[float] | None = None,
+):
+    """Thin wrapper around scipy.integrate.solve_ivp with a pure-Python fallback."""
+
+    if abs(config.t_stop - config.t_start) <= 1e-15:
+        return ODESolution(
+            t=[float(config.t_start)],
+            y=[[value] for value in initial_state],
+            success=True,
+            message="Skipped integration because t_start equals t_stop.",
+        )
+
+    try:
+        from scipy.integrate import solve_ivp
+    except ImportError:
+        return _runge_kutta_4(rhs, initial_state, config, t_eval)
+
+    return solve_ivp(
+        fun=rhs,
+        t_span=(config.t_start, config.t_stop),
+        y0=initial_state,
+        method=config.method,
+        rtol=config.rtol,
+        atol=config.atol,
+        t_eval=t_eval,
+    )

PythonModels/core/state.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass
+class VolumeState:
+    """Primary dynamic state for rigid adiabatic control volumes."""
+
+    m: float
+    U: float
+
+    def as_vector(self) -> list[float]:
+        return [self.m, self.U]
+
+    @classmethod
+    def from_vector(cls, values: list[float]) -> "VolumeState":
+        if len(values) != 2:
+            raise ValueError("VolumeState requires exactly two values: [m, U].")
+        return cls(m=values[0], U=values[1])
+