open_pub/Uni-Lab-OS
1
0
Fork 0
mirror of https://github.com/dptech-corp/Uni-Lab-OS.git synced 2025-12-17 04:51:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: unit conversion

Browse Source 
feat: better attr publisher log
This commit is contained in:
wznln
2025-04-23 15:04:08 +08:00
parent 0cd11fa46b
commit f476b40983
6 changed files with 39 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
unilabos/ros/nodes/base_device_node.py
View File

@@ -342,9 +342,15 @@ class BaseROS2DeviceNode(Node, Generic[T]):
else:
return getattr(self.driver_instance, attr_name)
except AttributeError as ex:
self.lab_logger().error(
f"publish error, {str(type(self.driver_instance))[8:-2]} has no attribute '{attr_name}'"
)
if ex.args[0].startswith(f"AttributeError: '{self.driver_instance.__class__.__name__}' object"):
self.lab_logger().error(
f"publish error, {str(type(self.driver_instance))[8:-2]} has no attribute '{attr_name}'"
)
else:
self.lab_logger().error(
f"publish error, when {str(type(self.driver_instance))[8:-2]} getting attribute '{attr_name}'"
)
self.lab_logger().error(traceback.format_exc())
self._property_publishers[attr_name] = PropertyPublisher(
self, attr_name, get_device_attr, msg_type, initial_period, self._print_publish

10
unilabos/ros/nodes/presets/serial_node.py
View File

@@ -47,7 +47,7 @@ class ROS2SerialNode(BaseROS2DeviceNode):
self.lab_logger().info(f"【ROS2SerialNode.__init__】创建串口写入服务: serialwrite")
def send_command(self, command: str):
self.lab_logger().info(f"【ROS2SerialNode.send_command】发送命令: {command}")
# self.lab_logger().debug(f"【ROS2SerialNode.send_command】发送命令: {command}")
with self._query_lock:
if self._closing:
self.lab_logger().error(f"【ROS2SerialNode.send_command】设备正在关闭，无法发送命令")
@@ -59,23 +59,23 @@ class ROS2SerialNode(BaseROS2DeviceNode):
response = self.hardware_interface.write(full_command_data)
# time.sleep(0.05)
output = self._receive(self.hardware_interface.read_until(b"\n"))
self.lab_logger().info(f"【ROS2SerialNode.send_command】接收响应: {output}")
# self.lab_logger().debug(f"【ROS2SerialNode.send_command】接收响应: {output}")
return output
def read_data(self):
self.lab_logger().debug(f"【ROS2SerialNode.read_data】读取数据")
# self.lab_logger().debug(f"【ROS2SerialNode.read_data】读取数据")
with self._query_lock:
if self._closing:
self.lab_logger().error(f"【ROS2SerialNode.read_data】设备正在关闭，无法读取数据")
raise RuntimeError
data = self.hardware_interface.read_until(b"\n")
result = self._receive(data)
self.lab_logger().debug(f"【ROS2SerialNode.read_data】读取到数据: {result}")
# self.lab_logger().debug(f"【ROS2SerialNode.read_data】读取到数据: {result}")
return result
def _receive(self, data: bytes):
ascii_string = "".join(chr(byte) for byte in data)
self.lab_logger().debug(f"【ROS2SerialNode._receive】接收数据: {ascii_string}")
# self.lab_logger().debug(f"【ROS2SerialNode._receive】接收数据: {ascii_string}")
return ascii_string
def handle_serial_request(self, request, response):