Coverage for backend/pinch_service/OpenPinch/src/analysis/site_analysis.py: 97%

78 statements  

« prev     ^ index     » next       coverage.py v7.10.7, created at 2025-11-06 23:27 +0000

1from typing import Tuple 

2from copy import deepcopy 

3from ..lib import * 

4from ..classes import Zone, Target, Stream, StreamCollection, ProblemTable 

5from .utility_targeting import get_zonal_utility_targets, calc_GGC_utility 

6from .support_methods import key_name 

7from .problem_table_analysis import problem_table_algorithm 

8from .process_analysis import get_process_pinch_targets, get_pinch_temperatures 

9 

10 

11__all__ = ["get_site_targets"] 

12 

13 

14####################################################################################################### 

15# Public API 

16####################################################################################################### 

17 

18def get_site_targets(site: Zone): 

19 """Targets a Total Site by systematically analysing individual zones and then performing TS-level analysis.""" 

20 

21 # Totally integrated analysis 

22 site = get_process_pinch_targets(site) 

23 

24 # Targets process level energy & exergy requirements 

25 z: Zone 

26 for z in site.subzones.values(): 

27 z = get_process_pinch_targets(z) 

28 

29 # Sums zonal targets 

30 site = _calc_total_zonal_targets(site) 

31 

32 # Calculates TS targets based on different approaches 

33 site.import_net_hot_and_cold_streams_from_sub_zones() 

34 site = _calc_site_net_utility_demand(site) 

35 

36 return site 

37 

38 

39####################################################################################################### 

40# Helper Functions 

41####################################################################################################### 

42 

43def _calc_total_zonal_targets(site: Zone) -> Zone: 

44 """Sums and records zonal targets.""" 

45 hot_utility_target = cold_utility_target = heat_recovery_target = 0.0 

46 utility_cost = num_units = area = capital_cost = total_cost = 0.0 

47 

48 hot_utilities = deepcopy(site.hot_utilities) 

49 cold_utilities = deepcopy(site.cold_utilities) 

50 hot_utilities, cold_utilities = _reset_utility_heat_flows(hot_utilities, cold_utilities) 

51 

52 for z in site.subzones.values(): 

53 z: Zone 

54 t: Target = z.targets[f"{z.name}/{TargetType.DI.value}"] 

55 hot_utility_target += t.hot_utility_target 

56 cold_utility_target += t.cold_utility_target 

57 heat_recovery_target += t.heat_recovery_target 

58 utility_cost += t.utility_cost 

59 

60 for j in range(len(hot_utilities)): 

61 hot_utilities[j].set_heat_flow( 

62 hot_utilities[j].heat_flow + t.hot_utilities[j].heat_flow 

63 ) 

64 

65 for j in range(len(cold_utilities)): 

66 cold_utilities[j].set_heat_flow( 

67 cold_utilities[j].heat_flow + t.cold_utilities[j].heat_flow 

68 ) 

69 

70 if area > ZERO: 70 ↛ 71line 70 didn't jump to line 71 because the condition on line 70 was never true

71 num_units += t.num_units 

72 area += t.area 

73 # capital_cost = t.capital_cost 

74 

75 # total_cost += t.total_cost 

76 

77 heat_recovery_limit = site.targets[f"{site.name}/{TargetType.DI.value}"].heat_recovery_limit 

78 

79 # Target co-generation of heat and power 

80 # if config.TURBINE_WORK_BUTTON: 

81 # st = get_power_cogeneration_above_pinch(st) 

82 # utility_cost = utility_cost - work_target / 1000 * config.ELECTRICITY_PRICE * config.ANNUAL_OP_TIME 

83 

84 target_values = _set_sites_targets(hot_utility_target, cold_utility_target, heat_recovery_target, heat_recovery_limit) 

85 site.add_target_from_results(TargetType.TZ.value, { 

86 "target_values": target_values, 

87 "hot_utilities": hot_utilities, 

88 "cold_utilities": cold_utilities, 

89 }) 

90 return site 

91 

92 

93def _reset_utility_heat_flows(hot_utilities: StreamCollection, cold_utilities: StreamCollection) -> Tuple[StreamCollection, StreamCollection]: 

94 for hu in hot_utilities: 

95 hu.heat_flow = 0.0 

96 for cu in cold_utilities: 

97 cu.heat_flow = 0.0 

