add MyPalletizer class.

Author: zlj-zz

pymycobot/__init__.py

@@ -3,14 +3,22 @@
 from __future__ import absolute_import
 import datetime
 
-from pymycobot.mycobot import MyCobot
 from pymycobot.generate import MycobotCommandGenerater
+from pymycobot.mycobot import MyCobot
+from pymycobot.mypalletizer import MyPalletizer
 from pymycobot.genre import Angle, Coord
 from pymycobot import utils
 
-__all__ = ["MyCobot", "MycobotCommandGenerater", "Angle", "Coord", "utils"]
+__all__ = [
+    "MyCobot",
+    "MycobotCommandGenerater",
+    "Angle",
+    "Coord",
+    "utils",
+    "MyPalletizer",
+]
 
-__version__ = "2.5.9"
+__version__ = "2.6.0.beta.1"
 __author__ = "Zachary zhang"
 __email__ = "[email protected]"
 __git_url__ = "https://github.com/elephantrobotics/pymycobot"

pymycobot/common.py

@@ -1,6 +1,7 @@
 # coding=utf-8
 
 from __future__ import division
+import time
 import struct
 
 
@@ -149,7 +150,7 @@ def _process_received(self, data, genre):
 
         # process valid data
         res = []
-        if data_len == 12:
+        if data_len == 12 or data_len == 8:
             for idx in range(0, len(valid_data), 2):
                 one = valid_data[idx : idx + 2]
                 res.append(self._decode_int16(one))
@@ -163,3 +164,21 @@ def _process_received(self, data, genre):
 
     def _process_single(self, data):
         return data[0] if data else -1
+
+
+def write(self, command):
+    self.log.debug("_write: {}".format(command))
+
+    self._serial_port.write(command)
+    self._serial_port.flush()
+    time.sleep(0.05)
+
+
+def read(self):
+    if self._serial_port.inWaiting() > 0:
+        data = self._serial_port.read(self._serial_port.inWaiting())
+        self.log.debug("_read: {}".format(data))
+    else:
+        self.log.debug("_read: no data can be read")
+        data = None
+    return data

pymycobot/generate.py

@@ -4,7 +4,6 @@
 import logging
 
 from pymycobot.log import setup_logging
-from pymycobot.error import calibration_parameters
 from pymycobot.common import ProtocolCode, DataProcessor
 
 
@@ -182,7 +181,7 @@ def send_angle(self, id, degree, speed):
             degree (float): angle value.
             speed (int): 0 ~ 100
         """
-        calibration_parameters(id=id, degree=degree, speed=speed)
+        self.calibration_parameters(id=id, degree=degree, speed=speed)
         return self._mesg(ProtocolCode.SEND_ANGLE, id, [self._angle2int(degree)], speed)
 
     # @check_parameters(Command.SEND_ANGLES)
@@ -193,9 +192,8 @@ def send_angles(self, degrees, speed):
             degrees (list): example [0.0, 0.0, 0.0, 0.0, 0.0, 0.0]
             speed (int): 0 ~ 100
         """
-        calibration_parameters(degrees=degrees, speed=speed)
+        self.calibration_parameters(degrees=degrees, speed=speed)
         degrees = [self._angle2int(degree) for degree in degrees]
-        # data = [degrees, speed]
         return self._mesg(ProtocolCode.SEND_ANGLES, degrees, speed)
 
     def get_coords(self):
@@ -214,7 +212,7 @@ def send_coord(self, id, coord, speed):
             coord(float): mm
             speed(int): 0 ~ 100
         """
-        calibration_parameters(id=id, speed=speed)
+        self.calibration_parameters(id=id, speed=speed)
         value = self._coord2int(coord) if id <= 3 else self._angle2int(coord)
         return self._mesg(ProtocolCode.SEND_COORD, id, [value], speed)
 
@@ -226,12 +224,12 @@ def send_coords(self, coords, speed, mode):
             speed(int);
             mode(int): 0 - normal, 1 - angluar, 2 - linear
         """
