Coverage for bim2sim/tasks/hvac/reduce.py: 16%

1import logging 

2 

3from bim2sim.elements.aggregation.hvac_aggregations import UnderfloorHeating, \ 

4 Consumer, PipeStrand, ParallelPump, ConsumerHeatingDistributorModule, \ 

5 GeneratorOneFluid 

6from bim2sim.elements.graphs.hvac_graph import HvacGraph 

7from bim2sim.kernel.decision import BoolDecision, DecisionBunch 

8from bim2sim.tasks.base import ITask 

9 

10 

11class Reduce(ITask): 

12 

13 reads = ('graph',) 

14 touches = ('graph',) 

15 

16 def run(self, graph: HvacGraph) -> (HvacGraph,): 

17 """Apply aggregations to reduce number of elements in the HVAC graph. 

18 

19 This task applies aggregations to the HVAC graph based on the specified 

20 aggregation classes. It logs information about the number of elements 

21 before and after applying aggregations, as well as the statistics for 

22 each aggregation class. The task also updates the graph and logs 

23 relevant information. If the code is running in debug mode, it plots 

24 the graph using different options. 

25 

26 Args: 

27 graph: The HVAC graph. 

28 

29 Returns: 

30 The updated HVAC graph. 

31 """ 

32 self.logger.info("Reducing elements by applying aggregations") 

33 

34 aggregations_cls = { 

35 'UnderfloorHeating': UnderfloorHeating, 

36 'Consumer': Consumer, 

37 'PipeStrand': PipeStrand, 

38 'ParallelPump': ParallelPump, 

39 'ConsumerHeatingDistributorModule': 

40 ConsumerHeatingDistributorModule, 

41 'GeneratorOneFluid': GeneratorOneFluid, 

42 } 

43 aggregations = [aggregations_cls[agg] for agg in 

44 self.playground.sim_settings.aggregations] 

45 

46 statistics = {} 

47 number_of_elements_before = len(graph.elements) 

48 

49 for agg_class in aggregations: 

50 name = agg_class.__name__ 

51 self.logger.info(f"Aggregating {name} ...") 

52 matches, metas = agg_class.find_matches(graph) 

53 i = 0 

54 for match, meta in zip(matches, metas): 

55 try: 

56 agg = agg_class(graph, match, **meta) 

57 except Exception as ex: 

58 self.logger.exception("Instantiation of '%s' failed", name) 

59 else: 

60 graph.merge( 

61 mapping=agg.get_replacement_mapping(), 

62 inner_connections=agg.inner_connections 

63 ) 

64 i += 1 

65 self.playground.update_graph(graph) 

66 statistics[name] = i 

67 if len(matches) > 0: 

68 self.logger.info( 

69 f"Found {len(matches)} Aggregations of type " 

70 f"{name} and was able to aggregate {i} of them.") 

71 else: 

72 self.logger.info(f"Found non Aggregations of type {name}") 

73 number_of_elements_after = len(graph.elements) 

74 

75 log_str = "Aggregations reduced number of elements from %d to %d:" % \ 

76 (number_of_elements_before, number_of_elements_after) 

77 for aggregation, count in statistics.items(): 

78 log_str += "\n - %s: %d" % (aggregation, count) 

79 self.logger.info(log_str) 

80 

81 if __debug__: 

82 self.logger.info("Plotting graph ...") 

83 graph.plot(self.paths.export) 

84 graph.plot(self.paths.export, ports=True) 

85 graph.plot(self.paths.export, ports=False, use_pyvis=True) 

86 

87 return graph, 

88 

89 @staticmethod 

90 def set_flow_sides(graph: HvacGraph): 

91 """ Set flow sides for ports in HVAC graph based on known flow sides. 

92 

93 This function iteratively sets flow sides for ports in the HVAC graph. 

94 It uses a recursive method (`recurse_set_unknown_sides`) to determine 

95 the flow side for each unset port. The function may prompt the user 

96 for decisions in case of conflicts or unknown sides. 

97 

98 Args: 

99 graph: The HVAC graph. 

100 

101 Yields: 

102 DecisionBunch: A collection of decisions may be yielded during the 

103 task. 

104 """ 

105 # TODO: needs testing! 

106 # TODO: at least one master element required 

107 accepted = [] 

108 while True: 

109 unset_port = None 

110 for port in graph.get_nodes(): 

111 if port.flow_side == 0 and graph.graph[port] \ 

112 and port not in accepted: 

113 unset_port = port 

114 break 

115 if unset_port: 

116 side, visited, masters = graph.recurse_set_unknown_sides( 

117 unset_port) 

118 if side in (-1, 1): 

119 # apply suggestions 

120 for port in visited: 

121 port.flow_side = side 

122 elif side == 0: 

123 # TODO: ask user? 

124 accepted.extend(visited) 

125 elif masters: 

126 # ask user to fix conflicts (and retry in next while loop) 

127 for port in masters: 

128 decision = BoolDecision( 

129 "Use %r as VL (y) or RL (n)?" % port, 

130 global_key= "Use_port_%s" % port.guid) 

131 yield DecisionBunch([decision]) 

132 use = decision.value 

133 if use: 

134 port.flow_side = 1 

135 else: 

136 port.flow_side = -1 

137 else: 

138 # can not be solved (no conflicting masters) 

139 # TODO: ask user? 

140 accepted.extend(visited) 

141 else: 

142 # done 

143 logging.info("Flow_side set") 

144 break