""" Python implementation of the task scheduling system """ from machine import WDT from machine import Pin import time from task_id import task_id from machine import Pin, Timer from IOT_4G import * # Constants TASK_MODE_ONCE = 0 TASK_MODE_PERIOD = 1 TIMER_MODE_TIMER = 0 TIMER_MODE_ONCEROUTINE = 1 TIMER_MODE_CYCROUTINE = 2 MAX_TIMER = 5 TICK_PER_S = 1000 # 1000 ticks per second # Compatibility functions for standard Python (replacing MicroPython specific functions) def ticks_ms(): """Get current time in milliseconds""" return int( time.ticks_ms()) def ticks_diff(end, start): """Calculate difference between two timestamps""" return end - start def sleep_ms(ms): """Sleep for given milliseconds""" time.sleep(ms / 1000.0) def seconds_to_hms(total_seconds): # 取整处理（可选） total_seconds = int(total_seconds) # 计算时分秒 hours, remainder = divmod(total_seconds, 3600) minutes, seconds = divmod(remainder, 60) # 格式化为两位数 return "{:02d}:{:02d}:{:02d}".format(hours, minutes, seconds) class task_Timer: """task_Timer class for task scheduling""" def __init__(self): self.start = 0 self.interval = 0 def set(self, interval): """Set timer with interval in milliseconds""" self.interval = interval self.start = ticks_ms() def expired(self): """Check if timer has expired""" return ticks_diff(ticks_ms(), self.start) >= self.interval class TaskData: """Task data structure""" def __init__(self, task_id, mode, name, last_time, task_proc): self.active = False self.mode = mode self.task_id = task_id self.name = name self.last_time = last_time self.period = last_time self.task_proc = task_proc class TimerData: """Timer data structure""" def __init__(self): self.in_use = False self.timer_id = 0 self.mode = 0 self.period = 0 self.last_time = 0 self.routine = None self.param = None interrupt_counter = 0 total_interrupts = 0 class TaskScheduler: """Task scheduler implementation""" def __init__(self): self._tasks = [] self._timers = [TimerData() for _ in range(MAX_TIMER)] for i,timer in enumerate(self._timers): if not timer.in_use: print(f"timer_id ={i}, timer = {timer.timer_id}") self._all_tick_count = 0 self.beep_cnt = 0 self.beep_On = True def active_task(self, task_id, period): """Activate a task with given period""" for task in self._tasks: if task.task_id == task_id: if not task.active: task.active = True task.period = period print(f"active_task {task.name} last_time={task.last_time}") return 0 return -1 def active_task_now(self, task_id, period): """Activate a task immediately""" for task in self._tasks: if task.task_id == task_id: if not task.active: task.active = True task.period = period task.last_time = 0 break def suspend_task(self, task_id): """Suspend a task""" for i, task in enumerate(self._tasks): if task.task_id == task_id: task.active = False return i return -1 def set_task_last_time(self, task_id, last_time): """Set last execution time for a task""" for i, task in enumerate(self._tasks): if task.task_id == task_id: task.last_time = last_time return i return -1 def start_timer(self, mode, period, routine, param=None): """Start a timer""" for i,timer in enumerate(self._timers): if not timer.in_use: if mode in (TIMER_MODE_TIMER, TIMER_MODE_ONCEROUTINE, TIMER_MODE_CYCROUTINE): timer.in_use = True timer.mode = mode timer.period = period timer.last_time = period timer.routine = routine timer.param = param timer.timer_id = i #print(f"start_timer id = {i} ") return i return -1 return -2 def stop_timer(self, timer_id): """Stop a timer""" if 0 <= timer_id < MAX_TIMER: self._timers[timer_id].in_use = False return 0 return -1 def stop_all_timers(self): """Stop all timers""" for timer in self._timers: timer.in_use = False def add_task(self, task_id, mode, name, period, task_proc): """Add a new task to the scheduler""" task = TaskData(task_id, mode, name, period, task_proc) print(f"add task = {name},period = {period}") self._tasks.append(task) def task_proc(self): """Main task processing function""" global interrupt_counter, total_interrupts if(interrupt_counter>0): self._all_tick_count = total_interrupts self.timer_proc(interrupt_counter) # Process tasks for task in self._tasks: if task.active: task.last_time-=interrupt_counter #print(f"current_time={time.ticks_ms()},_all_tick_count={self._all_tick_count}") interrupt_counter = 0 for task in self._tasks: if task.active: #print(f"task.last_time={task.last_time}\r") #print(f"ticks={time.ticks_ms()}\r") if task.last_time<=0: #print(f"ticks={time.ticks_ms()} task {task.name} proc running!\r") task.task_proc() # Execute task without pt if task.mode == TASK_MODE_ONCE: task.active = False task.last_time += task.period else: pass def get_system_ms(self): """Get system time in milliseconds""" return ticks_ms() def beep(self,beep_time ,beep_period=500): self.beep_cnt = beep_time self.beep_On ==True print("start beep!") self.active_task(task_id.TASK_ID_BEEP,beep_period) def task_beep(self): if self.beep_On ==True : beep_sta.value(1) #print("beep on\r") self.beep_cnt-=1 else: beep_sta.value(0) #print("beep off\r") self.beep_On = not self.beep_On if(self.beep_cnt==0): #print("beep stoped!\r") self.suspend_task(task_id.TASK_ID_BEEP) pass def timer_proc(self,dt=1): """Timer processing function""" for timer in self._timers: if timer.in_use: timer.last_time -=dt for timer in self._timers: if timer.in_use: #print(f"Timer {timer.timer_id} proc running! last_time ={timer.last_time}") if timer.last_time<= 0: # timer.period: if timer.mode == TIMER_MODE_TIMER: timer.last_time = timer.period timer.in_use = False elif timer.mode == TIMER_MODE_ONCEROUTINE: timer.in_use = False elif timer.mode == TIMER_MODE_CYCROUTINE: timer.last_time += timer.period #print(f"Timer {timer.timer_id} proc running!") if timer.routine: timer.routine(timer.param) # wdt = WDT(timeout=20000) # led_4G = Pin(38, Pin.OUT) # GPIO38 4G LED # led_wifi = Pin(39, Pin.OUT) # GPIO39 wifi LED # led_eth = Pin(40, Pin.OUT) # GPIO40 以太网 LED beep_sta = Pin(16, Pin.OUT) # GPIO16 beep def timer1_callback(timer): global interrupt_counter, total_interrupts interrupt_counter+=1 total_interrupts+=1 scheduler = TaskScheduler() timer = Timer(1) # 使用Timer1 def init_timer(): try: timer.init(period=1, mode=Timer.PERIODIC, callback=timer1_callback) print("1ms定时器中断已启动！") except Exception as e: print(f"定时器初始化失败: {e}") def init_task(): init_timer() print("app running") ''' # Define a task function def my_task(): print("MyTask running") # Add task to scheduler scheduler.add_task(task_id.TASK_ID_MyTask, TASK_MODE_PERIOD, "MyTask", 1000, my_task) scheduler.active_task(task_id.TASK_ID_MyTask, 1000) ''' led_cnt = 0 #闭包捕获,模拟static led_sta = Pin(21, Pin.OUT) # GPIO21 运行 LED def task_wdt(): #print("task_wdt running") #wdt.feed() pass def task_blink(): nonlocal led_cnt nonlocal led_sta led_cnt+=1 led_cnt%=10 if(led_cnt==0): led_sta.value(1) #print(f"ticks={time.ticks_ms()}") #print("app running") else: led_sta.value(0) pass scheduler.add_task(task_id.TASK_ID_WDT, TASK_MODE_PERIOD, "task_wdt", 500, task_wdt) scheduler.active_task(task_id.TASK_ID_WDT, 1000) scheduler.add_task(task_id.TASK_ID_LED, TASK_MODE_PERIOD, "task_led", 100, task_blink) scheduler.active_task(task_id.TASK_ID_LED, 100) scheduler.add_task(task_id.TASK_ID_BEEP, TASK_MODE_PERIOD, "task_beep", 100, scheduler.task_beep) #scheduler.active_task(task_id.TASK_ID_BEEP, 200) scheduler.add_task(task_id.TASK_ID_SYNC_TIMER, TASK_MODE_PERIOD, "task_sync", 1000, Sync_Timer) scheduler.active_task(task_id.TASK_ID_SYNC_TIMER, 1000*3540) #1h sysnc-1min def task_run(): # Main loop while True: scheduler.task_proc() def main_task(): init_task() def Timer_dec_Measure_Time(param): print("Timer_dec_Measure_Time running!") scheduler.start_timer(TIMER_MODE_ONCEROUTINE,100,Timer_dec_Measure_Time,None) scheduler.start_timer(TIMER_MODE_ONCEROUTINE,102,Timer_dec_Measure_Time,None) scheduler.beep(3,1000) task_run() if __name__ == '__main__': # 测试示例 runtime = 3665 # 单位：秒 formatted_time = seconds_to_hms(runtime) print(formatted_time) print(type(formatted_time)) print(f"运行时间:{formatted_time}\r") # 输出：01:01:05 main_task()