ROSNode

ROSNode wrapper for simulation and ROS2 deployment.

Two workflows:

Simulation — call step() directly in Python or Jupyter notebooks to test ros2py and/or py2ros translation functions. No ROS2 installation required.

Deployment — call write_ROSNode_to_rclpy(path) to emit a self-contained .py file that spins up a native rclpy node. Every ROS2 detail (QoS profiles, buffer depths, subscription callbacks, executor setup, etc…) is written out explicitly and can be modified as needed. The .py file also has no runtime dependency on dynamicalnodes beyond DynamicalSystem and whatever dependencies the f/h functions may have.

See also

DynamicalSystem

Core abstraction for modeling control components.

ros2py_py2ros

Message conversion utilities (ros2py_* / py2ros_* functions).

dynamicalnodes.rosnode.PubDict

Type alias for publication configuration dict

alias of Dict[str, Any]

class dynamicalnodes.rosnode.ROSNode(*, dynamical_system, state_name=None, subscribes_to=None, publishes_to=None, sync_mode=None, timer_hz=None)[source]

Bases: object

Wraps a DynamicalSystem for simulation and ROS2 deployment.

Use step() for notebook simulation; use write_ROSNode_to_rclpy() to generate a deployable rclpy node.

Parameters:

  • dynamical_system (DynamicalSystem) – The dynamical system to step.

  • subscribes_to (Optional[Sequence[Dict[str, Any]]]) –

    Each dict:

    • "topic" (str): ROS topic name.

    • "msg_type" (type): ROS message class.

    • "arg" (str): kwarg name passed to step().

    • "ros2py" (Callable): ROS msg → NumPy array converter. Import from dynamicalnodes.ros2py_py2ros or write your own.

    • "stale_after" (float, optional): Seconds before data expires.

    • "buffer_size" (int, optional): Queue depth (default 1).

    • "use_msg_timestamp" (bool, optional): When True, the subscription buffer stores the message’s own header.stamp as the timestamp instead of ROS-clock arrival time. Only valid for message types that have a header field. Useful when staleness should be relative to data capture time rather than network-arrival time (e.g. sensor fusion, bag replay). Defaults to False.

  • publishes_to (Optional[Sequence[Dict[str, Any]]]) –

    Each dict:

    • "topic" (str): ROS topic name.

    • "msg_type" (type): ROS message class.

    • "py2ros" (Callable): NumPy array → ROS msg converter. Import from dynamicalnodes.ros2py_py2ros.

    • "key" (str, optional): Key into h() return dict. Required when h() returns a dict and there are multiple publishers.

  • sync_mode (Optional[str]) – "any" — step whenever any subscription has fresh data. "all" — step only when every subscription has fresh data.

  • state_name (Optional[str]) – Name of the state parameter in f’s signature (e.g. "ck" for f(ck, ...)) . Required for stateful systems; ignored if f is None. step() injects self._state under this name before calling DynamicalSystem.step().

  • timer_hz (Optional[float]) – In the generated node: fires _run_step at this frequency instead of triggering from subscription callbacks.

Examples

Stateless filter (no state, single subscriber):

>>> from dynamicalnodes import DynamicalSystem, ROSNode
>>> from std_msgs.msg import Float64
>>> from dynamicalnodes.ros2py_py2ros import ros2py_float64, py2ros_float64
>>> import numpy as np
>>>
>>> smoother = DynamicalSystem(h=lambda imu: imu * 0.5)
>>> node = ROSNode(
...     dynamical_system=smoother,
...     subscribes_to=[{"topic": "/raw", "msg_type": Float64,
...                      "arg": "imu", "ros2py": ros2py_float64}],
...     publishes_to=[{"topic": "/smooth", "msg_type": Float64,
...                    "py2ros": py2ros_float64}],
... )
>>> out = node.step(imu=np.array([2.0]))
>>> out.data
1.0
property state: Any

Current state of the dynamical system.

step(**kwargs)[source]

Simulate one step without ROS2.

Converts ROS message inputs via their configured ros2py functions, steps the DynamicalSystem, and converts the output via py2ros. Use this to verify that message converters and system wiring are correct before deploying to ROS2.

Parameters:

**kwargs (Any) –

  • Subscription arg names: ROS messages or NumPy arrays.

  • Any extra parameters forwarded verbatim to the DynamicalSystem.

Returns:

Single publisher: the ROS message from h(). Multiple publishers: {topic: msg} dict. No publishers: raw h() output.

Return type:

Any

Examples

>>> import numpy as np
>>> from dynamicalnodes import DynamicalSystem, ROSNode
>>> from std_msgs.msg import Float64
>>> from dynamicalnodes.ros2py_py2ros import ros2py_float64, py2ros_float64
>>>
>>> double = DynamicalSystem(h=lambda x: x * 2)
>>> node = ROSNode(
...     dynamical_system=double,
...     subscribes_to=[{"topic": "/x", "msg_type": Float64,
...                      "arg": "x", "ros2py": ros2py_float64}],
...     publishes_to=[{"topic": "/y", "msg_type": Float64,
...                    "py2ros": py2ros_float64}],
... )
>>> node.step(x=Float64(data=3.0)).data
6.0
write_ROSNode_to_rclpy(path, *, node_name, deps=None, initial_state=None, initial_inputs=None, static_params=None, dynamic_params=None, sub_noise=None, pub_noise=None)[source]

