Coverage for backend/pinch_service/server.py: 53%

1from typing import Any 

2import json 

3import traceback 

4from http.server import BaseHTTPRequestHandler, HTTPServer 

5from cProfile import Profile 

6import pstats 

7import io 

8import argparse 

9 

10from pydantic import ValidationError 

11from serverFunctions import targeting_analysis, linearize_stream, visualise_analysis, get_t_h_curve 

12DEBUG_MODE = False 

13 

14class RequestHandler(BaseHTTPRequestHandler): 

15 """ 

16 Handles the HTTP Requests.  

17 """ 

18 

19 def response(self, status_code: int, result: Any) -> None: 

20 self.send_response(status_code) 

21 self.send_header('Content-type', 'text/json') 

22 self.end_headers() 

23 self.wfile.write(json.dumps(result).encode()) 

24 

25 

26 def xml_response(self, status_code: int, result: Any) -> None: 

27 self.send_response(status_code) 

28 self.send_header('Content-type', 'application/xml') 

29 self.end_headers() 

30 self.wfile.write(result) 

31 

32 

33 def do_POST(self) -> None: 

34 """  

35 Triggers Calculations on POST. 

36 .../calculate/ -> general pinch analysis calculations 

37 """ 

38 # Decode the JSON data 

39 content_length = int(self.headers['Content-Length']) 

40 post_data = self.rfile.read(content_length) 

41 try: 

42 data = json.loads(post_data) 

43 except json.JSONDecodeError: 

44 self.response(400, {"error": "Invalid JSON"}) 

45 return 

46 

47 if self.path == '/calculate': 

48 self._calculate_analysis(data) 

49 elif self.path == '/visualise': 49 ↛ 50line 49 didn't jump to line 50 because the condition on line 49 was never true

50 self._visualise_analysis(data) 

51 elif self.path == '/linearize': 

52 self._linearize_stream(data) 

53 elif self.path == '/generate_t_h_curve': 53 ↛ 56line 53 didn't jump to line 56 because the condition on line 53 was always true

54 self._generate_t_h_curve(data) 

55 else: 

56 self.response(404, {"error": "Not found"}) 

57 

58 def _calculate_analysis(self, data: Any) -> None: 

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

60 try: 

61 if DEBUG_MODE: # Profile on DEBUG 61 ↛ 62line 61 didn't jump to line 62 because the condition on line 61 was never true

62 target_profiler = Profile() 

63 target_profiler.enable() 

64 validated_data = targeting_analysis(data) 

65 

66 if DEBUG_MODE: 66 ↛ 67line 66 didn't jump to line 67 because the condition on line 66 was never true

67 target_profiler.disable() 

68 self._log_profile("Target Function", target_profiler) 

69 

70 # Return data 

71 self.response(200, json.loads(validated_data.model_dump_json())) 

72 

73 except ValidationError as ve: 

74 print("Validation Error:", ve) 

75 print("Traceback:", traceback.format_exc()) 

76 self.response(400, {"error": f"Invalid data format: {ve.json}", "traceback": traceback.format_exc()}) 

77 

78 except Exception as e: 

79 print("Error:", e) 

80 print("Traceback:", traceback.format_exc()) 

81 self.response(400, {"error": str(e), "traceback": traceback.format_exc()}) 

82 

83 def _visualise_analysis(self, data: Any) -> None: 

84 try: 

85 validated_data = visualise_analysis(data) 

86 

87 # Return data 

88 self.response(200, json.loads(validated_data.model_dump_json())) 

89 return 

90 

91 except ValidationError as ve: 

92 print("Validation Error:", ve) 

93 print("Traceback:", traceback.format_exc()) 

94 self.response(400, {"error": f"Invalid data format: {ve.json}", "traceback": traceback.format_exc()}) 

95 

96 except Exception as e: 

97 print("Error:", e) 

98 print("Traceback:", traceback.format_exc()) 

99 self.response(400, {"error": str(e), "traceback": traceback.format_exc()}) 

100 

101 def _linearize_stream(self, data: Any): 

102 try: 

103 validated_data = linearize_stream(data) 

104 

105 # Return data 

106 self.response(200, json.loads(validated_data.model_dump_json())) 

107 return 

108 

109 except ValidationError as ve: 

110 print("Validation Error:", ve) 

111 print("Traceback:", traceback.format_exc()) 

112 self.response(400, {"error": f"Invalid data format: {ve.json}", "traceback": traceback.format_exc()}) 

113 

114 except Exception as e: 

115 print("Error:", e) 

116 print("Traceback:", traceback.format_exc()) 

117 self.response(400, {"error": str(e), "traceback": traceback.format_exc()}) 

118 

119 def _generate_t_h_curve(self, data: Any): 

120 try: 

121 validated_data = get_t_h_curve(data) 

122 # Return data 

123 self.response(200, json.loads(validated_data.model_dump_json())) 

124 return 

125 

126 except ValidationError as ve: 

127 print("Validation Error:", ve) 

128 print("Traceback:", traceback.format_exc()) 

129 self.response(400, {"error": f"Invalid data format: {ve.json}", "traceback": traceback.format_exc()}) 

130 

131 except Exception as e: 

132 print("Error:", e) 

133 print("Traceback:", traceback.format_exc()) 

134 self.response(400, {"error": str(e), "traceback": traceback.format_exc()}) 

135 

136 def _log_profile(self, section_name: str, profiler: Profile) -> None: 

137 """Logs the profiling stats to the container logs""" 

138 s = io.StringIO() 

139 ps = pstats.Stats(profiler, stream=s) 

140 ps.strip_dirs().sort_stats('cumulative') # Sort by cumulative time 

141 ps.print_stats() 

142 log_output = f"Profiling Results for {section_name}:\n{s.getvalue()}" 

143 print(log_output) # Redirect to Docker container logs 

144 

145def run_server(debug_mode: bool): 

146 """Runs the server""" 

147 global DEBUG_MODE 

148 DEBUG_MODE = debug_mode 

149 

150 server_address = ('', 8082) 

151 httpd = HTTPServer(server_address, RequestHandler) 

152 print(f'Using port 8082 {"with debug mode" if DEBUG_MODE else ""}') 

153 httpd.serve_forever() 

154 

155 

156if(__name__ == '__main__'): 156 ↛ exitline 156 didn't exit the module because the condition on line 156 was always true

157 parser = argparse.ArgumentParser(description='Pinch Analysis Server') 

158 parser.add_argument('--debug', action='store_true', help='Enable debug mode for profiling and detailed logging') 

159 args = parser.parse_args() 

160 

161 print('Starting Pinch Analysis server...') 

162 run_server(debug_mode=args.debug)