simAGV/plc/cmd.py

#!/usr/bin/env python
# -*- coding: utf_8 -*-

import modbus_tk.defines as cst
from plc.tools import bytes_to_int_list, get_bit_bool, int_list_to_bytes
from plc.modbus import ModbusConnection
from conf.setting import AGVPLC
from models.path import Path
import time


class TaskAddr:
    
    # 起始地址 
    TASKTYPE_ADDR_BEGIN = 15000
    
    # AGV路径点集合
    PATH_ADDR_BEGIN = 15001
    # PATH_ADDR_LENGTH = 42

    #是否经过玻璃门
    PASS_DOOR_ADDR = 15051

    #货架第几层
    LAYER_ADDR = 15052

    #机械臂是否动作
    ARM_ADDR = 15053

    #任务数量
    TASK_NUM = 15060

    #任务取放开始地址 15061, 15062 -> 15079, 15080
    T1_ADDR_GET_ADDR = 15061
    T1_ADDR_RETURN_ADDR = 15062

    #任务截至地址
    T10_ADDR_GET_ADDR = 15079
    T10_ADDR_RETURN_ADDR = 15080

    #

    ######
    # Read Registers
    ######
    #任务是否完成
    TASK_DONE_ADDR = 16000
    #任务消息接收
    TASK_RECV_ADDR = 16001
    #AGV所在站点
    AGV_STOP_ADDR = 16002
    #AGV当前电量
    AGV_POWER_ADDR = 16003
    #AGV充电状态
    AGV_CHARGE_ADDR = 16004
    #设备忙
    BUSY_ADDR = 16005
    #设备异常
    ABNORM_ADDR = 16006
    #报警信息
    ALARM_ADDR = 16007

    #状态寄存器长度
    # 连续读取8字节，单字节长度为16位
    STATUS_ADDR_LENGTH = 8


#
def cmd_type(cmdType: int):
    try:
        with ModbusConnection(
                # 试剂amr机器人
                AGVPLC.HOST,
                AGVPLC.PORT
        ) as master:
            res = master.execute(
                1,
                cst.WRITE_SINGLE_REGISTER,
                starting_address=TaskAddr.TASKTYPE_ADDR_BEGIN,
                output_value=cmdType
            )
            print('cmd_type: ',res)
            if res:
                task_status = False

    except Exception as e:
        print('cmd_type信息指令异常：' + str(e))
        time.sleep(2)

async def setSubPaths(start: str, stop: str):
    path = await Path.filter(start = start, stop = stop, is_valid = True, is_active = True)

    subPaths = []

    if path and len(path) > 0:
        subPaths = path.paths

        try:
            with ModbusConnection(
                    # 试剂amr机器人
                    AGVPLC.HOST,
                    AGVPLC.PORT
            ) as master:
                res = master.execute(
                    1,
                    cst.WRITE_MULTIPLE_REGISTERS,
                    starting_address=TaskAddr.TASKTYPE_ADDR_BEGIN,
                    output_value=subPaths
                )
                print('cmd_type: ',res)
                if res:
                    task_status = False

        except Exception as e:
            print('cmd_type信息指令异常：' + str(e))
            time.sleep(2)



#
def pass_door(passDoor: int):
    try:
        with ModbusConnection(
                # 试剂amr机器人
                AGVPLC.HOST,
                AGVPLC.PORT
        ) as master:
            res = master.execute(
                1,
                cst.WRITE_SINGLE_REGISTER,
                starting_address=TaskAddr.PASS_DOOR_ADDR,
                output_value=passDoor
            )
            print('cmd_type: ',res)
            if res:
                task_status = False

    except Exception as e:
        print('cmd_type信息指令异常：' + str(e))
        time.sleep(2)

#
def setLayer(layer: int):
    try:
        with ModbusConnection(
                # 试剂amr机器人
                AGVPLC.HOST,
                AGVPLC.PORT
        ) as master:
            res = master.execute(
                1,
                cst.WRITE_SINGLE_REGISTER,
                starting_address=TaskAddr.LAYER_ADDR,
                output_value=layer
            )
            print('cmd_type: ',res)
            if res:
                task_status = False

    except Exception as e:
        print('cmd_type信息指令异常：' + str(e))
        time.sleep(2)

#
def moveArm(move: int):
    try:
        with ModbusConnection(
                # 试剂amr机器人
                AGVPLC.HOST,
                AGVPLC.PORT
        ) as master:
            res = master.execute(
                1,
                cst.WRITE_SINGLE_REGISTER,
                starting_address=TaskAddr.ARM_ADDR,
                output_value=move
            )
            print('cmd_type: ',res)
            if res:
                task_status = False

    except Exception as e:
        print('cmd_type信息指令异常：' + str(e))
        time.sleep(2)

#
def taskNum(num: int):
    try:
        with ModbusConnection(
                # 试剂amr机器人
                AGVPLC.HOST,
                AGVPLC.PORT
        ) as master:
            res = master.execute(
                1,
                cst.WRITE_SINGLE_REGISTER,
                starting_address=TaskAddr.TASK_NUM,
                output_value=num
            )
            print('cmd_type: ',res)
            if res:
                task_status = False

    except Exception as e:
        print('cmd_type信息指令异常：' + str(e))
        time.sleep(2)


def setTasks(tasks = []):

    try:
        with ModbusConnection(
                # 试剂amr机器人
                AGVPLC.HOST,
                AGVPLC.PORT
        ) as master:
            res = master.execute(
                1,
                cst.WRITE_MULTIPLE_REGISTERS,
                starting_address=TaskAddr.T1_ADDR_GET_ADDR,
                output_value=tasks
            )
            print('cmd_type: ',res)
            if res:
                task_status = False

    except Exception as e:
        print('cmd_type信息指令异常：' + str(e))
        time.sleep(2)


def readStatus():

    with ModbusConnection(
        AGVPLC.HOST,
        AGVPLC.PORT
    ) as master:
        status = master.execute(
            1,
            cst.READ_HOLDING_REGISTERS,
            TaskAddr.TASK_DONE_ADDR,
            TaskAddr.STATUS_ADDR_LENGTH
        )
        print('readStatus: ',status)
        return status