-        calibration_parameters(coords=coords, speed=speed)
+        self.calibration_parameters(coords=coords, speed=speed)
         coord_list = []
         for idx in range(3):
             coord_list.append(self._coord2int(coords[idx]))
-        for idx in range(3, 6):
-            coord_list.append(self._angle2int(coords[idx]))
+        for angle in coords[3:]:
+            coord_list.append(self._angle2int(angle))
         return self._mesg(ProtocolCode.SEND_COORDS, coord_list, speed, mode)
 
     def is_in_position(self, data, id=0):
@@ -247,14 +245,14 @@ def is_in_position(self, data, id=0):
             -1: error data
         """
         if id == 1:
-            calibration_parameters(coords=data)
+            self.calibration_parameters(coords=data)
             data_list = []
             for idx in range(3):
                 data_list.append(self._coord2int(data[idx]))
             for idx in range(3, 6):
                 data_list.append(self._angle2int(data[idx]))
         elif id == 0:
-            calibration_parameters(degrees=data)
+            self.calibration_parameters(degrees=data)
             data_list = [self._angle2int(i) for i in data]
         else:
             raise Exception("id is not right, please input 0 or 1")
@@ -353,7 +351,7 @@ def set_speed(self, speed):
         Args:
             speed (int): 0 - 100
         """
-        calibration_parameters(speed=speed)
+        self.calibration_parameters(speed=speed)
         return self._mesg(ProtocolCode.SET_SPEED, speed)
 
     """
@@ -369,11 +367,11 @@ def get_acceleration(self):
     """
 
     def get_joint_min_angle(self, joint_id):
-        calibration_parameters(id=joint_id)
+        self.calibration_parameters(id=joint_id)
         return self._mesg(ProtocolCode.GET_JOINT_MIN_ANGLE, joint_id, has_reply=True)
 
     def get_joint_max_angle(self, joint_id):
-        calibration_parameters(id=joint_id)
+        self.calibration_parameters(id=joint_id)
         return self._mesg(ProtocolCode.GET_JOINT_MAX_ANGLE, joint_id, has_reply=True)
 
     # Servo control
@@ -421,7 +419,7 @@ def set_color(self, r=0, g=0, b=0):
             b (int): 0 ~ 255
 
         """
-        calibration_parameters(rgb=[r, g, b])
+        self.calibration_parameters(rgb=[r, g, b])
         return self._mesg(ProtocolCode.SET_COLOR, r, g, b)
 
     def set_pin_mode(self, pin_no, pin_mode):
@@ -473,7 +471,7 @@ def set_gripper_value(self, value, speed):
             value (int): 0 ~ 100
             speed (int): 0 ~ 100
         """
-        calibration_parameters(speed=speed)
+        self.calibration_parameters(speed=speed)
         return self._mesg(ProtocolCode.SET_GRIPPER_VALUE, value, speed)
 
     def set_gripper_ini(self):

pymycobot/mycobot.py

@@ -7,7 +7,7 @@
 
 from pymycobot.log import setup_logging
 from pymycobot.generate import MycobotCommandGenerater
-from pymycobot.common import ProtocolCode
+from pymycobot.common import ProtocolCode, write, read
 from pymycobot.error import calibration_parameters
 
 
@@ -57,25 +57,15 @@ def __init__(self, port, baudrate="115200", timeout=0.1, debug=False):
         self.debug = debug
         setup_logging(self.debug)
         self.log = logging.getLogger(__name__)
-        import serial
 
-        self._serial_port = serial.Serial(port, baudrate, timeout=timeout)
+        self.calibration_parameters = calibration_parameters
 
-    def _write(self, command):
-        self.log.debug("_write: {}".format(command))
+        import serial
 
-        self._serial_port.write(command)
-        self._serial_port.flush()
-        time.sleep(0.05)
+        self._serial_port = serial.Serial(port, baudrate, timeout=timeout)
 
