Refactor/constants

frida_constants/CMakeLists.txt

@@ -0,0 +1,13 @@
+cmake_minimum_required(VERSION 3.8)
+project(frida_constants)
+
+# Find dependencies
+find_package(std_msgs REQUIRED)
+find_package(ament_cmake REQUIRED)
+find_package(ament_cmake_python REQUIRED)
+find_package(rclcpp REQUIRED)
+find_package(rclpy REQUIRED)
+
+ament_python_install_package(${PROJECT_NAME})
+
+ament_package()

frida_constants/frida_constants/__init__.py

@@ -0,0 +1,11 @@
+from enum import Enum
+
+import frida_constants.hri_constants as hri_constants
+from frida_constants.utils import parse_ros_config
+
+
+class ModuleNames(Enum):
+    HRI = hri_constants.__name__
+
+
+__all__ = [hri_constants.__name__, parse_ros_config.__name__, "ModuleNames"]

frida_constants/frida_constants/hri_constants.py

@@ -0,0 +1,8 @@
+SPEAK_SERVICE = "/speech/speak"
+HEAR_SERVICE = "/speech/STT"
+KEYWORD_TOPIC = "/speech/kws"
+COMMAND_INTERPRETER_SERVICE = "/nlp/command_interpreter"
+DATA_EXTRACTOR_SERVICE = "/nlp/data_extractor"
+SENTENCE_BUILDER_SERVICE = "/nlp/sentence_builder"
+ITEM_CATEGORIZATION_SERVICE = "/nlp/item_categorization"
+CONVESATION_SERVICE = "/nlp/conversation"

frida_constants/frida_constants/utils.py

@@ -0,0 +1,55 @@
+import importlib
+import inspect
+
+import yaml
+
+
+def get_constants_from_module(module_name):
+    """Get all constants from a module into a dict"""
+    module = importlib.import_module(module_name)
+
+    constants = {
+        name: value
+        for name, value in inspect.getmembers(module)
+        if not name.startswith("__")
+        and not inspect.ismodule(value)
+        and not inspect.isfunction(value)
+    }
+    return constants
+
+
+# Contains all constants from all modules
+all_constants = {}
+
+
+# Get a constant value from a module
+def get_constant(constant_name: str, module_names: list[str]):
+    for module_name in module_names:
+        if constant_name in all_constants[module_name]:
+            return all_constants[module_name][constant_name]
+
+    raise ValueError(f"Constant {constant_name} not found in any of the modules.")
+
+
+def process_dict_config(config_object, module_names: list[str]):
+    """Recursively replace all "REPLACE" strings with the corresponding constant value"""
+    for key, value in config_object.items():
+        if isinstance(value, dict):
+            process_dict_config(value, module_names)
+        elif isinstance(value, str):
+            if value == "REPLACE":
+                config_object[key] = get_constant(key, module_names)
+
+    return config_object
+
+
+def parse_ros_config(config_path: str, module_names: list[str]):
+    """Process a ROS-formatted yaml config file, replacing all "REPLACE" strings with the corresponding constant value"""
+    for module_name in module_names:
+        if module_name not in all_constants:
+            all_constants[module_name] = get_constants_from_module(module_name)
+
+    with open(config_path, "r") as file:
+        config = yaml.safe_load(file)
+
+    return process_dict_config(config, module_names)

frida_constants/package.xml

@@ -0,0 +1,18 @@
+<package format="3">
+  <name>frida_constants</name>
+  <version>0.1.0</version>
+  <description>Package containing shared constants and utilities.</description>
+  <maintainer email="[email protected]">RoBorregos</maintainer>
+  <license>Apache-2.0</license>
+
+  <!-- Build dependencies -->
+  <buildtool_depend>ament_cmake</buildtool_depend>
+  <buildtool_depend>ament_cmake_python</buildtool_depend>
+
+  <depend>rclcpp</depend>
+  <depend>rclpy</depend>
+
+  <export>
+    <build_type>ament_cmake</build_type>
+  </export>
+</package>

hri/packages/speech/config/speaker.yaml

@@ -4,7 +4,7 @@ say:
     SPEAKER_INPUT_CHANNELS: 32
     SPEAKER_OUT_CHANNELS: 32
 
-    speak_service: "/speech/speak"
+    SPEAK_SERVICE: REPLACE
     speak_topic: "/speech/speak_now"
     speaking_topic: "saying"
 

hri/packages/speech/launch/devices_launch.py

@@ -4,17 +4,23 @@
 from launch import LaunchDescription
 from launch_ros.actions import Node
 
+from frida_constants import ModuleNames, parse_ros_config
+
 
 def generate_launch_description():
     mic_config = os.path.join(
         get_package_share_directory("speech"), "config", "microphone.yaml"
     )
+
     hear_config = os.path.join(
         get_package_share_directory("speech"), "config", "hear.yaml"
     )
-    speaker_config = os.path.join(
-        get_package_share_directory("speech"), "config", "speaker.yaml"
-    )
+
+    speaker_config = parse_ros_config(
+        os.path.join(get_package_share_directory("speech"), "config", "speaker.yaml"),
+        [ModuleNames.HRI.value],
+    )["say"]["ros__parameters"]
+
     # respeaker_config = os.path.join(
     #     get_package_share_directory("speech"), "config", "respeaker.yaml"
     # )

hri/packages/speech/scripts/say.py

@@ -12,6 +12,7 @@
 from speech.wav_utils import WavUtils
 from std_msgs.msg import Bool, String
 
+from frida_constants.hri_constants import SPEAK_SERVICE
 from frida_interfaces.srv import Speak
 
 CURRENT_FILE_PATH = os.path.abspath(__file__)
@@ -26,7 +27,7 @@ def __init__(self):
         self.connected = False
         self.declare_parameter("speaking_topic", "/saying")
 
-        self.declare_parameter("speak_service", "/speech/speak")
+        self.declare_parameter("SPEAK_SERVICE", SPEAK_SERVICE)
         self.declare_parameter("speak_topic", "/speech/speak_now")
         self.declare_parameter("model", "en_US-amy-medium")
         self.declare_parameter("offline", True)
@@ -54,7 +55,7 @@ def __init__(self):
         )
 
         speak_service = (
-            self.get_parameter("speak_service").get_parameter_value().string_value
+            self.get_parameter("SPEAK_SERVICE").get_parameter_value().string_value
         )
         speak_topic = (
             self.get_parameter("speak_topic").get_parameter_value().string_value