Add dry run behaviour

Signed-off-by: Maria Teresa Ortega <teresa.ortega0903@gmail.com>

Author: teresa-ortega

examples/beluga/beluga_benchmark.py

@@ -24,6 +24,7 @@
 )
 @lambkin.option("--clock-rate", default=100.0)
 @lambkin.option("--sensor-topic", default="/scan")
+@lambkin.option("--dry-run", default=False)
 def nominal(ctx):
     """Run a nominal Beluga AMCL benchmark across sensor models and particle counts.
 
@@ -39,6 +40,7 @@ def nominal(ctx):
     print(f"sensor_topic: {ctx.options.sensor_topic}")
     print(f"source: {ctx.source}")
     print("---")
+    ctx.shell = lambkin.ShellProxy(dry_run=ctx.options.dry_run)
     ctx.shell.ros2.launch(
         ctx.source.path.parent / "beluga.launch.xml",
         f"sensor_model:={ctx.variant.sensor_model}",

src/lambkin/__init__.py

@@ -16,9 +16,11 @@
 
 from lambkin.common import named_product
 from lambkin.core.decorators import benchmark, option
+from lambkin.core.shell.proxy import ShellProxy
 
 __all__ = [
     "benchmark",
     "named_product",
     "option",
+    "ShellProxy",
 ]

src/lambkin/core/shell/proxy.py

@@ -27,13 +27,14 @@
 class _CommandProxy:
     """Builds a shell command lazily by chaining attribute access and calls."""
 
-    def __init__(self, parts: list[str]) -> None:
+    def __init__(self, parts: list[str], dry_run: bool = False) -> None:
         """Initialize the proxy with the command words accumulated so far."""
         self._parts = parts
+        self._dry_run = dry_run
 
     def __getattr__(self, name: str) -> _CommandProxy:
         """Append a new word to the command and return a new proxy."""
-        return _CommandProxy(self._parts + [name])
+        return _CommandProxy(self._parts + [name], self._dry_run)
 
     def __call__(self, *args: Any, **kwargs: Any) -> None:
         """Finalize and print the command.
@@ -59,12 +60,19 @@ def __call__(self, *args: Any, **kwargs: Any) -> None:
                 extra.extend([flag, shlex.quote(str(value))])
 
         command = " ".join(self._parts + extra)
-        print(f"[CMD]: {command}")
+
+        if self._dry_run:
+            # TODO: extend to dispatch commands to BackgroundProcess for real execution.
+            print(f"[CMD]: {command}")
 
 
 class ShellProxy:
     """Dry-run mock shell that prints commands instead of executing them."""
 
+    def __init__(self, dry_run: bool = False) -> None:
+        """Initialize the ShellProxy with the given dry_run flag."""
+        self._dry_run = dry_run
+
     def __getattr__(self, name: str) -> _CommandProxy:
         """Start building a new command from the given top-level tool name."""
-        return _CommandProxy([name])
+        return _CommandProxy([name], self._dry_run)