Generate a standalone rclpy Python file that can be run with ros2 run.

The output file is a self-contained native rclpy node: QoS profiles, buffer depths, subscription callbacks, publisher calls, and executor setup are all written out explicitly in plain rclpy — exactly as a ROS2 developer would write them by hand.

Parameters:

  • path (str) – Destination file path (e.g. "my_controller_node.py").

  • node_name (str) – ROS2 node name (used for the generated class name, super().__init__, and logger messages).

  • deps (Optional[List[Any]]) – Helper functions that f or h depend on. Each function’s source is inlined before the dynamics functions in the output file. Use this when your notebook or module defines utility functions that f/h call internally.

  • initial_state (Optional[Any]) – Embedded as self._state = ... in the generated node’s __init__. Supports NumPy arrays and any repr()-able value.

  • static_params (Optional[Dict[str, Any]]) – Parameters baked into the node at generation time. Embedded as self._static_params = {...} and merged into every step() call. Use for values that never change after deployment.

  • dynamic_params (Optional[Dict[str, Any]]) –

    Parameters declared on the ROS2 parameter server with their default values. Readable and writable at runtime via:

    ros2 param set /<node_name> <name> <value>

    Each key is emitted as declare_parameter(name, default) in __init__ and read via get_parameter(name).value in every _run_step call. Use for values you want to tune without restarting the node (e.g. controller gains).

  • initial_inputs (Optional[Dict[str, Any]]) – Cold-start fallback values for subscription inputs. Keys must match subscription arg names. A value is used only when the subscription buffer is completely empty — i.e. no message has ever arrived on that topic. Once the first message arrives the subscription value takes over permanently. If the subscription later goes stale the step is skipped (None is not replaced), preventing silent operation on dead data. Use this to break hard startup ordering dependencies between nodes.

  • sub_noise (Optional[Dict[str, float]]) – Zero-mean Gaussian noise injected into subscription data after ros2py conversion. Keys are topic strings; values are standard deviations (same units as the converted NumPy array). Useful for robustness testing and hardware-in-the-loop simulation. Example: {"/yk": 0.1} adds noise with std=0.1 to /yk.

  • pub_noise (Optional[Dict[str, float]]) – Zero-mean Gaussian noise injected into publication data before py2ros conversion. Same format as sub_noise. Example: {"/uk": 50.0} adds noise with std=50 to /uk.

Raises:

  • ValueError – If any function is a lambda (cannot be serialized).

  • ValueError – If the node has no subscriptions and no timer_hz — the generated node would never step.

  • OSError – If inspect.getsource() cannot retrieve a function’s source (e.g. C extensions). Use deps to wrap such functions.

  • SyntaxError – If the generated file contains invalid Python (indicates a bug in the generator).

Return type:

None

Notes

Function resolution

Return type:

None

Parameters:

  • path (str)

  • node_name (str)

  • deps (List[Any] | None)

  • initial_state (Any | None)

  • initial_inputs (Dict[str, Any] | None)

  • static_params (Dict[str, Any] | None)

  • dynamic_params (Dict[str, Any] | None)

  • sub_noise (Dict[str, float] | None)

  • pub_noise (Dict[str, float] | None)

Functions from dynamicalnodes.ros2py_py2ros are emitted as import statements. Functions from any other importable module (not __main__ or an IPython cell) are also imported by name. All other functions — typically those defined interactively in a notebook or script — are inlined verbatim via inspect.getsource().

Examples

>>> import numpy as np, tempfile, os
>>> from dynamicalnodes import DynamicalSystem, ROSNode
>>> from std_msgs.msg import Float64
>>> from dynamicalnodes.ros2py_py2ros import ros2py_float64, py2ros_float64
>>>
>>> def f_plant(xk, u_k):
...     return xk + 0.1 * u_k
>>> def h_plant(xk):
...     return xk
>>>
>>> sys = DynamicalSystem(f=f_plant, h=h_plant)
>>> node = ROSNode(
...     dynamical_system=sys,
...     subscribes_to=[{"topic": "/u_k", "msg_type": Float64,
...                      "arg": "u_k", "ros2py": ros2py_float64,
...                      "stale_after": 0.5}],
...     publishes_to=[{"topic": "/y_k", "msg_type": Float64,
...                    "py2ros": py2ros_float64}],
... )
>>> with tempfile.NamedTemporaryFile(suffix=".py", delete=False) as tmp:
...     path = tmp.name
>>> node.write_ROSNode_to_rclpy(path, node_name="plant",
...                             initial_state=np.array([0.0]))
>>> content = open(path).read()
>>> "def f_plant" in content and "def h_plant" in content
True
>>> "ros2py_float64" in content and "py2ros_float64" in content
True
>>> "create_subscription" in content and "create_publisher" in content
True
>>> "SingleThreadedExecutor" in content
True
>>> os.unlink(path)
dynamicalnodes.rosnode.SubDict

Type alias for subscription configuration dict

alias of Dict[str, Any]