修复已知BUG

Author: hejianxin79

LinkerHand/__pycache__/__init__.cpython-38.pyc

@@ -5,6 +5,7 @@
 import threading
 import numpy as np
 from enum import Enum
+from LinkerHand.utils.open_can import OpenCan
 
 class FrameProperty(Enum):
     INVALID_FRAME_PROPERTY = 0x00
@@ -25,7 +26,12 @@ class FrameProperty(Enum):
     MOTOR_TEMPERATURE_2 = 0x34
 
 class LinkerHandL10Can:
-    def __init__(self,can_id, can_channel='can0', baudrate=1000000, ):
+    def __init__(self,can_id, can_channel='can0', baudrate=1000000, yaml=""):
+        self.can_id = can_id
+        self.can_channel = can_channel
+        self.baudrate = baudrate
+        self.open_can = OpenCan(load_yaml=yaml)
+        
         self.x01 = [-1] * 5
         self.x02 = [-1] * 5
         self.x03 = [-1] * 5
@@ -70,12 +76,15 @@ def __init__(self,can_id, can_channel='can0', baudrate=1000000, ):
         self.receive_thread.start()
 
     def init_can_bus(self, channel, baudrate):
-        if sys.platform == "linux":
-            return can.interface.Bus(channel=channel, interface="socketcan", bitrate=baudrate)
-        elif sys.platform == "win32":
-            return can.interface.Bus(channel=channel, interface='pcan', bitrate=baudrate)
-        else:
-            raise EnvironmentError("Unsupported platform for CAN interface")
+        try:
+            if sys.platform == "linux":
+                return can.interface.Bus(channel=channel, interface="socketcan", bitrate=baudrate)
+            elif sys.platform == "win32":
+                return can.interface.Bus(channel=channel, interface='pcan', bitrate=baudrate)
+            else:
+                raise EnvironmentError("Unsupported platform for CAN interface")
+        except:
+            print("Please insert CAN device")
 
     def send_frame(self, frame_property, data_list,sleep=0.003):
         """Send a single CAN frame with specified properties and data."""
@@ -86,6 +95,16 @@ def send_frame(self, frame_property, data_list,sleep=0.003):
             self.bus.send(msg)
         except can.CanError as e:
             print(f"Failed to send message: {e}")
+            self.open_can.open_can(self.can_channel)
+            time.sleep(1)
+            self.is_can = self.open_can.is_can_up_sysfs(interface=self.can_channel)
+            time.sleep(1)
+            if self.is_can:
+                self.bus = can.interface.Bus(channel=self.can_channel, interface="socketcan", bitrate=self.baudrate)
+            else:
+                print("Reconnecting CAN devices ....")
+            # time.sleep(1)
+            # 
         time.sleep(sleep)
 
     def set_joint_positions(self, joint_angles):

LinkerHand/__pycache__/linker_hand_api.cpython-38.pyc

@@ -4,6 +4,7 @@
 import threading
 from enum import Enum
 import numpy as np
+from LinkerHand.utils.open_can import OpenCan
 
 class FrameProperty(Enum):
     INVALID_FRAME_PROPERTY = 0x00  # Invalid CAN frame property | No return
@@ -30,8 +31,12 @@ class FrameProperty(Enum):
 
 
 class LinkerHandL20Can:
-    def __init__(self, can_channel='can0', baudrate=1000000, can_id=0x28):
+    def __init__(self, can_channel='can0', baudrate=1000000, can_id=0x28,yaml=""):
         self.can_id = can_id
+        self.can_channel = can_channel
+        self.baudrate = baudrate
+        self.open_can = OpenCan(load_yaml=yaml)
+
         self.running = True
         self.x05, self.x06, self.x07 = [],[],[]
         # New pressure sensors
@@ -58,18 +63,21 @@ def __init__(self, can_channel='can0', baudrate=1000000, can_id=0x28):
         }
 
         # Initialize CAN bus according to operating system
