feat: node param override

better_launch/declarative.py

@@ -111,7 +111,10 @@ def exec_request(key: str, req: dict) -> Any:
 
             with res:
                 if isinstance(children, list):
-                    children = {f"{key}.children.{idx}": child for idx, child in enumerate(children)}
+                    children = {
+                        f"{key}.children.{idx}": child
+                        for idx, child in enumerate(children)
+                    }
 
                 if not isinstance(children, dict):
                     raise ValueError(
@@ -174,6 +177,7 @@ def _get_toml_args(toml: dict) -> list[DeclaredArg]:
 def launch_toml(
     path: str,
     launch_args: dict[str, str] = None,
+    extra_args: list[str] = None,
     *,
     # These should largely mirror the launch_this decorator
     ui: bool = None,
@@ -204,7 +208,7 @@ def launch_toml(
         executable = "my-node"
         name = "${name}"
 
-    Substitutions are also possible and use a similar syntax as in ROS1 (as shown for the `name` launch argument above). `if` and `unless` conditions can be added as well. 
+    Substitutions are also possible and use a similar syntax as in ROS1 (as shown for the `name` launch argument above). `if` and `unless` conditions can be added as well.
 
     All parameters below can be set through the launch file by declaring them on the global scope with a `bl_` prefix (i.e. `ui` becomes `bl_ui`).
 
@@ -310,6 +314,11 @@ def launch_toml(
             if arg is not None:
                 argv.extend([f"--{key}", str(arg)])
 
+    if extra_args:
+        if argv is None:
+            argv = []
+        argv.extend(extra_args)
+
     def launch_func(*args, **kwargs):
         _execute_toml(toml, eval_mode=eval_mode, **kwargs)
 

better_launch/launcher.py

@@ -117,6 +117,7 @@ class BetterLaunch(metaclass=BetterLaunchMeta):
 
     _launchfile: str = None
     _launch_func_args: dict[str, Any] = {}
+    _node_param_overrides: dict[str, dict[str, Any]] = {}
 
     def __init__(
         self,
@@ -224,6 +225,9 @@ def hello(self) -> None:
         # We don't want to log this
         print(msg)
 
+    def set_node_param_overrides(self, overrides: dict[str, dict[str, Any]]) -> None:
+        BetterLaunch._node_param_overrides = overrides or {}
+
     def spin(self, exit_with_last_node: bool = True) -> None:
         """Join the BetterLaunch thread until it terminates. You do **not** need to call this if you're using the [launch_this][] wrapper or the TUI.
 
@@ -1506,13 +1510,26 @@ def node(
         group = self.group_tip
         namespace = group.assemble_namespace()
 
+        # Merge CLI overrides (CLI takes precedence)
+        resolved_params: dict[str, Any]
+        if isinstance(params, str):
+            resolved_params = self.load_params(configfile=params)
+        elif params is None:
+            resolved_params = {}
+        else:
+            resolved_params = params
+
+        cli_overrides = self._node_param_overrides.get(name, {})
+        if cli_overrides:
+            resolved_params = {**resolved_params, **cli_overrides}
+
         node = Node(
             package,
             executable,
             name,
             namespace,
             remaps=remaps,
-            params=params,
+            params=resolved_params,
             cmd_args=cmd_args,
             env=env,
             isolate_env=isolate_env,

better_launch/utils/better_logging.py

@@ -13,7 +13,7 @@
 # Log format string for ROS so that we can identify and reformat its log messages.
 ROSLOG_PATTERN_ROS = "%%{severity}%%{time}%%{message}"
 
-# Regular expression matching ROSLOG_PATTERN_ROS. The named groups will be matched to 
+# Regular expression matching ROSLOG_PATTERN_ROS. The named groups will be matched to
 # logging.LogRecord attributes via their group names.
 ROSLOG_PATTERN_BL = r"%%(?P<levelname>\w+)%%(?P<created>[\d.]+)%%(?P<msg>[\s\S]*)"
 
@@ -221,15 +221,11 @@ def format(self, record: logging.LogRecord) -> str:
             record.levelno
         )
 
-        msg = record.getMessage()
-        if self.max_message_length > 0 and len(msg) > self.max_message_length:
-            msg = msg[: self.max_message_length] + "..."
-        record.msg = msg
-        # The message has already been formatted with its arguments above.
-        # Clearing record.args prevents the next formatter from attempting
-        # a second '%' substitution on the truncated text, which could crash
-        # if any format placeholders were removed during truncation.
-        record.args = None
+        if self.max_message_length > 0:
+            msg = record.getMessage()
+            if msg > self.max_message_length:
+                record.msg = msg[: self.max_message_length] + "…"
+
         return super().format(record)
 
 
@@ -302,26 +298,25 @@ def configure_logger(
     screen_formatter: logging.Formatter = None,
     file_formatter: logging.Formatter = None,
 ) -> None:
-    """Initialize the logging framework.
-    """
+    """Initialize the logging framework."""
     # TODO proper docstring
     if output:
-        if isinstance(output, Iterable) and not isinstance(output, str):
-            output = [output]
+        if isinstance(output, str) or not isinstance(output, Iterable):
+            sinks = [output]
+        else:
+            sinks = list(output)
 
-        for idx, sink in output:
-            if isinstance(sink, str):
-                output[idx] = LogSink[output.upper()]
-
-        output = set(output)
+        output_sinks = set(
+            LogSink[sink.upper()] if isinstance(sink, str) else sink for sink in sinks
+        )
     else:
-        output = {LogSink.SCREEN}
+        output_sinks = {LogSink.SCREEN}
 
     config = Settings()
     screen_filter = LevelFilter(config.screen_log_level)
     file_filter = LevelFilter(config.file_log_level)
 
-    for sink in output:
+    for sink in output_sinks:
         if sink == LogSink.SCREEN:
             screen_handler = roslog.launch_config.get_screen_handler()
             if screen_handler not in logger.handlers:

better_launch/utils/click.py

@@ -1,4 +1,5 @@
 from typing import Any, Type, Iterable, Callable
+import json
 from dataclasses import dataclass
 import click
 
@@ -53,20 +54,46 @@ def get_click_bl_options(expose: bool = False) -> list[click.Option]:
     list[click.Option]
         _description_
     """
+
     def update_value(ctx: click.Context, param: click.Parameter, value: Any):
         key = param.name[3:].replace("-", "_")
-        _update_settings(**{key: value})
+
+        if value is not None:
+            _update_settings(**{key: value})
+
+        if key == "node_param_override" and value:
+            ctx.command.allow_extra_args = True
+            ctx.allow_extra_args = True
+
+    def _bool_param_type(value: str) -> bool:
+        """Convert string to boolean for Click options."""
+        if isinstance(value, bool):
+            return value
+        val_lower = value.lower()
+        if val_lower in ("true", "1", "yes", "on"):
+            return True
+        if val_lower in ("false", "0", "no", "off"):
+            return False
+        raise click.BadParameter(f"Invalid boolean value: {value}")
 
     # XXX always keep these synchronized with our Settings class
     options = [
         click.Option(
             ["--bl-ui"],
-            type=bool,
+            type=_bool_param_type,
             default=None,
             help="Enforce or prevent starting the TUI",
             expose_value=expose,  # not passed to our run method
             callback=update_value,
         ),
+        click.Option(
+            ["--bl-node-param-override"],
+            type=_bool_param_type,
+            default=None,
+            help="Allow overriding node parameters from the command line",
+            expose_value=expose,
+            callback=update_value,
+        ),
         click.Option(
             ["--bl-colormode"],
             type=click.types.Choice([c.name for c in Colormode], case_sensitive=False),
@@ -138,11 +165,88 @@ def get_click_launch_command(
     allow_kwargs: bool = False,
 ) -> click.Command:
     click_cmd = click.Command(
-        cmd_name, callback=launch_func, params=options, help=cmd_help
+        cmd_name,
+        callback=launch_func,
+        params=options,
+        help=cmd_help,
+        context_settings={"ignore_unknown_options": True},
     )
 
-    if allow_kwargs:
-        click_cmd.allow_extra_args = True
-        click_cmd.ignore_unknown_options = True
+    click_cmd.allow_extra_args = allow_kwargs
+    click_cmd.ignore_unknown_options = True
 
     return click_cmd
+
+
+def args_to_dict(args: list[str]) -> dict[str, Any]:
+    """Convert a list of CLI arguments to a dictionary.
+
+    Parameters
+    ----------
+    args : list[str]
+        List of arguments, e.g. ["--foo", "bar", "--baz", "1.0"]
+
+    Returns
+    -------
+    dict[str, Any]
+        Dictionary of arguments, e.g. {"foo": "bar", "baz": 1.0}
+
+    Raises
+    ------
+    ValueError
+        If an argument does not start with "-" or if a value is missing.
+    """
+    result = {}
+    it = iter(args)
+    for arg in it:
+        if not arg.startswith("-"):
+            raise ValueError(f"Argument '{arg}' does not start with '-'")
+
+        key = arg.lstrip("-")
+        try:
+            value = next(it)
+        except StopIteration:
+            raise ValueError(f"Missing value for argument '{arg}'")
+
+        try:
+            value = json.loads(value)
+        except (json.JSONDecodeError, TypeError):
+            pass
+
+        result[key] = value
+
+    return result
+
+
+def parse_node_params(
+    kwargs: dict[str, Any],
+) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]:
+    """Separate node parameters from other keyword arguments.
+
+    Node parameters are identified by containing a dot in their key. The part
+    before the first dot is considered the node name, the part after the dot
+    is the parameter name.
+
+    Parameters
+    ----------
+    kwargs : dict[str, Any]
+        Dictionary of keyword arguments.
+
+    Returns
+    -------
+    tuple[dict[str, Any], dict[str, dict[str, Any]]]
+        Remaining keyword arguments and a dictionary of node parameters.
+    """
+    remaining_kwargs = {}
+    node_params = {}
+
+    for key, value in kwargs.items():
+        if "." in key:
+            node_name, param_name = key.split(".", 1)
+            if node_name not in node_params:
+                node_params[node_name] = {}
+            node_params[node_name][param_name] = value
+        else:
+            remaining_kwargs[key] = value
+
+    return remaining_kwargs, node_params

better_launch/utils/settings.py

@@ -46,6 +46,7 @@ def severity_to_loglevel(severity: str) -> int:
 @dataclass(frozen=True)
 class _Settings:
     ui: bool = False
+    node_param_override: bool = False
     colormode: Colormode = Colormode.DEFAULT
     print_limit: int = 0
     screen_log_level: int = logging.INFO
@@ -139,15 +140,15 @@ def get_env_variables(self) -> dict[str, Any]:
         return vars
 
     def as_dict(self) -> dict[str, Any]:
-        """Returns the settings as a dict.
-        """
+        """Returns the settings as a dict."""
         # A bit more comfortable than having to import dataclasses.asdict each time
         return asdict(self)
 
+
 def _update_settings(**overrides) -> None:
     """Replace the _SETTINGS object with a new instance with updated values. Only non-None values are applied.
 
-    This should only be called right after the launch process has started. 
+    This should only be called right after the launch process has started.
 
     Parameters
     ----------