Coverage for backend/django/PinchAnalysis/views/henNodeHelpers.py: 82%

1from PinchAnalysis.models.InputModels import StreamDataEntry 

2from PinchAnalysis.models.HenNode import HenNode 

3 

4def is_hot(stream: StreamDataEntry) -> bool: 

5 segments = stream.Segments.all() 

6 return segments.exists() and segments.first().t_supply > segments.first().t_target 

7 

8def is_cold(stream: StreamDataEntry) -> bool: 

9 segments = stream.Segments.all() 

10 return segments.exists() and segments.first().t_supply < segments.first().t_target 

11 

12def group_stream_by_unitop_type(streams, processed_stream_ids, hennodes_to_create) -> dict: 

13 # group streams by unitop type 

14 grouped_by_unitop = {} 

15 for stream in streams: 

16 if stream.id in processed_stream_ids: 

17 continue # already has a HenNode 

18 

19 display_type = getattr(getattr(stream.unitop, "schema", None), "displayType", None) 

20 if display_type in ["Heat Exchanger", "Heater", "Cooler"]: 20 ↛ 22line 20 didn't jump to line 22 because the condition on line 20 was always true

21 grouped_by_unitop.setdefault(display_type, {}).setdefault(stream.unitop.id, []).append(stream) 

22 elif not stream.unitop: 

23 # stream without unitop 

24 hennodes_to_create.append(HenNode( 

25 flowsheet=stream.flowsheet, 

26 stream_data_entry=stream, 

27 hot_connection=None, 

28 cold_connection=None 

29 )) 

30 processed_stream_ids.add(stream.id) 

31 

32 return grouped_by_unitop 

33 

34def set_hennode_connections(grouped_by_unitop, processed_stream_ids, hennodes_to_create): 

35 for unitop_type, groups in grouped_by_unitop.items(): 

36 for unitop_id, group in groups.items(): 

37 # Heat Exchangers - streamdataentry is null for heatexchangers ONLY. 

38 if unitop_type == "Heat Exchanger": 

39 hot = cold = None 

40 

41 # single pass to find first unprocessed hot&&cold 

42 for s in group: 42 ↛ 54line 42 didn't jump to line 54 because the loop on line 42 didn't complete

43 if s.id in processed_stream_ids: 43 ↛ 44line 43 didn't jump to line 44 because the condition on line 43 was never true

44 continue 

45 

46 if hot is None and is_hot(s): 

47 hot = s 

48 elif cold is None and is_cold(s): 48 ↛ 51line 48 didn't jump to line 51 because the condition on line 48 was always true

49 cold = s 

50 

51 if hot and cold: 

52 break 

53 

54 if hot and cold: 54 ↛ 36line 54 didn't jump to line 36 because the condition on line 54 was always true

55 hennodes_to_create.append(HenNode( 

56 flowsheet=hot.flowsheet, 

57 hot_connection=hot, 

58 cold_connection=cold, 

59 )) 

60 processed_stream_ids.update([hot.id, cold.id]) 

61 

62 # Heaters 

63 elif unitop_type == "Heater": 

64 for s in group: 

65 if s.id in processed_stream_ids: 65 ↛ 66line 65 didn't jump to line 66 because the condition on line 65 was never true

66 continue 

67 hennodes_to_create.append(HenNode( 

68 flowsheet=s.flowsheet, 

69 stream_data_entry=s, 

70 cold_connection=s, 

71 )) 

72 processed_stream_ids.add(s.id) 

73 

74 # Coolers 

75 elif unitop_type == "Cooler": 75 ↛ 36line 75 didn't jump to line 36 because the condition on line 75 was always true

76 for s in group: 

77 if s.id in processed_stream_ids: 77 ↛ 78line 77 didn't jump to line 78 because the condition on line 77 was never true

78 continue 

79 hennodes_to_create.append(HenNode( 

80 flowsheet=s.flowsheet, 

81 stream_data_entry=s, 

82 hot_connection=s, 

83 )) 

84 processed_stream_ids.add(s.id)