feat: integrate with ros2 connector

Author: rachwalk

src/rai/rai/agents/voice_agent.py

@@ -23,7 +23,12 @@
 from numpy.typing import NDArray
 
 from rai.agents.base import BaseAgent
-from rai.communication import AudioInputDeviceConfig, StreamingAudioInputDevice
+from rai.communication import (
+    AudioInputDeviceConfig,
+    ROS2ARIConnector,
+    ROS2ARIMessage,
+    StreamingAudioInputDevice,
+)
 from rai_asr.models import BaseTranscriptionModel, BaseVoiceDetectionModel
 
 
@@ -38,6 +43,7 @@ def __init__(
         self,
         microphone_device_id: int,  # TODO: Change to name based instead of id based identification
         microphone_config: AudioInputDeviceConfig,
+        ros2_name: str,
         transcription_model: BaseTranscriptionModel,
         vad: BaseVoiceDetectionModel,
         grace_period: float = 1.0,
@@ -51,7 +57,8 @@ def __init__(
         microphone.configure_device(
             target=str(microphone_device_id), config=microphone_config
         )
-        super().__init__(connectors={"microphone": microphone})
+        ros2_connector = ROS2ARIConnector(ros2_name)
+        super().__init__(connectors={"microphone": microphone, "ros2": ros2_connector})
         self.microphone_device_id = str(microphone_device_id)
         self.should_record_pipeline: List[BaseVoiceDetectionModel] = []
         self.should_stop_pipeline: List[BaseVoiceDetectionModel] = []
@@ -89,7 +96,10 @@ def add_detection_model(
     def run(self):
         self.running = True
         self.listener_handle = self.connectors["microphone"].start_action(
-            self.microphone_device_id, self.on_new_sample
+            action_data=None,
+            target=self.microphone_device_id,
+            on_feedback=self.on_new_sample,
+            on_done=lambda: None,
         )
 
     def stop(self):
@@ -184,8 +194,12 @@ def transcription_thread(self, identifier: str):
                     del self.buffer_reminders[identifier]
             # self.transcription_model.save_wav(f"{identifier}.wav")
             transcription = self.transcription_model.consume_transcription()
+            print("Transcription: ", transcription)
+            self.connectors["ros2"].send_message(
+                ROS2ARIMessage(
+                    {"data": transcription}, {"msg_type": "std_msgs/msg/String"}
+                ),
+                "/from_human",
+            )
             self.transcription_threads[identifier]["transcription"] = transcription
             self.transcription_threads[identifier]["event"].set()
-        # TODO: sending the transcription once https://github.com/RobotecAI/rai/pull/360 is merged
-        self.logger.info(f"transcription thread {identifier} finished")
-        print(transcription)

src/rai/rai/communication/__init__.py

@@ -15,6 +15,7 @@
 from .ari_connector import ARIConnector, ARIMessage
 from .base_connector import BaseConnector, BaseMessage
 from .hri_connector import HRIConnector, HRIMessage, HRIPayload
+from .ros2.connectors import ROS2ARIConnector, ROS2ARIMessage
 from .sound_device_connector import (
     AudioInputDeviceConfig,
     SoundDeviceError,
@@ -29,6 +30,8 @@
     "HRIConnector",
     "HRIMessage",
     "HRIPayload",
+    "ROS2ARIConnector",
+    "ROS2ARIMessage",
     "StreamingAudioInputDevice",
     "SoundDeviceError",
     "AudioInputDeviceConfig",

src/rai/rai/communication/ros2/connectors.py

@@ -14,7 +14,7 @@
 
 import threading
 import uuid
-from typing import Any, Callable, Dict, Optional
+from typing import Any, Callable, Dict, Optional, TypedDict
 
 from rclpy.executors import MultiThreadedExecutor
 from rclpy.node import Node
@@ -23,8 +23,14 @@
 from rai.communication.ros2.api import ROS2ActionAPI, ROS2ServiceAPI, ROS2TopicAPI
 
 
+class ROS2ARIPayload(TypedDict):
+    data: Any
+
+
 class ROS2ARIMessage(ARIMessage):
-    def __init__(self, payload: Any, metadata: Optional[Dict[str, Any]] = None):
+    def __init__(
+        self, payload: ROS2ARIPayload, metadata: Optional[Dict[str, Any]] = None
+    ):
         super().__init__(payload, metadata)