Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _run_command method in fMBT

Best Python code snippet using fMBT_python

controller.py

Source:controller.py

...9from pykeigan.utils import *10class Controller:11    def __init__(self):12        pass13    def _run_command(self,val,characteristics):14        pass15    @property16    def flash_memory_states(self):17        return {0:"FLASH_STATE_READY",1:"FLASH_STATE_TEACHING_PREPARE",2:"FLASH_STATE_TEACHING_DOING",3:"FLASH_STATE_PLAYBACK_PREPARE",4:"FLASH_STATE_PLAYBACK_DOING",5:"FLASH_STATE_PLAYBACK_PAUSING",6:"FLASH_STATE_TASKSET_RECORDING",7:"FLASH_STATE_TASKSET_DOING",8:"FLASH_STATE_TASKSET_PAUSING",20:"FLASH_STATE_IMU"}18    @property19    def motor_control_modes(self):20        return {0:"MOTOR_CONTROL_MODE_NONE",1:"MOTOR_CONTROL_MODE_VELOCITY",2:"MOTOR_CONTROL_MODE_POSITION",3:"MOTOR_CONTROL_MODE_TORQUE",0xFF:"MOTOR_CONTROL_MODE_OTHERS"}21    22    @property23    def baud_rates(self):24        return {0:"115200",1:"230400",2:"250000",3:"460800",4:"921600",5:"1000000"}25    @property26    def error_codes(self):27        return {28                0x00:"KM_SUCCESS",29                0x03:"KM_ERROR_NOT_FOUND",30                0x05:"KM_ERROR_INVALID_COMMAND",31                0x06:"KM_ERROR_INVALID_PARAM",32                0x07:"KM_ERROR_STORAGE_FULL",33                0x08:"KM_ERROR_INVALID_FLASH_STATE", 34                0x09:"KM_ERROR_INVALID_LENGTH",35                0x0A:"KM_ERROR_INVALID_CHECKSUM",36                0x0F:"KM_ERROR_FORBIDDEN",37                0x10:"KM_ERROR_INVALID_ADDR",38                0x14:"KM_ERROR_MOTOR_DISABLED",39                0x46:"KM_ERROR_NRF_DEVICE", 40                0x50:"KM_ERROR_WDT_EVENT",41                0x51:"KM_ERROR_OVER_HEAT",42                0x64:"KM_SUCCESS_ARRIVAL"43                }44    45    @property46    def command_names(self):47        return {48                0x00:"unknown",49                0x02:"set_max_speed",50                0x03:"set_min_speed",51                0x05:"set_curve_type",52                0x07:"set_acc",53                0x08:"set_dec",54                0x0e:"set_max_torque",55                0x16:"set_teaching_interval",56                0x17:"set_playback_interval",57                0x18:"set_qcurrent_p",58                0x19:"set_qcurrent_i",59                0x1A:"set_qcurrent_d",60                0x1B:"set_speed_p",61                0x1C:"set_speed_i",62                0x1D:"set_speed_d",63                0x1E:"set_position_p",64                0x1F:"set_position_i",65                0x20:"set_position_d",66                0x21:"set_pos_control_threshold",67                0x22:"reset_all_pid",68                0x23:"set_pid_table_value",69                0x24:"select_pid_table",70                0x25:"read_pid_table",71                0x27:"set_low_pass_filter",72                0x2B:"set_notify_pos_arrival_settings",73                0x2C:"set_motor_measurement_interval",74                0x2D:"set_motor_measurement_settings",75                0x2E:"set_interface",76                0x30:"set_response",77                0x31:"set_safe_run_settings",78                0x32:"set_safety_settings",79                0x33:"set_limit_current",80                0x37:"set_linkage_key",81                0x3A:"set_own_color",82                0x3C:"set_imu_measurement_interval",83                0x3D:"set_imu_measurement_settings",84                0x40:"read_register",85                0x41:"save_all_registers",86                0x46:"read_device_name",87                0x47:"read_device_info",88                0x4E:"reset_register",89                0x4F:"reset_all_registers",90                0x50:"disable_action",91                0x51:"enable_action",92                0x58:"set_speed",93                0x5A:"preset_position",94                0x5B:"get_position_offset",95                0x60:"run_forward",96                0x61:"run_reverse",97                0x62:"run_at_velocity",98                0x65:"move_to_pos",99                0x66:"move_to_pos",100                0x67:"move_by_dist",101                0x68:"move_by_dist",102                0x6A:"enable_action_task",103                0x6B:"disable_action_task",104                0x6C:"free_motor",105                0x6D:"stop_motor",106                0x72:"hold_torque",107                0x75:"move_to_pos_wait",108                0x76:"move_to_pos_wait",109                0x77:"move_by_dist_wait",110                0x78:"move_by_dist_wait",111                0x81:"start_doing_taskset",112                0x82:"stop_doing_taskset",113                0x85:"start_playback_motion",114                0x86:"prepare_playback_motion",115                0x87:"start_playback_motion_from_prep",116                0x88:"stop_playback_motion",117                0x90:"pause_queue",118                0x91:"resume_queue",119                0x92:"wait_queue",120                0x94:"erase_task",121                0x95:"clear_queue",122                0x96:"wait_queue_until_input",123                0x9A:"read_status",124                0xA0:"start_recording_taskset",125                0xA2:"stop_recording_taskset",126                0xA3:"erase_taskset",127                0xA4:"erase_all_tasksets",128                0xA5:"set_taskset_name",129                0xA6:"read_taskset_info",130                0xA7:"read_taskset_usage",131                0xA8:"set_trigger_taskset_settings",132                0xA9:"start_teaching_motion",133                0xAA:"prepare_teaching_motion",134                0xAB:"start_teaching_motion_from_prep",135                0xAC:"stop_teaching_motion",136                0xAD:"erase_motion",137                0xAE:"erase_all_motions",138                0xAF:"set_motion_name",139                0xB0:"read_motion_info",140                0xB1:"read_motion_usage",141                0xB2:"set_trigger_motion_settings",142                0xB7:"read_motion",143                0xB8:"write_motion_position",144                0xBC:"set_autostart_setting",145                0xBD:"set_button_setting",146                0xBE:"read_error",147                0xC0:"set_i2c_address",148                0xC3:"set_baud_rate",149                0xE0:"set_led",150                0xE6:"enable_motor_measurement",151                0xE7:"disable_motor_measurement",152                0xEA:"enable_imu_measurement",153                0xEB:"disable_imu_measurement",154                0xF0:"reboot",155                0xF3:"enable_check_sum"156                }157    158    @property159    def event_types(self):160        return {1:"button", 10:"gpio"}161    162    # Settings163    def set_max_speed(self,max_speed,identifier=b'\x00\x00'):164        """165        Set the maximum speed of rotation to the 'max_speed' in rad/sec.166        """167        command=b'\x02'168        values=float2bytes(max_speed)169        self._run_command(command+identifier+values,'motor_rx')170    def set_min_speed(self,min_speed,identifier=b'\x00\x00'):171        """172        Set the minimum speed of rotation to the 'min_speed' in rad/sec.173        """174        command=b'\x03'175        values=float2bytes(min_speed)176        self._run_command(command+identifier+values,'motor_rx')177    def set_curve_type(self,curve_type,identifier=b'\x00\x00'):178        """179        Set the acceleration or deceleration curve to the 'curve_type'.180        typedef enum curveType =181        {182            CURVE_TYPE_NONE = 0, // Turn off Motion control183            CURVE_TYPE_TRAPEZOID = 1, // Turn on Motion control with trapezoidal curve184            CURVE_TYPE_DIRECT_POS = 10 // Turn off Motion control (Direct position control)185        }186        """187        command=b'\x05'188        values=uint8_t2bytes(curve_type)189        self._run_command(command+identifier+values,'motor_rx')190    def set_acc(self,_acc,identifier=b'\x00\x00'):191        """192        Set the acceleration of rotation to the positive 'acc' in rad/sec^2.193        """194        command=b'\x07'195        values=float2bytes(_acc)196        self._run_command(command+identifier+values,'motor_rx')197    def set_dec(self,_dec,identifier=b'\x00\x00'):198        """199        Set the deceleration of rotation to the positive 'dec' in rad/sec^2.200        """201        command=b'\x08'202        values=float2bytes(_dec)203        self._run_command(command+identifier+values,'motor_rx')204    def set_max_torque(self,max_torque,identifier=b'\x00\x00'):205        """206        Set the maximum torque to the positive 'max_torque' in N.m.207        """208        command=b'\x0E'209        if max_torque < 0:210            raise ValueError("Value out of range")211        values=float2bytes(max_torque)212        self._run_command(command+identifier+values,'motor_rx')213    def set_teaching_interval(self,interval_ms,identifier=b'\x00\x00'):214        """215        Set the interval time of teaching to the positive integer "interval_ms" in ms.216        """217        command=b'\x16'218        values=uint32_t2bytes(interval_ms)219        self._run_command(command+identifier+values,'motor_rx')220    def set_playback_interval(self,interval_ms,identifier=b'\x00\x00'):221        """222        Set the interval time of playback to the positive integer "interval_ms" in ms.223        """224        command=b'\x17'225        values=uint32_t2bytes(interval_ms)226        self._run_command(command+identifier+values,'motor_rx')227    def set_qcurrent_p(self,q_current_p,identifier=b'\x00\x00'):228        """229        Set the q-axis current PID controller's Proportional gain to the postiive 'q_current_p'.230        """231        command=b'\x18'232        values=float2bytes(q_current_p)233        self._run_command(command+identifier+values,'motor_rx')234    def set_qcurrent_i(self,q_current_i,identifier=b'\x00\x00'):235        """236        Set the q-axis current PID controller's Integral gain to the positive 'q_current_i'.237        """238        command=b'\x19'239        values=float2bytes(q_current_i)240        self._run_command(command+identifier+values,'motor_rx')241    def set_qcurrent_d(self,q_current_d,identifier=b'\x00\x00'):242        """243        Set the q-axis current PID controller's Differential gain to the postiive 'q_current_d'.244        """245        command=b'\x1A'246        values=float2bytes(q_current_d)247        self._run_command(command+identifier+values,'motor_rx')248    def set_speed_p(self,speed_p,identifier=b'\x00\x00'):249        """250        Set the speed PID controller's Proportional gain to the positive 'speed_p'.251        """252        command=b'\x1B'253        values=float2bytes(speed_p)254        self._run_command(command+identifier+values,'motor_rx')255    def set_speed_i(self,speed_i,identifier=b'\x00\x00'):256        """257        Set the speed PID controller's Integral gain to the positive 'speed_i'.258        """259        command=b'\x1C'260        values=float2bytes(speed_i)261        self._run_command(command+identifier+values,'motor_rx')262    def set_speed_d(self,speed_d,identifier=b'\x00\x00'):263        """264        Set the speed PID controller's Deferential gain to the positive 'speed_d'.265        """266        command=b'\x1D'267        values=float2bytes(speed_d)268        self._run_command(command+identifier+values,'motor_rx')269    def set_position_p(self,position_p,identifier=b'\x00\x00'):270        """271        Set the position PID controller's Proportional gain to the positive 'position_p'.272        """273        command=b'\x1E'274        values=float2bytes(position_p)275        self._run_command(command+identifier+values,'motor_rx')276    def set_position_i(self,position_i,identifier=b'\x00\x00'):277        """278        Set the position PID controller's Integral gain to the positive 'position_i'.279        """280        command=b'\x1F'281        values=float2bytes(position_i)282        self._run_command(command+identifier+values,'motor_rx')283    def set_position_d(self,position_d,identifier=b'\x00\x00'):284        """285        Set the position PID controller's Differential gain to the positive 'position_d'.286        """287        command=b'\x20'288        values=float2bytes(position_d)289        self._run_command(command+identifier+values,'motor_rx')290    def set_pos_control_threshold(self,poscontrol_d,identifier=b'\x00\x00'):291        """292        Set the threshold of the deviation between the target and current position. While the deviation is more than the threshold, integral and differential gain is set to be zero.293        """294        command=b'\x21'295        values=float2bytes(poscontrol_d)296        self._run_command(command+identifier+values,'motor_rx')297    def reset_all_pid(self,identifier=b'\x00\x00'):298        """299        Reset all the PID parameters to the firmware default settings.300        """301        command=b'\x22'302        self._run_command(command+identifier,'motor_rx')303    def set_motor_measurement_interval(self,interval,identifier=b'\x00\x00'):304        """305        Set the notification interval of motor measurement values. It is valid only for USB(UART) and BLE.306        In case of BLE, the allowed minimum value is 100 [ms]. 307        -----------308        0:      2 [ms]309        1:      5 [ms]310        2:     10 [ms]311        3:     20 [ms]312        4:     50 [ms]313        5:    100 [ms]314        6:    200 [ms]315        7:    500 [ms]316        8:   1000 [ms]317        9:   2000 [ms]318        10:  5000 [ms]319        11: 10000 [ms]320        -----------        321        """322        command=b'\x2C'323        values = uint8_t2bytes(interval)324        self._run_command(command+identifier+values,'motor_rx')325    def set_motor_measurement_settings(self,flag,identifier=b'\x00\x00'):326        """327        Set the notification setting of motor measurement values.328        -----------329        bit7: -330        bit6: -331        bit5: -332        bit4: -333        bit3: -334        bit2: -335        bit1: Notification of motor measurement (1:ON, 0:OFF)336        bit0: To start Notification of motor measurement when booting(1:ON, 0:OFF)337        -----------338        """339        command=b'\x2D'340        values=uint8_t2bytes(flag)341        self._run_command(command+identifier+values,'motor_rx')342    @property343    def interface_type(self):344        return {345            "BLE": 0b1,346            "USB": 0b1000,347            "I2C": 0b10000,348            "BTN": 0b10000000,349        }350    def set_interface(self,flag,identifier=b'\x00\x00'):351        """352        You can enable or disable the data interfaces(physical 3 buttons, I2C, USB and BLE).353        The motor chooses the output interface of motor measurement values and IMU values as354        (High priority)BLE > USB > I2C(Low priority) by default.355        If you want to force it to send measurement values through USB,356        you need to set bit0(BLE) to OFF(0) and bit3(USB) to ON(1).357        For example, if you call set_interface(0b10001000), Physical 3 buttons: enabled, I2C: disabled, USB: enabled and BLE: disabled.358        To save this setting to the flash memory, ensure you call saveAllRegisters().@n359        -----------360        - bit7: Physical 3 buttons361        - bit6: UART2362        - bit5: Digital In363        - bit4: I2C(Wired)364        - bit3: UART1(USB, Wired)365        - bit2: -366        - bit1: Linkage367        - bit0: BLE(Wireless)368        -----------369        """370        command=b'\x2E'371        values=uint8_t2bytes(flag)372        self._run_command(command+identifier+values,'motor_rx')373    def set_safe_run_settings(self,isEnabled,timeout,stopAction,identifier=b'\x00\x00'):374        """set_safe_run_settings375        Set safe run mode to stop motor automatically when it cannot receive next command within a certain period (timeout).376        Args:377            isEnabled (bool): true if setting safe run mode enabled378            timeout (int): stop action will occur after this period379            stopAction (int):380                - 0: free_motor381                - 1: disable_motor_action382                - 2: stop_motor383                - 3: Fix to the current position (move_to)384        """385        command=b'\x31'386        values=uint8_t2bytes(isEnabled)+uint32_t2bytes(timeout)+uint8_t2bytes(stopAction)387        self._run_command(command+identifier+values,'motor_rx')388    def set_own_color(self,red,green,blue,identifier=b'\x00\x00'):389        """390        Set the own LED color(red:0-255, green:0-255, blue:0-255).391        """392        command=b'\x3A'393        values=uint8_t2bytes(red)+uint8_t2bytes(green)+uint8_t2bytes(blue)394        self._run_command(command+identifier+values,'motor_rx')395    def set_imu_measurement_interval(self,interval,identifier=b'\x00\x00'):396        """397        Set the notification interval of IMU measurement values. It is valid only for USB(UART) and BLE.398        In case of BLE, the allowed minimum value is 100 [ms]. 399        -----------400        0:     10 [ms]401        1:     20 [ms]402        2:     50 [ms]403        3:    100 [ms]404        4:    200 [ms]405        5:    500 [ms]406        6:   1000 [ms]407        7:   2000 [ms]408        9:   5000 [ms]409        10: 10000 [ms]410        -----------     411        """412        command=b'\x3C'413        values = uint8_t2bytes(interval)414        self._run_command(command+identifier+values,'motor_rx')415    def set_imu_measurement_settings(self,flag,identifier=b'\x00\x00'):416        """417        Set the notification setting of IMU.418        -----------419        bit7: -420        bit6: -421        bit5: -422        bit4: -423        bit3: -424        bit2: -425        bit1: Notification of IMU (1:ON, 0:OFF)426        bit0: To start Notification of IMU when booting(1:ON, 0:OFF)427        -----------428        """429        command=b'\x3D'430        values=uint8_t2bytes(flag)431        self._run_command(command+identifier+values,'motor_rx')432    def set_autostart_setting(self, mode, identifier=b'\x00\x00'):433        """434        Set the button setting  (Motor Firmware Ver. requires >= 2.42)435        -----------436        mode = 0:  Without auto start 437        mode = 1:  Auto start doing taskset (config by set_taskset_trigger_settings)438        mode = 2:  Auto start playback motion (config by set_motion_trigger_settings)439        -----------440        """441        command=b'\xBC'442        values=uint8_t2bytes(mode)443        self._run_command(command+identifier+values,'motor_rx')444    def set_button_setting(self, mode, identifier=b'\x00\x00'):445        """446        Set the button setting  (MotorFirmVer. >= 2.28)447        -----------448        mode = 0:  Manual button mode with Motion 449        mode = 1:  Manual button mode with Taskset450        mode = 30: Child button mode (Receive button action event from KeiganMotor)451        -----------452        """453        command=b'\xBD'454        values=uint8_t2bytes(mode)455        self._run_command(command+identifier+values,'motor_rx')456    def set_baud_rate(self, baud, identifier=b'\x00\x00'):457        """458        Set baud rate for serial communication setting 459        (baud rate unit is kbps)460        -----------461        0: 115200 462        1: 230400 463        2: 250000464        3: 460800465        4: 921600466        5: 1000000 (1M)467        -----------468        """469        command=b'\xC3'470        values=uint8_t2bytes(baud)471        self._run_command(command+identifier+values,'motor_rx')472    def set_notify_pos_arrival_settings(self, isEnabled, tolerance, settleTime, identifier=b'\x00\x00'):473        """474        Set notification settings when arriving at the target position 475        """476        command=b'\x2B'477        values=uint8_t2bytes(isEnabled)+float2bytes(tolerance)+uint32_t2bytes(settleTime)478        self._run_command(command+identifier+values,'motor_rx')479    def read_register(self,register,identifier=b'\x00\x00'):480        '''481        Read a specified setting (register).482        '''483        command=b'\x40'484        values=uint8_t2bytes(register)485        self._run_command(command+identifier+values,'motor_rx')486    def save_all_registers(self,identifier=b'\x00\x00'):487        """488        Save all settings (registers) in flash memory.489        """490        command=b'\x41'491        self._run_command(command+identifier,'motor_rx')492        time.sleep(3) # wait for next command to store registers in flash493    def reset_register(self,register,identifier=b'\x00\x00'):494        """495        Reset a specified register's value to the firmware default setting.496        """497        command=b'\x4E'498        values=uint8_t2bytes(register)499        self._run_command(command+identifier+values,'motor_rx')500    def reset_all_registers(self,identifier=b'\x00\x00'):501        """502        Reset all registers' values to the firmware default setting.503        """504        command=b'\x4F'505        self._run_command(command+identifier,'motor_rx')506    # Motor Action507    def disable_action(self,identifier=b'\x00\x00'):508        """509        Disable motor action.510        """511        command=b'\x50'512        self._run_command(command+identifier,'motor_tx')513    def enable_action(self,identifier=b'\x00\x00'):514        """515        Enable motor action.516        """517        command=b'\x51'518        self._run_command(command+identifier,'motor_tx')519    def set_speed(self,speed,identifier=b'\x00\x00'):520        """521        Set the speed of rotation to the positive 'speed' in rad/sec.522        """523        command=b'\x58'524        values=float2bytes(speed)525        self._run_command(command+identifier+values,'motor_tx')526    def preset_position(self,position,identifier=b'\x00\x00'):527        """528        Preset the current absolute position as the specified 'position' in rad. (Set it to zero when setting origin)529        """530        command=b'\x5A'531        values=float2bytes(position)532        self._run_command(command+identifier+values,'motor_tx')533    def get_position_offset(self,position,identifier=b'\x00\x00'):534        """535        Get the offset value of the absolute position.536        """537        command=b'\x5B'538        values=float2bytes(position)539        self._run_command(command+identifier+values,'motor_tx')540    def run_forward(self,identifier=b'\x00\x00'):541        """542        Rotate the motor forward (counter clock-wise) at the speed set by 0x58: speed.543        """544        command=b'\x60'545        self._run_command(command+identifier,'motor_tx')546    def run_reverse(self,identifier=b'\x00\x00'):547        """548        Rotate the motor reverse (clock-wise) at the speed set by 0x58: speed.549        """550        command=b'\x61'551        self._run_command(command+identifier,'motor_tx')552    def run_at_velocity(self,velocity,identifier=b'\x00\x00'):553        """554        Rotate the motor at the 'velocity'. The velocity can be positive or negative.555        """556        command=b'\x62'557        values=float2bytes(velocity)558        self._run_command(command+identifier+values,'motor_tx')559    def move_to_pos(self,position,speed=None,identifier=b'\x00\x00'):560        """561        Move the motor to the specified absolute 'position' at the 'speed'.562        If the speed is None, move at the speed set by 0x58: set_speed.563        """564        if speed is not None:565            command=b'\x65'566            values=float2bytes(position)+float2bytes(speed)567        else:568            command=b'\x66'569            values=float2bytes(position)570        self._run_command(command+identifier+values,'motor_tx')571    def move_by_dist(self,distance,speed=None,identifier=b'\x00\x00'):572        """573        Move the motor by the specified relative 'distance' from the current position at the 'speed'.574        If the speed is None, move at the speed set by 0x58: set_speed.575        """576        if speed is not None:577            command=b'\x67'578            values=float2bytes(distance)+float2bytes(speed)579        else:580            command=b'\x68'581            values=float2bytes(distance)582        self._run_command(command+identifier+values,'motor_tx')583    def move_to_pos_wait(self,position,speed=None,identifier=b'\x00\x00'):584        """585        Move the motor to the specified absolute 'position' at the 'speed'.586        It pauses the queue to handle commands in motor side at the same time.587        "Queue" will resume automatically if it arrives at the target position.588        See "set_notify_pos_arrival_settings" to change parameters such as the tolerance.589        If the speed is None, move at the speed set by 0x58: set_speed.590        """591        if speed is not None:592            command=b'\x75'593            values=float2bytes(position)+float2bytes(speed)594        else:595            command=b'\x76'596            values=float2bytes(position)597        self._run_command(command+identifier+values,'motor_tx')598    def move_by_dist_wait(self,distance,speed=None,identifier=b'\x00\x00'):599        """600        Move the motor by the specified relative 'distance' from the current position at the 'speed'.601        It pauses the queue to handle commands in motor side at the same time.602        The queue will be resumed automatically if it arrives at the target position.603        See "set_notify_pos_arrival_settings" to change parameters such as the tolerance.604        If the speed is None, move at the speed set by 0x58: set_speed.605        """606        if speed is not None:607            command=b'\x77'608            values=float2bytes(distance)+float2bytes(speed)609        else:610            command=b'\x78'611            values=float2bytes(distance)612        self._run_command(command+identifier+values,'motor_tx')613    def disable_action_task(self,identifier=b'\x00\x00'):614        """615        Disable motor action. (Firmware ver. requires >= 2.42)616        This function can be stored in taskset, while disable_action() cannot be stored.617        """       618        command=b'\x6A'619        self._run_command(command+identifier,'motor_tx') 620    def enable_action_task(self,identifier=b'\x00\x00'):621        """622        Enable motor action. (Firmware ver. requires >= 2.42)623        This function can be stored in taskset, while enable_action() cannot be stored.624        """       625        command=b'\x6B'626        self._run_command(command+identifier,'motor_tx') 627    def free_motor(self,identifier=b'\x00\x00'):628        """629        Stop the motor's excitation630        """631        command=b'\x6C'632        self._run_command(command+identifier,'motor_tx')633    def stop_motor(self,identifier=b'\x00\x00'):634        """635        Decelerate the speed to zero and stop.636        """637        command=b'\x6D'638        self._run_command(command+identifier,'motor_tx')639    def hold_torque(self,torque,identifier=b'\x00\x00'):640        """641        Keep and output the specified torque.642        """643        command=b'\x72'644        values=float2bytes(torque)645        self._run_command(command+identifier+values,'motor_tx')646    def start_doing_taskset(self,index,repeating,identifier=b'\x00\x00'):647        """648        Do taskset at the specified 'index' 'repeating' times.649        """650        command=b'\x81'651        values=uint16_t2bytes(index)+uint32_t2bytes(repeating)652        self._run_command(command+identifier+values,'motor_tx')653    def stop_doing_taskset(self,identifier=b'\x00\x00'):654        """655        Stop doing the current taskset.656        """657        command=b'\x82'658        self._run_command(command+identifier,'motor_tx')659    def start_playback_motion(self,index,repeating,option,identifier=b'\x00\x00'):660        """661        Start to playback motion at the specified 'index' 'repeating' times.662        """663        command=b'\x85'664        values=uint16_t2bytes(index)+uint32_t2bytes(repeating)+uint8_t2bytes(option)665        self._run_command(command+identifier+values,'motor_tx')666    def prepare_playback_motion(self,index,repeating,option,identifier=b'\x00\x00'):667        """668        Prepare to playback motion at the specified 'index' 'repeating' times.669        """670        command=b'\x86'671        values=uint16_t2bytes(index)+uint32_t2bytes(repeating)+uint8_t2bytes(option)672        self._run_command(command+identifier+values,'motor_tx')673    def start_playback_motion_from_prep(self,identifier=b'\x00\x00'):674        """675        Start to playback motion in the condition of the last preparePlaybackMotion.676        """677        command=b'\x87'678        self._run_command(command+identifier,'motor_tx')679    def stop_playback_motion(self,identifier=b'\x00\x00'):680        """681        Stop to playback motion.682        """683        command=b'\x88'684        self._run_command(command+identifier,'motor_tx')685    # Queue686    def pause_queue(self,identifier=b'\x00\x00'):687        """688        Pause the queue until 0x91: resume is executed.689        """690        command=b'\x90'691        self._run_command(command+identifier,'motor_tx')692    def resume_queue(self,identifier=b'\x00\x00'):693        """694        Resume the queue.695        """696        command=b'\x91'697        self._run_command(command+identifier,'motor_tx')698    def wait_queue(self,waittime,identifier=b'\x00\x00'):699        """700        Wait the queue or pause the queue for the specified 'time' in msec and resume it automatically.701        """702        command=b'\x92'703        values=uint32_t2bytes(waittime)704        self._run_command(command+identifier+values,'motor_tx')705    def clear_queue(self,identifier=b'\x00\x00'):706        """707        Clear the queue. Erase all tasks in the queue. This command works when 0x90: pause or 0x92: wait are executed.708        """709        command=b'\x95'710        self._run_command(command+identifier,'motor_tx')711    def wait_queue_until_input(self, pin, event=2, timeout=0, sub_pin1=0xFF, sub_state1=0, sub_pin2=0xFF, sub_state2=0, sub_pin3=0xFF, sub_state3=0, identifier=b'\x00\x00'):712        """713        Wait the queue or pause the queue for the specified GPIO digital pin event.714        It is resumd automatically if timeout[ms] passed.715        event parameter is as follows.716        -----------717        1: RISING (Low to High) 718        2: FALLING (High to Low)719        -----------720        """721        command=b'\x96'722        values=uint8_t2bytes(pin)+uint8_t2bytes(event)+uint32_t2bytes(timeout)+uint8_t2bytes(sub_pin1)+uint8_t2bytes(sub_state1)+uint8_t2bytes(sub_pin2)+uint8_t2bytes(sub_state2)+uint8_t2bytes(sub_pin3)+uint8_t2bytes(sub_state3)723        self._run_command(command+identifier+values,'motor_tx')        724    # Taskset725    def start_recording_taskset(self,index,identifier=b'\x00\x00'):726        """727        Start recording taskset at the specified 'index' in the flash memory.728        In the case of KM-1, index value is from 0 to 49 (50 in total).729        """730        command=b'\xA0'731        values=uint16_t2bytes(index)732        self._run_command(command+identifier+values,'motor_tx')733    def stop_recording_taskset(self,identifier=b'\x00\x00'):734        """735        Stop recording taskset.736        This command works while 0xA0: startRecordingTaskset is executed.737        """738        command=b'\xA2'739        self._run_command(command+identifier,'motor_tx')740    def erase_taskset(self,index,identifier=b'\x00\x00'):741        """742        Erase taskset at the specified index in the flash memory.743        In the case of KM-1, index value is from 0 to 49 (50 in total).744        """745        command=b'\xA3'746        values=uint16_t2bytes(index)747        self._run_command(command+identifier+values,'motor_tx')748    def erase_all_tasksets(self,identifier=b'\x00\x00'):749        """750        Erase all tasksets in the flash memory.751        """752        command=b'\xA4'753        self._run_command(command+identifier,'motor_tx')754    def set_trigger_taskset_settings(self,index,repeating,identifier=b'\x00\x00'):755        """756        Set taskset settings by trigger such as button757        """758        command=b'\xA8'759        values=uint16_t2bytes(index)+uint32_t2bytes(repeating)760        self._run_command(command+identifier+values,'motor_tx')761    # Teaching762    def start_teaching_motion(self,index,time_ms,identifier=b'\x00\x00'):763        """764        Start teaching motion without preparing by specifying the 'index' in the flash memory and recording 'time' in milliseconds.765        In the case of KM-1, index value is from 0 to 9 (10 in total).  Recording time cannot exceed 65408 [msec].766        """767        command=b'\xA9'768        values=uint16_t2bytes(index)+uint32_t2bytes(time_ms)769        self._run_command(command+identifier+values,'motor_tx')770    def prepare_teaching_motion(self,index,time_ms,identifier=b'\x00\x00'):771        """772        Prepare teaching motion by specifying the 'index' in the flash memory and recording 'time' in milliseconds.773        In the case of KM-1, index value is from 0 to 9 (10 in total).  Recording time cannot exceed 65408 [msec].774        """775        command=b'\xAA'776        values=uint16_t2bytes(index)+uint32_t2bytes(time_ms)777        self._run_command(command+identifier+values,'motor_tx')778    def start_teaching_motion_from_prep(self,identifier=b'\x00\x00'):779        """780        Start teaching motion in the condition of the last prepareTeachingMotion.781        This command works when the teaching index is specified by 0xAA: prepareTeachingMotion.782        """783        command=b'\xAB'784        self._run_command(command+identifier,'motor_tx')785    def stop_teaching_motion(self,identifier=b'\x00\x00'):786        """787        Stop teaching motion.788        """789        command=b'\xAC'790        self._run_command(command+identifier,'motor_tx')791    def erase_motion(self,index,identifier=b'\x00\x00'):792        """793        Erase motion at the specified index in the flash memory.794        In the case of KM-1, index value is from 0 to 9 (10 in total).795        """796        command=b'\xAD'797        values=uint16_t2bytes(index)798        self._run_command(command+identifier+values,'motor_tx')799    def erase_all_motions(self,identifier=b'\x00\x00'):800        """801        Erase all motions in the flash memory.802        """803        command=b'\xAE'804        self._run_command(command+identifier,'motor_tx')805    def set_trigger_motion_settings(self,index,repeating,option,autoStart,identifier=b'\x00\x00'):806        """807        Set playback motion settings by trigger such as button808        """809        command=b'\xB2'810        values=uint16_t2bytes(index)+uint32_t2bytes(repeating)+uint8_t2bytes(option)+uint8_t2bytes(autoStart)811        self._run_command(command+identifier+values,'motor_tx')812    def read_motion(self,index,read_comp_cb,identifier=b'\x00\x00'):813        """814        Externally output the motion stored by direct teaching. (Motor farm ver>=2.0)815        It is valid only for a wired connection(USB).816          callback signature817            read_comp_cb(motion_index,motion_value[])818        """819        command=b'\xB7'820        values = uint16_t2bytes(index)821        self._run_command(command + identifier + values, 'motor_tx')822        self.on_read_motion_read_comp_cb=read_comp_cb823        return True824    def write_motion_position(self,position,identifier=b'\x00\x00'):825        """826        Record one coordinate of movement from the outside. (Motor farm ver>=2.0)827        It is valid only for a wired connection(USB).828            â» This command can not be accepted unless it is in prepareTeaching state.What to do after running "prepareTeachingMotion"829            â» This command records coordinates one by one. If one record is made, it will be waiting to record the next coordinate.830            â» After recording coordinates with this command, recording can not be completed normally unless stopTeaching is executed.831        """832        command=b'\xB8'833        values = float2bytes(position)834        self._run_command(command + identifier + values, 'motor_tx')835    # LED836    def set_led(self,ledState,red,green,blue,identifier=b'\x00\x00'):837        """838        Set the LED state (off, solid, flash and dim) and color intensity (red, green and blue).839        typedef enum ledState =840        {841            LED_STATE_OFF = 0, // LED off842            LED_STATE_ON_SOLID = 1, // LED solid843            LED_STATE_ON_FLASH = 2, // LED flash844            LED_STATE_ON_DIM = 3 // LED dim845        }846        """847        command=b'\xE0'848        values=uint8_t2bytes(ledState)+uint8_t2bytes(red)+uint8_t2bytes(green)+uint8_t2bytes(blue)849        self._run_command(command+identifier+values,"motor_tx")850    def enable_continual_motor_measurement(self,identifier=b'\x00\x00'):851        """852        Enable the motor measurement and start continual notification of the measurement values.853        """854        command=b'\xE6'855        self._run_command(command+identifier,'motor_tx')856    def disable_continual_motor_measurement(self,identifier=b'\x00\x00'):857        """858        Disable the motor measurement and stop notification of the measurement values.859        """860        command=b'\xE7'861        self._run_command(command+identifier,'motor_tx')862    # IMU863    def enable_continual_imu_measurement(self,identifier=b'\x00\x00'):864        """865        Enable the IMU and start notification of the measurement values.866        """867        command=b'\xEA'868        self._run_command(command+identifier,'motor_tx')869    def disable_continual_imu_measurement(self,identifier=b'\x00\x00'):870        """871        Disable the IMU and stop notification of the measurement values.872        """873        command=b'\xEB'874        self._run_command(command+identifier,'motor_tx')875    # System876    def reboot(self,identifier=b'\x00\x00'):877        """878        Reboot the system.879        """880        command=b'\xF0'881        self._run_command(command+identifier,'motor_tx')882    def enable_check_sum(self,identifier=b'\x00\x00'):883        command=b'\xF3'884        self._run_command(command+identifier,'motor_tx')885    def enter_device_firmware_update(self,identifier=b'\x00\x00'):886        """887        Enter the device firmware update mode or bootloader mode. It goes with reboot.888        """889        command=b'\xFD'890        self._run_command(command+identifier,'motor_tx')891    def read_max_speed(self):892        """893        Read the current maximum speed of rotation 'max_speed' in rad/sec.894        """895        return self._read_setting_value(0x02)896    def read_min_speed(self):897        """898        Read the current minimum speed of rotation 'min_speed' in rad/sec.899        """900        return self._read_setting_value(0x03)901    def read_curve_type(self):902        """903        Read the current acceleration or deceleration curve 'curve_type'.904        typedef enum curveType =...

