<commit_before># -*- coding: utf-8 -*-
# __author: rock
# @time: 2021-07-19

import time
import struct
from .utils import xarm_is_connected, xarm_is_pause, check_modbus_baud, xarm_is_not_simulation_mode
from ..core.config.x_config import XCONF
from ..core.utils.log import logger
from ..core.utils import convert
from .code import APIState
from .gpio import GPIO

TRACK_BAUD = 2000000


class Track(GPIO):

    def __init__(self):
        super(Track, self).__init__()
        self._linear_track_error_code = 0
        self._linear_track_is_stop = False

    @property
    def linear_track_error_code(self):
        return self._linear_track_error_code

    @linear_track_error_code.setter
    def linear_track_error_code(self, val):
        self._linear_track_error_code = val

    @xarm_is_connected(_type='set')
    @xarm_is_not_simulation_mode(ret=0)
    @check_modbus_baud(baud=TRACK_BAUD, _type='set', default=None)
    def set_linear_track_enable(self, enable):
        value = convert.u16_to_bytes(int(enable))
        ret = self.arm_cmd.track_modbus_w16s(XCONF.ServoConf.CON_EN, value, 1)
        _, err = self.get_linear_track_error()
        ret[0] = self._check_modbus_code(ret, length=8, host_id=XCONF.LINEER_TRACK_HOST_ID)
        if ret[0] == 0 and self._linear_track_error_code == 0 and enable:
            self.linear_track_is_enabled = True
        else:
            self.linear_track_is_enabled = False
        self.get_linear_track_enable()
        self.log_api_info('API -> set_linear_track_enable(enable={}) -> code={}, err={}'.format(enable, ret[0], err), code=ret[0])
        return ret[0] if self._linear_track_error_code == 0 else APIState.END_EFFECTOR_HAS_FAULT

    @xarm_is_connected(_type='set')
    @xarm_is_not_simulation_mode(ret=0)
    @check_modbus_baud(baud=TRACK_BAUD, _type='set', default=None)
    def set_linear_track_back_origin(self, wait=True, **kwargs):
        auto_enable = kwargs.get('auto_enable', True)
        timeout = kwargs.get('timeout', 10)
        ret = self.arm_cmd.track_modbus_r16s(XCONF.ServoConf.BACK_ORIGIN, 1, 0x06)
        _, err = self.get_linear_track_error()
        ret[0] = self._check_modbus_code(ret, length=8, host_id=XCONF.LINEER_TRACK_HOST_ID)
        self.log_api_info('API -> set_linear_track_back_origin() -> code={}, err={}'.format(ret[0], err), code=ret[0])
        if ret[0] == 0 and wait:
            ret[0] = self.__wait_linear_track_back_origin(timeout)
        if auto_enable:
            ret[0] = self.set_linear_track_enable(True)
        return ret[0] if self._linear_track_error_code == 0 else APIState.END_EFFECTOR_HAS_FAULT

    @xarm_is_connected(_type='set')
    @xarm_is_not_simulation_mode(ret=0)
    @check_modbus_baud(baud=TRACK_BAUD, _type='set', default=None)
    def set_linear_track_pos(self, pos, speed=None, wait=True, timeout=100, **kwargs):
        auto_enable = kwargs.get('auto_enable', False)
        if auto_enable and not self.linear_track_is_enabled:
            self.set_linear_track_enable(auto_enable)
        if speed is not None and self.linear_track_speed != speed:
            self.set_linear_track_speed(speed)

        value = convert.int32_to_bytes(int(pos * 2000), is_big_endian=True)
        ret = self.arm_cmd.track_modbus_w16s(XCONF.ServoConf.TAGET_POS, value, 2)
        _, err = self.get_linear_track_error()
        ret[0] = self._check_modbus_code(ret, length=8, host_id=XCONF.LINEER_TRACK_HOST_ID)
        self.log_api_info('API -> set_track_pos(pos={}) -> code={}, err={}'.format(pos, ret[0], err), code=ret[0])

        if ret[0] == 0 and wait:
            return self.__wait_linear_track_stop(timeout)
        return ret[0] if self._linear_track_error_code == 0 else APIState.END_EFFECTOR_HAS_FAULT

    @xarm_is_connected(_type='set')
    @xarm_is_not_simulation_mode(ret=0)
    @check_modbus_baud(baud=TRACK_BAUD, _type='set', default=None)
    def set_linear_track_speed(self, speed):
        value = convert.u16_to_bytes(int(speed))
        ret = self.arm_cmd.track_modbus_w16s(XCONF.ServoConf.POS_SPD, value, 1)
        ret[0] = self._check_modbus_code(ret, length=8, host_id=XCONF.LINEER_TRACK_HOST_ID)
        _, err = self.get_linear_track_error()
        if ret[0] == 0 and self._linear_track_error_code == 0:
            self.linear_track_speed = speed
        self.log_api_info('API -> set_linear_track_speed(speed={}) -> code={}, err={}'.format(speed, ret[0], err), code=ret[0])
        return ret[0] if self._linear_track_error_code == 0 else APIState.END_EFFECTOR_HAS_FAULT

    @xarm_is_connected(_type='get')
    @xarm_is_not_simulation_mode(ret=(0, 0))
    @check_modbus_baud(baud=TRACK_BAUD, _type='get', default=None)
    def check_linear_track_on_zero(self):
        ret = self.arm_cmd.track_modbus_r16s(XCONF.ServoConf.CHECK_ON_ORIGIN, 1)
        _, err = self.get_linear_track_error()
        ret[0] = self._check_modbus_code(ret, length=7, host_id=XCONF.LINEER_TRACK_HOST_ID)
        code = ret[0] if self._linear_track_error_code == 0 else APIState.END_EFFECTOR_HAS_FAULT
        # self.log_api_info('API -> check_linear_track_on_zero() -> code={}, err={}'.format(ret[0], err), code=ret[0])
        return code, ret[-1] if code == 0 else 0

    @xarm_is_connected(_type='get')
    @xarm_is_not_simulation_mode(ret=(0, 0))
    @check_modbus_baud(baud=TRACK_BAUD, _type='get', default=-99)
    def get_linear_track_status(self):<commit_msg>fix speed<commit_after># -*- coding: utf-8 -*-
