Add nucleo_communicate_py package

device/nucleo_communicate_py/README.md

@@ -0,0 +1,20 @@
+# nucleo communication
+
+Nucleoとの通信周り
+
+TODO: いつかC++に書き換える
+
+## Nodes
+
+- `Receiver`: Nucleoからデータを受け取る
+- `Sender`: Nucleoにデータを送る
+
+## Executables
+
+- `receiver`: `Receiver`Nodeをspin
+- `sender`: `Sender`Nodeをspin
+- `main`: `Receiver`, `Sender`Node2つを同時にspin
+
+## Launches
+
+TODO

device/nucleo_communicate_py/nucleo_communicate_py/main.py

@@ -0,0 +1,30 @@
+# ref:
+# https://github.com/ros2/examples/blob/7e47aee/rclpy/executors/examples_rclpy_executors/composed.py
+
+import sys
+
+import rclpy
+from rclpy.executors import SingleThreadedExecutor, ExternalShutdownException
+
+from .receiver import Receiver
+from .sender import Sender
+
+
+def main(args=sys.argv):
+ rclpy.init(args)
+ try:
+ sender = Sender()
+ receiver = Receiver()
+
+ executor = SingleThreadedExecutor()
+ executor.add_node(sender)
+ executor.add_node(receiver)
+ executor.spin()
+ except KeyboardInterrupt:
+ pass
+ except ExternalShutdownException:
+ sys.exit(1)
+
+
+if __name__ == "__main__":
+ main()

device/nucleo_communicate_py/nucleo_communicate_py/receiver.py

@@ -0,0 +1,33 @@
+import sys
+
+import rclpy
+from rclpy.node import Node
+from packet_interfaces.msg import Current, Flex, Voltage
+
+
+class Receiver(Node):
+ def __init__(self) -> None:
+ super().__init__("receiver")
+ self._current_publisher = self.create_publisher(Current, "current", 10)
+ # NOTE: packet_interfaces/Composed の命名と揃える
+ self._flex1_publisher = self.create_publisher(Flex, "flexsensor1", 10)
+ self._flex2_publisher = self.create_publisher(Flex, "flexsensor2", 10)
+ self._voltage_publisher = self.create_publisher(Voltage, "voltage")
+
+ def _timer_callback(self) -> None:
+ # TODO
+ # https://github.com/rogy-AquaLab/2024_cavolinia_nucleo 参照
+ # UARTでNucleoからデータを取得してpublishする
+ self.get_logger().info("tick")
+
+
+def main(args=sys.argv):
+ rclpy.init(args)
+ receiver = Receiver()
+ rclpy.spin(receiver)
+ receiver.destroy_node()
+ rclpy.shutdown()
+
+
+if __name__ == "__main__":
+ main()

device/nucleo_communicate_py/nucleo_communicate_py/sender.py

@@ -0,0 +1,47 @@
+import sys
+
+import rclpy
+from rclpy.node import Node
+from std_msgs.msg import Empty
+from packet_interfaces.msg import Power
+
+
+class Sender(Node):
+ def __init__(self) -> None:
+ super().__init__("sender")
+ self._quit_subscription = self.create_subscription(
+ Empty,
+ "quit",
+ self._quit_callback,
+ 10
+ )
+ self._order_subscription = self.create_subscription(
+ Power,
+ "order/power",
+ self._order_callback,
+ 10
+ )
+
+ def _quit_callback(self, _quit: Empty) -> None:
+ # TODO: UARTで書き込み
+ # https://github.com/rogy-AquaLab/2024_cavolinia_nucleo 参照
+ # [0xFF] を送る
+ self.get_logger().info("Received quit order")
+
+ def _order_callback(self, _order: Power) -> None:
+ # TODO
+ # https://github.com/rogy-AquaLab/2024_cavolinia_nucleo 参照
+ # [0x00, ...] を送る
+ self.get_logger().info("Received power order")
+
+
+def main(args=sys.argv):
+ rclpy.init(args)
+ sender = Sender()
+ rclpy.spin(sender)
+ sender.destroy_node()
+ rclpy.shutdown()
+
+
+if __name__ == "__main__":
+ main()

device/nucleo_communicate_py/package.xml

@@ -0,0 +1,17 @@
+<?xml version="1.0"?>
+<?xml-model href="http://download.ros.org/schema/package_format3.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
+<package format="3">
+ <name>nucleo_communicate_py</name>
+ <version>0.1.0</version>
+ <description>Nucleoとの通信周り</description>
+ <maintainer email="[email protected]">h1rono</maintainer>
+ <license>MIT</license>
+
+ <exec_depend>rclpy</exec_depend>
+ <exec_depend>std_msgs</exec_depend>
+ <exec_depend>packet_interfaces</exec_depend>
+
+ <export>
+ <build_type>ament_python</build_type>
+ </export>
+</package>

device/nucleo_communicate_py/setup.cfg

@@ -0,0 +1,4 @@
+[develop]
+script_dir=$base/lib/nucleo_communicate_py
+[install]
+install_scripts=$base/lib/nucleo_communicate_py

device/nucleo_communicate_py/setup.py

@@ -0,0 +1,29 @@
+from setuptools import find_packages, setup
+
+
+package_name = 'nucleo_communicate_py'
+
+setup(
+ name=package_name,
+ version='0.1.0',
+ packages=find_packages(exclude=['test']),
+ data_files=[
+ ('share/ament_index/resource_index/packages',
+ ['resource/' + package_name]),
+ ('share/' + package_name, ['package.xml']),
+ ],
+ install_requires=['setuptools'],
+ zip_safe=True,
+ maintainer='h1rono',
+ maintainer_email='[email protected]',
+ description='Nucleoとの通信周り',
+ license='MIT',
+ tests_require=[],
+ entry_points={
+ 'console_scripts': [
+ "sender = nucleo_commnicate_py:sender.main",
+ "receiver = nucleo_communicate_py:receiver.main",
+ "main = nucleo_communicate_py:main.main"
+ ],
+ },
+)