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=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,34 @@
+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", 10)
+        self._timer = self.create_timer(0.5, self._timer_callback)
+
+    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=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=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_communicate_py.sender:main",
+            "receiver = nucleo_communicate_py.receiver:main",
+            "main = nucleo_communicate_py.main:main"
+        ],
+    },
+)