#!/opt/homebrew/bin/python3 # -*- coding:utf-8 -*- import json import tracemalloc import gc import time import base64 from PyQt6 import * from PyQt6.QtCore import * from logs import log from typing import Union from inspect import isfunction from common import common from influxDB import influxdb from logModel.logManager import logManager from instructionModel.instructionModel import InstructionModel from projectModel.projectManager import projectManager from userModel.userManager import userManager from taskModel.taskManager import taskManager from taskInstructionModel.taskInstructionManager import taskInstructionManager from taskModel.taskModel import TaskModel from interfaceSession.interfaceManager import interfaceManager class TaskActuator(QThread): TASK = 0 VIRTUAL = 1 INSTRUCTION = 2 logMsg = pyqtSignal(dict) updateProgress = pyqtSignal(str, int, int) updateInstructProgress = pyqtSignal(str, int, int) updateDetails = pyqtSignal(str, int, int) executeFinished = pyqtSignal(str) taskStop = pyqtSignal(str) # 添加任务停止信号 def __init__(self, id): super().__init__() self.currentIndex = 0 self.task_instruction_id = "" self.instructionList = [] self.currentTask = None self.parentId = "" self.running_lock = QMutex() # 新增互斥锁 self.running = False self.eventInfo = {} self.subTaskActuator = None self.taskScript = "" self.type = TaskActuator.INSTRUCTION self.id = str(id) self.name = "" self.taskInfo = {} def onUpdateProgress(self, taskId, current, total): self.updateProgress.emit(taskId, current, total) def onUpdateDetails(self, taskId, current, total): self.updateDetails.emit(taskId, current, total) def run(self): proId = None operator = None try: tracemalloc.start() loopStartTime = 0 currentInstruction = InstructionModel() session = None proId = projectManager.getCurrentProId() proInfo = projectManager.getCurrentProInfo() bucketName = "name" in proInfo and proInfo["name"] or "DEFAULT" influxdb.create(bucketName) operator = userManager.getCurrentUser() if proId != None and proId != "": projectManager.addHistory(proId, { "event": "执行", "type": self.type, "name": self.name, "operator": operator, }) instructionListLength = len(self.instructionList) if self.type == TaskActuator.TASK and instructionListLength > 0: self.currentTask = TaskModel(self.taskInfo) self.currentTask.setProInfo(proInfo) self.currentTask.logMsg.connect(self.onLogMsg, type=Qt.ConnectionType.DirectConnection) self.currentTask.updateProgress.connect(self.onUpdateProgress) loop = int(self.taskInfo.get("loop", 1)) delay = float(int(self.taskInfo.get("delay",0))/1000) taskIndex = 0 total = instructionListLength * loop self.updateProgress.emit(self.id, 0, total) self.updateDetails.emit(self.id, 0, total) self.updateDetails.emit(self.parentId, 0, total) self.currentTask.setTargetList(self.instructionList) self.currentTask.start() while not self.currentTask.isFinished and self.running: self.currentTask.callBack(0) self.currentIndex = 0 self.updateProgress.emit(self.id, 0, total) # 新一轮循环开始，重置所有子指令进度 for resetIdx in range(instructionListLength): resetInfo = self.instructionList[resetIdx] reset_tid = str(resetInfo["id"]) or "" reset_target_type = resetInfo["target_type"] or "" reset_loop = "loop" in resetInfo and int(resetInfo["loop"]) or 1 self.updateDetails.emit(str(self.parentId) + reset_tid, 0, reset_loop) # 执行任务 while not self.currentTask.isFinished and self.currentIndex < instructionListLength and self.running: taskInstructionInfo = self.instructionList[self.currentIndex] self.currentTask.setTargetId(taskInstructionInfo["target_id"]) self.currentTask.callBack(1) task_instruction_id = str(taskInstructionInfo["id"]) or "" target_id = str(taskInstructionInfo["target_id"]) or "" target_type = taskInstructionInfo["target_type"] or "" level = str(taskInstructionInfo["level"]) or "" target_loop = "loop" in taskInstructionInfo and int(taskInstructionInfo["loop"]) or 1 target_delay = "delay" in taskInstructionInfo and float(int(taskInstructionInfo["delay"])/1000) or 0 if self.parentId == "": self.parentId = self.id if target_type == "instruction": self.updateDetails.emit(str(self.parentId) + task_instruction_id, 0, target_loop) for j in range(target_loop): loopStartTime = time.time() self.currentTask.callBack(2) if self.currentTask.isFinished or not self.running: self.updateDetails.emit(str(self.parentId) + task_instruction_id, -1, -1) break progress_cb = lambda: self.updateProgress.emit(self.id, int(self.currentIndex + int(taskIndex * instructionListLength)), total) self._run_single_instruction(currentInstruction, taskInstructionInfo, proInfo, None, lambda: not self.currentTask.isFinished, progress_cb) self.updateDetails.emit(str(self.parentId) + task_instruction_id, j + 1, target_loop) tmpTime = time.time() if tmpTime < loopStartTime + target_delay: time.sleep(target_delay - (tmpTime - loopStartTime)) # time.sleep(target_delay) 指令延时在指令内部处理 elif target_type == "task": # 执行子任务 for j in range(target_loop): if not self.running: break self.currentTask.callBack(5) self.subTaskActuator = TaskActuator(target_id) self.subTaskActuator.parentId = str(self.parentId) + task_instruction_id self.subTaskActuator.task_instruction_id = str(self.parentId) + task_instruction_id taskInfo = taskManager.getInfo(target_id) taskInfo["params"] = taskInstructionInfo["target_param"] taskInstructions = taskInstructionManager.getInfo(target_id) common.getTaskDetails(taskInstructions) self.subTaskActuator.updateDetails.connect(self.onUpdateDetails, type=Qt.ConnectionType.DirectConnection) self.subTaskActuator.execute(taskInstructions, taskInfo) self.subTaskActuator.wait() self.currentTask.callBack(6) time.sleep(target_delay) # 子任务执行完毕 self.currentIndex += 1 if loop > 0: if self.currentTask.isFinished: self.updateProgress.emit(self.id,1,-1) # 根节点完成后保持绿色完成状态（value=total, max=total） self.updateDetails.emit(self.id, total, total) elif self.running: if self.task_instruction_id != "": self.updateDetails.emit( self.task_instruction_id, int( self.currentIndex + int(taskIndex*instructionListLength)), total) else: self.updateDetails.emit(self.id, int( self.currentIndex + int(taskIndex*instructionListLength)), total) self.updateProgress.emit(self.id, int( self.currentIndex + int(taskIndex*instructionListLength)), total) taskIndex += 1 self.currentTask.next() if loop > 1: time.sleep(delay) self.updateProgress.emit(str(self.taskInfo["id"]),1,-1) currentInstruction.deleteLater() elif self.type == TaskActuator.VIRTUAL and instructionListLength > 0: session = None currentInstruction = InstructionModel() self.currentIndex = 0 total = instructionListLength # 执行任务 while self.currentIndex < instructionListLength and self.running: taskInstructionInfo = self.instructionList[self.currentIndex] target_id = str(taskInstructionInfo["target_id"]) or "" target_type = taskInstructionInfo["target_type"] or "" target_loop = "loop" in taskInstructionInfo and int(taskInstructionInfo["loop"]) or 1 target_delay = "delay" in taskInstructionInfo and float(int(taskInstructionInfo["delay"])/1000) or 0 if target_type == "instruction": for j in range(target_loop): if not self.running: break self._run_single_instruction(currentInstruction, taskInstructionInfo, proInfo, None, lambda: True) time.sleep(target_delay) elif target_type == "task": # 执行子任务 for j in range(target_loop): self.subTaskActuator = TaskActuator(target_id) taskInfo = taskManager.getInfo(target_id) taskInstructions = taskInstructionManager.getInfo(target_id) common.getTaskDetails(taskInstructions) self.subTaskActuator.execute(taskInstructions, taskInfo) self.subTaskActuator.wait() time.sleep(target_delay) # 子任务执行完毕 self.currentIndex += 1 currentInstruction.deleteLater() elif self.type == TaskActuator.INSTRUCTION and instructionListLength == 1: session = None currentInstruction = InstructionModel() taskInstructionInfo = self.instructionList[0] target_id = str(taskInstructionInfo["target_id"]) or "" target_type = taskInstructionInfo["target_type"] or "" target_loop = "loop" in taskInstructionInfo and int(taskInstructionInfo["loop"]) or 1 target_delay = "delay" in taskInstructionInfo and float(int(taskInstructionInfo["delay"])/1000) or 0 if target_type == "instruction": self.updateInstructProgress.emit(target_id, 0, target_loop) for j in range(target_loop): if not self.running: break progress_cb = lambda: self.updateInstructProgress.emit(target_id, j, target_loop) self._run_single_instruction(currentInstruction, taskInstructionInfo, proInfo, None, lambda: True, progress_cb) time.sleep(target_delay) self.updateInstructProgress.emit(self.id,1,-1) currentInstruction.deleteLater() except Exception as e: try: common.showAlert.emit("EXEC错误", str(e), 0) except Exception as e: log.info(str(e)) finally: try: currentInstruction.ioCtrl.disconnect(session.ioctrl) currentInstruction.sendData.disconnect(session.send) currentInstruction.logMsg.disconnect(self.onLogMsg) session.newDataArrive.disconnect(currentInstruction.dataHandler) session.unlock() except: pass if proId != None and proId != "": projectManager.addHistory(proId, { "event": "结束", "type": self.type, "name": self.name, "operator": operator, }) self.updateInstructProgress.emit(self.id,1,-1) self.updateProgress.emit(self.id,1,-1) # 根节点完成后保持绿色完成状态 total = len(self.instructionList) * int(self.taskInfo.get("loop", 1)) self.updateDetails.emit(self.id, total, total) self.executeFinished.emit(self.id) self.quit() # currentInstruction.deleteLater() # currentInstruction = None # gc.collect() # 手动触发垃圾回收 # snapshot = tracemalloc.take_snapshot() # # 打印内存分配的前10条统计信息 # top_stats = snapshot.statistics('lineno') # print("内存分配的前10行：") # for stat in top_stats[:10]: # print(stat) def _run_single_instruction(self, currentInstruction, taskInstructionInfo, proInfo, session_ref, stop_check, progress_cb=None): """执行单次指令：连接信号、锁session、运行、解锁。返回使用的session。""" interfaceInfos = interfaceManager.getInfo(taskInstructionInfo["interface"]) if not interfaceInfos: return None currentInstruction.setInfo(taskInstructionInfo["instructionInfo"], taskInstructionInfo["target_param"]) currentInstruction.setProInfo(proInfo) currentInstruction.setUserInfo(userManager.getCurrentUser()) currentInstruction.setDevInfo(taskInstructionInfo["device"]) currentInstruction.setInterfaceInfo(interfaceInfos[0]) session = interfaceManager.getSession(taskInstructionInfo["interface"]) try: currentInstruction.logMsg.disconnect(self.onLogMsg) currentInstruction.ioCtrl.disconnect(session.ioctrl) currentInstruction.sendData.disconnect(session.send) session.newDataArrive.disconnect(currentInstruction.dataHandler) except Exception: pass currentInstruction.logMsg.connect(self.onLogMsg, type=Qt.ConnectionType.DirectConnection) currentInstruction.ioCtrl.connect(session.ioctrl, type=Qt.ConnectionType.DirectConnection) currentInstruction.sendData.connect(session.send, type=Qt.ConnectionType.DirectConnection) while not session.lock() and self.running and stop_check(): if progress_cb: progress_cb() QThread.msleep(1) if not self.running or not stop_check(): return None session.newDataArrive.connect(currentInstruction.dataHandler, type=Qt.ConnectionType.DirectConnection) currentInstruction.start() while not currentInstruction.isFinished and self.running and stop_check(): if progress_cb: progress_cb() currentInstruction.loop() QThread.msleep(1) try: currentInstruction.logMsg.disconnect(self.onLogMsg) currentInstruction.ioCtrl.disconnect(session.ioctrl) currentInstruction.sendData.disconnect(session.send) session.newDataArrive.disconnect(currentInstruction.dataHandler) except Exception: pass session.unlock() return session def onLogMsg(self, data): logManager.addLogMsg(data) self.logMsg.emit(data) # 执行指令或者队列 def execute(self, target, taskInfo = {}): if self.isRunning(): return False self.running = True if isinstance(target, list): #队列 self.instructionList = target self.taskInfo = taskInfo self.type = int(taskInfo.get("type", TaskActuator.TASK)) self.name = taskInfo.get("name", "") self.start() return True elif isinstance(target, dict): #指令 self.instructionList = [] self.name = target["instructionInfo"]["name"] self.instructionList.append(target) self.type = TaskActuator.INSTRUCTION self.start() return True return False def stopSub(self): self.running_lock.lock() self.running = False self.running_lock.unlock() # 停止子任务 if self.subTaskActuator and self.subTaskActuator.isRunning(): self.subTaskActuator.stopSub() self.updateDetails.emit(str(self.parentId) , -1, -1) self.quit() def stop(self): self.running_lock.lock() self.running = False self.running_lock.unlock() # 停止子任务 if self.subTaskActuator and self.subTaskActuator.isRunning(): self.subTaskActuator.stopSub() self.quit() def forceStop(self): self.running_lock.lock() self.running = False self.running_lock.unlock() if self.subTaskActuator and self.subTaskActuator.isRunning(): self.subTaskActuator.forceStop() if self.currentTask and hasattr(self.currentTask, 'isFinished'): self.currentTask.isFinished = True # 等待线程自然退出（最多500ms），避免 terminate 造成锁泄漏 if self.isRunning(): self.wait(500) # 超时仍在运行才强杀，并释放所有 session 锁 if self.isRunning(): for item in self.instructionList: try: iface = item.get("interface") if iface: interfaceManager.getSession(iface).unlock() except Exception: pass self.terminate() self.wait(200) self.updateDetails.emit(self.id, -1, -1) self.taskStop.emit(self.id) log.info(f"任务 {self.id} 已强制停止") defaultTaskActuator = TaskActuator("defaultTaskActuator")