代码仓库移植

src/solver.py

@@ -0,0 +1,115 @@
+"""
+Time-loop driver for the 0D-1D coupled tank-pipe simulation.
+
+Per time step (per spec §3.1):
+  1. Compute CFL-limited dt from pipe's max wave speed
+  2. Freeze ghost states from current tank states
+  3. Compute two boundary HLL fluxes (left and right)
+  4. Advance pipe by one step using those two fluxes (pipe.step handles
+     the internal fluxes itself)
+  5. Advance both tanks using the SAME two boundary fluxes * area
+     -> this "flux doubling" is the mechanism that makes system mass
+        and energy strictly conserved to machine precision
+  6. Advance time
+  7. Append snapshot to history
+"""
+import numpy as np
+
+
+def run(tank1, tank2, pipe, t_end, cfl, verbose=False, log_every=100):
+    """
+    Run the coupled tank-pipe simulation from t=0 to t=t_end.
+
+    Parameters
+    ----------
+    tank1, tank2 : Tank
+        Upstream and downstream tanks. tank1 connects to pipe.W[:, 0],
+        tank2 connects to pipe.W[:, -1].
+    pipe : Pipe
+        1D pipe instance with initial state already set.
+    t_end : float
+        End time in seconds.
+    cfl : float
+        CFL number in (0, 1].
+    verbose : bool, default False
+        If True, print step-progress info every `log_every` steps.
+    log_every : int, default 100
+        Logging interval when verbose=True.
+
+    Returns
+    -------
+    dict
+        History with keys 't', 'P1', 'T1', 'P2', 'T2' (all 1D arrays
+        of shape (n_steps,)), and 'W_hist' of shape (n_steps, 3, N).
+    """
+    history = {
+        't': [],
+        'P1': [], 'T1': [],
+        'P2': [], 'T2': [],
+        'W_hist': [],
+    }
+
+    t = 0.0
+    step = 0
+
+    while t < t_end:
+        # --- Phase 1: CFL time step ---
+        a_max = pipe.max_wave_speed()
+        dt = cfl * pipe.dx / a_max
+        dt = min(dt, t_end - t)
+        if dt < 1e-12:
+            raise RuntimeError(
+                f"dt degenerate at step {step}: dt={dt:.3e}, a_max={a_max:.3e}"
+            )
+
+        # --- Phase 2: freeze tank ghost states (snapshot for this step) ---
+        W_ghost_L = tank1.ghost_state()
+        W_ghost_R = tank2.ghost_state()
+
+        # --- Phase 3: two boundary fluxes (solver-level, same solver as pipe) ---
+        flux_L = pipe._flux_fn(W_ghost_L,     pipe.W[:, 0],  pipe.gamma)
+        flux_R = pipe._flux_fn(pipe.W[:, -1], W_ghost_R,     pipe.gamma)
+
+        # --- Phase 4: advance pipe (internal fluxes handled inside) ---
+        pipe.step(flux_L, flux_R, dt)
+
+        # --- Phase 5: advance tanks with the SAME boundary fluxes * area ---
+        fL_A = flux_L * pipe.area
+        fR_A = flux_R * pipe.area
+        # Left boundary flux is "rightward positive"; tank1 loses that mass
+        tank1.apply_flux(mdot=fL_A[0], edot=fL_A[2], dt=dt, sign=-1)
+        # Right boundary flux is "rightward positive"; tank2 gains that mass
+        tank2.apply_flux(mdot=fR_A[0], edot=fR_A[2], dt=dt, sign=+1)
+
+        # --- Phase 6: advance time ---
+        t += dt
+        step += 1
+
+        # --- Phase 7: record history ---
+        history['t'].append(t)
+        history['P1'].append(tank1.P)
+        history['T1'].append(tank1.T)
+        history['P2'].append(tank2.P)
+        history['T2'].append(tank2.T)
+        history['W_hist'].append(pipe.W.copy())
+
+        if verbose and step % log_every == 0:
+            _, u, _, _ = pipe.primitives()
+            print(
+                f"step={step:6d}  t={t:.5f}  dt={dt:.2e}  "
+                f"P1={tank1.P/1e6:7.4f}MPa  P2={tank2.P/1e6:7.4f}MPa  "
+                f"max|u|={float(np.max(np.abs(u))):7.1f}m/s"
+            )
+
+    if step == 0:
+        raise RuntimeError("solver.run() exited without taking any step")
+
+    # Convert lists to arrays for downstream consumers
+    history['t']      = np.asarray(history['t'])
+    history['P1']     = np.asarray(history['P1'])
+    history['T1']     = np.asarray(history['T1'])
+    history['P2']     = np.asarray(history['P2'])
+    history['T2']     = np.asarray(history['T2'])
+    history['W_hist'] = np.stack(history['W_hist'])   # shape (n_steps, 3, N)
+
+    return history