Expose per-cell temperatures as output ports on HeatExchanger and PFR

Author: milanofthe

docs/source/examples/heat_exchanger.ipynb

@@ -52,30 +52,26 @@
   {
    "cell_type": "markdown",
    "metadata": {},
-   "source": [
-    "Feed hot water at 370 K and cold water at 290 K. Record the outlet temperatures over time."
-   ]
+   "source": "Feed hot water at 370 K and cold water at 290 K. Record the outlet temperatures over time and the per-cell temperature profiles for spatial analysis."
   },
   {
    "cell_type": "code",
    "execution_count": null,
    "metadata": {},
    "outputs": [],
-   "source": "T_h_in = Source(func=lambda t: 370.0)   # hot inlet [K]\nT_c_in = Source(func=lambda t: 290.0)   # cold inlet [K]\n\nscp = Scope(labels=[\"T_h_out [K]\", \"T_c_out [K]\"])\n\nsim = Simulation(\n    blocks=[T_h_in, T_c_in, hx, scp],\n    connections=[\n        Connection(T_h_in, hx),          # T_h_in -> port 0\n        Connection(T_c_in, hx[1]),        # T_c_in -> port 1\n        Connection(hx, scp),              # T_h_out -> scope port 0\n        Connection(hx[1], scp[1]),        # T_c_out -> scope port 1\n    ],\n    dt=0.02,\n)\n\nsim.run(200)"
+   "source": "N = 10  # must match N_cells\n\nT_h_in = Source(func=lambda t: 370.0)   # hot inlet [K]\nT_c_in = Source(func=lambda t: 290.0)   # cold inlet [K]\n\nscp_out = Scope(labels=[\"T_h_out [K]\", \"T_c_out [K]\"])\nscp_hot = Scope(labels=[f\"T_h_{i+1}\" for i in range(N)])\nscp_cold = Scope(labels=[f\"T_c_{i+1}\" for i in range(N)])\n\nconnections = [\n    Connection(T_h_in, hx),           # T_h_in -> port 0\n    Connection(T_c_in, hx[1]),         # T_c_in -> port 1\n    Connection(hx, scp_out),           # T_h_out -> outlet scope\n    Connection(hx[1], scp_out[1]),     # T_c_out -> outlet scope\n]\n\n# Connect per-cell temperatures to profile scopes\nfor i in range(N):\n    connections.append(Connection(hx[2 + i], scp_hot[i]))       # T_h_i\n    connections.append(Connection(hx[2 + N + i], scp_cold[i]))  # T_c_i\n\nsim = Simulation(\n    blocks=[T_h_in, T_c_in, hx, scp_out, scp_hot, scp_cold],\n    connections=connections,\n    dt=0.02,\n)\n\nsim.run(200)"
   },
   {
    "cell_type": "code",
    "execution_count": null,
    "metadata": {},
    "outputs": [],
-   "source": "time, signals = scp.read()\n\nfig, ax = plt.subplots(figsize=(8, 5))\n\nax.plot(time, signals[0], label=\"Hot outlet\")\nax.plot(time, signals[1], label=\"Cold outlet\")\nax.axhline(370, color=\"gray\", ls=\"--\", alpha=0.4, label=\"Hot inlet\")\nax.axhline(290, color=\"gray\", ls=\"-.\", alpha=0.4, label=\"Cold inlet\")\nax.set_xlabel(\"Time [s]\")\nax.set_ylabel(\"Temperature [K]\")\nax.set_title(\"Heat Exchanger Outlet Temperatures\")\nax.legend()\nax.grid(True, alpha=0.3)\n\nplt.tight_layout()\nplt.show()"
+   "source": "time, out = scp_out.read()\n_, hot_profile = scp_hot.read()\n_, cold_profile = scp_cold.read()\n\nfig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 5))\n\n# Outlet temperatures over time\nax1.plot(time, out[0], label=\"Hot outlet\")\nax1.plot(time, out[1], label=\"Cold outlet\")\nax1.axhline(370, color=\"gray\", ls=\"--\", alpha=0.4, label=\"Hot inlet\")\nax1.axhline(290, color=\"gray\", ls=\"-.\", alpha=0.4, label=\"Cold inlet\")\nax1.set_xlabel(\"Time [s]\")\nax1.set_ylabel(\"Temperature [K]\")\nax1.set_title(\"Outlet Temperatures\")\nax1.legend()\nax1.grid(True, alpha=0.3)\n\n# Spatial temperature profile at steady state\ncells = np.arange(1, N + 1)\nT_h_ss = [hot_profile[i][-1] for i in range(N)]\nT_c_ss = [cold_profile[i][-1] for i in range(N)]\n\nax2.plot(cells, T_h_ss, \"o-\", color=\"tab:red\", label=\"Hot stream\")\nax2.plot(cells, T_c_ss, \"s-\", color=\"tab:blue\", label=\"Cold stream\")\nax2.set_xlabel(\"Cell number (hot flow direction →)\")\nax2.set_ylabel(\"Temperature [K]\")\nax2.set_title(\"Spatial Profile (Steady State)\")\nax2.legend()\nax2.grid(True, alpha=0.3)\n\nplt.tight_layout()\nplt.show()"
   },
   {
    "cell_type": "markdown",
    "metadata": {},
-   "source": [
-    "At startup, both sides are cold. The hot fluid quickly warms the exchanger from the inlet side while the cold fluid absorbs heat. At steady state, the counter-current arrangement creates the characteristic temperature cross — the cold outlet approaches the hot inlet and vice versa, achieving high thermal efficiency."
-   ]
+   "source": "At startup, both sides are cold. The hot fluid quickly warms the exchanger from the inlet side while the cold fluid absorbs heat. At steady state, the counter-current arrangement creates the characteristic temperature cross — the cold outlet approaches the hot inlet and vice versa. The spatial profile shows the classic counter-current pattern where both streams decrease along the hot flow direction."
   }
  ],
  "metadata": {

src/pathsim_chem/process/heat_exchanger.py

@@ -72,11 +72,6 @@ class HeatExchanger(DynamicalSystem):
         "T_c_in": 1,
     }
 