98 return hot_utilities, cold_utilities 

99 

100 

101def _set_sites_targets(hot_utility_target, cold_utility_target, heat_recovery_target, heat_recovery_limit) -> dict: 

102 """Assign thermal targets and integration degree to the zone based on site analysis methods.""" 

103 return { 

104 "hot_utility_target": hot_utility_target, 

105 "cold_utility_target": cold_utility_target, 

106 "heat_recovery_target": heat_recovery_target, 

107 "heat_recovery_limit": heat_recovery_limit, 

108 "degree_of_int": ( 

109 (heat_recovery_target / heat_recovery_limit) 

110 if heat_recovery_limit > 0 else 1.0 

111 ) 

112 } 

113 

114 

115def _calc_site_net_utility_demand(site: Zone) -> Zone: 

116 # Unified Total Zone Analysis - Amir's method 

117 s_tzt: Target = site.targets[key_name(site.name, TargetType.TZ.value)] 

118 hot_utilities = deepcopy(s_tzt.hot_utilities) 

119 cold_utilities = deepcopy(s_tzt.cold_utilities) 

120 

121 u: Stream 

122 u_h: Stream 

123 u_c: Stream 

124 ut_hr = utility_cost = 0.0 

125 

126 for u_h in hot_utilities: 

127 for u_c in cold_utilities: 

128 if abs(u_h.t_supply - u_c.t_target) < 1 and abs(u_h.t_target - u_c.t_supply) < 1: 

129 Q = min(u_h.heat_flow, u_c.heat_flow) 

130 u_h.set_heat_flow(u_h.heat_flow - Q) 

131 u_c.set_heat_flow(u_c.heat_flow - Q) 

132 ut_hr += Q 

133 

134 for u in hot_utilities + cold_utilities: 

135 utility_cost += u.ut_cost 

136 

137 pt, pt_real, _ = problem_table_algorithm(site.net_hot_streams, site.net_cold_streams, site.all_net_streams, site.config) 

138 pt, pt_real, hot_utilities, cold_utilities = get_zonal_utility_targets(pt, pt_real, hot_utilities, cold_utilities) 

139 

140 pt = calc_GGC_utility(pt, hot_utilities, cold_utilities, shifted=True) 

141 pt_real = calc_GGC_utility(pt_real, hot_utilities, cold_utilities, shifted=False) 

142 

143 hot_utility_target = pt.loc[0, PT.H_UT_NET.value] 

144 cold_utility_target = pt.loc[-1, PT.H_UT_NET.value] 

145 heat_recovery_target = s_tzt.heat_recovery_target + (s_tzt.hot_utility_target - hot_utility_target) 

146 hot_pinch, cold_pinch = get_pinch_temperatures(pt, col_H=PT.H_UT_NET.value) 

147 

148 # if config.TURBINE_WORK_BUTTON: 

149 # work_target = 0.0 

150 # if config.ABOVE_PINCH_CHECKBOX: 

151 # pass 

152 # # s_tsi = get_power_cogeneration_above_pinch(s_tsi) 

153 # utility_cost = utility_cost - work_target / 1000 * config.ELECTRICITY_PRICE * config.ANNUAL_OP_TIME 

154 

155 graphs = _save_graph_data(pt, pt_real) 

156 

157 target_values = _set_sites_targets(hot_utility_target, cold_utility_target, heat_recovery_target, s_tzt.heat_recovery_limit) 

158 site.add_target_from_results(TargetType.TS.value, { 

159 "pt": pt, 

160 "pt_real": pt_real, 

161 "target_values": target_values, 

162 "graphs": graphs, 

163 "hot_utilities": hot_utilities, 

164 "cold_utilities": cold_utilities, 

165 "hot_pinch": hot_pinch, 

166 "cold_pinch": cold_pinch, 

167 }) 

168 return site 

169 

170 

171def _save_graph_data(pt: ProblemTable, pt_real: ProblemTable) -> Zone: 

172 pt.round(decimals=4) 

173 pt_real.round(decimals=4) 

174 return { 

175 GT.TSP.value: pt[[PT.T.value, PT.H_HOT_NET.value, PT.H_COLD_NET.value, PT.H_HOT_UT.value, PT.H_COLD_UT.value]], 

176 GT.SUGCC.value: pt[[PT.T.value, PT.H_UT_NET.value]], 

177 }