simAGV/plc/test.py

#!/usr/bin/env python
# -*- coding: utf_8 -*-
import struct
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


'''
智能柜地址定义
'''
ADDR_DRAWER = 6051
ADDR_CABINET = 6052
ADDR_OPEN_DRAWER = 6053
ADDR_CLOSE_DRAWER = 6054


class Cabinet:
    '''
    智能柜配置
    '''
    HOST = '192.168.193.211'
    PORT = 6000

class Worktop:
    '''
    操作台配置
    '''
    HOST = '192.168.193.212'
    PORT = 502

class AmrDrug:
    '''
    药剂AMR配置PLC
    '''
    HOST = '192.168.193.202'
    PORT = 502
class AmrInstrument:
    '''
    仪器AMR配置PLC
    '''
    HOST = '192.168.193.203'
    PORT = 502


"""
操作台地址定义
"""
class WorktopAddr:
    ADDR_POWER_ONE = 6080
    ADDR_POWER_TWO  = 6081
    ADDR_TRANSFER_TRAY = 6082
    ADDR_POWER_ONE_STATUS = 6083
    ADDR_POWER_TWO_STATUS = 6084
    ADDR_TANSFER_TRAY_STATUS = 6085
    ADDR_WORKTOP_ERR = 6086
    TASK_ADDR_BEGIN = 6080   # 发送任务的起始未知
    STATUS_ADDR_BEGIN = 6083 # 状态的起始状态
    STATUS_ADDR_LENGTH = 4

class AmrAddr:
    
    # 起始地址 
    TASK_ADDR_BEGIN = 6001

    # 门状态
    CABIENT_STATUS_ADDR_BEGIN = 6061

    # 抓试剂点位的起始地址
    GRAB_ADDR_BEGIN = 7000
    # 抓取校验点的起始地址
    GRAB_MARK_ADDR_BEGIN = 7600

    # 放试剂点位的起始地址
    DEPOSIT_ADDR_BEGIN= 9000
    # 松抓校验点的起始地址
    DEPOSIT_MARK_ADDR_BEGIN= 7900

    
    # 状态起始地址及长度
    STATUS_ADDR_BEGIN = 6100
    STATUS_ADDR_LENGTH = 42

    # 取货起始地址及长度
    LOAD_LOCATION_BEGIN = 7000
    LOAD_LOCATION_COUNT = 25

    # 卸货起始地址及长度
    UNLOAD_LOCATION_BEGIN = 7300
    UNLOAD_LOCATION_COUNT = 25

    TASK_STATUS = 6100  # 状态显示1
    AMR_STATUS = 6101  # 状态显示2
    INSTRUMENT_STATUS_1 = 6102  # 状态显示3
    D6103 = 6103
    D6104 = 6104
    INSTRUMENT_STATUS_2 = 6105
    D6105 = 6105
    D6106 = 6106
    D6107 = 6107
    D6108 = 6108
    D6109 = 6109
    INSTRUMENT_CAM_HIGH = 6110
    CUR_LOAD_LAYER_COMPLETE_NUM = 6112  # 当前层已取试剂数量
    CUR_LOAD_LAYER_COMPLETE = 6113  # 当前层试剂已取完
    CUR_LOAD_LAYER_ID = 6114  # 当前在取层数
    LOAD_NUMBER = 6115  # 合计一共已取试剂数量
    D6116 = 6116
    D6117 = 6117
    D6118 = 6118
    D6119 = 6119
    ROBOT_CUR_NAV_LOCATION = 6120  # 机器人当前导航站点
    ROBOT_LOCATING_STATUS = 6121  # 机器人定位状态
    ROBOT_CUR_NAV_STATUS = 6122  # 机器人当前导航状态
    ROBOT_CUR_NAV_TYPE = 6123  # 机器人当前导航类型
    ROBOT_CONFIDENCE_LEVEL = 6124  # 机器人定位置信度
    BATTERY_STATUS = 6126  # 电池电量
    BATTERY_TEM = 6128  # 电池温度
    BATTERY_VOLTAGE = 6130  # 电池电压
    ROBOT_CUR_LOCATION = 6132  # 机器人当前所在站点
    ROBOT_PRE_LOCATION = 6133  # 机器人上一个所在站点
    ROBOT_NXT_LOCATION = 6134  # 机器人下一个要经过的站点
    BLOCK_REASON = 6135  # 被阻挡的原因
    DECELERATION_REASON = 6136  # 机器人减速原因
    ROBOT_CUR_STATUS = 6137  # 状态指示



def get_worktop_status():
    """
    获取操作台状态[0,0,88,0]
    """
    with ModbusConnection(
            Worktop.HOST,
            Worktop.PORT
    ) as master:
        worktop_list = master.execute(
            1,
            cst.READ_HOLDING_REGISTERS,
            WorktopAddr.STATUS_ADDR_BEGIN,
            WorktopAddr.STATUS_ADDR_LENGTH,
        )
        print('获取旋转托盘状态：：：',worktop_list)
        # return parser_worktop_status(worktop_list)



def get_drawer_status():
    '''
    解析抽屉状态
    '''
    with ModbusConnection(
        Cabinet.HOST, 
        Cabinet.PORT
    ) as master:
        drawer_short_list = master.execute(
            1, cst.READ_HOLDING_REGISTERS, ADDR_DRAWER, 1)
        print('get_drawer_status:::', drawer_short_list)
        # return parser_drawer_status(drawer_short_list[0])


def get_cabinet_task_status():
    '''
    解析任务状态
    '''
    with ModbusConnection(
        Cabinet.HOST,
        Cabinet.PORT
    ) as master:
        task_status_data_list = master.execute(
            1, cst.READ_HOLDING_REGISTERS, ADDR_CABINET, 1)
        print('get_cabinet_task_status:::', task_status_data_list)
        # return parser_task_status(task_status_data_list[0])


def get_instrument_amr_status():
    '''
    获取试剂amr 并解析AMR状态
    '''
    try:
        with ModbusConnection(
            AmrInstrument.HOST,
            AmrInstrument.PORT
        ) as master:
            drawer_short_list = master.execute(
                1,
                cst.READ_HOLDING_REGISTERS,
                AmrAddr.STATUS_ADDR_BEGIN,
                AmrAddr.STATUS_ADDR_LENGTH
            )
            print('get_instrument_amr_status::',drawer_short_list)
            # return parser_ams_status(int_list_to_bytes(drawer_short_list))
    except Exception as e :
        # amr_status = AmrStatus()
        # return amr_status

        print("get_instrument_amr_status Exception: ", e)


def get_drug_amr_status():
    '''
    获取试剂amr 并解析AMR状态
    '''

    with ModbusConnection(
        AmrDrug.HOST,
        AmrDrug.PORT
    ) as master:
        drawer_short_list = master.execute(
            1,
            cst.READ_HOLDING_REGISTERS,
            AmrAddr.STATUS_ADDR_BEGIN,
            AmrAddr.STATUS_ADDR_LENGTH
        )
        print('get_drug_amr_status::',drawer_short_list)
        # return parser_ams_status(int_list_to_bytes(drawer_short_list))




get_worktop_status()
get_drawer_status()
get_cabinet_task_status()
get_instrument_amr_status()
get_drug_amr_status()