Coverage for backend/django/pinch_factory/pinch_factory.py: 85%
159 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1import json
2import os
3import traceback
4from typing import Any, Dict
5import requests
7from PinchAnalysis.models.OutputModels import GraphDataPoint, HeatReceiverUtilitySummary, HeatSupplierUtilitySummary, PinchCurve, PinchGraph, PinchGraphSet, PinchTemp, TargetSummary
8from PinchAnalysis.models.StreamDataProject import StreamDataProject
9from PinchAnalysis.models.InputModels import PinchUtility, Segment
10from core.auxiliary.enums.generalEnums import AbstractionType
11from core.auxiliary.enums import pinchEnums
12from PinchAnalysis.serializers.PinchInputSerializers import SegmentSerializer
13from core.auxiliary.models.Flowsheet import Flowsheet
15class PinchFactory:
16 def __init__(self, flowsheet_id: int, num_intervals: int = 20, t_min: float = 1) -> None:
17 self.project = Flowsheet.objects.get(pk=flowsheet_id).current_state.StreamDataProject
18 self.flowsheet_state = self.project.flowsheet_state
20 # Sampling intervals for stream linearisation
21 self.num_intervals = num_intervals
22 # Maximum temperature difference between actual stream samples and linearisation curve
23 self.t_min = t_min
25 def build_calculate_request(self, excluded_segments: list[int]) -> dict:
26 """
27 Extract and prepare inputs from the project data.
28 """
29 request_data = {}
30 request_data['streams'] = []
31 request_data['utilities'] = []
32 for stream_data_entry in self.project.StreamDataEntries.all():
33 for segment in stream_data_entry.Segments.all():
34 if segment.id in excluded_segments:
35 continue
36 request_data['streams'].append({
37 "zone": segment.zone,
38 "name": segment.name,
39 "t_supply": segment.t_supply,
40 "t_target": segment.t_target,
41 "heat_flow": segment.heat_flow,
42 "dt_cont": segment.dt_cont,
43 "htc": segment.htc,
44 })
45 for utility in self.project.Inputs.PinchUtilities.all():
46 request_data['utilities'].append({
47 "name": utility.name,
48 "type": utility.type,
49 "t_supply": utility.t_supply,
50 "t_target": utility.t_target,
51 "heat_flow": utility.heat_flow,
52 "dt_cont": utility.dt_cont,
53 "htc": utility.htc,
54 "price": utility.price,
55 })
57 request_data['options'] = {
58 'main': [prop.key for prop in self.project.Options.selections.containedProperties.all() if prop.get_value() is True],
59 'turbine': [{"key": prop.key, "value": prop.get_value()} for prop in self.project.Options.turbine_options.properties.containedProperties.all()]
60 }
61 request_data['zone_tree'] = self.build_zone_structure()
63 return request_data
65 def build_zone_structure(self) -> list:
66 # Step 1: Collect all groups related to the StreamDataEntries
67 all_groups = {}
68 for stream_data_entry in self.project.StreamDataEntries.all():
69 group = stream_data_entry.group
70 zone = stream_data_entry.zone
71 all_groups[zone] = {
72 "children": [],
73 "group": group,
74 }
75 # Step 2: Organize into tree
76 root_node = None
77 parent_zones = {}
78 for zone_data in all_groups.values():
79 group = zone_data["group"]
80 parent_group = group.get_parent_group()
81 if parent_group:
82 parent_zone = parent_group.simulationObject.componentName
83 if parent_zone in all_groups: 83 ↛ 86line 83 didn't jump to line 86 because the condition on line 83 was always true
84 all_groups[parent_zone]["children"].append(zone_data)
85 else:
86 if parent_zone not in parent_zones:
87 parent_zones[parent_zone] = {
88 "children": [],
89 "group": parent_group,
90 }
91 parent_zones[parent_zone]["children"].append(zone_data)
92 for parent_zone_name, parent_zone_data in parent_zones.items(): 92 ↛ 93line 92 didn't jump to line 93 because the loop on line 92 never started
93 all_groups[parent_zone_name] = parent_zone_data
94 for zone_data in all_groups.values():
95 group = zone_data["group"]
96 parent_group = group.get_parent_group()
97 if not parent_group:
98 root_node = zone_data
100 # Step 3: Format it nicely
101 def clean_node(node):
102 return {
103 "name": node["group"].simulationObject.componentName,
104 "type": node["group"].abstractionType,
105 "children": [clean_node(child) for child in node["children"]],
106 }
108 if root_node is None: 108 ↛ 109line 108 didn't jump to line 109 because the condition on line 108 was never true
109 return None
110 return clean_node(root_node)
112 def build_linearize_request(
113 self,
114 t_h_data,
115 streams_io_props: list[dict],
116 ) -> dict:
117 """Build the request accepted by OpenPinch's linearization endpoint."""
119 return {
120 't_h_data': t_h_data,
121 't_min': self.t_min,
122 'streams': streams_io_props,
123 }
125 def build_t_h_request(self, streams_io_props: list[dict], mole_flow: float, ppKey: str, prev_states = None) -> dict:
126 return (
127 {
128 't_min': self.t_min,
129 'num_intervals': self.num_intervals,
130 'streams': streams_io_props,
131 'mole_flow': mole_flow,
132 'ppKey': ppKey,
133 'prev_states': prev_states,
134 }
135 )
137 def clear_outputs(self) -> None:
138 """
139 Removes the previous outputs from the project
140 """
141 output_owner = self.project.Outputs
142 output_owner.targets.all().delete()
143 output_owner.graph_sets.all().delete()
145 def run_calculate(self, excluded_segments: list[int]) -> Dict[str, Any]:
146 """
147 Format data and send request to pinch service
148 """
149 try:
150 request_data = self.build_calculate_request(excluded_segments)
151 # print(request_data)
152 url = (os.getenv('PINCH_SERVICE_URL') or "http://localhost:8082") + "/" + "calculate"
153 result = requests.post(url, json=request_data)
154 if result.status_code != 200: 154 ↛ 155line 154 didn't jump to line 155 because the condition on line 154 was never true
155 raise Exception(result.json())
156 self.clear_outputs()
157 response_data = result.json()
158 self.serialize_return_data(response_data)
160 except Exception as e:
161 print("Error during calculation:", e)
162 print("Traceback:", traceback.format_exc())
163 raise RuntimeError("Calculation error occurred.") from e
165 def run_linearize(self, t_h_data, streams_io_props: list[dict], **_):
166 """
167 Linearizes a stream curve
168 """
169 try:
170 url = (os.getenv('PINCH_SERVICE_URL') or "http://localhost:8082") + "/" + "linearize"
171 request_data = self.build_linearize_request(t_h_data, streams_io_props)
172 # print(request_data)
173 result = requests.post(url, json=request_data)
174 if result.status_code != 200: 174 ↛ 175line 174 didn't jump to line 175 because the condition on line 174 was never true
175 raise Exception(result.json())
176 response_data = result.json()
177 return self.get_linear_streams(response_data)
178 except Exception as e:
179 print("Error during calculation:", e)
180 print("Traceback:", traceback.format_exc())
181 raise RuntimeError("Calculation error occurred.") from e
183 def serialize_return_data(self, response_data):
184 """
185 Converts output data to db entries, including nested objects.
186 """
187 try:
188 output_owner = self.project.Outputs
189 targets = response_data.get('targets', None)
190 graphs = response_data.get('graphs', None)
192 # Target Objects
193 if targets: 193 ↛ 239line 193 didn't jump to line 239 because the condition on line 193 was always true
194 target_list = []
195 heat_suppliers = []
196 heat_receivers = []
197 for entry in targets:
198 # Pop data that should not be included in the target creation
199 temp_pinch_data = entry.pop('temp_pinch', None)
200 hot_utilities = entry.pop('hot_utilities', [])
201 cold_utilities = entry.pop('cold_utilities', [])
203 # print('temp_pinch_data', temp_pinch_data)
205 temp_pinch = PinchTemp.objects.create(**temp_pinch_data, flowsheet_state=self.flowsheet_state) if temp_pinch_data else None
207 # Create TargetSummary
208 target = TargetSummary(
209 output_owner=output_owner,
210 temp_pinch=temp_pinch,
211 **entry,
212 flowsheet_state=self.flowsheet_state
213 )
214 # print(target)
215 target.save()
217 for supplier_utility in hot_utilities:
218 heat_suppliers.append(HeatSupplierUtilitySummary(
219 summary_owner=target,
220 **supplier_utility,
221 flowsheet_state=self.flowsheet_state
222 ))
224 for receiver_utility in cold_utilities:
225 heat_receivers.append(HeatReceiverUtilitySummary(
226 summary_owner=target,
227 **receiver_utility,
228 flowsheet_state=self.flowsheet_state
229 ))
231 target_list.append(target)
233 # Bulk create objects
234 TargetSummary.objects.bulk_create(target_list, ignore_conflicts=True)
235 HeatSupplierUtilitySummary.objects.bulk_create(heat_suppliers)
236 HeatReceiverUtilitySummary.objects.bulk_create(heat_receivers)
238 # Graphs
239 if graphs: 239 ↛ exitline 239 didn't return from function 'serialize_return_data' because the condition on line 239 was always true
240 graph_set_list = []
241 graph_list = []
242 curve_list = []
243 data_point_list = []
245 for key, graph_set_data in graphs.items():
246 graph_set = PinchGraphSet(output_owner=output_owner, name=graph_set_data.get('name'), flowsheet_state=self.flowsheet_state)
247 graph_set_list.append(graph_set)
249 # Create nested graphs
250 for graph_data in graph_set_data.get('graphs', []):
251 graph = PinchGraph(
252 graph_set=graph_set,
253 name=graph_data.get('name'),
254 type=graph_data.get('type', pinchEnums.GraphType.CC),
255 flowsheet_state=self.flowsheet_state
256 )
257 graph_list.append(graph)
259 # Create nested curves
260 for segment_data in graph_data.get('segments', []):
261 curve = PinchCurve(
262 graph=graph,
263 title=segment_data.get('title'),
264 colour=segment_data.get('colour', pinchEnums.LineColour.Hot),
265 arrow=segment_data.get('arrow', pinchEnums.ArrowHead.NO_ARROW),
266 flowsheet_state=self.flowsheet_state
267 )
268 curve_list.append(curve)
270 for point in segment_data.get('data_points', []):
271 data_point = GraphDataPoint(
272 curve=curve,
273 x=point.get('x'),
274 y=point.get('y'),
275 flowsheet_state=self.flowsheet_state
276 )
277 data_point_list.append(data_point)
279 # Bulk create the objects
280 PinchGraphSet.objects.bulk_create(graph_set_list)
281 PinchGraph.objects.bulk_create(graph_list)
282 PinchCurve.objects.bulk_create(curve_list)
283 GraphDataPoint.objects.bulk_create(data_point_list)
285 except Exception as e:
286 raise RuntimeError("Serialization error occurred.") from e
288 def get_linear_streams(self, response_data):
289 return response_data['streams']
291 def run_get_t_h_data(self, streams_io_props: list[dict], mole_flow: float, ppKey: str, prev_states = None, **_):
292 """
293 Get t_h data from streams
294 """
295 try:
296 url = (os.getenv('PINCH_SERVICE_URL') or "http://localhost:8082") + "/" + "generate_t_h_curve"
297 request_data = self.build_t_h_request(streams_io_props, mole_flow, ppKey, prev_states)
298 result = requests.post(url, json=request_data)
299 if result.status_code != 200: 299 ↛ 300line 299 didn't jump to line 300 because the condition on line 299 was never true
300 raise Exception(result.json())
301 response_data = result.json()
302 return response_data
303 except Exception as e:
304 print("Error during calculation:", e)
305 print("Traceback:", traceback.format_exc())
306 raise RuntimeError("Calculation error occurred.") from e