以下是基于您提供的需求书写的Python代码： ```python import threading from time import sleep import serial # 定义全局变量 WW = 0 RR = 0 obj_type = "" address = None data_value = None DD_R_01, DD_R_02, DD_R_03, DD_R_04, DD_R_05, DD_R_06, DD_R_07, DD_R_08, DD_R_09, DD_R_10 = [0] * 10 DD_R_11, DD_R_12, DD_R_13, DD_R_14, DD_R_15, DD_R_16, DD_R_17, DD_R_18, DD_R_19, DD_R_20 = [0] * 10 DD_R_21, DD_R_22, DD_R_23, DD_R_24, DD_R_25, DD_R_26, DD_R_27, DD_R_28, DD_R_29 = [0] * 9 DD_W = 0 PMS_W, PMS_R, XX_W, XX_R, YY_W, YY_R, SS_W, SS_R, MM_W, MM_R = [0] * 10 TT_W, TT_R, CC_W, CC_R, DD_W, DD_R, CN_0_W, CN_0_R, CN_2_W, CN_2_R, TN_W, TN_R = [0] * 12 def set_global_vars(var_name): """设置对应变量为1，并将其余同组变量置零""" global PMS_W, PMS_R, XX_W, XX_R, YY_W, YY_R, SS_W, SS_R, MM_W, MM_R, \ TT_W, TT_R, CC_W, CC_R, DD_W, DD_R, CN_0_W, CN_0_R, CN_2_W, CN_2_R, TN_W, TN_R all_vars = [ 'PMS_W', 'PMS_R', 'XX_W', 'XX_R', 'YY_W', 'YY_R', 'SS_W', 'SS_R', 'MM_W', 'MM_R', 'TT_W', 'TT_R', 'CC_W', 'CC_R', 'DD_W', 'DD_R', 'CN_0_W', 'CN_0_R', 'CN_2_W', 'CN_2_R', 'TN_W', 'TN_R' ] for var in all_vars: globals()[var] = 1 if var == var_name else 0 def user_input_thread(): """用户交互线程：处理用户的输入和主程序部分的功能""" while True: try: # 第一次输入 - 工作状态(W/R) choice = input("请选择工作状态 W 或 R (quit退出): ").strip().upper() if choice.lower() == "quit": print("程序即将退出...") break if choice not in ["W", "R"]: print("无效选项，请重新输入!") continue global WW, RR WW = 1 if choice == "W" else 0 RR = 1 if choice == "R" else 0 # 第二次输入 - 对象类别(P/X/Y/S/M/T/C/D/V/F) obj_choice = input("请选择对象类别 P X Y S M T C D V F: ").strip().upper() global obj_type obj_type = obj_choice.upper() # 特殊情况：如果选择了"D" if obj_choice.strip().upper() == "D": set_global_vars('DD_W') print(f"赋值DD_W=1并打印显示‘星期六’") # 第三次输入 - 地址（十进制） address_val = int(input("请输入地址（十进制数字）: ")) global address address = address_val # 第四次输入 - 数据（十进制） data_val = int(input("请输入数据（十进制数字）: ")) global data_value data_value = data_val # 主程序运行逻辑 run_main_program() except Exception as e: print(e) def rs485_serial_thread(): """RS485通信线程：负责发送和接收数据""" ser = serial.Serial(port="COM6", baudrate=9600, bytesize=serial.EIGHTBITS, parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, timeout=0.2) def receive_data(): nonlocal ser received_count = 0 buffer = [] while received_count < 29: raw_data = ser.read_until(expected=b'\r\n') # MODBUS ASCII协议结束符 \r\n decoded_data = raw_data.decode().replace('\r\n', '').lstrip(':').rstrip(' ').strip('0x') if len(decoded_data) > 0 and decoded_data.isdigit(): value = int(decoded_data) buffer.append(value) # 如果接收到特殊数值，则触发对应的事件 if value == 254: print("操作员=A") elif value == 36 and DD_R_01 <= 200: print("盈利富足") received_count += 1 # 输出当前已接收到的所有值 dd_list = ['DD_R_01', 'DD_R_02', 'DD_R_03', 'DD_R_04', 'DD_R_05', 'DD_R_06', 'DD_R_07', 'DD_R_08', 'DD_R_09', 'DD_R_10', 'DD_R_11', 'DD_R_12', 'DD_R_13', 'DD_R_14', 'DD_R_15', 'DD_R_16', 'DD_R_17', 'DD_R_18', 'DD_R_19', 'DD_R_20', 'DD_R_21', 'DD_R_22', 'DD_R_23', 'DD_R_24', 'DD_R_25', 'DD_R_26', 'DD_R_27', 'DD_R_28', 'DD_R_29'] # 更新全局变量 for i, val in enumerate(buffer[:len(dd_list)]): globals()[dd_list[i]] = val # 补充剩余未填充的区域为0 for j in range(len(buffer), 29): globals()[dd_list[j]] = 0 def send_data(): nonlocal ser while True: key_press = input("") if key_press.upper() == "B": # B按键直接显示内容 print(f"WW={WW}, RR={RR}") elif key_press.upper() == "A": # A按键发送固定字符串 command = b"02 45 30 30 34 31 39 30 30 34 03 44 41\r\n" ser.write(command) threads = ( threading.Thread(target=receive_data), threading.Thread(target=send_data) ) for t in threads: t.start() for t in threads: t.join() def run_main_program(): """运行主体程序功能""" global RR, DD_W if RR == 1 and DD_W == 1: print("星期五") if __name__ == "__main__": thread_user_input = threading.Thread(target=user_input_thread) thread_rs485_communication = threading.Thread(target=rs485_serial_thread) thread_user_input.start() thread_rs485_communication.start() thread_user_input.join() thread_rs485_communication.join() ``` ### §相关问题§ 1. 如何修改此程序以便支持更多种类的对象类别？ 2. 此程序如何优化以减少CPU占用率？ 3. 是否可以在现有基础上加入异常日志记录功能？