Coverage for bim2sim/elements/aggregation/hvac_aggregations.py: 77%
583 statements
« prev ^ index » next coverage.py v7.6.12, created at 2025-03-12 17:09 +0000
1import math
2from functools import partial
3from typing import List, Iterable, Dict, Union, Tuple, Optional
5import networkx as nx
6import numpy as np
8from bim2sim.elements import hvac_elements as hvac
9from bim2sim.elements.aggregation import AggregationMixin, logger
10from bim2sim.elements.base_elements import ProductBased
11from bim2sim.elements.graphs.hvac_graph import HvacGraph
12from bim2sim.elements.hvac_elements import HVACPort
13from bim2sim.elements.mapping import attribute
14from bim2sim.elements.mapping.units import ureg
15from bim2sim.kernel.decision import BoolDecision, DecisionBunch
18def verify_edge_ports(func):
19 """Decorator to verify edge ports"""
21 def wrapper(agg_instance, *args, **kwargs):
22 ports = func(agg_instance, *args, **kwargs)
23 for port in ports:
24 if not port.connection:
25 continue
26 if port.connection.parent in agg_instance.elements:
27 raise AssertionError("%s (%s) is not an edge port of %s" % (
28 port, port.guid, agg_instance))
29 return ports
31 return wrapper
34class HVACAggregationPort(HVACPort):
35 """Port for Aggregation"""
36 guid_prefix = 'AggPort'
38 def __init__(self, originals, *args, **kwargs):
39 super().__init__(*args, **kwargs)
40 # TODO / TBD: DJA: can one Port replace multiple? what about position?
42 if not type(originals) == list:
43 originals = [originals]
44 if not all(isinstance(n, hvac.HVACPort) for n in originals):
45 raise TypeError("originals must by HVACPorts")
46 self.originals = originals
47 self.flow_direction = self.flow_direction_from_original()
49 def flow_direction_from_original(self):
50 if len(self.originals) > 1:
51 flow_directions = set(
52 [original.flow_direction for original in self.originals])
53 if len(flow_directions) > 1:
54 raise NotImplementedError(
55 'Aggregation of HVACPorts with different flow directions'
56 'is not implemented.')
57 else:
58 return list(flow_directions)[0]
59 else:
60 originals = self.originals[0]
61 while originals:
62 if hasattr(originals, 'originals'):
63 if len(originals.originals) > 1:
64 raise NotImplementedError(
65 'Aggregation with more than one original is not '
66 'implemented.')
67 originals = originals.originals[0]
68 else:
69 return originals.flow_direction
71 def _calc_position(self, name):
72 """Position of original port"""
73 return self.originals.position
76class HVACAggregationMixin(AggregationMixin):
77 """ Mixin class for all HVACAggregations.
79 Adds some HVAC specific functionality to the AggregationMixin.
81 Args:
82 base_graph: networkx graph that should be searched for aggregations
83 match_graph: networkx graph that only holds matches
84 """
86 def __init__(self, base_graph: nx.Graph, match_graph: nx.Graph, *args,
87 **kwargs):
88 # make get_ports signature match_graph ProductBased.get_ports
89 self.get_ports = partial(self.get_ports, base_graph, match_graph)
90 graph_elements = list(set([node.parent for node in match_graph.nodes]))
91 super().__init__(graph_elements, *args, **kwargs)
93 @verify_edge_ports
94 def get_ports(self, base_graph: HvacGraph, match_graph: HvacGraph
95 ) -> List[HVACPort]:
96 """ Get the edge ports based on the difference between base_graph and
97 match_graph,
99 Args:
100 base_graph: The base graph.
101 match_graph: The matching graph.
103 Returns:
104 A list of HVACPort objects representing the edge ports.
105 """
106 # edges of g excluding all relations to s
107 e1 = base_graph.subgraph(base_graph.nodes - match_graph.nodes).edges
109 # if graph and match_graph are identical
110 if not e1:
111 # ports with only one connection are edge ports in this case
112 edge_ports = [v for v, d in match_graph.degree() if d == 1]
113 else:
114 # all edges related to s
115 e2 = base_graph.edges - e1
116 # related to s but not s exclusive
117 e3 = e2 - match_graph.edges
118 # get only edge_ports that belong to the match_graph graph
119 edge_ports = list(
120 set([port for port in [e for x in list(e3) for e in x]
121 if port in match_graph]))
122 ports = [HVACAggregationPort(port, parent=self) for port in edge_ports]
123 return ports
125 @classmethod
126 def get_empty_mapping(cls, elements: Iterable[ProductBased]):
127 """ Get information to remove elements.
129 Args:
130 elements:
132 Returns:
133 mapping: tuple of mapping dict with original ports as values and
134 None as keys.
135 connections: connection list of outer connections.
136 """
137 ports = [port for element in elements for port in element.ports]
138 mapping = {port: None for port in ports}
139 # TODO: len > 1, optimize
140 external_ports = []
141 for port in ports:
142 if port.connection and port.connection.parent not in elements:
143 external_ports.append(port.connection)
145 mapping[external_ports[0].connection] = external_ports[1]
146 mapping[external_ports[1].connection] = external_ports[0]
147 connections = []
149 return mapping, connections
151 def get_replacement_mapping(self) \
152 -> Dict[HVACPort, Union[HVACAggregationPort, None]]:
153 """ Get replacement dict for existing ports."""
154 mapping = {port: None for element in self.elements
155 for port in element.ports}
156 for port in self.ports:
157 for original in port.originals:
158 mapping[original] = port
159 return mapping
161 @classmethod
162 def find_matches(cls, base_graph: HvacGraph
163 ) -> Tuple[List[nx.Graph], List[dict]]:
164 """ Find all matches for aggregation in HVAC graph.
166 Args:
167 base_graph: The HVAC graph that is searched for potential
168 matches.
170 Returns:
171 matches_graphs: List of HVAC graphs that matches the aggregation.
172 metas: List of dict with metas information. One element for each
173 matches_graphs.
175 Raises:
176 NotImplementedError: If method is not implemented.
177 """
178 raise NotImplementedError(
179 "Method %s.find_matches not implemented" % cls.__name__)
181 def _calc_has_pump(self, name) -> bool:
182 """ Determines if aggregation contains pumps.
184 Returns:
185 True, if aggregation has pumps
186 """
187 has_pump = False
188 for ele in self.elements:
189 if hvac.Pump is ele.__class__:
190 has_pump = True
191 break
192 return has_pump
195class PipeStrand(HVACAggregationMixin, hvac.Pipe):
196 """ Aggregates pipe strands, i.e. pipes, pipe fittings and valves.
198 This aggregation reduces the number of elements by merging straight
199 connected elements with just two ports into one PipeStrand. The length and
200 a medium diameter are calculated based on the aggregated elements to
201 maintain meaningful parameters for pressure loss calculations.
202 """
203 aggregatable_classes = {hvac.Pipe, hvac.PipeFitting, hvac.Valve}
204 multi = ('length', 'diameter')
206 @classmethod
207 def find_matches(cls, base_graph: HvacGraph
208 ) -> Tuple[List[HvacGraph], List[dict]]:
209 """ Find all matches for PipeStrand in HvacGraph.
211 Args:
212 base_graph: The Hvac graph to search for matches in.
214 Returns:
215 A tuple containing two lists:
216 - matches_graphs: List of HvacGraphs that hold PipeStrands
217 - metas: List of dict with meta information. One element for
218 each match.
219 """
220 pipe_strands = HvacGraph.get_type_chains(
221 base_graph.element_graph, cls.aggregatable_classes,
222 include_singles=True)
223 matches_graphs = [base_graph.subgraph_from_elements(pipe_strand)
224 for pipe_strand in pipe_strands]
226 metas = [{} for x in matches_graphs] # no metadata calculated
227 return matches_graphs, metas
229 @attribute.multi_calc
230 def _calc_avg(self) -> dict:
231 """ Calculates the total length and average diameter of all pipe-like
232 elements.
233 """
235 total_length = 0
236 avg_diameter = 0
237 diameter_times_length = 0
239 for pipe in self.elements:
240 length = getattr(pipe, "length")
241 diameter = getattr(pipe, "diameter")
242 if not (length and diameter):
243 logger.warning("Ignored '%s' in aggregation", pipe)
244 continue
246 diameter_times_length += diameter * length
247 total_length += length
249 if total_length != 0:
250 avg_diameter = diameter_times_length / total_length
252 result = dict(
253 length=total_length,
254 diameter=avg_diameter
255 )
256 return result
258 diameter = attribute.Attribute(
259 description="Average diameter of aggregated pipe",
260 functions=[_calc_avg],
261 unit=ureg.millimeter,
262 dependant_elements='elements'
263 )
265 length = attribute.Attribute(
266 description="Length of aggregated pipe",
267 functions=[_calc_avg],
268 unit=ureg.meter,
269 dependant_elements='elements'
270 )
273class UnderfloorHeating(PipeStrand):
274 """ Class for aggregating underfloor heating systems.
276 The normal pitch (spacing) between pipes is typically between 0.1m and 0.2m.
277 """
279 @classmethod
280 def find_matches(cls, base_graph: HvacGraph
281 ) -> Tuple[List[HvacGraph], List[dict]]:
282 """ Finds matches of underfloor heating systems in a given graph.
284 Args:
285 base_graph: An HvacGraph that should be checked for underfloor
286 heating systems.
288 Returns:
289 A tuple containing two lists:
290 - matches_graphs: A list of HvacGraphs that contain underfloor
291 heating systems.
292 - metas: A list of dict with meta information for each
293 underfloor heating system. One element for each match.
294 """
295 chains = HvacGraph.get_type_chains(
296 base_graph.element_graph, cls.aggregatable_classes,
297 include_singles=True)
298 matches_graphs = []
299 metas = []
300 for chain in chains:
301 meta = cls.check_conditions(chain)
302 if meta:
303 metas.append(meta)
304 matches_graphs.append(base_graph.subgraph_from_elements(chain))
305 return matches_graphs, metas
307 @staticmethod
308 def check_number_of_elements(chain: nx.classes.reportviews.NodeView,
309 tolerance: int = 20) -> bool:
310 """ Check if the targeted chain has more than 20 elements.
312 This method checks if a given chain has more than the specified number
313 of elements.
315 Args:
316 chain: Possible chain of consecutive elements to be an underfloor
317 heating.
318 tolerance: Integer tolerance value to check the number of elements.
319 Default is 20.
321 Returns:
322 True if the chain has more than the specified number of elements,
323 False otherwise.
324 """
325 return len(chain) >= tolerance
327 @staticmethod
328 def check_pipe_strand_horizontality(ports_coors: np.ndarray,
329 tolerance: float = 0.8) -> bool:
330 """ Checks the horizontality of a pipe strand.
332 This method checks if the pipe strand is located horizontally, meaning
333 it is parallel to the floor and most elements are in the same z plane.
335 Args:
336 ports_coors: An array with pipe strand port coordinates.
337 tolerance: Tolerance to check pipe strand horizontality.
338 Default is 0.8.
340 Returns:
341 True, if check succeeds and False if check fails.
342 """
343 counts = np.unique(ports_coors[:, 2], return_counts=True)
344 # TODO: cluster z coordinates
345 idx_max = np.argmax(counts[1])
346 return counts[1][idx_max] / ports_coors.shape[0] >= tolerance
348 @staticmethod
349 def get_pipe_strand_attributes(ports_coors: np.ndarray,
350 chain: nx.classes.reportviews.NodeView
351 ) -> [*(ureg.Quantity,) * 5]:
352 """ Gets the attributes of a pipe strand.
354 This method retrieves the attributes of a pipe strand in order to
355 perform further checks and calculations.
357 Args:
358 ports_coors: An array with pipe strand port coordinates.
359 chain: A possible chain of elements to be an Underfloor heating.
361 Returns:
362 heating_area: Underfloor heating area.
363 total_length: Underfloor heating total pipe length.
364 avg_diameter: Average underfloor heating diameter.
365 dist_x: Underfloor heating dimension in x.
366 dist_y: Underfloor heating dimension in y.
367 """
368 total_length = sum(segment.length for segment in chain if
369 segment.length is not None)
370 avg_diameter = (sum(segment.diameter ** 2 * segment.length for segment
371 in chain if segment.length is not
372 None) / total_length) ** 0.5
373 length_unit = total_length.u
374 x_coord, y_coord = ports_coors[:, 0], ports_coors[:, 1]
375 min_x = ports_coors[np.argmin(x_coord)][:2]
376 max_x = ports_coors[np.argmax(x_coord)][:2]
377 min_y = ports_coors[np.argmin(y_coord)][:2]
378 max_y = ports_coors[np.argmax(y_coord)][:2]
379 if min_x[1] == max_x[1] or min_y[0] == max_y[0]:
380 dist_x = (max_x[0] - min_x[0]) * length_unit
381 dist_y = (max_y[1] - min_y[1]) * length_unit
382 heating_area = (dist_x * dist_y)
383 else:
384 dist_x = (np.linalg.norm(min_y - max_x)) * length_unit
385 dist_y = (np.linalg.norm(min_y - min_x)) * length_unit
386 heating_area = (dist_x * dist_y)
388 return heating_area, total_length, avg_diameter, dist_x, dist_y
390 @staticmethod
391 def get_ufh_type():
392 # TODO: function to obtain the underfloor heating form based on issue
393 # #211
394 raise NotImplementedError
396 @classmethod
397 def get_pipe_strand_spacing(cls,
398 chain: nx.classes.reportviews.NodeView,
399 dist_x: ureg.Quantity,
400 dist_y: ureg.Quantity,
401 tolerance: int = 10
402 ) -> [*(ureg.Quantity,) * 2]:
403 """ Sorts the pipe elements according to their angle in the horizontal
404 plane. Necessary to calculate subsequently the underfloor heating
405 spacing.
407 Args:
408 chain: possible chain of elements to be an Underfloor heating
409 dist_x: Underfloor heating dimension in x
410 dist_y: Underfloor heating dimension in y
411 tolerance: integer tolerance to get pipe strand spacing
413 Returns:
414 x_spacing: Underfloor heating pitch in x,
415 y_spacing: Underfloor heating pitch in y
416 """
417 # ToDo: what if multiple pipe elements on the same line? Collinear
418 # algorithm, issue #211
419 orientations = {}
420 for element in chain:
421 if type(element) is hvac.Pipe:
422 a = abs(element.ports[0].position[1] -
423 element.ports[1].position[1])
424 b = abs(element.ports[0].position[0] -
425 element.ports[1].position[0])
426 if b != 0:
427 theta = int(math.degrees(math.atan(a / b)))
428 else:
429 theta = 90
430 if theta not in orientations:
431 orientations[theta] = []
432 orientations[theta].append(element)
433 for orient in orientations.copy():
434 if len(orientations[orient]) < tolerance:
435 del orientations[orient]
436 orientations = list(sorted(orientations.items()))
437 x_spacing = dist_x / (len(orientations[0][1]) - 1)
438 y_spacing = dist_y / (len(orientations[1][1]) - 1)
439 return x_spacing, y_spacing
441 @staticmethod
442 def check_heating_area(heating_area: ureg.Quantity,
443 tolerance: ureg.Quantity =
444 1e6 * ureg.millimeter ** 2) -> bool:
445 """ Check if the total area of the underfloor heating is greater than
446 the tolerance value - just as safety factor.
448 Args:
449 heating_area: Underfloor heating area,
450 tolerance: Quantity tolerance to check heating area
452 Returns:
453 None: if check fails
454 True: if check succeeds
455 """
456 return heating_area >= tolerance
458 @staticmethod
459 def check_spacing(x_spacing: ureg.Quantity,
460 y_spacing: ureg.Quantity,
461 tolerance: tuple = (90 * ureg.millimeter,
462 210 * ureg.millimeter)
463 ) -> Optional[bool]:
464 """ Check if the spacing between adjacent elements with the same
465 orientation is between the tolerance values.
466 Args:
467 x_spacing: Underfloor heating pitch in x
468 y_spacing: Underfloor heating pitch in y
469 tolerance: tuple tolerance to check underfloor heating spacing
471 Returns:
472 None: if check fails
473 True: if check succeeds
474 """
475 if not ((tolerance[0] < x_spacing < tolerance[1])
476 or (tolerance[0] < y_spacing < tolerance[1])):
477 return
478 return True
480 @staticmethod
481 def check_kpi(total_length: ureg.Quantity,
482 avg_diameter: ureg.Quantity,
483 heating_area: ureg.Quantity,
484 tolerance: tuple = (0.09, 0.01)) -> Optional[bool]:
485 """ Check if the quotient between the cross-sectional area of the pipe
486 strand (x-y plane) and the total heating area is between the
487 tolerance values - area density for underfloor heating.
489 Args:
490 total_length: Underfloor heating total pipe length,
491 avg_diameter: Average Underfloor heating diameter,
492 heating_area: Underfloor heating area,
493 tolerance: tuple tolerance to check underfloor heating kpi
495 Returns:
496 None: if check fails
497 True: if check succeeds
498 """
499 kpi_criteria = (total_length * avg_diameter) / heating_area
500 return tolerance[0] > kpi_criteria > tolerance[1]
502 @classmethod
503 def check_conditions(cls, chain: nx.classes.reportviews.NodeView
504 ) -> Optional[dict]:
505 """ Checks ps_elements and returns instance of UnderfloorHeating if all
506 following criteria are fulfilled:
507 0. minimum of 20 elements
508 1. the pipe strand is located horizontally
509 2. the pipe strand elements located in an specific z-coordinate
510 3. the spacing tolerance
511 4. underfloor heating area tolerance
512 5. kpi criteria
514 Args:
515 chain: possible chain of elements to be an Underfloor heating
517 Returns:
518 None: if check fails
519 meta: dict with calculated values if check succeeds
520 """
521 # TODO: use only floor heating pipes and not connecting pipes
522 if not cls.check_number_of_elements(chain):
523 return
524 ports_coordinates = np.array(
525 [p.position for e in chain for p in e.ports])
526 if not cls.check_pipe_strand_horizontality(ports_coordinates):
527 return
529 heating_area, total_length, avg_diameter, dist_x, dist_y = \
530 cls.get_pipe_strand_attributes(ports_coordinates, chain)
531 x_spacing, y_spacing = cls.get_pipe_strand_spacing(
532 chain, dist_x, dist_y)
534 if not cls.check_heating_area(heating_area):
535 return
536 if not cls.check_spacing(x_spacing, y_spacing):
537 return
538 if not cls.check_kpi(total_length, avg_diameter, heating_area):
539 return
541 meta = dict(
542 length=total_length,
543 diameter=avg_diameter,
544 heating_area=heating_area,
545 x_spacing=x_spacing,
546 y_spacing=y_spacing
547 )
548 return meta
550 def is_consumer(self):
551 return True
553 heating_area = attribute.Attribute(
554 unit=ureg.meter ** 2,
555 description='Heating area',
556 )
557 x_spacing = attribute.Attribute(
558 unit=ureg.meter,
559 description='Spacing in x',
560 )
561 y_spacing = attribute.Attribute(
562 unit=ureg.meter,
563 description='Spacing in y',
564 )
565 rated_power = attribute.Attribute(
566 unit=ureg.kilowatt,
567 description="rated power"
568 )
569 rated_mass_flow = attribute.Attribute(
570 description="Rated mass flow of pump",
571 unit=ureg.kg / ureg.s,
572 )
575class ParallelPump(HVACAggregationMixin, hvac.Pump):
576 """ Aggregates pumps in parallel."""
577 aggregatable_classes = {hvac.Pump, hvac.Pipe, hvac.PipeFitting, PipeStrand}
578 whitelist_classe = {hvac.Pump}
580 multi = ('rated_power', 'rated_height', 'rated_volume_flow', 'diameter',
581 'diameter_strand', 'length')
583 @classmethod
584 def find_matches(cls, base_graph: HvacGraph
585 ) -> Tuple[List[HvacGraph], List[dict]]:
586 """ Find matches of parallel pumps in the given graph.
588 Args:
589 base_graph: HVAC graph that should be checked for parallel pumps.
591 Returns:
592 A tuple containing two lists:
593 matches_graph: List of HVAC graphs that hold the parallel pumps.
594 metas: List of dict with meta information. One element for
595 each match.
596 """
597 element_graph = base_graph.element_graph
598 inert_classes = cls.aggregatable_classes - cls.whitelist_classe
599 parallel_pump_strands = HvacGraph.get_parallels(
600 element_graph, cls.whitelist_classe, inert_classes,
601 grouping={'rated_power': 'equal'},
602 grp_threshold=1)
603 matches_graph = [base_graph.subgraph_from_elements(parallel.nodes)
604 for parallel in parallel_pump_strands]
605 metas = [{} for x in matches_graph] # no metadata calculated
606 return matches_graph, metas
609 @attribute.multi_calc
610 def _calc_avg(self) -> dict:
611 """Calculates the total length and average diameter of all pump-like
612 elements."""
613 avg_diameter_strand = 0
614 total_length = 0
615 diameter_times_length = 0
617 for item in self.not_pump_elements:
618 if hasattr(item, "diameter") and hasattr(item, "length"):
619 length = item.length
620 diameter = item.diameter
621 if not (length and diameter):
622 logger.info("Ignored '%s' in aggregation", item)
623 continue
625 diameter_times_length += length * diameter
626 total_length += length
627 else:
628 logger.info("Ignored '%s' in aggregation", item)
630 if total_length != 0:
631 avg_diameter_strand = diameter_times_length / total_length
633 result = dict(
634 length=total_length,
635 diameter_strand=avg_diameter_strand
636 )
637 return result
639 def get_replacement_mapping(self) \
640 -> Dict[HVACPort, Union[HVACAggregationPort, None]]:
641 mapping = super().get_replacement_mapping()
643 # TODO: cant this be solved in find_matches?
644 # search for aggregations made during the parallel pump construction
645 new_aggregations = [element.aggregation for element in self.elements if
646 element.aggregation is not self]
647 for port in (p for a in new_aggregations for p in a.ports):
648 for original in port.originals:
649 mapping[original] = port
650 return mapping
652 @property
653 def pump_elements(self) -> list:
654 """list of pump-like elements present on the aggregation"""
655 return [ele for ele in self.elements if isinstance(ele, hvac.Pump)]
657 def _calc_rated_power(self, name) -> ureg.Quantity:
658 """Calculate the rated power adding the rated power of the pump-like
659 elements"""
660 if all(ele.rated_power for ele in self.pump_elements):
661 return sum([ele.rated_power for ele in self.pump_elements])
662 else:
663 return None
665 rated_power = attribute.Attribute(
666 unit=ureg.kilowatt,
667 description="rated power",
668 functions=[_calc_rated_power],
669 dependant_elements='pump_elements'
670 )
672 def _calc_rated_height(self, name) -> ureg.Quantity:
673 """Calculate the rated height, using the maximal rated height of
674 the pump-like elements"""
675 return max([ele.rated_height for ele in self.pump_elements])
677 rated_height = attribute.Attribute(
678 description='rated height',
679 functions=[_calc_rated_height],
680 unit=ureg.meter,
681 dependant_elements='pump_elements'
682 )
684 def _calc_volume_flow(self, name) -> ureg.Quantity:
685 """Calculate the volume flow, adding the volume flow of the pump-like
686 elements"""
687 return sum([ele.rated_volume_flow for ele in self.pump_elements])
689 rated_volume_flow = attribute.Attribute(
690 description='rated volume flow',
691 functions=[_calc_volume_flow],
692 unit=ureg.meter ** 3 / ureg.hour,
693 dependant_elements='pump_elements'
694 )
696 def _calc_diameter(self, name) -> ureg.Quantity:
697 """Calculate the diameter, using the pump-like elements diameter"""
698 return sum(item.diameter ** 2 for item in self.pump_elements) ** 0.5
700 diameter = attribute.Attribute(
701 description='diameter',
702 functions=[_calc_diameter],
703 unit=ureg.millimeter,
704 dependant_elements='pump_elements'
705 )
707 @property
708 def not_pump_elements(self) -> list:
709 """list of not-pump-like elements present on the aggregation"""
710 return [ele for ele in self.elements if not isinstance(ele, hvac.Pump)]
712 length = attribute.Attribute(
713 description='length of aggregated pipe elements',
714 functions=[_calc_avg],
715 unit=ureg.meter,
716 dependant_elements='not_pump_elements'
717 )
719 diameter_strand = attribute.Attribute(
720 description='average diameter of aggregated pipe elements',
721 functions=[_calc_avg],
722 unit=ureg.millimeter,
723 dependant_elements='not_pump_elements'
724 )
727class Consumer(HVACAggregationMixin, hvac.HVACProduct):
728 """ A class that aggregates a Consumer.
730 This class represents a Consumer system in an HVAC graph, which can
731 contain various elements such as space heaters, pipes, pumps, and
732 valves. It aggregates these elements into a single entity, called
733 Consumer.
735 Attributes:
736 multi: A tuple of attribute names that can have multiple
737 values for a Consumer.
738 aggregatable_classes: A dict of element classes that can be
739 aggregated into a Consumer.
740 whitelist_classes: A dict of element classes that should be included
741 when searching for a Consumer in an HVAC graph.
742 blacklist_classes: A dict of element classes that should be excluded
743 when searching for a Consumer in an HVAC graph.
744 boarder_classes: A dictionary of element classes that define the
745 border of a Consumer system.
746 """
748 aggregatable_classes = {
749 hvac.SpaceHeater, hvac.Pipe, hvac.PipeFitting, hvac.Junction,
750 hvac.Pump, hvac.Valve, hvac.ThreeWayValve, PipeStrand,
751 UnderfloorHeating}
752 whitelist_classes = {hvac.SpaceHeater, UnderfloorHeating}
753 blacklist_classes = {hvac.Chiller, hvac.Boiler, hvac.CoolingTower,
754 hvac.HeatPump, hvac.Storage, hvac.CHP}
755 boarder_classes = {hvac.Distributor}
756 multi = ('has_pump', 'rated_power', 'rated_pump_power', 'rated_height',
757 'rated_volume_flow', 'temperature_inlet',
758 'temperature_outlet', 'volume', 'description')
760 @classmethod
761 def find_matches(cls, base_graph: HvacGraph
762 ) -> Tuple[List[HvacGraph], List[dict]]:
763 """ Find matches of consumer in the given base HVAC graph.
765 Args:
766 base_graph: The HVAC graph to search for consumers in.
768 Returns:
769 A tuple with two lists
770 - matches_graph: A list of HVAC graphs that contain
771 consumers, and the second list contains
772 - metas: A list of dict with meta information about each
773 consumer
774 """
775 # remove boarder_classes nodes from base_graph to separate cycles
776 graph = HvacGraph.remove_classes_from(base_graph, cls.boarder_classes)
777 cycles = nx.connected_components(graph)
779 matches_graphs = []
780 for cycle in cycles:
781 cycle_graph = graph.subgraph(cycle)
782 # check for blacklist_classes in cycle, i.e. generators
783 generator = {ele for ele in cycle_graph.elements if
784 ele.__class__ in cls.blacklist_classes}
785 if generator:
786 # check for whitelist_classes in cycle, i.e. consumers
787 gen_con = {ele for ele in cycle_graph.elements if
788 ele.__class__ in cls.whitelist_classes}
789 if gen_con:
790 # TODO: Consumer separieren
791 pass
792 else:
793 consumer = {ele for ele in cycle_graph.elements if
794 ele.__class__ in cls.whitelist_classes}
795 if consumer:
796 matches_graphs.append(cycle_graph)
798 metas = [{} for x in matches_graphs]
799 return matches_graphs, metas
801 @property
802 def pump_elements(self) -> list:
803 """ List of pump-like elements present on the aggregation."""
804 return [ele for ele in self.elements if isinstance(ele, hvac.Pump)]
806 @property
807 def not_pump_elements(self) -> list:
808 """ List of not-pump-like elements present on the aggregation."""
809 return [ele for ele in self.elements if not isinstance(ele, hvac.Pump)]
811 def _calc_TControl(self, name):
812 return any([isinstance(ele, hvac.ThreeWayValve) for ele in self.elements])
814 @property
815 def whitelist_elements(self) -> list:
816 """ List of whitelist_classes elements present on the aggregation."""
817 return [ele for ele in self.elements
818 if type(ele) in self.whitelist_classes]
820 def _calc_rated_power(self, name) -> ureg.Quantity:
821 """ Calculate the rated power adding the rated power of the
822 whitelist_classes elements.
823 """
824 return sum([ele.rated_power for ele in self.whitelist_elements])
826 rated_power = attribute.Attribute(
827 description="rated power",
828 unit=ureg.kilowatt,
829 functions=[_calc_rated_power],
830 dependant_elements='whitelist_elements'
831 )
833 has_pump = attribute.Attribute(
834 description="Cycle has a pump system",
835 functions=[HVACAggregationMixin._calc_has_pump]
836 )
838 def _calc_rated_pump_power(self, name) -> ureg.Quantity:
839 """ Calculate the rated pump power adding the rated power of the
840 pump-like elements.
841 """
842 return sum([ele.rated_power for ele in self.pump_elements])
844 rated_pump_power = attribute.Attribute(
845 description="rated pump power",
846 unit=ureg.kilowatt,
847 functions=[_calc_rated_pump_power],
848 dependant_elements='pump_elements'
849 )
851 def _calc_volume_flow(self, name) -> ureg.Quantity:
852 """ Calculate the volume flow, adding the volume flow of the pump-like
853 elements.
854 """
855 return sum([ele.rated_volume_flow for ele in self.pump_elements])
857 rated_volume_flow = attribute.Attribute(
858 description="rated volume flow",
859 unit=ureg.meter ** 3 / ureg.hour,
860 functions=[_calc_volume_flow],
861 dependant_elements='pump_elements'
862 )
864 def _calc_flow_temperature(self, name) -> ureg.Quantity:
865 """ Calculate the flow temperature, using the flow temperature of the
866 whitelist_classes elements.
867 """
868 return sum(ele.flow_temperature.to_base_units() for ele
869 in self.whitelist_elements) / len(self.whitelist_elements)
871 flow_temperature = attribute.Attribute(
872 description="temperature inlet",
873 unit=ureg.kelvin,
874 functions=[_calc_flow_temperature],
875 dependant_elements='whitelist_elements'
876 )
878 def _calc_return_temperature(self, name) -> ureg.Quantity:
879 """ Calculate the return temperature, using the return temperature of
880 the whitelist_classes elements.
881 """
882 return sum(ele.return_temperature.to_base_units() for ele
883 in self.whitelist_elements) / len(self.whitelist_elements)
885 return_temperature = attribute.Attribute(
886 description="temperature outlet",
887 unit=ureg.kelvin,
888 functions=[_calc_return_temperature],
889 dependant_elements='whitelist_elements'
890 )
892 def _calc_dT_water(self, name):
893 """ Water dt of consumer."""
894 return self.flow_temperature - self.return_temperature
896 dT_water = attribute.Attribute(
897 description="Nominal temperature difference",
898 unit=ureg.kelvin,
899 functions=[_calc_dT_water],
900 )
902 def _calc_body_mass(self, name):
903 """ Body mass of consumer."""
904 return sum(ele.body_mass for ele in self.whitelist_elements)
906 body_mass = attribute.Attribute(
907 description="Body mass of Consumer",
908 functions=[_calc_body_mass],
909 unit=ureg.kg,
910 )
912 def _calc_heat_capacity(self, name):
913 """ Heat capacity of consumer."""
914 return sum(ele.heat_capacity for ele in
915 self.whitelist_elements)
917 heat_capacity = attribute.Attribute(
918 description="Heat capacity of Consumer",
919 functions=[_calc_heat_capacity],
920 unit=ureg.joule / ureg.kelvin,
921 )
923 def _calc_demand_type(self, name):
924 """ Demand type of consumer."""
925 return 1 if self.dT_water > 0 else -1
927 demand_type = attribute.Attribute(
928 description="Type of demand if 1 - heating, if -1 - cooling",
929 functions=[_calc_demand_type],
930 )
932 volume = attribute.Attribute(
933 description="volume",
934 unit=ureg.meter ** 3,
935 )
937 def _calc_rated_height(self, name) -> ureg.Quantity:
938 """ Calculate the rated height power, using the maximal rated height of
939 the pump-like elements.
940 """
941 return max([ele.rated_height for ele in self.pump_elements])
943 rated_height = attribute.Attribute(
944 description="rated volume flow",
945 unit=ureg.meter,
946 functions=[_calc_rated_height],
947 dependant_elements='pump_elements'
948 )
950 def _calc_description(self, name) -> str:
951 """ Obtains the aggregation description using the whitelist_classes
952 elements.
953 """
954 con_types = {}
955 for ele in self.whitelist_elements:
956 if type(ele) not in con_types:
957 con_types[type(ele)] = 0
958 con_types[type(ele)] += 1
960 return ', '.join(['{1} x {0}'.format(k.__name__, v) for k, v
961 in con_types.items()])
963 description = attribute.Attribute(
964 description="String with number of Consumers",
965 functions=[_calc_description],
966 dependant_elements='whitelist_elements'
967 )
969 t_control = attribute.Attribute(
970 description="Bool for temperature control cycle.",
971 functions=[_calc_TControl]
972 )
975class ConsumerHeatingDistributorModule(HVACAggregationMixin, hvac.HVACProduct):
976 """ A class that aggregates (several) consumers including the distributor.
978 Attributes:
979 multi: A Tuple of attributes to consider in aggregation.
980 aggregatable_classes: A dict of element classes that can be
981 aggregated into a ConsumerDistributorModule.
982 whitelist_classes: A dict of element classes that should be included
983 when searching for a ConsumerDistributorModule in an HVAC graph.
984 blacklist_classes: A dict of element classes that should be excluded
985 when searching for a ConsumerDistributorModule in an HVAC graph.
986 boarder_classes: Dictionary of classes that are used as boarders.
987 """
989 multi = (
990 'medium', 'use_hydraulic_separator', 'hydraulic_separator_volume',
991 'temperature_inlet', 'temperature_outlet')
992 # TODO: Abused to not just sum attributes from elements
993 aggregatable_classes = {
994 hvac.SpaceHeater, hvac.Pipe, hvac.PipeFitting, hvac.Distributor,
995 PipeStrand, Consumer, hvac.Junction, hvac.ThreeWayValve }
996 whitelist_classes = {
997 hvac.SpaceHeater, UnderfloorHeating, Consumer}
998 blacklist_classes = {hvac.Chiller, hvac.Boiler, hvac.CoolingTower}
999 boarder_classes = {hvac.Distributor}
1001 def __init__(self, base_graph, match_graph, *args, **kwargs):
1002 self.undefined_consumer_ports = kwargs.pop(
1003 'undefined_consumer_ports', None)
1004 self._consumer_cycles = kwargs.pop('consumer_cycles', None)
1005 self.consumers = {con for consumer in self._consumer_cycles for con in
1006 consumer if con.__class__ in self.whitelist_classes}
1007 self.open_consumer_pairs = self._register_open_consumer_ports()
1008 super().__init__(base_graph, match_graph, *args, **kwargs)
1009 # add open consumer ports to found ports by get_ports()
1010 for con_ports in self.open_consumer_pairs:
1011 self.ports.append(HVACAggregationPort(con_ports, parent=self))
1013 def _register_open_consumer_ports(self):
1014 """ This function registers open consumer ports by pairing up loose
1015 ends at the distributor. If there is an odd number of loose ends,
1016 it raises a NotImplementedError.
1018 Returns:
1019 list: A list of pairs of open consumer ports.
1021 Raises:
1022 NotImplementedError: If there is an odd number of loose ends at
1023 the distributor.
1024 """
1025 if (len(self.undefined_consumer_ports) % 2) == 0:
1026 consumer_ports = self.undefined_consumer_ports
1027 else:
1028 raise NotImplementedError(
1029 "Odd Number of loose ends at the distributor.")
1030 return consumer_ports
1032 @classmethod
1033 def find_matches(cls, base_graph: HvacGraph
1034 ) -> Tuple[List[HvacGraph], List[dict]]:
1035 """ Finds matches of consumer heating distributor modules in the given
1036 graph.
1038 Args:
1039 base_graph: The graph to be checked for consumer heating distributor
1040 modules.
1042 Returns:
1043 A tuple containing two lists:
1044 - matches_graphs: contains the HVAC graphs that hold the
1045 consumer heating distributor module.
1046 The second list
1047 - metas: contains the meta information for each consumer
1048 heating distributor modules as a dictionary.
1049 """
1050 distributors = {ele for ele in base_graph.elements
1051 if type(ele) in cls.boarder_classes}
1052 matches_graphs = []
1053 metas = []
1054 for distributor in distributors:
1055 _graph = base_graph.copy()
1056 _graph.remove_nodes_from(distributor.ports)
1057 consumer_cycle_elements = []
1058 metas.append({'undefined_consumer_ports': [],
1059 'consumer_cycles': []})
1060 cycles = nx.connected_components(_graph)
1061 for cycle in cycles:
1062 cycle_graph = base_graph.subgraph(cycle)
1063 # check for blacklist_classes in cycle, i.e. generators
1064 generator = {ele for ele in cycle_graph.elements if
1065 ele.__class__ in cls.blacklist_classes}
1066 if generator:
1067 # check for whitelist_classes in cycle that contains a
1068 # generator
1069 gen_con = {ele for ele in cycle_graph.elements if
1070 ele.__class__ in cls.whitelist_classes}
1071 if gen_con:
1072 # TODO: separate consumer (maybe recursive function?)
1073 pass
1074 else:
1075 consumer_cycle = {ele for ele in cycle_graph.elements if
1076 ele.__class__ in cls.whitelist_classes}
1077 if consumer_cycle:
1078 consumer_cycle_elements.extend(cycle_graph.elements)
1079 metas[-1]['consumer_cycles'].append(
1080 consumer_cycle_elements)
1081 else:
1082 # cycle does not hold a consumer might be undefined
1083 # consumer ports
1084 metas[-1]['undefined_consumer_ports'].extend(
1085 [neighbor for cycle_node in list(cycle_graph.nodes)
1086 for neighbor in base_graph.neighbors(cycle_node)
1087 if neighbor.parent == distributor])
1089 match_graph = base_graph.subgraph_from_elements(
1090 consumer_cycle_elements + [distributor])
1091 matches_graphs.append(match_graph)
1093 return matches_graphs, metas
1095 @attribute.multi_calc
1096 # TODO fix hardcoded values
1097 def _calc_avg(self):
1098 result = dict(
1099 medium=None,
1100 use_hydraulic_separator=False,
1101 hydraulic_separator_volume=1,
1102 )
1103 return result
1105 medium = attribute.Attribute(
1106 description="Medium of the DestributerCycle",
1107 functions=[_calc_avg]
1108 )
1110 @property
1111 def whitelist_elements(self) -> list:
1112 """list of whitelist_classes elements present on the aggregation"""
1113 return [ele for ele in self.elements if type(ele) in self.whitelist_classes]
1115 def _calc_flow_temperature(self, name) -> list:
1116 """Calculate the flow temperature, using the flow temperature of the
1117 whitelist_classes elements"""
1118 return [ele.flow_temperature.to_base_units() for ele
1119 in self.whitelist_elements]
1121 def _calc_has_pump(self, name) -> list[bool]:
1122 """Returns a list with boolean for every consumer if it has a pump."""
1123 return [con.has_pump for con in self.whitelist_elements]
1125 flow_temperature = attribute.Attribute(
1126 description="temperature inlet",
1127 unit=ureg.kelvin,
1128 functions=[_calc_flow_temperature],
1129 dependant_elements='whitelist_elements'
1130 )
1132 has_pump = attribute.Attribute(
1133 description="List with bool for every consumer if it has a pump",
1134 functions=[_calc_has_pump]
1135 )
1137 def _calc_return_temperature(self, name) -> list:
1138 """Calculate the return temperature, using the return temperature of the
1139 whitelist_classes elements"""
1140 return [ele.return_temperature.to_base_units() for ele
1141 in self.whitelist_elements]
1143 return_temperature = attribute.Attribute(
1144 description="temperature outlet",
1145 unit=ureg.kelvin,
1146 functions=[_calc_return_temperature],
1147 dependant_elements='whitelist_elements'
1148 )
1150 def _calc_dT_water(self, name):
1151 """water dt of consumer"""
1152 return [ele.dT_water.to_base_units() for ele
1153 in self.whitelist_elements]
1155 dT_water = attribute.Attribute(
1156 description="Nominal temperature difference",
1157 unit=ureg.kelvin,
1158 functions=[_calc_dT_water],
1159 dependant_elements='whitelist_elements'
1160 )
1162 def _calc_body_mass(self, name):
1163 """heat capacity of consumer"""
1164 return [ele.body_mass for ele in self.whitelist_elements]
1166 body_mass = attribute.Attribute(
1167 description="Body mass of Consumer",
1168 functions=[_calc_body_mass],
1169 unit=ureg.kg,
1170 )
1172 def _calc_heat_capacity(self, name):
1173 """heat capacity of consumer"""
1174 return [ele.heat_capacity for ele in self.whitelist_elements]
1176 heat_capacity = attribute.Attribute(
1177 description="Heat capacity of Consumer",
1178 functions=[_calc_heat_capacity],
1179 unit=ureg.joule / ureg.kelvin,
1180 )
1182 def _calc_demand_type(self, name):
1183 """demand type of consumer"""
1184 return [ele.demand_type for ele in self.whitelist_elements]
1186 demand_type = attribute.Attribute(
1187 description="Type of demand if 1 - heating, if -1 - cooling",
1188 functions=[_calc_demand_type],
1189 dependant_elements='whitelist_elements'
1190 )
1192 def calc_mass_flow(self, name):
1193 """Returns the mass flow, in the form of a list with the mass flow of
1194 the whitelist_classes-like elements"""
1195 return [ele.rated_mass_flow for ele in self.whitelist_elements]
1197 rated_mass_flow = attribute.Attribute(
1198 description="rated mass flow",
1199 functions=[calc_mass_flow],
1200 unit=ureg.kg / ureg.s,
1201 dependant_elements='whitelist_elements'
1202 )
1204 use_hydraulic_separator = attribute.Attribute(
1205 description="boolean if there is a hdydraulic seperator",
1206 functions=[_calc_avg]
1207 )
1209 hydraulic_separator_volume = attribute.Attribute(
1210 description="Volume of the hdydraulic seperator",
1211 functions=[_calc_avg],
1212 unit=ureg.m ** 3,
1213 )
1215 def _calc_rated_power(self, name):
1216 """Returns the rated power, as a list of the rated power of the
1217 whitelist_elements elements"""
1218 return [ele.rated_power for ele in self.whitelist_elements]
1220 rated_power = attribute.Attribute(
1221 description="Rated heating power of all consumers",
1222 unit=ureg.kilowatt,
1223 functions=[_calc_rated_power],
1224 dependant_elements='whitelist_elements'
1225 )
1227 def _calc_TControl(self, name) -> list[bool]:
1228 return [con.t_control for con in self.whitelist_elements]
1230 t_control = attribute.Attribute(
1231 description="List with bool for every consumer if it has a feedback "
1232 "cycle for temperature control.",
1233 functions=[_calc_TControl]
1234 )
1236 def _calc_temperature_array(self):
1237 return [self.flow_temperature, self.return_temperature]
1239 temperature_array = attribute.Attribute(
1240 description="Array of flow and return temperature",
1241 functions=[_calc_temperature_array]
1242 )
1245class GeneratorOneFluid(HVACAggregationMixin, hvac.HVACProduct):
1246 """ Aggregates generator modules with only one fluid cycle (CHPs, Boilers,
1247 ...)
1249 Not for Chillers or Heat-pumps!
1250 """
1251 aggregatable_classes = {
1252 hvac.Pump, PipeStrand, hvac.Pipe, hvac.PipeFitting, hvac.Distributor,
1253 hvac.Boiler, ParallelPump, hvac.Valve, hvac.Storage,
1254 hvac.ThreeWayValve, hvac.Junction, ConsumerHeatingDistributorModule,
1255 Consumer}
1256 whitelist_classes = {hvac.Boiler, hvac.CHP}
1257 boarder_classes = {hvac.Distributor, ConsumerHeatingDistributorModule}
1258 multi = ('rated_power', 'has_bypass', 'rated_height', 'volume',
1259 'rated_volume_flow', 'rated_pump_power', 'has_pump')
1261 def __init__(self, base_graph, match_graph, *args, **kwargs):
1262 self.non_relevant = kwargs.pop('non_relevant', set())
1263 self.has_parallel = kwargs.pop('has_parallel', False)
1264 self.bypass_elements = kwargs.pop('bypass_elements', set())
1265 self.has_bypass = True if self.bypass_elements else False
1266 super().__init__(base_graph, match_graph, *args, **kwargs)
1268 @classmethod
1269 def find_matches(cls, base_graph: HvacGraph
1270 ) -> Tuple[List[HvacGraph], List[dict]]:
1271 """ Finds matches of generators with one fluid.
1273 Non-relevant elements like bypasses are added to metas
1274 information to delete later.
1276 Args:
1277 base_graph: HVAC graph that should be checked for one fluid
1278 generators
1280 Returns:
1281 A tuple containing two lists:
1282 - matches_graphs: List of HVAC graphs that hold a generator
1283 cycle including the distributor.
1284 - metas: List of dict with meta information for each generator
1285 as a dictionary. In this case it holds non_relevant
1286 nodes, which have to be deleted later but are not
1287 contained in the match_graph .Because we are currently not
1288 able to distinguish to which graph these non_relevant
1289 nodes belong, we just output the complete list of
1290 non-relevant nodes for every element_graph.
1291 """
1292 element_graph = base_graph.element_graph
1293 inerts = cls.aggregatable_classes - cls.whitelist_classes
1294 _graph = HvacGraph.remove_not_wanted_nodes(
1295 element_graph, cls.whitelist_classes, inerts)
1296 dict_all_cycles_wanted = HvacGraph.get_all_cycles_with_wanted(
1297 _graph, cls.whitelist_classes)
1298 list_all_cycles_wanted = [*dict_all_cycles_wanted.values()]
1300 # create flat lists to subtract for non-relevant
1301 generator_flat = set()
1302 wanted_flat = set()
1304 # check for generation cycles
1305 generator_cycles = []
1306 for cycles_list in list_all_cycles_wanted:
1307 generator_cycle = list(
1308 nx.subgraph(_graph, cycle) for cycle in cycles_list
1309 if any(type(node) == block for block in
1310 cls.boarder_classes for node in cycle))
1311 if generator_cycle:
1312 generator_cycles.extend(generator_cycle)
1313 generator_flat.update(generator_cycle[0].nodes)
1314 wanted_flat.update(
1315 [item for sublist in cycles_list for item in sublist])
1317 cleaned_generator_cycles = []
1318 metas = []
1319 if generator_flat:
1320 non_relevant = wanted_flat - generator_flat
1322 # remove overlapping elements in GeneratorCycles
1323 for gen_cycle in generator_cycles:
1324 pseudo_lst = gen_cycle.copy()
1325 for gen_cycle_two in generator_cycles:
1326 if gen_cycle == gen_cycle_two:
1327 continue
1328 pseudo_lst.remove_nodes_from(gen_cycle_two)
1329 cleaned_generator_cycles.append(pseudo_lst)
1330 _graph = base_graph.copy()
1332 # match_graph bypass elements from non relevant elements
1333 for i in range(len(cleaned_generator_cycles)):
1334 metas.append(dict())
1335 metas[i]['bypass_elements'] = []
1336 for cycle in list_all_cycles_wanted[i]:
1337 if len(cycle - cleaned_generator_cycles[i].nodes
1338 - non_relevant) > 0:
1339 continue
1340 bypass_elements = cycle - cleaned_generator_cycles[i].nodes
1341 cleaned_generator_cycles[i].add_nodes_from(bypass_elements)
1342 non_relevant.difference_update(bypass_elements)
1343 metas[i]['bypass_elements'].append(bypass_elements)
1345 if len(metas) > 0:
1346 metas[0]['non_relevant'] = non_relevant
1348 matches_graphs = []
1349 for cycle in cleaned_generator_cycles:
1350 match_graph = base_graph.subgraph_from_elements(list(cycle.nodes))
1351 match_graph = HvacGraph.remove_classes_from(
1352 match_graph, cls.boarder_classes)
1353 matches_graphs.append(match_graph)
1354 return matches_graphs, metas
1356 @attribute.multi_calc
1357 def _calc_avg(self):
1358 """ Calculates the parameters of all the below listed elements."""
1359 avg_diameter_strand = 0
1360 total_length = 0
1361 diameter_times_length = 0
1363 for element in self.not_whitelist_elements:
1364 if hasattr(element, "diameter") and hasattr(element, "length"):
1365 length = element.length
1366 diameter = element.diameter
1367 if not (length and diameter):
1368 logger.info("Ignored '%s' in aggregation", element)
1369 continue
1371 diameter_times_length += diameter * length
1372 total_length += length
1374 else:
1375 logger.info("Ignored '%s' in aggregation", element)
1377 if total_length != 0:
1378 avg_diameter_strand = diameter_times_length / total_length
1380 result = dict(
1381 length=total_length,
1382 diameter_strand=avg_diameter_strand
1383 )
1384 return result
1386 @attribute.multi_calc
1387 def _calc_has_bypass(self):
1388 decision = BoolDecision(
1389 "Does the generator %s has a bypass?" % self.name,
1390 global_key=self.guid + '.bypass',
1391 allow_save=True,
1392 allow_load=True,
1393 related=[element.guid for element in self.elements], )
1394 has_bypass = decision.decide()
1395 return dict(has_bypass=has_bypass)
1397 @classmethod
1398 def find_bypasses(cls, graph):
1399 """ Finds bypasses based on the graphical network.
1401 Currently not used, might be removed in the future.
1402 """
1403 # todo remove if discussed
1404 wanted = set(cls.whitelist_classes)
1405 boarders = set(cls.boarder_classes)
1406 inerts = set(cls.aggregatable_elements) - wanted
1407 bypass_nodes = HvacGraph.detect_bypasses_to_wanted(
1408 graph, wanted, inerts, boarders)
1409 return bypass_nodes
1411 def _calc_has_bypass_decision(self):
1412 """ Checks if bypass exists based on decision. Currently not used as
1413 only possible with workaround. See todo documentation"""
1414 # todo remove if discussed, see #184
1415 # todo more elegant way? Problem is that cant yield from attributes
1416 # or cached propertys @earnsdev. Maybe using
1417 # get_pending_attribute_decisions() in combination with request?
1418 decisions = DecisionBunch()
1419 cur_decision = BoolDecision(
1420 "Does the generator %s has a bypass?" % self.guid,
1421 key=self,
1422 global_key=self.guid + '.bypass',
1423 related=[element.guid for element in self.elements], )
1424 decisions.append(cur_decision)
1425 yield decisions
1426 answers = decisions.to_answer_dict()
1427 has_bypass = list(answers.values())[0]
1428 self.has_bypass = has_bypass
1429 return has_bypass
1431 @property
1432 def whitelist_elements(self) -> list:
1433 """ List of whitelist_classes elements present on the aggregation"""
1434 return [ele for ele in self.elements if type(ele)
1435 in self.whitelist_classes]
1437 @property
1438 def not_whitelist_elements(self) -> list:
1439 """ List of not-whitelist_classes elements present on the aggregation"""
1440 return [ele for ele in self.elements if type(ele) not
1441 in self.whitelist_classes]
1443 def _calc_rated_power(self, name) -> ureg.Quantity:
1444 """ Calculate the rated power adding the rated power of the
1445 whitelist_classes elements."""
1446 return sum([ele.rated_power for ele in self.whitelist_elements])
1448 rated_power = attribute.Attribute(
1449 unit=ureg.kilowatt,
1450 description="rated power",
1451 functions=[_calc_rated_power],
1452 dependant_elements='whitelist_elements'
1453 )
1455 def _calc_min_power(self, name):
1456 """ Calculates the min power, adding the min power of the
1457 whitelist_elements."""
1458 return sum([ele.min_power for ele in self.whitelist_elements])
1460 min_power = attribute.Attribute(
1461 unit=ureg.kilowatt,
1462 description="min power",
1463 functions=[_calc_min_power],
1464 dependant_elements='whitelist_elements'
1465 )
1467 def _calc_min_PLR(self, name):
1468 """ Calculates the min PLR, using the min power and rated power."""
1469 return self.min_power / self.rated_power
1471 min_PLR = attribute.Attribute(
1472 description="Minimum part load ratio",
1473 unit=ureg.dimensionless,
1474 functions=[_calc_min_PLR],
1475 )
1477 def _calc_flow_temperature(self, name) -> ureg.Quantity:
1478 """ Calculate the flow temperature, using the flow temperature of the
1479 whitelist_classes elements."""
1480 return sum(ele.flow_temperature.to_base_units() for ele
1481 in self.whitelist_elements) / len(self.whitelist_elements)
1483 flow_temperature = attribute.Attribute(
1484 description="Nominal inlet temperature",
1485 unit=ureg.kelvin,
1486 functions=[_calc_flow_temperature],
1487 dependant_elements='whitelist_elements'
1488 )
1490 def _calc_return_temperature(self, name) -> ureg.Quantity:
1491 """ Calculate the return temperature, using the return temperature of
1492 the whitelist_classes elements."""
1493 return sum(ele.return_temperature.to_base_units() for ele
1494 in self.whitelist_elements) / len(self.whitelist_elements)
1496 return_temperature = attribute.Attribute(
1497 description="Nominal outlet temperature",
1498 unit=ureg.kelvin,
1499 functions=[_calc_return_temperature],
1500 dependant_elements='whitelist_elements'
1501 )
1503 def _calc_dT_water(self, name):
1504 """ Rated power of boiler."""
1505 return abs(self.return_temperature - self.flow_temperature)
1507 dT_water = attribute.Attribute(
1508 description="Nominal temperature difference",
1509 unit=ureg.kelvin,
1510 functions=[_calc_dT_water],
1511 )
1513 def _calc_diameter(self, name) -> ureg.Quantity:
1514 """ Calculate the diameter, using the whitelist_classes elements
1515 diameter."""
1516 return sum(
1517 item.diameter ** 2 for item in self.whitelist_elements) ** 0.5
1519 diameter = attribute.Attribute(
1520 description='diameter',
1521 unit=ureg.millimeter,
1522 functions=[_calc_diameter],
1523 dependant_elements='whitelist_elements'
1524 )
1526 length = attribute.Attribute(
1527 description='length of aggregated pipe elements',
1528 functions=[_calc_avg],
1529 unit=ureg.meter,
1530 )
1532 diameter_strand = attribute.Attribute(
1533 description='average diameter of aggregated pipe elements',
1534 functions=[_calc_avg],
1535 unit=ureg.millimeter,
1536 )
1538 has_pump = attribute.Attribute(
1539 description="Cycle has a pumpsystem",
1540 functions=[HVACAggregationMixin._calc_has_pump]
1541 )
1543 @property
1544 def pump_elements(self) -> list:
1545 """ List of pump-like elements present on the aggregation"""
1546 return [ele for ele in self.elements if isinstance(ele, hvac.Pump)]
1548 def _calc_rated_pump_power(self, name) -> ureg.Quantity:
1549 """ Calculate the rated pump power adding the rated power of the
1550 pump-like elements."""
1551 if all(ele.rated_power for ele in self.pump_elements):
1552 return sum([ele.rated_power for ele in self.pump_elements])
1553 else:
1554 return None
1556 rated_pump_power = attribute.Attribute(
1557 description="rated pump power",
1558 unit=ureg.kilowatt,
1559 functions=[_calc_rated_pump_power],
1560 dependant_elements='pump_elements'
1561 )
1563 def _calc_volume_flow(self, name) -> ureg.Quantity:
1564 """ Calculate the volume flow, adding the volume flow of the pump-like
1565 elements."""
1566 return sum([ele.rated_volume_flow for ele in self.pump_elements])
1568 rated_volume_flow = attribute.Attribute(
1569 description="rated volume flow",
1570 unit=ureg.m ** 3 / ureg.s,
1571 functions=[_calc_volume_flow],
1572 dependant_elements='pump_elements'
1573 )
1575 def _calc_volume(self, name):
1576 """ Calculates volume of GeneratorOneFluid."""
1577 return NotImplementedError
1579 volume = attribute.Attribute(
1580 description="Volume of Boiler",
1581 unit=ureg.m ** 3,
1582 functions=[_calc_volume]
1583 )
1585 def _calc_rated_height(self, name) -> ureg.Quantity:
1586 """ Calculate the rated height power, using the maximal rated height of
1587 the pump-like elements."""
1588 return max([ele.rated_height for ele in self.pump_elements])
1590 rated_height = attribute.Attribute(
1591 description="rated volume flow",
1592 unit=ureg.m,
1593 functions=[_calc_rated_height],
1594 dependant_elements='pump_elements'
1595 )