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

1from typing import Any 

2import json 

3import traceback 

4from http.server import BaseHTTPRequestHandler, HTTPServer 

5from cProfile import Profile 

6from pydantic import ValidationError 

7from OpenPinch import target, visualise 

8from docs.types.LinearizeTypes import LinearizeRequest, LinearizeResponse, THCurveRequest, THCurveResponse 

9from docs.types.VisualiseTypes import VisualiseRequest, VisualiseResponse 

10from OpenPinch.src.lib.schema import TargetRequest, TargetResponse 

11from linearization.linearize_stream import generate_t_h_curve, piecewise_curve, serialize_states 

12 

13 

14def targeting_analysis(data: Any, parent_fs_name: str = 'Project') -> TargetResponse: 

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

16 # Validate request data using Pydantic model 

17 request_data = TargetRequest.model_validate(data) 

18 

19 # Perform advanced pinch analysis and total site analysis 

20 return_data = target( 

21 zone_tree=request_data.zone_tree, 

22 streams=request_data.streams, 

23 utilities=request_data.utilities, 

24 options=request_data.options, 

25 name=parent_fs_name, 

26 ) 

27 

28 # Validate response data 

29 validated_data = TargetResponse.model_validate(return_data) 

30 

31 # Return data 

32 return validated_data 

33 

34 

35def visualise_analysis(data: Any) -> VisualiseResponse: 

36 # Validate request data using Pydantic model 

37 request_data = VisualiseRequest.model_validate(data) 

38 

39 # Create visualisation of pinch graphs 

40 return_data = visualise(request_data.zones) 

41 

42 # Validate response data 

43 validated_data = VisualiseResponse.model_validate(return_data) 

44 

45 # Return data 

46 return validated_data 

47 

48 

49def linearize_stream(data: Any) -> LinearizeResponse: 

50 # Validate request data using Pydantic model 

51 request_data = LinearizeRequest.model_validate(data) 

52 return_data = {"streams": []} 

53 

54 # Create and Linearize stream 

55 for stream in request_data.streams: 

56 hot_stream = stream.t_supply > stream.t_target 

57 curve_points = data["t_h_data"][0] 

58 mask_points = piecewise_curve( 

59 curve=curve_points, epsilon=request_data.t_min, hot_stream=hot_stream) 

60 return_data["streams"] =mask_points.tolist() 

61 

62 # Validate response data 

63 validated_data = LinearizeResponse.model_validate(return_data) 

64 

65 # Return data 

66 return validated_data 

67 

68 

69def get_t_h_curve(data: Any) -> THCurveResponse: 

70 # Validate request data using Pydantic model 

71 request_data = THCurveRequest.model_validate(data) 

72 return_data = {"curve_points": [ 

73 None] * len(request_data.streams), "states": [None] * len(request_data.streams)} 

74 

75 # Create and Linearize stream 

76 for index, stream in enumerate(request_data.streams): 

77 curve_points, state = generate_t_h_curve( 

78 ppKey=request_data.ppKey, 

79 composition=stream.composition, 

80 mole_flow=request_data.mole_flow, 

81 t_supply=stream.t_supply, 

82 t_target=stream.t_target, 

83 p_supply=stream.p_supply, 

84 p_target=stream.p_target, 

85 h_supply=stream.h_supply, 

86 h_target=stream.h_target, 

87 num_points=request_data.num_intervals 

88 ) 

89 

90 return_data["curve_points"][index] = curve_points.tolist() 

91 return_data["states"][index] = serialize_states(state) 

92 

93 # Validate response data 

94 validated_data = THCurveResponse.model_validate(return_data) 

95 

96 # Return data 

97 return validated_data