from typing import Optional from flask import Flask, request, jsonify from flask_cors import CORS from program_core import ProgramCore from program_public_tools import ProgramPublicTools import time class Server: def __init__(self, server_ip: str = '0.0.0.0', server_port: Optional[int] = None): self.__pubtools: ProgramPublicTools = ProgramPublicTools() self.__server_ip: str = server_ip self.__server_port: Optional[int] = server_port self.__server_ip_init() self.__core: ProgramCore = ProgramCore(self.__pubtools) self.__core.start_core() self.__app = Flask(__name__) CORS(self.__app) self.__setup_routes() self.__send_server_completed_flag() self.__buffer_error_rate_percent_1_value = 0 self.__buffer_error_rate_percent_1_time = time.time() self.__buffer_error_rate_percent_2_value = 0 self.__buffer_error_rate_percent_2_time = time.time() self.__buffer_speed_kbps_1_value = 0 self.__buffer_speed_kbps_1_time = time.time() self.__buffer_speed_kbps_2_value = 0 self.__buffer_speed_kbps_2_time = time.time() @staticmethod def __send_server_completed_flag(): print("[Server] Server program is ready......") def __server_ip_init(self): if self.__server_port is None: self.__server_port = self.__pubtools.find_available_server_port(number=1, address=self.__server_ip)[0] print(f"Server is running on port [{self.__server_port}].") print(f"Server is running on port [{self.__server_port}].") print(f"Server is running on port [{self.__server_port}].") def start(self): self.__app.run(host=self.__server_ip, port=self.__server_port, debug=False) self.__server_debug_output() def __server_debug_output(self): _, _, url = self.__core.instrument_controller.services.digital_oscilloscope.keep_listening() print(f"[Server Debug Output]: OSC Stream Server is running on <{url}>") print(f"[Server Debug Output]: OSC Stream Server is running on <{url}>") print(f"[Server Debug Output]: OSC Stream Server is running on <{url}>") print(f"[Server Debug Output]: OSC Stream Server is running on <{url}>") print(f"[Server Debug Output]: OSC Stream Server is running on <{url}>") @staticmethod def __instrument_get_not_connect_string(not_connect_list: list[str]) -> str: response_string: str = "not_connect_list:" for not_connect_item in not_connect_list: response_string = response_string + " " + not_connect_item return response_string def __setup_routes(self): @self.__app.route("/") def home(): return "Server is running." @self.__app.route("/instrument/not_connect_list", methods=['GET']) def instrument_not_connect_list(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/not_connect_list" + ".") not_connect_list: list[str] = self.__core.instrument_controller.services.check() return self.__instrument_get_not_connect_string(not_connect_list) @self.__app.route("/instrument/digital_multimeter/connection_status", methods=['GET']) def instrument_digital_multimeter_connection_status(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/digital_multimeter/connection_status" + ".") connection_status: bool = (self.__core.instrument_controller.services.digital_multimeter is not None) return f"connection_status: {connection_status}" @self.__app.route("/instrument/digital_oscilloscope/connection_status", methods=['GET']) def instrument_digital_oscilloscope_connection_status(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/digital_oscilloscope/connection_status" + ".") connection_status: bool = (self.__core.instrument_controller.services.digital_oscilloscope is not None) return f"connection_status: {connection_status}" @self.__app.route("/instrument/waveform_generator/connection_status", methods=['GET']) def instrument_waveform_generator_connection_status(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/waveform_generator/connection_status" + ".") connection_status: bool = (self.__core.instrument_controller.services.waveform_generator is not None) return f"connection_status: {connection_status}" @self.__app.route("/instrument/analog_electronic_load/connection_status", methods=['GET']) def instrument_analog_electronic_load_connection_status(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/analog_electronic_load/connection_status" + ".") connection_status: bool = (self.__core.instrument_controller.services.analog_electronic_load is not None) return f"connection_status: {connection_status}" @self.__app.route("/instrument/digital_multimeter/keep_listening", methods=['GET']) def instrument_digital_multimeter_keep_listening(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/digital_multimeter/keep_listening" + ".") if self.__core.instrument_controller.services.digital_multimeter is not None: self.__core.instrument_controller.services.digital_multimeter.keep_listening() return f"keep_listening" else: return "None" @self.__app.route("/instrument/digital_multimeter/get_range", methods=['GET']) def instrument_digital_multimeter_get_range(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/digital_multimeter/get_range" + ".") if self.__core.instrument_controller.services.digital_multimeter is not None: range_list: list[str] = self.__core.instrument_controller.services.digital_multimeter.get_range() response_string: str = "get_range:" for rang_type_item in range_list: response_string = response_string + " " + rang_type_item return response_string else: return "None" @self.__app.route("/instrument/digital_multimeter/set_range", methods=['GET']) def instrument_digital_multimeter_set_range(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/digital_multimeter/set_range" + ".") if self.__core.instrument_controller.services.digital_multimeter is not None: range_string = request.args.get('range') range_type = self.__core.instrument_controller.services.digital_multimeter.solve_range_string( range_string) self.__core.instrument_controller.services.digital_multimeter.set_range(range_type) return f"set_range: {range_string}" else: return "None" @self.__app.route("/instrument/digital_multimeter/get_value", methods=['GET']) def instrument_digital_multimeter_get_value(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/digital_multimeter/get_value" + ".") if self.__core.instrument_controller.services.digital_multimeter is not None: get_value = self.__core.instrument_controller.services.digital_multimeter.get_value_latest() return f"get_value: {get_value}" else: return "None" @self.__app.route("/instrument/digital_oscilloscope/keep_listening", methods=['GET']) def instrument_digital_oscilloscope_keep_listening(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/digital_oscilloscope/keep_listening" + ".") if self.__core.instrument_controller.services.digital_oscilloscope is not None: _, v_port, v_url = self.__core.instrument_controller.services.digital_oscilloscope.keep_listening() return f"keep_listening: {v_port} {v_url}" else: return "None" @self.__app.route("/instrument/digital_oscilloscope/export_image", methods=['GET']) def instrument_digital_oscilloscope_export_image(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/digital_oscilloscope/export_image" + ".") if self.__core.instrument_controller.services.digital_oscilloscope is not None: get_time: str = self.__core.instrument_controller.services.digital_oscilloscope.save_img() return get_time else: return "None" # self.__core.reboot_serial_controller() # return "success" @self.__app.route("/instrument/waveform_generator/output_start", methods=['GET']) def instrument_waveform_generator_output_start(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/waveform_generator/output_start" + ".") if self.__core.instrument_controller.services.waveform_generator is not None: self.__core.instrument_controller.services.waveform_generator.output_start() return f"output_start" else: return "None" @self.__app.route("/instrument/waveform_generator/output_restart", methods=['GET']) def instrument_waveform_generator_output_restart(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/waveform_generator/output_restart" + ".") if self.__core.instrument_controller.services.waveform_generator is not None: self.__core.instrument_controller.services.waveform_generator.output_restart() return f"output_restart" else: return "None" @self.__app.route("/instrument/waveform_generator/output_pause", methods=['GET']) def instrument_waveform_generator_output_pause(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/waveform_generator/output_pause" + ".") if self.__core.instrument_controller.services.waveform_generator is not None: self.__core.instrument_controller.services.waveform_generator.output_pause() return f"output_pause" else: return "None" @self.__app.route("/instrument/waveform_generator/config/apply_config", methods=['GET']) def instrument_waveform_generator_config_apply_config(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/waveform_generator/config/apply_config" + ".") if self.__core.instrument_controller.services.waveform_generator is not None: self.__core.instrument_controller.services.waveform_generator.apply_config(True) return "apply_config" else: return "None" @self.__app.route("/instrument/waveform_generator/config/set_enable", methods=['GET']) def instrument_waveform_generator_config_set_enable(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/waveform_generator/config/set_enable" + ".") channel_string: str = request.args.get('channel') enable_string: str = request.args.get('enable').lower() if enable_string == "true": enable: bool = True else: enable: bool = False if self.__core.instrument_controller.services.waveform_generator is not None: channel = self.__core.instrument_controller.services.waveform_generator.set_enable(channel_string, enable) return f"set_enable: {channel} {enable}" else: return f"set_enable: None None" @self.__app.route("/instrument/waveform_generator/config/get_type", methods=['GET']) def instrument_waveform_generator_config_get_type(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/waveform_generator/config/get_type" + ".") if self.__core.instrument_controller.services.waveform_generator is not None: res: str = self.__core.instrument_controller.services.waveform_generator.config_channel_1.get_type_string() return f"get_type: " + res else: return f"get_type: " + "None" @self.__app.route("/instrument/waveform_generator/config/set_type", methods=['GET']) def instrument_waveform_generator_config_set_type(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/waveform_generator/config/set_type" + ".") channel_string: str = request.args.get('channel') type_string: str = request.args.get('type') if self.__core.instrument_controller.services.waveform_generator is not None: channel, config = self.__core.instrument_controller.services.waveform_generator.get_config( channel_string) config.type = config.match_type_string(type_string) return f"set_type: {channel} {type_string}" else: return f"set_type: None None" @self.__app.route("/instrument/waveform_generator/config/set_freq", methods=['GET']) def instrument_waveform_generator_config_set_freq(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/waveform_generator/config/set_freq" + ".") channel_string: str = request.args.get('channel') freq: float = float(request.args.get('freq')) if self.__core.instrument_controller.services.waveform_generator is not None: channel, config = self.__core.instrument_controller.services.waveform_generator.get_config( channel_string) config.freq = freq return f"set_freq: {channel} {freq}" else: return f"set_freq: None None" @self.__app.route("/instrument/waveform_generator/config/get_freq_unit", methods=['GET']) def instrument_waveform_generator_config_get_freq_unit(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/waveform_generator/config/get_freq_unit" + ".") if self.__core.instrument_controller.services.waveform_generator is not None: _, config = self.__core.instrument_controller.services.waveform_generator.get_config("1") unit_type_string: str = config.get_freq_unit_type_string() return f"get_freq_unit: {unit_type_string}" else: return f"get_freq_unit: None" @self.__app.route("/instrument/waveform_generator/config/set_freq_unit", methods=['GET']) def instrument_waveform_generator_config_set_freq_unit(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/waveform_generator/config/set_freq_unit" + ".") channel_string: str = request.args.get('channel') freq_unit_string: str = request.args.get('freq_unit') if self.__core.instrument_controller.services.waveform_generator is not None: channel, config = self.__core.instrument_controller.services.waveform_generator.get_config( channel_string) config.freq_unit = config.match_freq_unit_type_string(freq_unit_string) return f"set_freq_unit: {channel} {freq_unit_string}" else: return f"set_freq_unit: None None" @self.__app.route("/instrument/waveform_generator/config/set_high_level", methods=['GET']) def instrument_waveform_generator_config_set_high_level(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/waveform_generator/config/set_high_level" + ".") channel_string: str = request.args.get('channel') high_level: float = float(request.args.get('high_level')) if self.__core.instrument_controller.services.waveform_generator is not None: channel, config = self.__core.instrument_controller.services.waveform_generator.get_config( channel_string) config.high_level = high_level return f"set_high_level: {channel} {high_level}" else: return f"set_high_level: None None" @self.__app.route("/instrument/waveform_generator/config/set_low_level", methods=['GET']) def instrument_waveform_generator_config_set_low_level(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/waveform_generator/config/set_low_level" + ".") channel_string: str = request.args.get('channel') low_level: float = float(request.args.get('low_level')) if self.__core.instrument_controller.services.waveform_generator is not None: channel, config = self.__core.instrument_controller.services.waveform_generator.get_config( channel_string) config.low_level = low_level return f"set_low_level: {channel} {low_level}" else: return f"set_low_level: None None" @self.__app.route("/instrument/waveform_generator/config/get_level_unit", methods=['GET']) def instrument_waveform_generator_config_get_level_unit(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/waveform_generator/config/get_level_unit" + ".") if self.__core.instrument_controller.services.waveform_generator is not None: _, config = self.__core.instrument_controller.services.waveform_generator.get_config("1") unit_type_string: str = config.get_freq_unit_type_string() return f"get_freq_unit: {unit_type_string}" else: return f"get_freq_unit: None" @self.__app.route("/instrument/waveform_generator/config/set_high_level_unit", methods=['GET']) def instrument_waveform_generator_config_set_high_level_unit(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/waveform_generator/config/set_high_level_unit" + ".") channel_string: str = request.args.get('channel') high_level_unit_string: str = request.args.get('high_level_unit') if self.__core.instrument_controller.services.waveform_generator is not None: channel, config = self.__core.instrument_controller.services.waveform_generator.get_config( channel_string) config.high_level_unit = config.match_level_unit_type_string(high_level_unit_string) return f"set_high_level_unit: {channel} {high_level_unit_string}" else: return f"set_high_level_unit: None {high_level_unit_string}" @self.__app.route("/instrument/waveform_generator/config/set_low_level_unit", methods=['GET']) def instrument_waveform_generator_config_set_low_level_unit(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/waveform_generator/config/set_low_level_unit" + ".") channel_string: str = request.args.get('channel') low_level_unit_string: str = request.args.get('low_level_unit') if self.__core.instrument_controller.services.waveform_generator is not None: channel, config = self.__core.instrument_controller.services.waveform_generator.get_config( channel_string) config.low_level_unit = config.match_level_unit_type_string(low_level_unit_string) return f"set_low_level_unit: {channel} {low_level_unit_string}" else: return f"set_low_level_unit: None {low_level_unit_string}" @self.__app.route("/instrument/waveform_generator/config/apply_all_setting", methods=['GET']) def instrument_waveform_generator_config_apply_all_setting(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/waveform_generator/config/apply_all_setting" + ".") if self.__core.instrument_controller.services.waveform_generator is not None: waveform_type: str = request.args.get('type') waveform_freq: float = float(request.args.get('freq')) waveform_freq_unit: str = request.args.get('freq_unit') waveform_voltage: float = float(request.args.get('voltage')) waveform_voltage_unit: str = request.args.get('voltage_unit') enable_ch1: bool = bool(request.args.get('ch1')) enable_ch2: bool = bool(request.args.get('ch2')) generator = self.__core.instrument_controller.services.waveform_generator _, config_ch1 = generator.get_config("1") _, config_ch2 = generator.get_config("2") for config_chx in [config_ch1, config_ch2]: config_chx.type = config_chx.match_type_string(waveform_type) config_chx.type_string = waveform_type config_chx.freq = waveform_freq config_chx.freq_unit = config_chx.match_freq_unit_type_string(waveform_freq_unit) config_chx.high_level = waveform_voltage config_chx.low_level = - waveform_voltage config_chx.high_level_unit = config_chx.match_level_unit_type_string(waveform_voltage_unit) config_chx.low_level_unit = config_chx.match_level_unit_type_string(waveform_voltage_unit) generator.set_enable("1", enable_ch1) generator.set_enable("2", enable_ch2) generator.apply_config(immediate_start=False) generator.output_start() return "True" else: return "None" @self.__app.route("/instrument/analog_electronic_load/set_resistance", methods=['GET']) def instrument_analog_electronic_load_set_resistance(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/instrument/analog_electronic_load/set_resistance" + ".") if self.__core.instrument_controller.services.analog_electronic_load is not None: resistance: str = request.args.get('set_resistance') self.__core.instrument_controller.services.analog_electronic_load.set_resistance(resistance) return "True" else: return "False" @self.__app.route("/serial_device/port_list", methods=['GET']) def serial_device_port_list(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/serial_device/port_list" + ".") port_list: list[str] = self.__core.serial_controller.get_port_list() response_string: str = "port_list:" if len(port_list) == 0: response_string = response_string + " " + "None" else: response_string = response_string + " " + "Empty" for port_name in port_list: response_string = response_string + " " + port_name return response_string @self.__app.route("/serial_device/auto_connect", methods=['GET']) def serial_device_auto_connect(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/serial_device/auto_connect" + ".") matched_port_list, not_matched_port_list, cnc_matched = self.__core.serial_controller.auto_connect() response_data = { 'method': 'auto_connect', 'cnc_matched': cnc_matched, 'port_list': { 'matched': matched_port_list, 'not_matched': not_matched_port_list } } return jsonify(response_data), 200 @self.__app.route("/serial_device/cnc/config_connection", methods=['GET']) def serial_device_cnc_config_connection(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/serial_device/cnc/config_connection" + ".") port_name: str = request.args.get('port_name') com_connection = self.__core.serial_controller.try_build_connection(port_name, if_add_source_to_lib=True) if_success = self.__core.serial_controller.services.cnc_attenuator.match_connection(com_connection) return f"{if_success}" @self.__app.route("/serial_device/cnc/set", methods=['GET']) def serial_device_cnc_set(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/serial_device/cnc/set" + ".") set_value: str = request.args.get('value') if_success = self.__core.serial_controller.services.cnc_attenuator.set(float(set_value)) return f"{if_success}" @self.__app.route("/serial_device/plc/config_connection", methods=['GET']) def serial_device_plc_connect_sender(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/serial_device/plc/config_connection" + ".") tester = self.__core.serial_controller.get_tester() tester.pause_test_service() port_name: str = request.args.get('port_name') plc_com_role: str = request.args.get('plc_com_role').lower() if_success: bool = False if (port_name is not None) and (port_name not in ["None", "", " "]): if port_name not in ["Empty", "empty"]: com_connection = self.__core.serial_controller.try_build_connection(port_name, if_add_source_to_lib=True) if com_connection is not None: if plc_com_role == "sender": if_success = self.__core.serial_controller.services.plc_sender.match_connection(com_connection) elif plc_com_role == "receiver": if_success = self.__core.serial_controller.services.plc_receiver.match_connection( com_connection) elif plc_com_role == "receiver2": if_success = self.__core.serial_controller.services.plc_receiver2.match_connection( com_connection) if if_success is True: self.__core.serial_controller.services.enable_multi_receiver = True else: if_success = False else: if plc_com_role == "receiver2": if_success = self.__core.serial_controller.services.plc_receiver2.match_connection(None) self.__core.serial_controller.services.enable_multi_receiver = False tester.task_add_reset_error_rate() else: if_success: bool = True # if_success = True # ports = self.__core.serial_controller.get_port_list() # self.__core.serial_controller.services.plc_sender.match_connection() get_sender_port_name: str = "None" if self.__core.serial_controller.services.plc_sender.com is not None: get_sender_port_name = self.__core.serial_controller.services.plc_sender.com.connection_port_name get_receiver_port_name: str = "None" if self.__core.serial_controller.services.plc_receiver.com is not None: get_receiver_port_name = self.__core.serial_controller.services.plc_receiver.com.connection_port_name self.__core.serial_controller.services.com_usr_config_success = if_success response_data = { 'method': 'config_connection', 'plc_com_role': plc_com_role, 'port_name': port_name, 'if_success': if_success, 'new_config': { 'sender_port': f"{get_sender_port_name}", 'receiver_port': f"{get_receiver_port_name}", }, 'received_config': { 'role': f"{plc_com_role}", 'port': f"{port_name}", } } return jsonify(response_data), 200 @self.__app.route("/serial_device/plc/check_connection", methods=['GET']) def serial_device_plc_check_connection(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/serial_device/plc/check_connection" + ".") sender_port: Optional[str] = self.__core.serial_controller.services.plc_sender.com.connection_port_name receiver_port: Optional[str] = self.__core.serial_controller.services.plc_receiver.com.connection_port_name sender_connection_flag: bool = (sender_port is not None) receiver_connection_flag: bool = (receiver_port is not None) connection_flag: bool = (sender_connection_flag is True) and (receiver_connection_flag is True) response_data = { 'method': 'check_connection', 'connection_status': connection_flag, 'config': { 'sender_port': f"{self.__core.serial_controller.services.plc_sender.com.connection_port_name}", 'receiver_port': f"{self.__core.serial_controller.services.plc_receiver.com.connection_port_name}", } } return jsonify(response_data), 200 @self.__app.route("/serial_device/plc/start", methods=['GET']) def serial_device_plc_start(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/serial_device/plc/start" + ".") self.__pubtools.debug_output("[Server] Response " + "/serial_device/plc/start" + ".") if self.__core.serial_controller.services.com_usr_config_success is True: package_config: Optional[str] = request.args.get('package_config') sender = self.__core.serial_controller.services.plc_sender receiver = self.__core.serial_controller.services.plc_receiver sender_port, sender_url = sender.keep_send() receiver_port, receiver_url = receiver.keep_receive() receiver2 = None if self.__core.serial_controller.services.enable_multi_receiver is True: receiver2 = self.__core.serial_controller.services.plc_receiver2 receiver2.keep_receive() tester = self.__core.serial_controller.get_tester() tester.start_test_service(sender, receiver, receiver2) if package_config == "单包1kbit": tester_frame_new: bytes = tester.send_package_1kbit elif package_config == "单包16kbit": tester_frame_new: bytes = tester.send_package_16kbit else: tester_frame_new: bytes = package_config.encode() if len(tester_frame_new) % 3 == 0: tester_frame_new += b'12\n' elif len(tester_frame_new) % 3 == 1: # b'1' tester_frame_new += b'1\n' elif len(tester_frame_new) % 3 == 2: # b'12' tester_frame_new += b'\n' else: tester_frame_new = b'12\n' tester.reset_frame_std_and_restart(tester_frame_new) else: sender_port, sender_url, receiver_port, receiver_url = 0, 0, 0, 0 response_data = { 'method': 'start', 'sender': { 'port': f"{sender_port}", 'url': sender_url, }, 'receiver': { 'port': f"{receiver_port}", 'url': receiver_url, } } return jsonify(response_data), 200 @self.__app.route("/serial_device/plc/pause", methods=['GET']) def serial_device_plc_pause(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/serial_device/plc/pause" + ".") # self.__core.serial_controller.services.plc_sender.pause() # self.__core.serial_controller.services.plc_receiver.pause() tester = self.__core.serial_controller.get_tester() tester.pause_test_service() return "pause" @self.__app.route("/serial_device/plc/clear", methods=['GET']) def serial_device_plc_clear(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/serial_device/plc/clear" + ".") self.__core.serial_controller.get_tester().task_add_reset_error_rate() return "clear" @self.__app.route("/serial_device/plc/realtime_performance", methods=['GET']) def serial_device_plc_realtime_performance(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/serial_device/plc/realtime_performance" + ".") tester = self.__core.serial_controller.get_tester() error_rate_percent: float = tester.error_rate_percent error_rate_percent2: float = tester.error_rate_percent2 if self.__buffer_error_rate_percent_1_value == error_rate_percent: if abs(self.__buffer_error_rate_percent_1_time - time.time()) > 3: if error_rate_percent <= 99.5: error_rate_percent = 0 else: pass else: self.__buffer_error_rate_percent_1_value = error_rate_percent self.__buffer_error_rate_percent_1_time = time.time() if self.__buffer_error_rate_percent_2_value == error_rate_percent2: if abs(self.__buffer_error_rate_percent_2_time - time.time()) > 3: if error_rate_percent2 <= 99.5: error_rate_percent2 = 0 else: pass else: self.__buffer_error_rate_percent_2_value = error_rate_percent2 self.__buffer_error_rate_percent_2_time = time.time() return f"realtime_performance: error_rate_percent={error_rate_percent} error_rate_percent2={error_rate_percent2}" @self.__app.route("/serial_device/plc/realtime_performance_speed", methods=['GET']) def serial_device_plc_realtime_performance_speed(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/serial_device/plc/realtime_performance_speed" + ".") tester = self.__core.serial_controller.get_tester() speed_kbps: float = tester.speed_kbps speed_kbps2: float = tester.speed_kbps2 if self.__buffer_speed_kbps_1_value == speed_kbps: if abs(self.__buffer_speed_kbps_1_time - time.time()) > 3: speed_kbps = 0 else: pass else: self.__buffer_speed_kbps_1_value = speed_kbps self.__buffer_speed_kbps_1_time = time.time() if self.__buffer_speed_kbps_2_value == speed_kbps2: if abs(self.__buffer_speed_kbps_2_time - time.time()) > 3: speed_kbps2 = 0 else: pass else: self.__buffer_speed_kbps_2_value = speed_kbps2 self.__buffer_speed_kbps_2_time = time.time() return f"realtime_performance_speed: speed_kbps={speed_kbps} speed_kbps2={speed_kbps2}" @self.__app.route("/serial_device/plc/motor_mode", methods=['GET']) def serial_device_plc_motor_mode(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/serial_device/plc/motor_mode" + ".") get_type: Optional[str] = request.args.get('type') mode_number: Optional[str] = request.args.get('mode') if get_type is not None and mode_number is not None: tester = self.__core.serial_controller.get_tester() use_msg = None if get_type == "motor": use_msg_list = tester.motor_msg_list if mode_number in ["1", "2", "3", "4"]: try: index = int(mode_number) - 1 use_msg = use_msg_list[index] except ValueError: return "failed" except IndexError: return "failed" elif get_type == "relay": use_msg_list = tester.relay_msg_list if mode_number in ["1", "2"]: try: index = int(mode_number) - 1 use_msg = use_msg_list[index] except ValueError: return "failed" except IndexError: return "failed" else: return "failed" if use_msg is not None: tester.set_circle_hook_to_msg_once_mode(use_msg) return "success" return "failed" @self.__app.route("/files/documents/start", methods=['GET']) def files_documents_start(): # print("[Server] Response " + "" + ".") print("[Server] Response " + "/files/documents/start" + ".") host, port = self.__core.local_file_service.start() return f"documents: host={host} port={port}" if __name__ == "__main__": server: Server = Server(server_port=2023) server.start()