from modbus import * from eeprom import EEPROM from config import Config from DwinTest import DwinTest from machine import Pin from machine import RTC import time from task_id import task_id from fuzzy import AI_Control from fuzzy import PWM_PERIOD #from task import * from task import TASK_MODE_ONCE from task import TASK_MODE_PERIOD from task import TIMER_MODE_TIMER from task import TIMER_MODE_ONCEROUTINE from task import TIMER_MODE_CYCROUTINE from task import scheduler import task import IOT_4G import mqtt coeff_flow = 10 # coeff_Flow_k = 100 #触摸屏的K 是显示值*100传送上来 显示1.00，传递100 coeff_Flow_b = 10 #触摸屏的B 是显示值*10传送上来 显示1.00，传递10 #MODBUS BUS 流量单位是1/1000 #MODBUS BUS 温度是1/100 #触摸屏设定流量单位 1/100 #设定温度单位 1 #将所有温度单位统一为0.01度 MEASURE_TEMP_COFF = 1 #温度的小数点为0.01 CONFIG_TEMP_COFF = 100 #配置温度的小数点为1 #在Ctrl里统一基准流量为 0.001 #在Ctrl里统一温度 为 0.01 LED_SLAVE = Pin(16, Pin.OUT) #D02维电器【GPI016】 #高电平加热 ,高电平亮灯 LED_Heat = Pin(40, Pin.OUT) # GPIO40 以太网 LED 用来指示加热 LED_RELAY = Pin(15, Pin.OUT) # GPIO15 加热控制继电器 class Oxygen_Ctrl: def __init__(self, port=2, baudrate = 9600): self.eeprom = EEPROM() self.config = Config(self.eeprom) self.modbus_485 = modbus_host(port, baudrate) #modbus part self.org_temperature = [0] * 4 #回读原始温度值 self.org_flow_feed = [0] * 12 #回读原始流量值 self.org_flow_out = [2500] * 12 #输出原始流量值 #Dwin part self.dWin = DwinTest(self.eeprom) self.dWin.init() self.dWin.reg_callback("_update_cfg_callback",self.update_cfg_callback) self.dWin.reg_callback("_handle_start_test", self.handle_start_test) self.dWin.reg_callback("_handle_stop_test", self.handle_stop_test) #temperature and flowmeter true value (after calibration) self.cali_temperature = [0] * 4 #回读校准 温度值 self.cali_flow_feed = [0] * 12 #回读校准 流量值 self.Measure_Time = self.config.SET_TestTime*3600 #测量时间 单位为秒,设定的为小时 LED_SLAVE.value(1) self.state_id = 0 self.pre_state_id = 0 self.state_str_list = [ "Awaiting oxidation", "Awaiting oxidation", "Start heating ", "Start heating", "Temperature is OK", "Start oxidation", "7th", "8th", "9th", "11th", "12th"] self.Timer_Measure_id = -1 #运行状态机计时 self.ai_ctrl = AI_Control() self.ai_ctrl.set_p_out = self.set_p_out self.Timer_heat = -1; # 后续使用本地时间计算差值 self.t_start = time.localtime() print(f"t_start ={self.t_start}\r\r") current_ts = time.time() # 当前时间戳（秒） new_ts = current_ts + self.Measure_Time # 新时间戳 self.t_end = time.localtime(new_ts) self.t_dwin_update = time.localtime() print(f"t_end ={self.t_end}\r\r") hhmmss = task.seconds_to_hms(self.Measure_Time) self.dWin.write_const_string(hhmmss,0x170) self.mqtt_ClientX = mqtt.init_mqtt() self.mqtt_ClientX.reg_callback("_recv_callback",self.recv_callback) def recv_callback(self,cloudmsg): #decode package and control stove if("Emergency_stop" in cloudmsg): if(cloudmsg["Emergency_stop"] == True ): #为真时关闭设备 self.handle_stop_test() print(f"Emergency_stop = {cloudmsg["Emergency_stop"]}") else: self.handle_stop_test() print(f"Emergency_stop = {cloudmsg["Emergency_stop"]}") pass def Heat_ctrl(self,bHeat): if bHeat: LED_Heat.value(1) LED_RELAY.value(1) #print("LED_Heat ON!\r") else: LED_Heat.value(0) LED_RELAY.value(0) #print("LED_Heat OFF!\r") def set_p_out(self,pwm_width): if pwm_width >0: self.Heat_ctrl(True) #def set_p_off(param): # self.Heat_ctrl(False) if(pwm_width>PWM_PERIOD): pwm_width = PWM_PERIOD if(pwm_width=expiretime or diff<0: self.t_dwin_update = time.localtime() self.dWin.write_var_32(0x0020, Measure_Time)# 保存剩余时间的秒数 #Measure_Time = self.dWin.read_var(0x0020, 2)# 保存剩余时间的秒数 self.dWin.write_var_32(0x0070,self.state_id)# 保存状态 #print(f"Measure_Time = {Measure_Time} {self.state_id}" ) else: pass def readBackDwinTimeAndState(self): ret = False try : print("readBackDwinTimeAndState" ) self.Measure_Time = self.dWin.read_var(0x0020,2)# 回读剩余时间的秒数 #self.state_id = self.dWin.read_var(0x0070,2)# 回读状态 ret = True except Exception as e: print(f"readBackDwinTimeAndState Exception: {e}") return ret def getremainTime(self): self.Measure_Time = self.config.SET_TestTime*3600 if(self.state_id<4): self.t_start = time.localtime() new_ts = time.mktime(self.t_start) + self.Measure_Time # 新时间戳 self.t_end = time.localtime(new_ts) #self.Measure_Time -=1 #递减1s now = time.localtime() Measure_Time = time.mktime(self.t_end)-time.mktime(now) #print(f"self.Measure_Time={self.Measure_Time}") return Measure_Time def handle_start_test(self): #如果在等待，启动测试 scheduler.beep(1,1000) if (self.state_id ==1): #启动加热 self.dWin.write_const_string("Start heating") self.state_id = 2 if (self.state_id ==4): #启动测试 #启动定时器，对self.Measure_Time 递减 self.dWin.write_const_string("Start oxidation") self.state_id = 5 # 对测量时间计时 def Timer_dec_Measure_Time(param): Measure_Time = self.getremainTime() #self.updateDwinTimeAndState(Measure_Time) hhmmss = task.seconds_to_hms(Measure_Time) self.dWin.write_const_string(hhmmss,0x170) if(Measure_Time<=0):#时间递减到0，停止测试 self.handle_stop_test() #lambada self.t_start = time.localtime() current_ts = time.time() # 当前时间戳（秒） new_ts = current_ts + self.Measure_Time # 新时间戳 self.t_end = time.localtime(new_ts) print(f"t_end ={self.t_end}\r\r") self.Timer_Measure_id = scheduler.start_timer(TIMER_MODE_CYCROUTINE,1000,Timer_dec_Measure_Time,None) pass def handle_stop_test(self): #吧测试状态归0 if (self.state_id >0): self.state_id = 0 if(self.Timer_Measure_id!=-1): scheduler.stop_timer(self.Timer_Measure_id) self.Timer_Measure_id = -1 self.ai_ctrl.stop_heat() #停止加热 scheduler.beep(3,1000) #停止报警 pass def Task_fuzzy(self): self.ai_ctrl.task_fuzzy() pass def state_machine(self): hhmmss = "00:00:00" #self.dWin.write_const_string(self.state_str_list[self.state_id]) #print(f"state_machine self.state_id = {self.state_id}\r") if self.state_id == 0: ''' # used for wdt #如果是复位启动，在此加载 状态 ret = self.readBackDwinTimeAndState() if ret == True and self.Measure_Time != self.config.SET_TestTime*3600: #如果在走字，启动控温，并将状态机切换为4 self.state_id ==4 #切换到升温状态 self.handle_start_test() else: self.dWin.write_const_string("Awaiting oxidation") self.state_id = 1 ''' self.dWin.write_const_string("Awaiting oxidation") self.state_id = 1 hhmmss = task.seconds_to_hms(self.Measure_Time)# #self.dWin.write_var_32(0x0020, self.Measure_Time) self.dWin.write_const_string(hhmmss,0x170) #while True: # pass elif self.state_id ==1: # 等待按键跳转 pass elif self.state_id ==2: self.dWin.write_const_string("Start heating") #启动控温 self.ai_ctrl.ctrl_stove(True , self.cali_temperature, self.config.SetTempra*CONFIG_TEMP_COFF) self.state_id =3 elif self.state_id ==3: if(self.cali_temperature[0]*MEASURE_TEMP_COFF >= self.config.SetTempra*CONFIG_TEMP_COFF): self.dWin.write_const_string("Temperature is OK") #启动测试 self.state_id = 4 pass pass elif self.state_id ==4: #等待启动测试按钮 pass elif self.state_id ==5: #已经按下测试按钮 #对运行时间开始计数 pass # if(self.pre_state_id!=self.state_id): self.updateMqtt_dwin(0) self.pre_state_id = self.state_id print(f"self.state_id={self.state_id}") else: t2 = 2*60 self.updateMqtt_dwin(t2) def updateMqtt_dwin(self,updatePeriod): Measure_Time = self.getremainTime() #self.updateDwinTimeAndState(Measure_Time) hhmmss = task.seconds_to_hms(Measure_Time) dic_info = { "Flow01": f"{self.cali_flow_feed[0]/1000:.3f}", "Flow02": f"{self.cali_flow_feed[1]/1000:.3f}", "Flow03": f"{self.cali_flow_feed[2]/1000:.3f}", "Flow04": f"{self.cali_flow_feed[3]/1000:.3f}", "Flow05": f"{self.cali_flow_feed[4]/1000:.3f}", "Flow06": f"{self.cali_flow_feed[5]/1000:.3f}", "Flow07": f"{self.cali_flow_feed[6]/1000:.3f}", "Flow08": f"{self.cali_flow_feed[7]/1000:.3f}", "Flow09": f"{self.cali_flow_feed[8]/1000:.3f}", "Flow10": f"{self.cali_flow_feed[9]/1000:.3f}", "Flow11": f"{self.cali_flow_feed[10]/1000:.3f}", "Flow12": f"{self.cali_flow_feed[11]/1000:.3f}", "Temperature":self.cali_temperature[0]/CONFIG_TEMP_COFF, "Status":f"{self.state_str_list[self.state_id]}", "TimeRemaining":f"{hhmmss}" } self.mqtt_ClientX.send_info(dic_info,updatePeriod) #*60 pass def add_task(self): #添加任务 scheduler.add_task(task_id.TASK_ID_MODBUS_IN, TASK_MODE_PERIOD, "task_modbus", 200, Task_refresh_modbus_in) scheduler.active_task(task_id.TASK_ID_MODBUS_IN, 100) scheduler.add_task(task_id.TASK_ID_DWIN, TASK_MODE_PERIOD, "task_dwin", 5, Task_refresh_dwin_input) scheduler.active_task(task_id.TASK_ID_DWIN, 5) scheduler.add_task(task_id.TASK_ID_MACHINE, TASK_MODE_PERIOD, "task_Machine", 1000, Task_state_machine) scheduler.active_task(task_id.TASK_ID_MACHINE, 100) scheduler.add_task(task_id.TASK_ID_FUZZY, TASK_MODE_PERIOD, "task_Fuzzy", 1000, self.Task_fuzzy) scheduler.active_task(task_id.TASK_ID_FUZZY, PWM_PERIOD) pass instrument_ctrl = Oxygen_Ctrl() def Task_refresh_modbus_in(): instrument_ctrl.Ai_read() instrument_ctrl.update_Anlog_in_value() instrument_ctrl.Ai_Dwin_show() #instrument_ctrl.dWin.write_const_string(instrument_ctrl.state_str_list[Pre]) #print("org_temperature:", instrument_ctrl.org_temperature) #print("org_flow_feed:", instrument_ctrl.org_flow_feed) #print("cali_temperature:", instrument_ctrl.cali_temperature) #print("cali_flow_feed:", instrument_ctrl.cali_flow_feed) def Task_refresh_dwin_input(): instrument_ctrl.dWin.run() def Task_state_machine(): instrument_ctrl.state_machine(); pass def init_task(): instrument_ctrl.add_task() pass def Connect4G_NTP(): if not IOT_4G.Sync_Timer(): instrument_ctrl.dWin.write_const_string("LTE connect fail!") else: instrument_ctrl.dWin.write_const_string("LTE connect OK!") time.sleep(4) pass def ctrl_run(): task.init_task() init_task() Connect4G_NTP() #print("App run") task.task_run() if __name__ == '__main__': ctrl_run() ''' # instrument_ctrl.refresh_Flowout() #time.sleep(1) # 弄一个任务不停的去刷它 Task_refresh_modbus_in() Task_refresh_dwin_input() Task_state_machine() '''