-        if sys.platform == "linux":
-            self.bus = can.interface.Bus(
-                channel=can_channel, interface="socketcan", bitrate=baudrate, 
-                can_filters=[{"can_id": can_id, "can_mask": 0x7FF}]
-            )
-        elif sys.platform == "win32":
-            self.bus = can.interface.Bus(
-                channel=can_channel, interface='pcan', bitrate=baudrate, 
-                can_filters=[{"can_id": can_id, "can_mask": 0x7FF}]
-            )
-        else:
-            raise EnvironmentError("Unsupported platform for CAN interface")
+        try:
+            if sys.platform == "linux":
+                self.bus = can.interface.Bus(
+                    channel=can_channel, interface="socketcan", bitrate=baudrate, 
+                    can_filters=[{"can_id": can_id, "can_mask": 0x7FF}]
+                )
+            elif sys.platform == "win32":
+                self.bus = can.interface.Bus(
+                    channel=can_channel, interface='pcan', bitrate=baudrate, 
+                    can_filters=[{"can_id": can_id, "can_mask": 0x7FF}]
+                )
+            else:
+                raise EnvironmentError("Unsupported platform for CAN interface")
+        except:
+            print("Please insert CAN device")
 
         # Initialize data storage
         self.x01, self.x02, self.x03, self.x04 = [[0.0] * 5 for _ in range(4)]
@@ -108,6 +116,14 @@ def receive_response(self):
                     self.process_response(msg)
             except can.CanError as e:
                 print(f"Error receiving message: {e}")
+                self.open_can.open_can(self.can_channel)
+                time.sleep(1)
+                self.is_can = self.open_can.is_can_up_sysfs(interface=self.can_channel)
+                time.sleep(1)
+                if self.is_can:
+                    self.bus = can.interface.Bus(channel=self.can_channel, interface="socketcan", bitrate=self.baudrate)
+                else:
+                    print("Reconnecting CAN devices ....")
 
     def set_finger_base(self, angles):
         self.send_command(FrameProperty.JOINT_PITCH_R, angles)

LinkerHand/core/__pycache__/__init__.cpython-38.pyc

@@ -4,6 +4,7 @@
 import threading
 import numpy as np
 from enum import Enum
+from LinkerHand.utils.open_can import OpenCan
 current_dir = os.path.dirname(os.path.abspath(__file__))
 target_dir = os.path.abspath(os.path.join(current_dir, ".."))
 sys.path.append(target_dir)
@@ -106,8 +107,12 @@ class FrameProperty(Enum):
     FINGER_TEMPERATURE = 0x84  # Finger joint temperatures
 
 class LinkerHandL21Can:
-    def __init__(self, can_channel='can0', baudrate=1000000, can_id=0x28):
+    def __init__(self, can_channel='can0', baudrate=1000000, can_id=0x28,yaml=""):
         self.can_id = can_id
+        self.can_channel = can_channel
+        self.baudrate = baudrate
+        self.open_can = OpenCan(load_yaml=yaml)
+
         self.running = True
         self.last_thumb_pos, self.last_index_pos,self.last_ring_pos,self.last_middle_pos, self.last_little_pos = None,None,None,None,None
         self.x01, self.x02, self.x03, self.x04,self.x05,self.x06,self.x07, self.x08,self.x09,self.x0A,self.x0B,self.x0C,self.x0D,self.x0E,self.speed = [],[],[],[],[],[],[],[],[],[],[],[],[],[],[]
@@ -145,18 +150,21 @@ def __init__(self, can_channel='can0', baudrate=1000000, can_id=0x28):
             176: 11,
         }
         # Initialize CAN bus according to operating system
-        if sys.platform == "linux":
-            self.bus = can.interface.Bus(
-                channel=can_channel, interface="socketcan", bitrate=baudrate, 
-                can_filters=[{"can_id": can_id, "can_mask": 0x7FF}]
-            )
-        elif sys.platform == "win32":
-            self.bus = can.interface.Bus(
-                channel=can_channel, interface='pcan', bitrate=baudrate, 
-                can_filters=[{"can_id": can_id, "can_mask": 0x7FF}]
-            )
-        else:
-            raise EnvironmentError("Unsupported platform for CAN interface")
+        try:
+            if sys.platform == "linux":
+                self.bus = can.interface.Bus(
+                    channel=can_channel, interface="socketcan", bitrate=baudrate, 
+                    can_filters=[{"can_id": can_id, "can_mask": 0x7FF}]
+                )
+            elif sys.platform == "win32":
+                self.bus = can.interface.Bus(
+                    channel=can_channel, interface='pcan', bitrate=baudrate, 
+                    can_filters=[{"can_id": can_id, "can_mask": 0x7FF}]
+                )
+            else:
+                raise EnvironmentError("Unsupported platform for CAN interface")
+        except:
+            print("Please insert CAN device")
 
         # Start receive thread
         self.receive_thread = threading.Thread(target=self.receive_response)
