Coverage for backend/pinch_service/serverFunctions/pinchServerFunctions.py: 89%

1from typing import Any 

2import json 

3import traceback 

4from http.server import BaseHTTPRequestHandler, HTTPServer 

5from cProfile import Profile 

6from pydantic import ValidationError 

7 

8from heat_exchanger_profiler import ( 

9 get_heat_exchanger_length_states, 

10 THCurveRequest, 

11 THCurveResponse, 

12) 

13 

14import OpenPinch 

15from OpenPinch import ( 

16 TargetInput, 

17 TargetOutput, 

18 VisualiseInput, 

19 VisualiseOutput, 

20 LineariseInput, 

21 LineariseOutput, 

22) 

23 

24 

25def targeting_analysis(data: Any, parent_fs_name: str = 'Project') -> TargetOutput: 

26 """Calculates targets and outputs from inputs and options""" 

27 # Validate request data using Pydantic model 

28 request_data = TargetInput.model_validate(data) 

29 

30 # Perform advanced pinch analysis and total site analysis 

31 return_data = OpenPinch.pinch_analysis_service( 

32 data=request_data, 

33 project_name=parent_fs_name, 

34 ) 

35 

36 # Validate response data 

37 validated_data = TargetOutput.model_validate(return_data) 

38 

39 # Return data 

40 return validated_data 

41 

42 

43def visualise_analysis(data: Any) -> VisualiseOutput: 

44 # Validate request data using Pydantic model 

45 request_data = VisualiseInput.model_validate(data) 

46 

47 # Create visualisation of pinch graphs 

48 return_data = OpenPinch.get_visualise( 

49 request_data.zones 

50 ) 

51 

52 # Validate response data 

53 validated_data = VisualiseOutput.model_validate(return_data) 

54 

55 # Return data 

56 return validated_data 

57 

58 

59def linearize_stream(data: Any) -> LineariseOutput: 

60 # Validate request data using Pydantic model 

61 request_data = LineariseInput.model_validate(data) 

62 

63 # Get piecewise linearisation of extracted hot and cold streams 

64 return_data = OpenPinch.get_piecewise_linearisation_for_streams( 

65 streams=request_data.streams, 

66 t_h_data=request_data.t_h_data, 

67 dt_diff_max=request_data.t_min, 

68 ) 

69 

70 # Validate response data 

71 validated_data = LineariseOutput.model_validate({"streams": return_data["t_h_points"]}) 

72 

73 # Return data 

74 return validated_data 

75 

76 

77def get_heat_exchanger_profile(data: Any) -> THCurveResponse: 

78 # Validate request data using Pydantic model 

79 request_data = THCurveRequest.model_validate(data) 

80 

81 # Create the structure of the return data 

82 return_data = { 

83 "curve_points": [None] * len(request_data.streams), 

84 "states": [None] * len(request_data.streams) 

85 } 

86 

87 # Get the heat exchanger profile from states  

88 for i, s in enumerate(request_data.streams): 

89 t_h_curve_points, serialised_states = get_heat_exchanger_length_states( 

90 ppKey=request_data.ppKey, 

91 composition=s.composition, 

92 mole_flow=request_data.mole_flow, 

93 t_supply=s.t_supply, 

94 t_target=s.t_target, 

95 p_supply=s.p_supply, 

96 p_target=s.p_target, 

97 h_supply=s.h_supply, 

98 h_target=s.h_target, 

99 num_points=request_data.num_intervals, 

100 prev_state=request_data.prev_states[i] if isinstance(request_data.prev_states, list) else None, 

101 ) 

102 

103 return_data["curve_points"][i] = t_h_curve_points 

104 return_data["states"][i] = serialised_states 

105 

106 # Validate response data 

107 validated_data = THCurveResponse.model_validate(return_data) 

108 

109 # Return data 

110 return validated_data