util.py

Source:util.py

2    '''3    Splice testing exception4    '''5    pass6def _run_command(connection, command, timeout=60, get_pty=False):7    """ Run a command and raise exception in case of timeout or none-zero result """8    status = connection.recv_exit_status(command, timeout, get_pty=get_pty)9    if status is not None and status != 0:10        raise SpliceTestFailed("Failed to run %s: got %s return value\nSTDOUT: %s\nSTDERR: %s" % (command, status, connection.last_stdout, connection.last_stderr))11    elif status is None:12        raise SpliceTestFailed("Failed to run %s: got timeout %s\nSTDOUT: %s\nSTDERR: %s" % (command, timeout, connection.last_stdout, connection.last_stderr))13def run_sst(connection, spacewalk_only=False, splice_only=False, timeout=120):14    """ Run spacewalk-splice-tool """15    command = "sudo -u splice spacewalk-splice-checkin"16    if spacewalk_only:17        command += " --spacewalk-sync"18    elif splice_only:19        command += " --splice-sync"20    # changing system date21    _run_command(connection, "[ ! -z \"`spacewalk-report fake-checkin-date`\" ] && date -s \"`spacewalk-report fake-checkin-date`\" && katello-service restart && sleep 10 ||:", 3 * timeout)22    _run_command(connection, command, timeout, get_pty=True)23def fake_spacewalk_env(connection, test_name):24    """ Select test with fake spacewalk """25    _run_command(connection, "spacewalk-report-set %s" % test_name)26def fake_spacewalk_step(connection):27    """ Step to next data """28    _run_command(connection, "spacewalk-report-set -n")29def sst_step(connection, fake_spacewalk_connection=None, timeout=120):30    """ Run sst and step to next data (iteration) """31    run_sst(connection, timeout=timeout)32    if fake_spacewalk_connection is None:33        fake_spacewalk_step(connection)34    else:35        fake_spacewalk_step(fake_spacewalk_connection)36def cleanup_katello(connection, katello, keep_splice=False, full_reset=False):37    """ Clean up katello and splice databases """38    if full_reset:39        _run_command(connection, "katello-configure --no-bars --user-pass='%s' --reset-data=YES" % katello.password, 900)40    else:41        for org in katello.list_organizations():42            if org['id'] != 1:43                # keeping default org44                katello.delete_organization(org['label'])45        for role in katello.list_roles():46            if not role['id'] in [1, 2]:47                # keeping default roles48                katello.delete_role(role['id'])49    if not keep_splice:50        _run_command(connection, "service splice_all stop ||:")51        _run_command(connection, "mongo checkin_service --eval 'db.dropDatabase()'")52        _run_command(connection, "service splice_all start ||:")53def parse_report_json(report):54    """ Read json report and figure out important facts """55    result = {}56    result["total_number_of_instances"] = len(report)57    result["number_of_current"] = sum(1 for instance in report if instance['entitlement_status']['status'] == 'valid')58    result["number_of_insufficient"] = sum(1 for instance in report if instance['entitlement_status']['status'] == 'partial')59    result["number_of_invalid"] = sum(1 for instance in report if instance['entitlement_status']['status'] == 'invalid')...

