Coverage for bim2sim/tasks/hvac/expansiontanks.py: 78%

1from bim2sim.kernel.decision import BoolDecision, DecisionBunch 

2from bim2sim.elements.hvac_elements import Storage 

3from bim2sim.elements.graphs.hvac_graph import HvacGraph 

4from bim2sim.tasks.base import ITask 

5from bim2sim.tasks.base import Playground 

6 

7 

8class ExpansionTanks(ITask): 

9 

10 reads = ('graph',) 

11 touches = ('graph',) 

12 

13 def run(self, graph: HvacGraph, force: bool = False) -> HvacGraph: 

14 """Analyses graph network for expansion tanks and removes them. 

15 

16 This task performs the following steps: 

17 1. Identifies potential expansion tanks in the HVAC graph. 

18 2. Logs the number of identified potential expansion tanks. 

19 3. Prompts and yields decisions regarding the removal of expansion 

20 tanks and updates the graph accordingly. 

21 4. Logs the number of elements removed because they were identified as 

22 expansion tanks. 

23 

24 Args: 

25 graph: The HVAC graph containing elements and ports. 

26 force: If True, forcefully remove expansion tanks without 

27 confirmation. Defaults to False. 

28 

29 Returns: 

30 The updated HVAC graph after handling expansion tanks. 

31 """ 

32 self.logger.info("Inspecting for expansion tanks") 

33 playground = self.playground 

34 potential_expansion_tanks = self.identify_expansion_tanks(graph) 

35 self.logger.info( 

36 f"Found {potential_expansion_tanks} " 

37 f"potential expansion tanks in network.") 

38 graph, n_removed = yield from self.decide_expansion_tanks( 

39 graph, potential_expansion_tanks, playground, force) 

40 self.logger.info( 

41 f"Removed {n_removed} elements because they were expansion tanks.") 

42 return graph, 

43 

44 @staticmethod 

45 def identify_expansion_tanks(graph: HvacGraph) -> set: 

46 """Identify potential expansion tanks in the HVAC graph. 

47 

48 Expansion tanks are all tanks with only one port. 

49 

50 Args: 

51 graph: HVAC graph to be analysed. 

52 

53 Returns: 

54 set of potential expansion tanks. 

55 """ 

56 element_graph = graph.element_graph 

57 potential_expansion_tanks = {node for node in element_graph.nodes 

58 if (isinstance(node, Storage) and 

59 len(node.neighbors) < 2)} 

60 return potential_expansion_tanks 

61 

62 @staticmethod 

63 def decide_expansion_tanks( 

64 graph: HvacGraph, 

65 potential_expansion_tanks: set, 

66 playground: Playground = None, 

67 force: bool = False) -> [HvacGraph, int]: 

68 """Decide and handle the removal of potential expansion tanks. 

69 

70 This method evaluates potential expansion tanks and prompts the user 

71 for decisions on removal. If force is True, expansion tanks are 

72 removed without confirmation. If playground is provided, the graph is 

73 updated and visualized during the process. 

74 

75 Args: 

76 graph: HVAC graph where the expansion tank should be removed. 

77 potential_expansion_tanks: set of potential expansion tank elements 

78 to be evaluated. 

79 playground: BIM2SIM Playground instance. Defaults to None. 

80 force: If True, forcefully remove expansion tanks without 

81 confirmation. Defaults to False. 

82 

83 Yields: 

84 Decision bunch with BoolDecisions for confirming expansion tank 

85 removal. 

86 

87 Returns: 

88 Tuple containing the updated HVAC graph where expansions tanks 

89 are removed and the number of removed expansion tanks. 

90 """ 

91 if force: 

92 n_removed = len(potential_expansion_tanks) 

93 remove_ports = [ 

94 port for pot_tank in potential_expansion_tanks 

95 for port in pot_tank.ports] 

96 graph.remove_nodes_from(remove_ports) 

97 if playground: 

98 playground.update_graph(graph) 

99 else: 

100 decisions = DecisionBunch() 

101 for tank in potential_expansion_tanks: 

102 cur_decision = BoolDecision( 

103 f"Found {tank} which is a possible expansion tank and" 

104 f" therefore should be deleted", 

105 key=tank, 

106 global_key="expansionTank.%s" % tank.guid, 

107 allow_skip=True, 

108 related={tank.guid}, 

109 ) 

110 decisions.append(cur_decision) 

111 yield decisions 

112 answers = decisions.to_answer_dict() 

113 n_removed = 0 

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

115 if answer: 

116 remove = element.ports 

117 n_removed += 1 

118 graph.remove_nodes_from(remove) 

119 if playground: 

120 playground.update_graph(graph) 

121 else: 

122 raise NotImplementedError() 

123 # TODO: handle real storages 

124 # maybe add ports to the storage and select the related 

125 # dead end, with which the new storage port should be 

126 # connected 

127 return graph, n_removed