From 5bff590a0566f9f14d51af2cc13ca0f2449f3f12 Mon Sep 17 00:00:00 2001
From: Tiffany Cappellari <156013635+tcappellari-bdai@users.noreply.github.com>
Date: Thu, 16 Jan 2025 08:24:08 -0800
Subject: [PATCH] [SW-1836] Move values.py from bdai_ros to synchros2 (#138)

---
 synchros2/synchros2/launch/values.py | 65 ++++++++++++++++++++++++++++
 synchros2/synchros2/subscription.py  |  4 +-
 2 files changed, 67 insertions(+), 2 deletions(-)
 create mode 100644 synchros2/synchros2/launch/values.py

diff --git a/synchros2/synchros2/launch/values.py b/synchros2/synchros2/launch/values.py
new file mode 100644
index 0000000..0bbd8dd
--- /dev/null
+++ b/synchros2/synchros2/launch/values.py
@@ -0,0 +1,65 @@
+"""Provides a helper class for accessing the values of launch arguments from a LaunchConfiguration
+
+Copyright (c) 2024 Boston Dynamics AI Institute Inc.  All rights reserved.
+"""
+# mypy: ignore-errors
+from launch import LaunchContext
+from launch.substitutions import LaunchConfiguration
+
+from synchros2.launch.actions import convert_to_bool
+
+
+class LaunchConfigurationValues:
+    """Helper class for accessing the values of launch arguments from a LaunchConfiguration.
+
+    This helper serves the following purposes:
+    1. Avoid spending a line of code on each of a potentially large number of values retrieved from the launch config
+    2. Enable easily accessing the values of boolean arguments as actual `bool` types when needed,
+       while still normally treating them as `str` the way downstream launch operations usually expect.
+
+    Example:
+    ----------------------------------------------------------------
+    def launch_setup(context: LaunchContext):
+        launch_elements = []
+        vals = LaunchConfigurationValues(context)
+        vals["robot_name"]  # str
+        vals.bool("use_afterburner")  # bool
+
+        # The "condition" kwarg expects a launch operation on a string-type boolean, not an actual bool
+        launch_elements.append(
+            Node(
+                [...]
+                condition=IfCondition(vals["use_afterburner"]),  # Argument is in ["true", "false"], not [True, False]
+            ),
+        )
+
+        if vals.bool("use_afterburner"):  # Access value as one of [True, False]
+            # Add an additional launch element
+            [...]
+    ----------------------------------------------------------------
+    """
+
+    def __init__(self, context: LaunchContext):
+        """Initialize a LaunchConfigurationValues helper for a given launch context
+
+        Args:
+            context : The LaunchContext being used in the launch_setup function where this helper class is needed.
+        """
+        self._context = context
+        self._string_values: dict[str, str] = {}
+        self._bool_values: dict[str, bool] = {}
+
+    def __getitem__(self, launch_arg_name: str) -> str:
+        """Retrieve the string value of the specified launch argument"""
+        if launch_arg_name not in self._string_values:
+            self._string_values[launch_arg_name] = LaunchConfiguration(launch_arg_name).perform(self._context)
+        return self._string_values[launch_arg_name]
+
+    def bool(self, launch_arg_name: str) -> bool:  # noqa: A003
+        """Retrieve the boolean value of the specified launch argument.
+
+        This method should fail if the argument's string value is not one of the expected options for a boolean.
+        """
+        if launch_arg_name not in self._bool_values:
+            self._bool_values[launch_arg_name] = convert_to_bool(launch_arg_name, self.__getitem__(launch_arg_name))
+        return self._bool_values[launch_arg_name]
diff --git a/synchros2/synchros2/subscription.py b/synchros2/synchros2/subscription.py
index 808a625..895f668 100644
--- a/synchros2/synchros2/subscription.py
+++ b/synchros2/synchros2/subscription.py
@@ -130,8 +130,8 @@ def topic_name(self) -> str:
         return self._topic_name
 
     # Aliases for improved readability
-    cancel = MessageFeed.close
-    unsubscribe = MessageFeed.close
+    cancel = MessageFeed.close  # type: ignore
+    unsubscribe = MessageFeed.close  # type: ignore
 
 
 def wait_for_message_async(