-    def _read(self):
-        if self._serial_port.inWaiting() > 0:
-            data = self._serial_port.read(self._serial_port.inWaiting())
-            self.log.debug("_read: {}".format(data))
-        else:
-            self.log.debug("_read: no data can be read")
-            data = None
-        return data
+    _write = write
+    _read = read
 
     def _mesg(self, genre, *args, **kwargs):
         """
@@ -124,7 +114,10 @@ def _mesg(self, genre, *args, **kwargs):
                     return r
                 else:
                     return res
-            elif genre in [ProtocolCode.GET_JOINT_MIN_ANGLE, ProtocolCode.GET_JOINT_MAX_ANGLE]:
+            elif genre in [
+                ProtocolCode.GET_JOINT_MIN_ANGLE,
+                ProtocolCode.GET_JOINT_MAX_ANGLE,
+            ]:
                 return self._int2angle(res[0]) if res else 0
             else:
                 return res

pymycobot/mypalletizer.py

@@ -0,0 +1,92 @@
+# coding=utf-8
+
+import logging
+
+from .log import setup_logging
+from .generate import MycobotCommandGenerater
+from .common import ProtocolCode, read, write
+
+
+def calibration_parameters(**kwargs):
+    pass
+
+
+class MyPalletizer(MycobotCommandGenerater):
+    def __init__(self, port, baudrate="115200", timeout=0.1, debug=False):
+        """
+        Args:
+            port     : port string
+            baudrate : baud rate string, default '115200'
+            timeout  : default 0.1
+            debug    : whether show debug info
+        """
+        super(MyPalletizer, self).__init__(debug)
+        self.debug = debug
+        setup_logging(self.debug)
+        self.log = logging.getLogger(__name__)
+
+        self.calibration_parameters = calibration_parameters
+
+        import serial
+
+        self._serial_port = serial.Serial(port, baudrate, timeout=timeout)
+
+    _write = write
+    _read = read
+
+    def _mesg(self, genre, *args, **kwargs):
+        """
+
+        Args:
+            genre: command type (Command)
+            *args: other data.
+                   It is converted to octal by default.
+                   If the data needs to be encapsulated into hexadecimal,
+                   the array is used to include them. (Data cannot be nested)
+            **kwargs: support `has_reply`
+                has_reply: Whether there is a return value to accept.
+        """
+        real_command, has_reply = super(MyPalletizer, self)._mesg(
+            genre, *args, **kwargs
+        )
+        self._write(self._flatten(real_command))
+
+        if has_reply:
+            data = self._read()
+            res = self._process_received(data, genre)
+            if genre in [
+                ProtocolCode.IS_POWER_ON,
+                ProtocolCode.IS_CONTROLLER_CONNECTED,
+                ProtocolCode.IS_PAUSED,
+                ProtocolCode.IS_IN_POSITION,
+                ProtocolCode.IS_MOVING,
+                ProtocolCode.IS_SERVO_ENABLE,
+                ProtocolCode.IS_ALL_SERVO_ENABLE,
+                ProtocolCode.GET_SERVO_DATA,
+                ProtocolCode.GET_DIGITAL_INPUT,
+                ProtocolCode.GET_GRIPPER_VALUE,
+                ProtocolCode.IS_GRIPPER_MOVING,
+                ProtocolCode.GET_SPEED,
+                ProtocolCode.GET_ENCODER,
+                ProtocolCode.GET_BASIC_INPUT,
+            ]:
+                return self._process_single(res)
+            elif genre in [ProtocolCode.GET_ANGLES]:
+                return [self._int2angle(angle) for angle in res]
+            elif genre in [ProtocolCode.GET_COORDS]:
+                if res:
+                    r = []
+                    for idx in range(3):
+                        r.append(self._int2coord(res[idx]))
+                    r.append(self._int2angle(res[3]))
+                    return r
+                else:
+                    return res
+            elif genre in [
+                ProtocolCode.GET_JOINT_MIN_ANGLE,
+                ProtocolCode.GET_JOINT_MAX_ANGLE,
+            ]:
+                return self._int2angle(res[0]) if res else 0
+            else:
+                return res
+        return None

tests/test_generator.py

@@ -1,4 +1,3 @@
-from ensurepip import version
 import os
 import sys
 import pytest