# __author: rock
# @time: 2021-07-19

import time
import struct
from .utils import xarm_is_connected, xarm_is_pause, check_modbus_baud, xarm_is_not_simulation_mode
from ..core.config.x_config import XCONF
from ..core.utils.log import logger
from ..core.utils import convert
from .code import APIState
from .gpio import GPIO

TRACK_BAUD = 2000000


class Track(GPIO):

    def __init__(self):
        super(Track, self).__init__()
        self._linear_track_error_code = 0
        self._linear_track_is_stop = False

    @property
    def linear_track_error_code(self):
        return self._linear_track_error_code

    @linear_track_error_code.setter
    def linear_track_error_code(self, val):
        self._linear_track_error_code = val

    @xarm_is_connected(_type='set')
    @xarm_is_not_simulation_mode(ret=0)
    @check_modbus_baud(baud=TRACK_BAUD, _type='set', default=None)
    def set_linear_track_enable(self, enable):
        value = convert.u16_to_bytes(int(enable))
        ret = self.arm_cmd.track_modbus_w16s(XCONF.ServoConf.CON_EN, value, 1)
        _, err = self.get_linear_track_error()
        ret[0] = self._check_modbus_code(ret, length=8, host_id=XCONF.LINEER_TRACK_HOST_ID)
        if ret[0] == 0 and self._linear_track_error_code == 0 and enable:
            self.linear_track_is_enabled = True
        else:
            self.linear_track_is_enabled = False
        self.get_linear_track_enable()
        self.log_api_info('API -> set_linear_track_enable(enable={}) -> code={}, err={}'.format(enable, ret[0], err), code=ret[0])
        return ret[0] if self._linear_track_error_code == 0 else APIState.END_EFFECTOR_HAS_FAULT

    @xarm_is_connected(_type='set')
    @xarm_is_not_simulation_mode(ret=0)
    @check_modbus_baud(baud=TRACK_BAUD, _type='set', default=None)
    def set_linear_track_back_origin(self, wait=True, **kwargs):
        auto_enable = kwargs.get('auto_enable', True)
        timeout = kwargs.get('timeout', 10)
        ret = self.arm_cmd.track_modbus_r16s(XCONF.ServoConf.BACK_ORIGIN, 1, 0x06)
        _, err = self.get_linear_track_error()
        ret[0] = self._check_modbus_code(ret, length=8, host_id=XCONF.LINEER_TRACK_HOST_ID)
        self.log_api_info('API -> set_linear_track_back_origin() -> code={}, err={}'.format(ret[0], err), code=ret[0])
        if ret[0] == 0 and wait:
            ret[0] = self.__wait_linear_track_back_origin(timeout)
        if auto_enable:
            ret[0] = self.set_linear_track_enable(True)
        return ret[0] if self._linear_track_error_code == 0 else APIState.END_EFFECTOR_HAS_FAULT

    @xarm_is_connected(_type='set')
    @xarm_is_not_simulation_mode(ret=0)
    @check_modbus_baud(baud=TRACK_BAUD, _type='set', default=None)
    def set_linear_track_pos(self, pos, speed=None, wait=True, timeout=100, **kwargs):
        auto_enable = kwargs.get('auto_enable', False)
        if auto_enable and not self.linear_track_is_enabled:
            self.set_linear_track_enable(auto_enable)
        if speed is not None and self.linear_track_speed != speed:
            self.set_linear_track_speed(speed)

        value = convert.int32_to_bytes(int(pos * 2000), is_big_endian=True)
        ret = self.arm_cmd.track_modbus_w16s(XCONF.ServoConf.TAGET_POS, value, 2)
        _, err = self.get_linear_track_error()
        ret[0] = self._check_modbus_code(ret, length=8, host_id=XCONF.LINEER_TRACK_HOST_ID)
        self.log_api_info('API -> set_track_pos(pos={}) -> code={}, err={}'.format(pos, ret[0], err), code=ret[0])

        if ret[0] == 0 and wait:
            return self.__wait_linear_track_stop(timeout)
        return ret[0] if self._linear_track_error_code == 0 else APIState.END_EFFECTOR_HAS_FAULT

    @xarm_is_connected(_type='set')
    @xarm_is_not_simulation_mode(ret=0)
    @check_modbus_baud(baud=TRACK_BAUD, _type='set', default=None)
    def set_linear_track_speed(self, speed):
        value = convert.u16_to_bytes(int(speed*6.667))
        ret = self.arm_cmd.track_modbus_w16s(XCONF.ServoConf.POS_SPD, value, 1)
        ret[0] = self._check_modbus_code(ret, length=8, host_id=XCONF.LINEER_TRACK_HOST_ID)
        _, err = self.get_linear_track_error()
        if ret[0] == 0 and self._linear_track_error_code == 0:
            self.linear_track_speed = speed
        self.log_api_info('API -> set_linear_track_speed(speed={}) -> code={}, err={}'.format(speed, ret[0], err), code=ret[0])
        return ret[0] if self._linear_track_error_code == 0 else APIState.END_EFFECTOR_HAS_FAULT

    @xarm_is_connected(_type='get')
    @xarm_is_not_simulation_mode(ret=(0, 0))
    @check_modbus_baud(baud=TRACK_BAUD, _type='get', default=None)
    def check_linear_track_on_zero(self):
        ret = self.arm_cmd.track_modbus_r16s(XCONF.ServoConf.CHECK_ON_ORIGIN, 1)
        _, err = self.get_linear_track_error()
        ret[0] = self._check_modbus_code(ret, length=7, host_id=XCONF.LINEER_TRACK_HOST_ID)
        code = ret[0] if self._linear_track_error_code == 0 else APIState.END_EFFECTOR_HAS_FAULT
        # self.log_api_info('API -> check_linear_track_on_zero() -> code={}, err={}'.format(ret[0], err), code=ret[0])
        return code, ret[-1] if code == 0 else 0

    @xarm_is_connected(_type='get')
    @xarm_is_not_simulation_mode(ret=(0, 0))
    @check_modbus_baud(baud=TRACK_BAUD, _type='get', default=-99)
    def get_linear_track_status(self):
{'fraction_non_alphanumeric': 0.12029657089898053, 'fraction_numerical': 0.019833178869323446, 'mean_word_length': 3.975103734439834, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '717880', 'n_tokens_mistral': 4034, 'n_tokens_neox': 3919, 'n_words': 622}