Coverage for pyrc\postprocessing\heat.py: 13%
150 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-08-03 16:07 +0200
1# -------------------------------------------------------------------------------
2# Copyright (C) 2026 Joel Kimmich, Tim Jourdan
3# ------------------------------------------------------------------------------
4# License
5# This file is part of PyRC, distributed under GPL-3.0-or-later.
6# ------------------------------------------------------------------------------
7from __future__ import annotations
9from datetime import datetime
10from typing import TYPE_CHECKING
12import numpy as np
13from matplotlib import pyplot as plt
15from pyrc.core.components.templates import solution_object
16from pyrc.tools.functions import seconds_to_dates
17from leanplot import TimePlot
19if TYPE_CHECKING:
20 from pyrc.core.nodes import Node, MassFlowNode, ChannelNode
21 from pyrc.core.components.capacitor import Capacitor
22 from pyrc.core.components.resistor import Resistor
23 from pyrc.core.inputs import BoundaryCondition, InternalHeatSource
24 from pyrc.core.components.templates import RCSolution
27def get_accumulated_heat_flux(node: Capacitor):
28 """
29 Sums up the heat flux of all connected resistors.
31 Used to get the heat flux that is brought in through `BoundaryCondition` s.
33 Parameters
34 ----------
35 node : NoteTemplate
36 The Node of which the heat flux is being calculated.
37 For this, all connected resistors are summed up.
39 Returns
40 -------
41 float :
42 The heat flux of all connected resistors.
43 """
44 resistors: Resistor = node.neighbours
45 seen_nodes = set()
46 resistors_without_parallel = []
47 for resistor in resistors:
48 connected_node = resistor.get_connected_node(node)
49 if connected_node not in seen_nodes:
50 seen_nodes.add(connected_node)
51 resistors_without_parallel.append(resistor)
53 heat_flux = [r.heat_flux(node) for r in resistors_without_parallel]
55 return sum(heat_flux)
58class HeatFlux:
59 def __init__(self, rc_solution: RCSolution = solution_object):
60 self.rc_solution = rc_solution
62 def parse_time_step_index(self, time_step_index):
63 if time_step_index is None:
64 time_step_index = list(range(len(self.rc_solution.time_steps)))
65 elif (
66 not isinstance(time_step_index, list)
67 and not isinstance(time_step_index, slice)
68 and not isinstance(time_step_index, np.ndarray)
69 ):
70 time_step_index = [time_step_index]
71 return time_step_index
73 def boundary(self, boundary: BoundaryCondition, time_step_index=None):
74 resistors: Resistor = boundary.neighbours
76 time_step_index = self.parse_time_step_index(time_step_index)
78 bc_temp = self.rc_solution.input_vectors[time_step_index, boundary.index]
79 heat_flux_sum = np.zeros_like(bc_temp)
81 for resistor in resistors:
82 other_node_index = resistor.get_connected_node(boundary).index
83 node_temp = self.rc_solution.result_vectors[time_step_index, other_node_index]
84 heat_flux_sum += 1 / resistor.resistance * (bc_temp - node_temp)
85 return self.rc_solution.time_steps[time_step_index], heat_flux_sum
87 def preheat(self, distributor: MassFlowNode, collector: MassFlowNode):
88 """
89 Returns the preheat in Kelvin.
91 Returns
92 -------
93 np.ndarray
94 """
95 time_step_index = self.parse_time_step_index(None)
96 temp_dist: np.ndarray = self.rc_solution.result_vectors[time_step_index, distributor.index]
97 temp_col: np.ndarray = self.rc_solution.result_vectors[time_step_index, collector.index]
98 return temp_col - temp_dist
100 def calculate_heat_flux(self, node, resistors: list | Resistor, time_step_index=None) -> np.ndarray | float:
101 """
102 Returns the heat flux going into the node by the passed resistors.
104 Parameters
105 ----------
106 node : NoteTemplate
107 The Node of which the heat flux is being calculated.
108 resistors : list | Resistor
109 The resistors used to calculate the heat flow.
110 time_step_index : int | slice | list, optional
111 The time steps that should be calculated.
112 If None, all are calculated.
114 Returns
115 -------
116 float | np.ndarray :
117 The heat fluxes for every time step.
118 """
119 from pyrc.core.components.resistor import Resistor
121 if isinstance(resistors, Resistor):
122 resistors = [resistors]
123 time_step_index = self.parse_time_step_index(time_step_index)
125 result = 0
126 node_temp = self.rc_solution.result_vectors[time_step_index, node.index]
128 from pyrc.core.components.input import Input
130 for resistor in resistors:
131 other_node = resistor.get_connected_node(node)
132 if isinstance(other_node, Input):
133 result += (
134 1
135 / resistor.equivalent_resistance
136 * (self.rc_solution.input_vectors[time_step_index, other_node.index] - node_temp)
137 )
138 else:
139 result += (
140 1
141 / resistor.equivalent_resistance
142 * (self.rc_solution.result_vectors[time_step_index, other_node.index] - node_temp)
143 )
144 return result
146 def internal_heat_source(self, ihc: InternalHeatSource | list, time_step_index=None):
147 if not isinstance(ihc, list):
148 ihc = [ihc]
149 ihc_index = [i.index for i in ihc]
150 time_step_index = self.parse_time_step_index(time_step_index)
151 ihc_values = self.rc_solution.input_vectors[time_step_index, :][:, ihc_index]
152 return self.rc_solution.time_steps[time_step_index], ihc_values
154 def heat_flux_directions(
155 self,
156 nodes: list[Node],
157 directions: list | np.ndarray = np.array([0, 0, -1]),
158 except_resistor_types=None,
159 time_step_index=None,
160 ) -> tuple | np.ndarray:
161 """
162 Returns the heat flux through the layer in the desired direction of all nodes in the given list.
164 E.g. if the direction is (0,0,1) the heat flux in positive z direction is calculated. The heat flux is
165 positive if going in the same direction as the desired one.
167 Parameters
168 ----------
169 nodes : list[Node]
170 The Nodes of which the heat flux is being calculated.
171 directions : list | np.ndarray, optional
172 The direction(s) in which the heat flux is calculated.
173 Can be a list then each direction is calculated and returned.
174 except_resistor_types
175 time_step_index
177 Returns
178 -------
179 np.ndarray | tuple :
180 The result of one direction as np.ndarray or the result of all directions as tuple.
181 """
182 if except_resistor_types is None:
183 except_resistor_types = []
184 if isinstance(directions, np.ndarray):
185 directions = [directions]
186 results = []
187 for direction in directions:
188 nodes_and_resistors = [
189 (node, node.resistors_in_direction_filtered(direction, except_resistor_types=except_resistor_types))
190 for node in nodes
191 if node is not None
192 ]
194 result = 0
195 for node, resistors in nodes_and_resistors:
196 result += self.calculate_heat_flux(node, resistors, time_step_index)
197 results.append(result)
199 if len(results) > 1:
200 return tuple(results)
201 return results[0]
204def fluxes_data(
205 boundaries: list = None,
206 boundaries_sum: list = None,
207 time_step_index=None,
208 balance: bool = False,
209 rc_solution: RCSolution = None,
210):
211 if rc_solution is not None:
212 heat_flux = HeatFlux(rc_solution=rc_solution)
213 else:
214 heat_flux = HeatFlux()
216 results = []
217 labels = []
219 sum_vector = None
220 time_steps = None
222 def add_to_sum(result: np.ndarray, _sum_vector):
223 if _sum_vector is None:
224 _sum_vector = np.zeros_like(result)
225 return _sum_vector + result
227 if boundaries is not None:
228 for bc in boundaries:
229 time_steps, y = heat_flux.boundary(bc, time_step_index)
230 results.append(y)
231 labels.append(bc.__class__.__name__)
232 sum_vector = add_to_sum(y, sum_vector)
233 if boundaries_sum is not None:
234 if not isinstance(boundaries_sum[0], list):
235 boundaries_sum = [boundaries_sum]
236 for i, bc_groups in enumerate(boundaries_sum):
237 time_steps, bc_sum = heat_flux.internal_heat_source(bc_groups[0], time_step_index)
238 for bc in bc_groups[1:]:
239 _, bc_add_on = heat_flux.internal_heat_source(bc, time_step_index)
240 bc_sum += bc_add_on
241 results.append(bc_sum)
242 labels.append(f"{bc_groups[0].__class__.__name__}s: {i}")
243 sum_vector = add_to_sum(bc_sum.reshape(-1, 1), sum_vector.reshape(-1, 1))
244 if balance and time_steps is not None and sum_vector is not None:
245 results.append(sum_vector)
246 labels.append("Balance")
247 return time_steps, results, labels
250def plot_channel_balance(
251 channel_nodes: list[Node] | ChannelNode,
252 distributor: MassFlowNode,
253 collector: MassFlowNode,
254 time_step_index=None,
255 start_date=datetime(2023, 1, 1),
256 y_scale=1,
257):
258 """
259 Plots the balance of the given channel nodes.
261 Parameters
262 ----------
263 channel_nodes : list[ChannelNode] | ChannelNode
264 The channel nodes to plot the sum of.
265 distributor : MassFlowNode
266 The distributor before the channel nodes.
267 Used to calculate the heat flux that goes into the mass flow within the channel nodes.
268 collector : MassFlowNode
269 The collector after the channel nodes.
270 Used to calculate the heat flux that goes into the mass flow within the channel nodes.
271 time_step_index : int | slice | list, optional
272 The time steps that should be calculated.
274 """
275 from pyrc.core.nodes import ChannelNode
277 if isinstance(channel_nodes, ChannelNode):
278 channel_nodes = [channel_nodes]
279 heat_flux = HeatFlux()
280 time_step_index = heat_flux.parse_time_step_index(time_step_index)
281 exterior_direction = np.array([0, 0, -1])
282 interior_direction = np.array([0, 0, 1])
283 upper_direction = np.array([0, -1, 0])
284 lower_direction = np.array([0, 1, 0])
285 ex, interior, up, lo = heat_flux.heat_flux_directions(
286 channel_nodes,
287 directions=[exterior_direction, interior_direction, upper_direction, lower_direction],
288 time_step_index=time_step_index,
289 )
290 # side_heat_flux_dist = heat_flux.heat_flux_directions([distributor],
291 # directions=[np.array([1, 0, 0])],
292 # time_step_index=time_step_index,
293 # except_resistor_types=[MassTransport]
294 # )
295 # side_heat_flux_col = heat_flux.heat_flux_directions([collector],
296 # directions=[np.array([-1, 0, 0])],
297 # time_step_index=time_step_index,
298 # except_resistor_types=[MassTransport]
299 # )
300 temp_dist = heat_flux.rc_solution.result_vectors[time_step_index, distributor.index]
301 temp_col = heat_flux.rc_solution.result_vectors[time_step_index, collector.index]
302 mass_flow = distributor.mass_flow
303 spec_capacity = distributor.material.heat_capacity
304 channel_heat_flux = mass_flow * spec_capacity * (temp_col - temp_dist)
306 x = heat_flux.rc_solution.time_steps[time_step_index]
307 x = seconds_to_dates(x, start_date)
309 time_plot = TimePlot(
310 x=x,
311 ys=[ex, interior, (up + lo)],
312 labels=["exterior", "interior", "inbetween"],
313 y_scale=y_scale,
314 y_title="Heat Flux / W",
315 )
316 time_plot.plot_stack()
317 time_plot.ax.plot(
318 time_plot.x,
319 channel_heat_flux * time_plot.y_scale,
320 label=["channel"],
321 color="black",
322 linewidth=time_plot.line_width,
323 )
324 time_plot.format()
325 time_plot.show()
328def plot_fluxes_accumulated(
329 boundaries: list = None, boundaries_sum: list = None, time_step_index=None, plot_balance: bool = False
330):
331 x, ys, labels = fluxes_data(boundaries, boundaries_sum, time_step_index, plot_balance)
333 fig, (ax1, ax2) = plt.subplots(nrows=2, ncols=1, sharex=True)
335 for i, y in enumerate(ys):
336 if labels[i] == "Balance":
337 pass
338 ax1.plot(x, y, labels[i])
340 fig.tight_layout()
341 ax1.set_xlabel("Time")
342 ax1.set_ylabel("Positive Heat flux / W")
343 ax2.set_ylabel("Negative Heat flux / W")
344 for ax in (ax1, ax2):
345 ax.grid(True)
346 ax.legend()
348 plt.show()