Metadata-Version: 2.3
Name: robo-ws-bridge
Version: 0.9.0
Summary: Python library implementing the Foxglove WebSocket protocol for streaming robotics data
Keywords: websocket,foxglove,robotics,ros,ros2,streaming
Author: Marko Bausch
License: GPL-3.0
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Science/Research
Classifier: License :: OSI Approved :: GNU General Public License v3 (GPLv3)
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Internet
Classifier: Topic :: Scientific/Engineering
Classifier: Typing :: Typed
Requires-Dist: typing-extensions>=4.15.0
Requires-Dist: websockets>=15.0.1
Requires-Python: >=3.10
Project-URL: Homepage, https://github.com/mrkbac/robotic-tools
Project-URL: Issues, https://github.com/mrkbac/robotic-tools/issues
Project-URL: Repository, https://github.com/mrkbac/robotic-tools
Description-Content-Type: text/markdown

# robo-ws-bridge

A Python library implementing the Foxglove WebSocket protocol for streaming robotics data.

## Installation

```bash
uv add robo-ws-bridge
```

## Features

- **WebSocketBridgeServer**: Async server for publishing robotics data over WebSocket
- **WebSocketBridgeClient**: Async client for subscribing to data streams
- Native channel streaming, time synchronization, playback control, subscriptions, and status messages
- Extensible handlers for parameters, services, client publishing, connection graphs, and assets

## Usage

### Server Example

```python
import asyncio
from robo_ws_bridge.server import Channel
from robo_ws_bridge import WebSocketBridgeServer

async def main():
    server = WebSocketBridgeServer(host="0.0.0.0", port=8765, name="my-server")

    # Define and advertise a channel
    channel = Channel(
        id=1,
        topic="/sensor/data",
        encoding="json",
        schema_name="SensorData",
        schema='{"type": "object"}',
    )

    await server.start()
    await server.advertise_channel(channel)

    # Publish messages
    while True:
        data = b'{"temperature": 22.5}'
        await server.publish_message(channel.id, data)
        await asyncio.sleep(0.1)

asyncio.run(main())
```

### Playback control

Playback control uses our native async server implementation; it does not depend on the Foxglove SDK.
Supplying the recording time range advertises the `playbackControl` capability; the server accepts
the `foxglove.sdk.v1` subprotocol by default.

```python
from robo_ws_bridge import (
    PlaybackCommand,
    PlaybackControlRequest,
    PlaybackState,
    PlaybackStatus,
    WebSocketBridgeServer,
)

server = WebSocketBridgeServer(playback_time_range=(start_time_ns, end_time_ns))

async def control(request: PlaybackControlRequest) -> PlaybackState:
    if request.seek_time is not None:
        await recording.seek(request.seek_time)
    await recording.set_speed(request.playback_speed)
    if request.playback_command is PlaybackCommand.PLAY:
        await recording.play()
    else:
        await recording.pause()
    return PlaybackState(
        status=PlaybackStatus.PLAYING if recording.is_playing else PlaybackStatus.PAUSED,
        current_time=recording.current_time,
        playback_speed=recording.speed,
        did_seek=request.seek_time is not None,
    )

server.on_playback_control(control)
```

Each server endpoint has a unique session ID so reconnecting Foxglove clients
discard stale visualization state. Rotate it explicitly when switching runs:

```python
await server.clear_session()
```

Status IDs provide the same small lifecycle for the Problems panel:

```python
from robo_ws_bridge import StatusLevel

await server.send_status(StatusLevel.ERROR, "Playback failed", status_id="playback")
await server.remove_status(["playback"])
```

### Client Example

```python
import asyncio
from robo_ws_bridge import WebSocketBridgeClient

async def main():
    client = WebSocketBridgeClient("ws://localhost:8765")

    async def handle_message(channel, timestamp, data):
        print(f"{channel['topic']}: {data}")

    async def handle_advertise(channel):
        await client.subscribe(channel["topic"])

    client.on_message(handle_message)
    client.on_advertised_channel(handle_advertise)

    await client.connect()

    # Keep running until interrupted
    try:
        while True:
            await asyncio.sleep(1)
    finally:
        await client.disconnect()

asyncio.run(main())
```

## Protocol Reference

See the [Foxglove WebSocket Protocol](https://docs.foxglove.dev/docs/connecting-to-data/frameworks/ws-protocol/) for protocol details.
