Metadata-Version: 2.5
Name: ros-pydantic-gen
Version: 0.2.0
Summary: Generate pydantic v2 models from a live ROS graph via rosbridge
License: MIT
Requires-Python: >=3.9
Requires-Dist: pydantic>=2.0
Requires-Dist: roslibpy>=1.5
Provides-Extra: dev
Requires-Dist: pytest>=7; extra == 'dev'
Requires-Dist: ruff>=0.4; extra == 'dev'
Description-Content-Type: text/markdown

# ros-pydantic-gen

Point it at a running robot, get typed Python back.

`ros-pydantic-gen` introspects a live ROS graph through
[rosbridge](http://wiki.ros.org/rosbridge_suite) and writes a single
self-contained module of [Pydantic v2](https://docs.pydantic.dev) models — one
model per message type (nested types included), a topic → model registry, and a
snapshot of the parameter server.

```python
# before: what shape is this, exactly?
listener.subscribe(lambda msg: print(msg["pose"]["pose"]["position"]["x"]))

# after: autocomplete, type checking, validation at the boundary
ros_models.subscribe(client, "/odom", lambda m: print(m.pose.pose.position.x))
```

It talks to ROS over a WebSocket, so it needs **no local ROS installation** — it
runs happily from WSL, macOS, or a laptop pointed at a robot on the network.

---

## Contents

- [Requirements](#requirements)
- [Installation](#installation)
- [Preparing the ROS side](#preparing-the-ros-side)
- [Quick start](#quick-start)
- [What gets generated](#what-gets-generated)
- [Using the generated module](#using-the-generated-module)
- [CLI reference](#cli-reference)
- [Recipes](#recipes)
- [How it handles ROS's awkward bits](#how-it-handles-ross-awkward-bits)
- [Troubleshooting](#troubleshooting)
- [Project layout](#project-layout)
- [Development](#development)

---

## Requirements

| | |
| --- | --- |
| Python | 3.9+ (developed and tested on 3.12) |
| Runtime deps | `pydantic>=2.0`, `roslibpy>=1.5` |
| On the robot | `rosbridge_server` **and** `rosapi` |
| ROS version | ROS 1 fully supported; ROS 2 works — see [ROS 2 notes](#ros-2) |

Nothing here requires `rospy`, `catkin`, or a sourced ROS workspace on your
machine.

---

## Installation

### From the archive

```bash
unzip ros-pydantic-gen.zip
cd ros-pydantic-gen

python -m venv .venv
source .venv/bin/activate          # Windows: .venv\Scripts\activate

pip install -e ".[dev]"            # editable install + pytest/ruff
```

`-e` (editable) is what you want while the generator is still being adapted to
your robot's message set — edits take effect without reinstalling. Drop the
`[dev]` extra if you don't need the test suite:

```bash
pip install -e .
```

### Verify

```bash
ros-pydantic-gen --version         # -> ros-pydantic-gen 0.2.0
ros-pydantic-gen --help
pytest                             # 81 tests, no ROS required
```

The console script and `python -m ros_pydantic_gen` are equivalent; use the
latter if you'd rather not install at all:

```bash
PYTHONPATH=src python -m ros_pydantic_gen --help
```

### Into an existing project

```bash
pip install /path/to/ros-pydantic-gen
```

Then either call the CLI, or drive it from Python:

```python
from pathlib import Path
from ros_pydantic_gen import ConnectionConfig, RosIntrospector, connect, generate

with connect(ConnectionConfig(host="172.17.0.1", port=9001)) as client:
    graph = RosIntrospector(client).snapshot()

Path("ros_models.py").write_text(generate(graph))
```

---

## Preparing the ROS side

The generator reads the graph through `rosapi`, a node that ships with
`rosbridge_suite`. It is launched automatically by the standard launch file, so
in most cases this is all you need:

**ROS 1**

```bash
sudo apt install ros-noetic-rosbridge-suite
roslaunch rosbridge_server rosbridge_websocket.launch
# listening on ws://0.0.0.0:9090
```

**ROS 2**

```bash
sudo apt install ros-humble-rosbridge-suite
ros2 launch rosbridge_server rosbridge_websocket_launch.xml
```

**Non-default port** (Duckietown, for instance, commonly exposes 9001):

```bash
roslaunch rosbridge_server rosbridge_websocket.launch port:=9001
```

**Docker.** If the bridge runs in a container, the host reaches it on the docker
bridge address — `172.17.0.1` on Linux — provided the port is published
(`-p 9001:9001`). From WSL, the same address usually works when the container
runs inside WSL's Docker.

Check the bridge is reachable before generating anything:

```bash
python -c "import roslibpy; r=roslibpy.Ros('172.17.0.1', 9001); r.run(); \
print('topics:', len(r.get_topics())); r.terminate()"
```

---

## Quick start

```bash
ros-pydantic-gen --host 172.17.0.1 --port 9001 -o ros_models.py
```

```
connected to ws://172.17.0.1:9001
  found 33 topics / 23 distinct types
  resolving duckietown_msgs/BoolStamped
  resolving duckietown_msgs/Twist2DStamped
  ...
  found 41 parameters
wrote ros_models.py (57 models, 33 topics)
```

Progress goes to **stderr** and the module to the path you gave, so `-q`
silences the chatter for scripting without affecting the output.

---

## What gets generated

One module. No package, no imports beyond `pydantic`. Roughly:

```python
class RosMessage(BaseModel):
    """Base class for every generated ROS message model."""
    model_config = ConfigDict(populate_by_name=True, extra="ignore",
                              validate_assignment=True)
    ROS_TYPE: ClassVar[str] = ""

    def to_ros(self) -> dict[str, Any]: ...


class RosgraphMsgsLog(RosMessage):
    """ROS message ``rosgraph_msgs/Log``."""

    ROS_TYPE: ClassVar[str] = "rosgraph_msgs/Log"
    DEBUG: ClassVar[Any] = 1
    INFO: ClassVar[Any] = 2
    WARN: ClassVar[Any] = 4

    header: StdMsgsHeader = Field(default_factory=StdMsgsHeader)  # std_msgs/Header
    level: int = 0  # byte
    msg: str = ""  # string
    topics: list[str] = Field(default_factory=list)  # string[]


TOPICS: dict[str, type[RosMessage]] = {
    "/odom": NavMsgsOdometry,
    "/rosout": RosgraphMsgsLog,
    ...
}

TOPIC_TYPES: dict[str, str] = {"/odom": "nav_msgs/Odometry", ...}


class RosParams(BaseModel):
    use_sim_time: bool = Field(False, alias="/use_sim_time")
    ...
```

Every field carries a trailing comment with its original ROS type
(`# float64[36]`), so the generated file doubles as a readable message
reference. `examples/ros_models.py` is a complete one.

---

## Using the generated module

### Subscribing

```python
import roslibpy
import ros_models

client = roslibpy.Ros("172.17.0.1", 9001)
client.run()

def on_odom(msg: ros_models.NavMsgsOdometry) -> None:
    print(msg.pose.pose.position.x, msg.twist.twist.angular.z)

ros_models.subscribe(client, "/odom", on_odom)
```

`subscribe` looks the message type up in the registry, wires up the
`roslibpy.Topic` for you, and validates each incoming payload into a model
instance before your callback sees it.

### Publishing

```python
twist = ros_models.GeometryMsgsTwist()
twist.linear.x = 0.2
twist.angular.z = -0.4

ros_models.publish(client, "/cmd_vel", twist)
```

Defaults mirror ROS's own zero-initialisation, so `GeometryMsgsTwist()` is a
valid all-zeros message — no need to spell out every field. That holds for
messages with fixed-length arrays too: `NavMsgsOdometry()` gives you a
36-element zero covariance, not a validation error.

### Validating by hand

Useful when you already have a subscriber, or you're parsing a bag dump:

```python
model = ros_models.model_for("/odom")
msg = model.model_validate(raw_dict)          # raises ValidationError on mismatch
payload = msg.to_ros()                        # back to a ROS-shaped dict
```

`to_ros()` emits **alias** names, so reserved-word fields go back out as `from`,
`class`, and so on — round-tripping is byte-identical.

### Type checking

The module is fully annotated. Under mypy or pyright, `msg.pose.pose.positon.x`
(note the typo) becomes a caught error rather than a `KeyError` at 3am on a
moving robot.

---

## CLI reference

```
ros-pydantic-gen [--host H] [--port P] [--secure] [--timeout S]
                 [--include RE] [--exclude RE] [--no-params]
                 [-o PATH] [--no-defaults] [--raw-uint8]
                 [--dump-json PATH] [--from-json PATH] [-q]
```

**Connection**

| Flag | Default | Effect |
| --- | --- | --- |
| `--host` | `localhost` | rosbridge host |
| `--port` | `9090` | rosbridge port |
| `--secure` | off | use `wss://` instead of `ws://` |
| `--timeout` | `10.0` | per-call rosapi timeout, seconds |

**Selection**

| Flag | Effect |
| --- | --- |
| `--include RE` | keep only topics whose name matches the regex |
| `--exclude RE` | drop topics matching the regex (applied after `--include`) |
| `--no-params` | skip the parameter server entirely |

**Output**

| Flag | Effect |
| --- | --- |
| `-o`, `--output` | module path (default `ros_models.py`); parent dirs are created |
| `--no-defaults` | emit required fields instead of ROS zero-defaults |
| `--raw-uint8` | type `uint8[]` as `list[int]` rather than base64 |
| `--dump-json` | also write the raw graph snapshot to this path |
| `--from-json` | regenerate from a snapshot; no connection made |
| `-q`, `--quiet` | suppress progress output |

Exit codes: `0` success, `1` connection failure or missing `roslibpy`.

---

## Recipes

**Only the topics you care about.** A busy robot advertises a lot of noise;
regexes keep the generated module reviewable.

```bash
ros-pydantic-gen --include '^/(camera|odom|cmd_vel)' -o ros_models.py
ros-pydantic-gen --exclude '(_debug|/rosout)' -o ros_models.py
```

**Capture once, iterate offline.** The single most useful workflow when the
robot is in a lab and you are not.

```bash
ros-pydantic-gen --host 172.17.0.1 --port 9001 --dump-json graph.json -o ros_models.py
# later, on a train, robot switched off:
ros-pydantic-gen --from-json graph.json -o ros_models.py
```

The snapshot is plain JSON — commit it, diff it, and you have a record of
exactly what the robot's interface looked like on a given day.

**Strict mode for ingest pipelines.** By default fields carry ROS zero-defaults,
which makes construction easy but lets a truncated payload validate. When you'd
rather catch that:

```bash
ros-pydantic-gen --no-defaults -o ros_models_strict.py
```

**Detect interface drift in CI.**

```bash
ros-pydantic-gen --from-json graph.json -o /tmp/current.py
diff <(grep -v 'Generated :' ros_models.py) <(grep -v 'Generated :' /tmp/current.py)
```

The `Generated :` timestamp is the only nondeterministic line; strip it and the
output is byte-stable, so a non-empty diff means the interface really changed.

**Binary image topics.** If you've configured rosbridge with CBOR compression,
byte arrays arrive as real lists rather than base64:

```bash
ros-pydantic-gen --raw-uint8 -o ros_models.py
```

---

## How it handles ROS's awkward bits

- **Nested types** — `rosapi/MessageDetails` returns the full tree, so one call
  per top-level type pulls in every nested message. `geometry_msgs/Quaternion`
  gets a model even though no topic publishes one directly.
- **Base64 arrays** — rosbridge JSON-encodes `uint8[]` as base64, so
  `sensor_msgs/Image.data` is typed `Union[Base64Bytes, list[int]]`; you get
  real `bytes` in, base64 back out, round-tripping intact.
- **Reserved words** — `from`, `class`, `lambda`, `2nd_value`, `schema` become
  `from_`, `class_`, `lambda_`, `f_2nd_value`, `schema_`, each carrying
  `Field(alias=...)`. With `populate_by_name=True`, either name constructs.
- **Fixed-length arrays** — `float64[36]` becomes
  `Annotated[list[float], Field(min_length=36, max_length=36)]`, so a malformed
  covariance matrix fails validation instead of propagating. The default is 36
  zeros, matching ROS, so `NavMsgsOdometry()` still constructs.
- **Constants** — `rosgraph_msgs/Log.WARN` and friends land as `ClassVar`s, not
  fields.
- **ROS 1 and ROS 2 naming** — `geometry_msgs/msg/Pose` and `geometry_msgs/Pose`
  collapse to the same class name.
- **Dependency ordering** — topological sort; cycles degrade to forward
  references via `from __future__ import annotations`.
- **Partial failures** — a type the bridge can't describe is logged, listed as a
  comment block in the output, and does not abort the run.

---

## Troubleshooting

**`Could not connect to rosbridge at ws://...`**
The bridge isn't reachable. Confirm the node is up (`rosnode list | grep
rosbridge`), the port is published if containerised, and no firewall sits in
between. From WSL, `localhost` refers to WSL itself — use the container or host
IP.

**Topics appear in the comment block instead of as models.**

```python
# No model could be generated for these topics - the bridge could not
# describe their message type ...
#   /duckie/lane_pose: duckietown_msgs/LanePose
```

`message_details` resolves types against the *bridge's* Python environment, not
yours. If `duckietown_msgs` isn't on the rosbridge node's `PYTHONPATH`, it
cannot describe those types. Fix it on the robot — source the workspace that
defines the messages before launching the bridge — and regenerate.

**`AttributeError: 'str' object has no attribute 'get'`**
This was a real bug, now fixed and regression-tested. `roslibpy`'s
`ServiceResponse` subclasses `collections.UserDict`, **not** `dict` — so an
`isinstance(response, dict)` check silently falls through, and iterating the
response yields its *keys* as bare strings. If you extend the introspection
layer, test against `collections.abc.Mapping`. See
`tests/test_introspect.py::TestUnwrapTypedefs::test_userdict_response`.

**Timeouts on a large graph.** Each type costs a round trip. On a slow link,
raise `--timeout 30` and narrow the work with `--include`.

**`RosParams` values look stale.** They are — deliberately. `RosParams` is a
schema whose defaults happen to be the values that were live at generation time.
For current values, read the server at runtime with
`roslibpy.Param(client, name).get()`.

**Validation errors on live data.** Usually a genuine mismatch: the robot is
running a different build of the message than the bridge described. Extra keys
are ignored by design; *missing* keys are what surface here.

<a name="ros-2"></a>
**ROS 2.** roslibpy's ROS 2 support is still maturing. Message and parameter
introspection work; the notable difference is the header type
(`std_msgs/Header` loses its `seq` field), which the generator reflects
faithfully because it reads the definition rather than assuming one.

---

## Project layout

```
src/ros_pydantic_gen/
├── graph.py        # RosField / TypeDef / RosGraph - the domain model
├── introspect.py   # the only module that talks to roslibpy
├── naming.py       # ROS identifiers -> Python identifiers
├── rostypes.py     # ROS primitive -> Python type tables
├── config.py       # ConnectionConfig, GeneratorOptions
├── cli.py          # argument parsing and orchestration
├── codegen/
│   ├── messages.py # one pydantic model per message type
│   ├── params.py   # the RosParams model
│   └── module.py   # section assembly
└── templates/      # boilerplate of the emitted module, as real files
```

Dependency flow is one-directional: `cli → introspect → graph → codegen`.

`graph.RosGraph` is the seam. Code generation never sees a rosapi payload, and
introspection never sees a line of generated Python. That separation is what
makes `--from-json` and the offline test suite possible, and it means adding a
new output format only touches `codegen/`.

---

## Development

```bash
pip install -e ".[dev]"
pytest                    # 81 tests, ~1s, no ROS needed
ruff check .
```

The suite runs entirely offline against `tests/fixtures/fake_ros.py`, which
mimics rosapi's responses — including the `UserDict` wrapper, a typedef with
`fieldarraylen` missing, a type that raises, and one that returns nothing.

Generated modules are written to a temp directory, imported for real via
`importlib`, and exercised with realistic payloads. A syntax error or a bad
annotation therefore fails a test rather than surfacing on a robot.

`tests/test_packaging.py` builds an actual wheel and inspects its contents,
because `PYTHONPATH=src pytest` passing tells you nothing about whether the
*installed* distribution works. It needs `pip install build hatchling`; without
them the wheel test skips rather than failing.

If you touch `[tool.hatch.build]`, note that `packages = ["src/ros_pydantic_gen"]`
already ships every file under that directory, `templates/*.tmpl` included.
Adding a `force-include` for them duplicates each path and hatchling aborts the
build.

**Extending it.** A few natural next steps and where they'd go:

| Change | Where |
| --- | --- |
| Support services / actions | `introspect.py` (`get_service_*`), then `codegen/` |
| Emit dataclasses or TypedDicts instead | a new module under `codegen/` |
| Change a type mapping | `rostypes.py` |
| Adjust the generated boilerplate | `templates/*.py.tmpl` |

When adding a rosapi call, remember the `Mapping` rule above, and add a fixture
case to `fake_ros.py` so it stays testable without a robot.

---

## License

MIT.