services.py

Source:services.py

...5from pyholecli import settings6class RunnableBaseClass:7    def __init__(self, connection):8        self._c = connection9    def _run_command(self, *args, **kwargs):10        return run_command(self._c, *args, **kwargs)11    def _put_file(self, local, remote, *args, **kwargs):12        return put_file(self._c, local, remote, *args, **kwargs)13class HostnameUtility(RunnableBaseClass):14    hosts_file = settings.LOCAL_HOSTNAME_FILENAME15    @property16    def hosts(self):17        if not hasattr(self, '_hosts'):18            result = self._run_command('cat', self.hosts_file, hide=True)19            self._hosts = Hosts.from_hosts_string(result.stdout)20        return self._hosts21    def get_host(self, hostname):22        try:23            return self.hosts[hostname]24        except KeyError:25            message = 'Could not find hostname {}'.format(hostname)26            raise HostnameNotFound(message)27    def set_host(self, hostname, ip):28        self.hosts[hostname] = Host(hostname, ip)29        self._write_hosts()30    def remove_hosts(self, hostnames):31        hostnames = [hostnames] if isinstance(hostnames, str) else hostnames32        for hostname in hostnames:33            try:34                del self.hosts[hostname]35            except KeyError:36                message = 'Could not find hostname {}'.format(hostname)37                raise HostnameNotFound(message)38        self._write_hosts()39    def _write_hosts(self):40        hosts_fd = BytesIO(str(self.hosts).encode('utf-8'))41        self._put_file(hosts_fd, self.hosts_file)42        self._run_command('pihole', 'restartdns')43class PiholeCLI(RunnableBaseClass):44    def _run_command(self, *args, **kwargs):45        return super()._run_command('pihole', *args, **kwargs)46    def status(self):47        return self._run_command('status')48    def enable(self):49        return self._run_command('enable')50    def disable(self):51        return self._run_command('disable')52    def query(self, domain):53        return self._run_command('-q', domain)54    def chronometer(self):55        return self._run_command('-c', pty=True)56    def tail(self):57        return self._run_command('-t', pty=True)58    def blacklist(self, domains):59        return self._run_command('-b', domains)60    def get_blacklisted_domains(self):61        return self._run_command('-b', '-l')62    def remove_blacklisted_domains(self, domains):63        return self._run_command('-b', '-d', domains)64    def whitelist(self, domains):65        return self._run_command('-w', domains)66    def get_whitelisted_domains(self):67        return self._run_command('-w', '-l')68    def remove_whitelisted_domains(self, domains):69        return self._run_command('-w', '-d', domains)70    def wildcard_blacklist(self, domains):71        return self._run_command('-wild', domains)72    def get_wildcard_blacklist(self):73        return self._run_command('-wild', '-l')74    def remove_wildcard_blacklist(self, domains):...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.