kraken/backend/device/fubo_pneumatic_finger.py

'''
@Author  :   liujunshen
@Ide     :   vscode
@File    :   fubo_pneumatic_finger.py
@Time    :   2023/04/03 16:49:11
'''


import logging
import time

import serial
from serial.tools.list_ports import comports


logger = logging.getLogger(__name__)


def get_serial_ports():
    """获取可选端口"""
    ports = list(comports(include_links=False))
    available_ports_list = [port.device for port in ports]
    return available_ports_list


class FuboPneumaticFingerClient:
    """富伯客户端"""

    FLEX_CMD = b"F"
    EXTEND_CMD = b"E"
    BALL_CMD = b"B"
    CYLINDER_CMD = b"C"
    DOUBLE_CMD = b"D"
    TREBLE_CMD = b"T"
    RELEASE_CMD = b"R"

    def __init__(self, init_params=None):
        self.baud_rate = 9600
        self.data_bite = 8
        self.timeout = 1
        self.stop_bit = serial.STOPBITS_ONE  # 停止位
        self.parity_bit = serial.PARITY_NONE  # 校验位
        self.is_connected = False
        self.ser = None  # 连接对象
        self.port = "COM 4"
        if init_params:
            self.port = init_params["port"]
        
        self.connect()

    def __del__(self):
        self.ser.close()

    def connect(self):
        try:
            self.ser = serial.Serial(port=self.port,
                                     baudrate=self.baud_rate,
                                     parity=self.parity_bit,
                                     stopbits=self.stop_bit,
                                     timeout=self.timeout)
            if self.ser.is_open:
                self.is_connected = True
                logger.info("connect")
                return 1
            else:
                self.is_connected = False
                logger.warning("open failed")
                return 0
        except OSError as e:
            warning_info = f"pneumatic finger connect failed: {e}"
            logger.warning(warning_info)
            return 0
    
    def release(self):
        self.ser.write(self.RELEASE_CMD)
        return self.ser.read()

    def extend(self):
        self.ser.write(self.EXTEND_CMD)
        return self.ser.read()

    def reconnect(self):
        self.close()
        return self.connect()
    
    def start(self, model=None):
        if (model == "flex") or (model is None):
            self.ser.write(self.FLEX_CMD)
        elif model == "ball":
            self.ser.write(self.BALL_CMD)
        elif model == "cylinder":
            self.ser.write(self.CYLINDER_CMD)
        elif model == "double":
            self.ser.write(self.DOUBLE_CMD)
        elif model == "treble":
            self.ser.write(self.TREBLE_CMD)
        return self.ser.read()

    def start_round(self, model=None, time_interval=5):        
        self.start(model)
        time.sleep(time_interval)
        self.extend()
        return 1

    def stop(self):
        return 1

    def status(self):
        status = {"is_connected": self.is_connected}
        return status

    def close(self):
        if self.ser:
            self.ser.close()
            self.is_connected = False
        return {"is_connected": self.is_connected}