-    output_port_labels = {
-        "T_h_out": 0,
-        "T_c_out": 1,
-    }
-
     def __init__(self, N_cells=5, F_h=0.1, F_c=0.1, V_h=0.5, V_c=0.5,
                  UA=500.0, rho_h=1000.0, Cp_h=4184.0, rho_c=1000.0, Cp_c=4184.0,
                  T_h0=370.0, T_c0=300.0):
@@ -107,6 +102,12 @@ def __init__(self, N_cells=5, F_h=0.1, F_c=0.1, V_h=0.5, V_c=0.5,
 
         N = self.N_cells
 
+        # dynamic output port labels: outlets + per-cell profiles
+        self.output_port_labels = {"T_h_out": 0, "T_c_out": 1}
+        for i in range(N):
+            self.output_port_labels[f"T_h_{i+1}"] = 2 + i
+            self.output_port_labels[f"T_c_{i+1}"] = 2 + N + i
+
         # initial state: interleaved [T_h1, T_c1, T_h2, T_c2, ...]
         x0 = np.empty(2 * N)
         x0[0::2] = T_h0
@@ -192,9 +193,16 @@ def _jc_d(x, u, t):
 
             return J
 
-        # output: hot outlet = last cell hot, cold outlet = first cell cold
+        # output: outlets + per-cell profiles
+        # [T_h_out, T_c_out, T_h_1..T_h_N, T_c_1..T_c_N]
         def _fn_a(x, u, t):
-            return np.array([x[2*(self.N_cells - 1)], x[1]])
+            N = self.N_cells
+            out = np.empty(2 + 2 * N)
+            out[0] = x[2 * (N - 1)]   # T_h_out
+            out[1] = x[1]              # T_c_out
+            out[2:N + 2] = x[0::2]    # hot profile
+            out[N + 2:] = x[1::2]     # cold profile
+            return out
 
         super().__init__(
             func_dyn=_fn_d,

src/pathsim_chem/process/pfr.py

@@ -76,11 +76,6 @@ class PFR(DynamicalSystem):
         "T_in": 1,
     }
 
-    output_port_labels = {
-        "C_out": 0,
-        "T_out": 1,
-    }
-
     def __init__(self, N_cells=5, V=1.0, F=0.1, k0=1e6, Ea=50000.0, n=1.0,
                  dH_rxn=-50000.0, rho=1000.0, Cp=4184.0,
                  C0=0.0, T0=300.0):
@@ -110,6 +105,12 @@ def __init__(self, N_cells=5, V=1.0, F=0.1, k0=1e6, Ea=50000.0, n=1.0,
 
         N = self.N_cells
 
+        # dynamic output port labels: outlets + per-cell profiles
+        self.output_port_labels = {"C_out": 0, "T_out": 1}
+        for i in range(N):
+            self.output_port_labels[f"C_{i+1}"] = 2 + i
+            self.output_port_labels[f"T_{i+1}"] = 2 + N + i
+
         # initial state: interleaved [C_1, T_1, C_2, T_2, ...]
         x0 = np.empty(2 * N)
         x0[0::2] = C0
@@ -196,10 +197,16 @@ def _jc_d(x, u, t):
 
             return J
 
-        # output: last cell values
+        # output: outlets + per-cell profiles
+        # [C_out, T_out, C_1..C_N, T_1..T_N]
         def _fn_a(x, u, t):
             N = self.N_cells
-            return np.array([x[2*(N-1)], x[2*(N-1) + 1]])
+            out = np.empty(2 + 2 * N)
+            out[0] = x[2 * (N - 1)]       # C_out
+            out[1] = x[2 * (N - 1) + 1]   # T_out
+            out[2:N + 2] = x[0::2]        # concentration profile
+            out[N + 2:] = x[1::2]         # temperature profile
+            return out
 
         super().__init__(
             func_dyn=_fn_d,

tests/process/test_heat_exchanger.py

@@ -60,8 +60,14 @@ def test_port_labels(self):
         """Test port label definitions."""
         self.assertEqual(HeatExchanger.input_port_labels["T_h_in"], 0)
         self.assertEqual(HeatExchanger.input_port_labels["T_c_in"], 1)
-        self.assertEqual(HeatExchanger.output_port_labels["T_h_out"], 0)
-        self.assertEqual(HeatExchanger.output_port_labels["T_c_out"], 1)
+        hx = HeatExchanger(N_cells=5)
+        self.assertEqual(hx.output_port_labels["T_h_out"], 0)
+        self.assertEqual(hx.output_port_labels["T_c_out"], 1)
+        # per-cell ports
+        self.assertEqual(hx.output_port_labels["T_h_1"], 2)
+        self.assertEqual(hx.output_port_labels["T_c_1"], 7)
+        self.assertEqual(hx.output_port_labels["T_h_5"], 6)
+        self.assertEqual(hx.output_port_labels["T_c_5"], 11)
 
     def test_state_size(self):
         """Test that state vector has correct size."""

tests/process/test_pfr.py

@@ -60,8 +60,14 @@ def test_port_labels(self):
         """Test port label definitions."""
         self.assertEqual(PFR.input_port_labels["C_in"], 0)
         self.assertEqual(PFR.input_port_labels["T_in"], 1)
-        self.assertEqual(PFR.output_port_labels["C_out"], 0)
-        self.assertEqual(PFR.output_port_labels["T_out"], 1)
+        pfr = PFR(N_cells=5)
+        self.assertEqual(pfr.output_port_labels["C_out"], 0)
+        self.assertEqual(pfr.output_port_labels["T_out"], 1)
+        # per-cell ports
+        self.assertEqual(pfr.output_port_labels["C_1"], 2)
+        self.assertEqual(pfr.output_port_labels["T_1"], 7)
+        self.assertEqual(pfr.output_port_labels["C_5"], 6)
+        self.assertEqual(pfr.output_port_labels["T_5"], 11)
 
     def test_state_size(self):
         """Test that state vector has correct size."""