@@ -176,6 +184,14 @@ def send_command(self, frame_property, data_list,sleep_time=0.003):
             self.bus.send(msg)
         except can.CanError as e:
             print(f"Failed to send message: {e}")
+            self.open_can.open_can(self.can_channel)
+            time.sleep(1)
+            self.is_can = self.open_can.is_can_up_sysfs(interface=self.can_channel)
+            time.sleep(1)
+            if self.is_can:
+                self.bus = can.interface.Bus(channel=self.can_channel, interface="socketcan", bitrate=self.baudrate)
+            else:
+                print("Reconnecting CAN devices ....")
         time.sleep(sleep_time)
 
     def receive_response(self):

LinkerHand/core/can/__pycache__/__init__.cpython-38.pyc

@@ -4,6 +4,7 @@
 import threading
 import numpy as np
 from enum import Enum
+from LinkerHand.utils.open_can import OpenCan
 current_dir = os.path.dirname(os.path.abspath(__file__))
 target_dir = os.path.abspath(os.path.join(current_dir, ".."))
 sys.path.append(target_dir)
@@ -109,8 +110,12 @@ class FrameProperty(Enum):
     WHOLE_FRAME = 0xF0  # Whole frame transmission | Returns one byte frame property + the entire structure for 485 and network transmission only
 
 class LinkerHandL25Can:
-    def __init__(self, can_channel='can0', baudrate=1000000, can_id=0x28):
+    def __init__(self, can_channel='can0', baudrate=1000000, can_id=0x28,yaml=""):
         self.can_id = can_id
+        self.can_channel = can_channel
+        self.baudrate = baudrate
+        self.open_can = OpenCan(load_yaml=yaml)
+
         self.running = True
         self.last_thumb_pos, self.last_index_pos,self.last_ring_pos,self.last_middle_pos, self.last_little_pos = None,None,None,None,None
         self.x01, self.x02, self.x03, self.x04,self.x05,self.x06,self.x07, self.x08,self.x09,self.x0A,self.x0B,self.x0C,self.x0D,self.x0E,self.speed = [],[],[],[],[],[],[],[],[],[],[],[],[],[],[]
@@ -148,18 +153,21 @@ def __init__(self, can_channel='can0', baudrate=1000000, can_id=0x28):
             176: 11,
         }
         # 根据操作系统初始化 CAN 总线
-        if sys.platform == "linux":
-            self.bus = can.interface.Bus(
-                channel=can_channel, interface="socketcan", bitrate=baudrate, 
-                can_filters=[{"can_id": can_id, "can_mask": 0x7FF}]
-            )
-        elif sys.platform == "win32":
-            self.bus = can.interface.Bus(
-                channel=can_channel, interface='pcan', bitrate=baudrate, 
-                can_filters=[{"can_id": can_id, "can_mask": 0x7FF}]
-            )
-        else:
-            raise EnvironmentError("Unsupported platform for CAN interface")
+        try:
+            if sys.platform == "linux":
+                self.bus = can.interface.Bus(
+                    channel=can_channel, interface="socketcan", bitrate=baudrate, 
+                    can_filters=[{"can_id": can_id, "can_mask": 0x7FF}]
+                )
+            elif sys.platform == "win32":
+                self.bus = can.interface.Bus(
+                    channel=can_channel, interface='pcan', bitrate=baudrate, 
+                    can_filters=[{"can_id": can_id, "can_mask": 0x7FF}]
+                )
+            else:
+                raise EnvironmentError("Unsupported platform for CAN interface")
+        except:
+            print("Please insert CAN device")
 
         # 启动接收线程
         self.receive_thread = threading.Thread(target=self.receive_response)
@@ -180,6 +188,14 @@ def send_command(self, frame_property, data_list):
             #print(f"Message sent: ID={hex(self.can_id)}, Data={data}")
         except can.CanError as e:
             print(f"Failed to send message: {e}")
+            self.open_can.open_can(self.can_channel)
+            time.sleep(1)
+            self.is_can = self.open_can.is_can_up_sysfs(interface=self.can_channel)
+            time.sleep(1)
+            if self.is_can:
+                self.bus = can.interface.Bus(channel=self.can_channel, interface="socketcan", bitrate=self.baudrate)
+            else:
+                print("Reconnecting CAN devices ....")
         time.sleep(0.001)
 
     def receive_response(self):