TG-PlatformPlus/taskInstructionModel/taskInstructionManager.py

#!/opt/homebrew/bin/python3
# -*- coding:utf-8 -*-
import json
import uuid
from PyQt6 import *
from PyQt6.QtCore import *
from logs import log
from typing import Union
from models import TaskInstruction, Session

class TaskInstructionManager(QObject):
 
    def __init__(self, parent=None):
        super().__init__()
    
    @pyqtSlot(QVariant, result=str)
    def create(self, param):
        if not type(param) == dict:
            param = param.toVariant()
        param["id"] = str(uuid.uuid4())
        ok, res = Session.addByClass(TaskInstruction, param)
        if ok:
            return str(res["id"])
        return ""
    
    @pyqtSlot(QVariant, result=str)
    def create_from_json(self, param):
        if not isinstance(param, dict):
            param = param.toVariant()
        ok, res = Session.addByClass(TaskInstruction, param)
        if ok:
            return str(res["id"])
        return ""

    @pyqtSlot(QVariant, result=str)
    def create_from_git_json(self, param):
        if not isinstance(param, dict):
            param = param.toVariant()
        ok, res = Session.addByClassFromGit(TaskInstruction, param)
        if ok:
            return str(res["id"])
        return ""

    @pyqtSlot(str, QVariant, result=bool)
    def update(self, id, param=None):
        if not type(param) == dict:
            param = param.toVariant()
        data = {}
        if "task_id" in param:
            data["task_id"] = param["task_id"]
        if "device_id" in param:
            data["device_id"] = param["device_id"]
        if "loop" in param:
            data["loop"] = param["loop"]
        if "delay" in param:
            data["delay"] = param["delay"]
        if "target_type" in param:
            data["target_type"] = param["target_type"]
        if "target_id" in param:
            data["target_id"] = param["target_id"]
        if "target_param" in param:
            data["target_param"] = param["target_param"]
        if "interface_index" in param:
            data["interface_index"] = param["interface_index"]
        if "level" in param:
            data["level"] = param["level"]
        ok,res = Session.updateById(TaskInstruction, id, data)
        return ok

    @pyqtSlot(str, result=bool)
    @pyqtSlot(list, result=bool)  
    def delete(self, id):
        if isinstance(id, str): 
            ok,res = Session.deleteById(TaskInstruction, id)
            return ok
        else:
            ok,res = Session.deleteByIds(TaskInstruction, id)
            return ok
    
    @pyqtSlot(str, result=str)
    def copy(self, id, copy_task_id):
        ok,res = Session.queryById(TaskInstruction, id)
        if ok:
            res["task_id"] = copy_task_id
            return self.create(res)
    
    @pyqtSlot(str, result=bool)
    def copyTask(self, task_id, new_task_id):
        instructionLst = self.getTaskInstructions(task_id)
        for instruction in instructionLst:
            self.copy(instruction["id"], new_task_id)
        return True

    # 获取所有属于设备模型的指令
    @pyqtSlot(int, result=QVariant)
    def getTaskInstructions(self, task_id):
        ok, res = Session.queryByTaskId(TaskInstruction, task_id)
        if ok:
            return res
        else:
            return []

    @pyqtSlot(str, result=QVariant)
    @pyqtSlot(str, result=list)
    def getInstructionInfo(self, id = "all"):
        if id == "all": 
            ok,res = Session.queryByAll(TaskInstruction)
            if ok:
                return res
            else:
                return []
        else:
            ok,res = Session.queryById(TaskInstruction, id)
            if ok:
                return res
            return None
    @pyqtSlot(str, result=QVariant)
    @pyqtSlot(str, result=list)
    def getInfo(self, task_id = "all"):
        if task_id == "all": 
            ok,res = Session.queryByAll(TaskInstruction)
            if ok:
                return res
            else:
                return []
        else:
            ok,res = Session.queryByTaskId(TaskInstruction, task_id)
            if ok:
                return res
            return []

    # 获取全部包含子任务的任务
    def getIncludingsubtasks(self):
        ok,res = Session.queryTaskByTargetType(TaskInstruction, "task")
        if ok:
            return res
        return []

    def getPageInfo(self, task_id, start_row=0, limit=50):
        ok,res = Session.pagedQueryByTaskId(TaskInstruction, task_id, start_row, limit)
        if ok:
            return res
        return []
taskInstructionManager = TaskInstructionManager()