Coverage for backend/django/core/auxiliary/views/ExtractSegmentDataFromFS.py: 87%
133 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1import traceback
2from typing import Union, Tuple
3from rest_framework.decorators import api_view
4from rest_framework.response import Response
5from drf_spectacular.utils import extend_schema
6from rest_framework import serializers, status
7from idaes_factory.unit_conversion.unit_conversion import convert_value
8from core.validation import api_view_validate
9from core.managers import get_flowsheet_access
11from PinchAnalysis.models.InputModels import Segment, StreamDataEntry
12from flowsheetInternals.unitops.services.edit_operations.recorder import (
13 tracked_bulk_create,
14)
15from PinchAnalysis.models.HenNode import HenNode
16from PinchAnalysis.models.StreamDataProject import StreamDataProject
17from PinchAnalysis.views.henNodeHelpers import group_stream_by_unitop_type, set_hennode_connections
18from flowsheetInternals.unitops.models.SimulationObject import SimulationObject
19from core.auxiliary.enums.pinchEnums import StreamType
20from core.auxiliary.enums import SimulationObjectClass
21from pinch_factory.pinch_factory import PinchFactory
22from django.db.models import Q
24DECIMAL_PLACES = 3
25VARIANCE = 0.01
28def get_compounds(stream, include_null: bool = False) -> set[tuple[str, float]]:
29 """
30 Returns a set of tuples containing the index (key) and value of all property
31 value objects in the stream's mole_frac_comp. Used for composition comparisons.
33 Returns:
34 - set[tuple[str, float]]: A set of (key, value) tuples.
35 """
36 mole_frac_comp = stream.properties.get_property("mole_frac_comp")
37 property_values = mole_frac_comp.values.all()
38 result = [
39 (prop.get_index("compound").key, prop.value)
40 for prop in property_values
41 if (prop.value not in [None, ""] or include_null)
42 ]
43 return result
46def create_he_streams(sim_obj, group) -> None:
47 streamDataProject = group.flowsheet_state.StreamDataProject
48 stream_ls: list[StreamDataEntry] = []
50 for key in sim_obj.schema.propertyPackagePorts.keys():
51 if key != "__none__": 51 ↛ 50line 51 didn't jump to line 50 because the condition on line 51 was always true
52 stream_ls.append(
53 StreamDataEntry(
54 flowsheet_state=group.flowsheet_state,
55 streamDataProject=streamDataProject,
56 unitop=sim_obj,
57 group=group,
58 property_package_mapping=key,
59 )
60 )
61 tracked_bulk_create(StreamDataEntry.objects, stream_ls)
64def compare_compositions(stream_1: SimulationObject, stream_2: SimulationObject) -> bool:
65 """
66 Compares the composition of two streams
67 Returns: boolean indicating equality - True if compositions are equal
68 """
69 stream_1_set = get_compounds(stream_1)
70 stream_2_set = get_compounds(stream_2)
71 return stream_1_set == stream_2_set
73# This needs to be revised and maybe moved elsewhere. The dT is WRONG. also it needs to be ln(dT).
74# specifically, we need dT of cold segment, and dT of hot segment (pairs from exchanger(?), and we dont have that ehre.)
75def _calc_area(htc: float, q_kw: float, t_supply_c: float, t_target_c: float) -> float:
76 U = float(htc or 0)
77 Q = float(q_kw or 0)
78 dT = abs(float(t_supply_c or 0) - float(t_target_c or 0))
79 if U <= 0 or dT <= 0: 79 ↛ 80line 79 didn't jump to line 80 because the condition on line 79 was never true
80 return 0.0
81 return Q / (U * dT)
84def _get_io_stream_properties(streamDataEntry: StreamDataEntry, prop_arg: str, tar_unit: str) -> Tuple[float,float]:
85 i, o = getattr(streamDataEntry, prop_arg)
86 i_val = i.get_value()
87 o_val = o.get_value()
88 if i_val is None or o_val is None: 88 ↛ 89line 88 didn't jump to line 89 because the condition on line 88 was never true
89 return None, None
90 else:
91 supply = convert_value(i_val, i.unit, tar_unit)
92 target = convert_value(o_val, o.unit, tar_unit)
93 return supply, target
96def _get_stream_type(streamDataEntry: StreamDataEntry):
97 if streamDataEntry.unitop.objectType == SimulationObjectClass.HeatExchanger:
98 if streamDataEntry.property_package_mapping == "Cold Side":
99 stream_type = StreamType.Cold.value
100 else:
101 stream_type = StreamType.Hot.value
102 else:
103 if streamDataEntry.unitop.objectType == SimulationObjectClass.Heater:
104 stream_type = StreamType.Cold.value
105 else:
106 stream_type = StreamType.Hot.value
107 return stream_type
110def _get_terminal_states(streamDataEntry: StreamDataEntry) -> dict:
111 inlet_stream, _ = streamDataEntry.inlet_outlet_stream
112 if not inlet_stream: 112 ↛ 113line 112 didn't jump to line 113 because the condition on line 112 was never true
113 return None
115 mole_flow = convert_value(
116 inlet_stream.properties.get_property("flow_mol").get_value(),
117 inlet_stream.properties.get_property("flow_mol").unit,
118 "mol/s",
119 )
120 service_T_supply, service_T_target = _get_io_stream_properties(streamDataEntry, "temperatures", "degK")
121 service_P_supply, service_P_target = _get_io_stream_properties(streamDataEntry, "pressures", "Pa")
122 service_H_supply, service_H_target = _get_io_stream_properties(streamDataEntry, "enthalpies", "J/mol")
124 if service_T_supply == None or service_P_supply == None or service_H_target == None: 124 ↛ 125line 124 didn't jump to line 125 because the condition on line 124 was never true
125 return None
126 else:
127 return {
128 "streams_io_props": [{
129 "t_supply": service_T_supply,
130 "t_target": service_T_target,
131 "p_supply": service_P_supply,
132 "p_target": service_P_target,
133 "h_supply": service_H_supply,
134 "h_target": service_H_target,
135 "composition": get_compounds(inlet_stream),
136 }],
137 "ppKey": streamDataEntry.unitop.get_property_package(),
138 "mole_flow": mole_flow,
139 "comp_name": inlet_stream.componentName,
140 "stream_type": _get_stream_type(streamDataEntry),
141 "streamDataEntry": streamDataEntry,
142 }
145def _stream_segment_creator(comp_name: str, points: list, stream_type: str, streamDataEntry: StreamDataEntry, **_):
146 # Create segments after linearization
147 segments = []
148 for index in range(len(points) - 1):
149 t_supply = convert_value(points[index][1], "degK", "degC")
150 t_target = check_target_temperature_validity(
151 t_supply,
152 convert_value(points[index + 1][1], "degK", "degC"),
153 stream_type
154 )
155 heat_flow = abs(
156 convert_value(points[index + 1][0] - points[index][0], "W", "kW")
157 )
158 htc=1
159 segments.append(
160 Segment(
161 stream_data_entry=streamDataEntry,
162 name=f"{comp_name} ({index + 1})" if len(points) > 1 else comp_name,
163 t_supply=t_supply,
164 t_target=t_target,
165 heat_flow=heat_flow,
166 htc=htc,
167 area=_calc_area(htc,heat_flow, t_supply, t_target),
168 flowsheet_state=streamDataEntry.flowsheet_state,
169 )
170 )
171 return segments
174@api_view_validate
175@api_view(['POST'])
176def extract_stream_data(request) -> Response:
177 try:
178 flowsheet_id = request.GET.get("flowsheet")
179 access_state = get_flowsheet_access(request.user, flowsheet_id)
180 if access_state.has_read_access and not access_state.has_write_access:
181 return Response(
182 {"error": "This flowsheet is shared with read-only access."},
183 status=status.HTTP_403_FORBIDDEN,
184 )
185 factory = PinchFactory(flowsheet_id)
187 streamDataEntries = StreamDataEntry.objects.filter(
188 flowsheet_state__flowsheet_id=flowsheet_id,
189 custom=False,
190 )
192 segments = []
193 # i'll just delete hennodes and create new ones for now
194 HenNode.objects.filter(
195 flowsheet_state__flowsheet_id=flowsheet_id
196 ).delete()
198 for streamDataEntry in streamDataEntries:
199 streamDataEntry.Segments.all().delete()
200 terminal_data = _get_terminal_states(streamDataEntry)
201 if terminal_data is None: 201 ↛ 202line 201 didn't jump to line 202 because the condition on line 201 was never true
202 continue
203 terminal_data["prev_states"] = streamDataEntry.states if hasattr(streamDataEntry, "states") else None
204 t_h_data = factory.run_get_t_h_data(**terminal_data)
206 streamDataEntry.t_h_data = t_h_data["curve_points"]
207 streamDataEntry.states = t_h_data["states"]
208 streamDataEntry.save()
210 linearised_points = factory.run_linearize(t_h_data["curve_points"], **terminal_data)
211 new_segments = _stream_segment_creator(points=linearised_points, **terminal_data)
212 segments.extend(new_segments)
214 tracked_bulk_create(Segment.objects, segments)
216 # create hennodes
217 created_segments = list(Segment.objects.filter(id__in=[s.id for s in segments]))
218 sdes = {seg.stream_data_entry for seg in created_segments if seg.stream_data_entry}
220 processed_stream_ids = set(HenNode.objects.filter(
221 Q(stream_data_entry__in=sdes) |
222 Q(hot_connection__in=sdes) |
223 Q(cold_connection__in=sdes)
224 ).values_list('stream_data_entry_id', flat=True))
226 hennodes_to_create = []
228 # group streams by unitop type
229 grouped_by_unitop = group_stream_by_unitop_type(sdes, processed_stream_ids, hennodes_to_create)
231 # set connections for the hennodes
232 set_hennode_connections(grouped_by_unitop, processed_stream_ids, hennodes_to_create)
234 HenNode.objects.bulk_create(hennodes_to_create)
236 # link segments to hennodes
237 for segment in created_segments:
238 sde = segment.stream_data_entry
239 if sde: 239 ↛ 237line 239 didn't jump to line 237 because the condition on line 239 was always true
240 hn = HenNode.objects.filter(
241 Q(stream_data_entry=sde) |
242 Q(hot_connection=sde) |
243 Q(cold_connection=sde)
244 ).first()
245 if hn: 245 ↛ 237line 245 didn't jump to line 237 because the condition on line 245 was always true
246 segment.hen_node = hn
247 segment.save(update_fields=['hen_node'])
249 return Response(status=200)
251 except Exception as e:
252 tb_info = traceback.format_exc()
253 print(tb_info)
254 error_message = str(e)
255 response_data = {
256 "status": "error",
257 "message": error_message,
258 "traceback": tb_info
259 }
260 return Response(response_data, status=500)
262def check_target_temperature_validity(t_supply: float, t_target: float, stream_type: StreamType, min_delta_t: float = 0.0001) -> tuple[float, float]:
263 if abs(t_supply - t_target) < min_delta_t: 263 ↛ 264line 263 didn't jump to line 264 because the condition on line 263 was never true
264 t_target = t_supply - min_delta_t if stream_type == StreamType.Hot.value else t_supply + min_delta_t
265 return t_target