Coverage for bim2sim/tasks/hvac/dead_ends.py: 79%

1from bim2sim.kernel.decision import BoolDecision, DecisionBunch 

2from bim2sim.elements.graphs.hvac_graph import HvacGraph 

3from bim2sim.tasks.base import ITask 

4from bim2sim.tasks.base import Playground 

5 

6 

7class DeadEnds(ITask): 

8 

9 reads = ('graph',) 

10 touches = ('graph',) 

11 

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

13 """Analyses graph for dead ends and removes ports due to dead ends. 

14 

15 This task performs the following steps: 

16 1. Identifies potential dead ends in the HVAC graph. 

17 2. Logs the number of identified potential dead ends. 

18 3. Prompts and yields decisions regarding the removal of dead ends and 

19 updates the graph accordingly. 

20 4. Logs the number of ports removed due to dead ends. 

21 5. Optionally, plots the HVAC graph in debug mode. 

22 

23 Args: 

24 graph: HVAC graph containing elements and ports. 

25 

26 Returns: 

27 Updated HVAC graph after handling dead ends. 

28 """ 

29 self.logger.info("Inspecting for dead ends") 

30 pot_dead_ends = self.identify_dead_ends(graph) 

31 self.logger.info("Found %s possible dead ends in network." 

32 % len(pot_dead_ends)) 

33 graph, n_removed = yield from self.decide_dead_ends( 

34 graph, pot_dead_ends, False) 

35 self.logger.info("Removed %s ports due to found dead ends." % n_removed) 

36 if __debug__: 

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

38 graph.plot(self.paths.export) 

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

40 return graph, 

41 

42 @staticmethod 

43 def identify_dead_ends(graph: HvacGraph) -> list: 

44 """Identify potential dead ends in graph. 

45 

46 This method identifies potential dead end ports in the HVAC graph by 

47 considering two cases: 

48 1. Ports that are not connected to any other port. 

49 2. Ports that are connected to only one other port. 

50 

51 Args: 

52 graph: HVAC graph to be analysed. 

53 

54 Returns: 

55 pot_dead_ends: List of potential dead ends. 

56 """ 

57 

58 uncoupled_graph = graph.copy() 

59 element_graph = uncoupled_graph.element_graph 

60 for node in element_graph.nodes: 

61 inner_edges = node.inner_connections 

62 uncoupled_graph.remove_edges_from(inner_edges) 

63 # find first class dead ends (ports which are not connected to any other 

64 # port) 

65 pot_dead_ends_1 = [v for v, d in uncoupled_graph.degree() if d == 0] 

66 # find second class dead ends (ports which are connected to one side 

67 # only) 

68 pot_dead_ends_2 = [v for v, d in graph.degree() if d == 1] 

69 pot_dead_ends = list(set(pot_dead_ends_1 + pot_dead_ends_2)) 

70 return pot_dead_ends 

71 

72 @staticmethod 

73 def decide_dead_ends(graph: HvacGraph, pot_dead_ends: list, 

74 playground: Playground = None, 

75 force: bool = False) -> [{HvacGraph}, int]: 

76 """Decide if potential dead ends are consumers or dead ends. 

77 

78 This method evaluates potential dead end ports and prompts the user for 

79 decisions on removal. If force is True, dead ends are removed 

80 without confirmation. If playground is provided, the graph is updated 

81 and visualized during the process. 

82 

83 Args: 

84 graph: HVAC graph to be analysed. 

85 pot_dead_ends: List of potential dead end ports to be evaluated. 

86 playground: BIM2SIM Playground instance. Defaults to None. 

87 force: If True, then all potential dead ends are removed without 

88 confirmation. Defaults to False. 

89 

90 Yields: 

91 Decision bunch with BoolDecisions for confirming dead ends. 

92 

93 Returns: 

94 Tuple containing the updated HVAC graph where dead ends are 

95 removed and the number of removed ports. 

96 """ 

97 n_removed = 0 

98 remove_ports = {} 

99 for dead_end in pot_dead_ends: 

100 if len(dead_end.parent.ports) > 2: 

101 # dead end at > 2 ports -> remove port but keep element 

102 remove_ports[dead_end] = ([dead_end], [dead_end.parent]) 

103 continue 

104 else: 

105 # TODO: how to handle devices where we might want to connect 

106 # dead ends instead delete 

107 remove_ports_strand = [] 

108 remove_elements_strand = [] 

109 # find if there are more elements in strand to be removed 

110 strand_ports = HvacGraph.get_path_without_junctions( 

111 graph, dead_end, include_edges=True) 

112 strand = graph.subgraph(strand_ports).element_graph 

113 for port in strand_ports: 

114 remove_ports_strand.append(port) 

115 for element in strand: 

116 remove_elements_strand.append(element) 

117 remove_ports[dead_end] = (remove_ports_strand, 

118 remove_elements_strand) 

119 

120 if force: 

121 for dead_end in pot_dead_ends: 

122 remove = remove_ports[dead_end][0] 

123 n_removed += len(set(remove)) 

124 graph.remove_nodes_from([n for n in graph if n in set(remove)]) 

125 

126 else: 

127 decisions = DecisionBunch() 

128 for dead_end, (port_strand, element_strand) in remove_ports.items(): 

129 dead_end_port = dead_end 

130 if hasattr(dead_end_port, "originals"): 

131 while hasattr(dead_end_port, "originals"): 

132 related_guid = dead_end_port.originals[0].parent.guid 

133 dead_end_port = dead_end_port.originals[0] 

134 else: 

135 related_guid = dead_end.parent.guid 

136 cur_decision = BoolDecision( 

137 question="Found possible dead end at port %s in system, " 

138 "please check if it is a dead end" % dead_end, 

139 console_identifier="GUID: %s" % dead_end.guid, 

140 key=dead_end, 

141 global_key="deadEnd.%s" % dead_end.guid, 

142 allow_skip=False, 

143 related={related_guid}, context=set( 

144 element.guid for element in element_strand)) 

145 decisions.append(cur_decision) 

146 yield decisions 

147 answers = decisions.to_answer_dict() 

148 n_removed = 0 

149 for element, answer in answers.items(): 

150 if answer: 

151 remove = remove_ports[element][0] 

152 n_removed += len(set(remove)) 

153 graph.remove_nodes_from( 

154 [n for n in graph if n in set(remove)]) 

155 if playground: 

156 playground.update_graph(graph) 

157 else: 

158 raise NotImplementedError() 

159 # TODO connect elements that are open but not dead ends 

160 # (missing connections) #769 

161 # TODO: handle consumers 

162 # dead end identification with guid decision 

163 # (see issue97 add_gui_decision) 

164 # build clusters with position for the rest of open ports 

165 # decision to to group these open ports to consumers 

166 # delete the rest of open ports afterwards 

167 return graph, n_removed