# NATS with NOVA OS > Learn about NATS messaging system integration with Wandelbots NOVA OS for real-time data communication and event-driven architectures in your robot cell. [NATS](https://nats.io/) is a high-performance messaging system integrated into Wandelbots NOVA OS. It enables real-time communication between different components in your robot cell. As a lightweight and fast message broker, NATS handles data exchange, event notifications, and service coordination across your robotics infrastructure. For more details about NATS capabilities, check out the [official NATS documentation](https://docs.nats.io/). ### Key features - **High Performance**: Handles millions of messages per second with minimal latency - **Flexible Messaging**: Supports pub-sub, request-reply, and queuing patterns - **JetStream Persistence**: Store and replay events with NATS JetStream for reliable delivery - **CloudEvents Compliant**: Uses the industry-standard CloudEvents v1.0 specification for event structure - **Easy Integration**: Client libraries available for multiple programming languages ## Why NATS with Wandelbots NOVA? Wandelbots NOVA uses NATS to provide: - **Real-time robot state streaming**: Monitor robot controller states, motion group positions, joint values, and I/O signals as they change - **System and cell status updates**: Get instant notifications about service status, system updates, and cell configurations - **Event-driven architectures**: Build applications that react to system events like update starts and completions - **Collision detection data**: Access collision setups and safety zone information for path planning - **I/O communication**: Read and write input/output values for both BUS I/O services and robot controllers ## Key concepts > **Info:** The API documentation can be found [here](https://portal.wandelbots.io/docs/api/v2/async). ### Subjects Messages in NATS are published to and subscribed from **subjects**, which are logical channels that organize communication. NOVA uses a hierarchical naming convention with dot-separated tokens. Subject patterns look like this: ``` {instance}.v2.cells.{cell} {instance}.v2.cells.{cell}.controllers.{controller}.state {instance}.v2.cells.{cell}.controllers.{controller}.ios {instance}.v2.cells.{cell}.bus-ios.ios {instance}.v2.events.system.update.started {instance}.v2.system.status ``` Typically, `{instance}` would be `nova` and `{cell}`, `{controller}` would be your configured identifiers. ### Publish-Subscribe This is the fundamental messaging pattern: - **Publishers** send messages to subjects without knowing who receives them. For example, NOVA publishes robot state updates. - **Subscribers** listen to subjects they care about and receive all published messages. Your application can subscribe to state updates. - **Wildcards** let you subscribe to multiple subjects at once. For instance, `nova.v2.cells.*.status` subscribes to all cell statuses. ### Request-Reply This pattern handles synchronous communication: - A client sends a request message with a `replyTo` header - A service processes the request and responds to the reply address - The client receives and processes the reply - NOVA uses this for operations like setting I/O values or selecting which I/Os to stream ### JetStream persistence NOVA uses NATS JetStream for persistent storage: - **system-state stream**: Keeps the latest message per subject for cell configurations, controller states, and service status - **system-events stream**: Stores up to 10,000 event messages (max 100MiB) for 30 days ## Available data streams NATS with NOVA OS gives you access to these real-time data streams: ### Cell & System information - **Cell configurations**: Robot controllers, applications, payloads, and cell descriptions - **System status**: Status of all system services and cell resources - **Service health**: Service severity levels (INFO, WARNING, ERROR) and current phases - **Core applications**: Information about running applications in the cell ### Robot controller data - **Controller state**: System mode, operation mode, safety state, and velocity override - **Motion group state**: Joint positions, velocities, torques, currents, flange pose, TCP pose - **Motion execution details**: Jogging or trajectory execution status with time-to-end and current location - **Motion group descriptions**: TCPs, mounting, safety zones, limits, DH parameters - **I/O values**: Digital (boolean), analog (float), and integer I/O signals ### BUS I/O service - **I/O status**: Connection state and service messages - **I/O values**: Input and output values from BUS communication protocols - **Set outputs**: Write output values (also inputs for virtual services) ### Collision world - **Collision setups**: Colliders, link chains, tool shapes, and self-collision detection settings ### Program data - **Program details**: ID, execution ID, execution state, timestamps - **Program events**: Start, stop, pause, resume - **Error reporting**: Error codes - **Program logs**: Log messages, timestamps - **Cycle details**: Cycle times, error rates, cycle counts, task times, velocities ### System events - **Update lifecycle**: System update started and completed events with version information - **CloudEvents format**: Follows CloudEvents v1.0 specifications for event structure ## Set up NATS with Python This guide shows you how to connect to NOVA's NATS server using the [nats.py](https://github.com/nats-io/nats.py) Python SDK and subscribe to real-time data streams from your robot cell. ### Prerequisites - Python 3.8 or higher - Access to a NOVA instance with NATS enabled - Basic knowledge of Python async/await > **Warning:** Ensure your network configuration allows NATS communication. > NATS typically uses port 4222 for client connections and port 8222 for HTTP monitoring. ### Installation #### Step 1: Install NATS Python client ```bash pip install nats-py ``` For WebSocket support (useful for web applications or restricted networks): ```bash pip install nats-py[aiohttp] ``` #### Step 2: Verify your NATS connection details NOVA's NATS server is accessible via WebSocket: - **WebSocket URL**: `ws://{controller_ip}:80/api/nats` (or `wss://` for TLS) - Replace `{controller_ip}` with your NOVA instance IP address - **Authentication**: Bearer token authentication (if required by your NOVA instance) > **Info:** Check with your system administrator for the exact NATS connection details and authentication requirements for your NOVA installation. ### Basic connection Here's how to connect to NOVA's NATS server via WebSocket: **Without Authentication** ```python import asyncio import nats async def main(): # Connect to NOVA's NATS server via WebSocket # Replace 'your-nova-ip' with your actual NOVA instance IP nc = await nats.connect("ws://your-nova-ip:80/api/nats") print(f"Connected to NATS at {nc.connected_url.netloc}") # Close connection when done await nc.close() if __name__ == '__main__': asyncio.run(main()) ``` **With Token Authentication** ```python import asyncio import nats async def main(): nats_url = "ws://your-nova-ip:80/api/nats" # Get your authentication token # This could come from environment variables, a config file, or API token = "your-bearer-token" # Connect with token authentication connect_kwargs = {"servers": [nats_url]} if token: connect_kwargs["token"] = token nc = await nats.connect(**connect_kwargs) print(f"Connected to NATS at {nc.connected_url.netloc}") # Close connection when done await nc.close() if __name__ == '__main__': asyncio.run(main()) ``` > **Info:** If your NOVA instance requires authentication, use the token-based approach. The token is typically a Bearer token obtained from your authentication system. #### Environment variable setup Set your token as an environment variable: ```bash # Linux/macOS export NOVA_NATS_TOKEN="your-bearer-token" # Windows PowerShell $env:NOVA_NATS_TOKEN="your-bearer-token" # Windows CMD set NOVA_NATS_TOKEN=your-bearer-token ``` ### Authentication NOVA supports token-based authentication for NATS connections. Here's a practical example of how to manage authentication: ```python import asyncio import nats import os async def connect_to_nats(nats_url: str, token: str = None): """Connect to NATS with optional token authentication.""" try: print(f"Connecting to NATS broker: {nats_url}") # Prepare connection parameters connect_kwargs = {"servers": [nats_url]} # Use Bearer token authentication if available if token: connect_kwargs["token"] = token print("Using token authentication") # Connect to NATS nc = await nats.connect(**connect_kwargs) print("Successfully connected to NATS broker") return nc except Exception as e: print(f"Failed to connect to NATS: {e}") return None async def main(): nats_url = "ws://your-nova-ip:80/api/nats" token = os.getenv("NOVA_NATS_TOKEN") nc = await connect_to_nats(nats_url, token) if nc: # Use the connection print(f"Connected to {nc.connected_url.netloc}") # Don't forget to close when done await nc.close() if __name__ == '__main__': asyncio.run(main()) ``` Then run your Python script: ```bash python your_script.py ``` > **Warning:** Never hardcode authentication tokens in your source code. Always use environment variables, configuration files, or secure secret management solutions. ### Subscribing to cobot controller state Stream a real-time robot controller state including joint positions, velocities, and operation mode. > **Info:** The `controller` in the subject is the **robot controller name** you created in the NOVA Setup app, not a literal string. Replace it with your actual robot name (e.g., `ur5`, `abb-robot`, etc.). ```python import asyncio import nats import json async def main(): nc = await nats.connect("ws://your-nova-ip:80/api/nats") async def state_handler(msg): # Parse the robot controller state state = json.loads(msg.data) print(f"Controller: {state['controller']}") print(f"Mode: {state['mode']}") print(f"Operation Mode: {state['operation_mode']}") # Access motion group data for mg in state['motion_groups']: print(f"Motion Group: {mg['motion_group']}") print(f"Joint Position: {mg['joint_position']}") print(f"Standstill: {mg['standstill']}") # IMPORTANT: Replace these with your actual identifiers from NOVA Setup cell_name = "cell" # Your cell name (usually "cell") controller_name = "ur5-robot" # Your robot controller name from Setup app # Build the subject dynamically subject = f"nova.v2.cells.{cell_name}.controllers.{controller_name}.state" await nc.subscribe(subject, cb=state_handler) print(f"Listening for robot controller state updates on: {subject}") # Keep the connection alive try: await asyncio.Event().wait() except KeyboardInterrupt: pass finally: await nc.close() if __name__ == '__main__': asyncio.run(main()) ``` ### Streaming I/O values Monitor input and output signals from your robot controller or BUS I/O service: ```python import asyncio import nats import json async def main(): nc = await nats.connect("ws://your-nova-ip:80/api/nats") async def io_handler(msg): data = json.loads(msg.data) print(f"Timestamp: {data['timestamp']}") print(f"Sequence: {data['sequence_number']}") # Process I/O values for io_value in data['io_values']: io_id = io_value['io'] value = io_value['value'] value_type = io_value['value_type'] print(f" {io_id} ({value_type}): {value}") # Replace with your actual identifiers from NOVA Setup cell_name = "cell" controller_name = "ur5-robot" # Your robot controller name # Subscribe to controller I/O updates subject = f"nova.v2.cells.{cell_name}.controllers.{controller_name}.ios" await nc.subscribe(subject, cb=io_handler) # Or subscribe to BUS I/O updates # bus_io_subject = f"nova.v2.cells.{cell_name}.bus-ios.ios" # await nc.subscribe(bus_io_subject, cb=io_handler) print("Listening for I/O updates...") try: await asyncio.Event().wait() except KeyboardInterrupt: pass finally: await nc.close() if __name__ == '__main__': asyncio.run(main()) ``` ### Setting I/O output values Write to output signals using a request-reply pattern: ```python import asyncio import nats import json async def main(): nc = await nats.connect("ws://your-nova-ip:80/api/nats") # Replace with your actual identifiers cell_name = "cell" controller_name = "ur5-robot" # Your robot controller name # Prepare I/O values to set io_values = [ { "io": "output1", "value": True, "value_type": "boolean" }, { "io": "output2", "value": "42", "value_type": "integer" } ] # Build the subject for setting I/O subject = f"nova.v2.cells.{cell_name}.controllers.{controller_name}.ios.set" # Set I/O values (with reply) try: response = await nc.request( subject, json.dumps(io_values).encode(), timeout=2.0 ) # Check for errors in response if response.data: result = json.loads(response.data) if 'message' in result: print(f"Error: {result['message']}") else: print("I/O values set successfully") except asyncio.TimeoutError: print("Request timed out") await nc.close() if __name__ == '__main__': asyncio.run(main()) ``` ### Subscribing with wildcards Use wildcards to subscribe to multiple subjects at once: ```python import asyncio import nats import json async def main(): nc = await nats.connect("ws://your-nova-ip:80/api/nats") async def multi_handler(msg): print(f"Received on {msg.subject}") data = json.loads(msg.data) # Process the message print(json.dumps(data, indent=2)) # Subscribe to all cell statuses (wildcard for any cell) await nc.subscribe("nova.v2.cells.*.status", cb=multi_handler) # Subscribe to all controller states in a specific cell # The * wildcard matches any controller name cell_name = "cell" await nc.subscribe( f"nova.v2.cells.{cell_name}.controllers.*.state", cb=multi_handler ) # Subscribe to all system events await nc.subscribe("nova.v2.events.>", cb=multi_handler) print("Listening for multiple subjects...") try: await asyncio.Event().wait() except KeyboardInterrupt: pass finally: await nc.close() if __name__ == '__main__': asyncio.run(main()) ``` ### Using JetStream for persistence Access persisting state and events using JetStream: ```python import asyncio import nats async def main(): nc = await nats.connect("ws://your-nova-ip:80/api/nats") # Create JetStream context js = nc.jetstream() # Pull the latest system state try: # Create a pull consumer psub = await js.pull_subscribe( "nova.v2.system.status", stream="system-state" ) # Fetch the latest message msgs = await psub.fetch(1, timeout=2.0) for msg in msgs: print("Latest system status:") print(msg.data.decode()) await msg.ack() except Exception as e: print(f"Error accessing JetStream: {e}") await nc.close() if __name__ == '__main__': asyncio.run(main()) ``` ### Monitoring system events Track system update lifecycle events: ```python import asyncio import nats import json async def main(): nc = await nats.connect("ws://your-nova-ip:80/api/nats") async def event_handler(msg): event = json.loads(msg.data) # CloudEvents v1.0 structure print(f"Event Type: {event['type']}") print(f"Event ID: {event['id']}") print(f"Source: {event['source']}") print(f"Time: {event['time']}") # Event-specific data if 'data' in event: data = event['data'] if event['type'] == 'nova.v2.events.system.update.started': print(f"Update started: {data['oldVersion']} -> {data['newVersion']}") elif event['type'] == 'nova.v2.events.system.update.completed': if 'error' in data: print(f"Update failed: {data['error']}") else: print(f"Update completed: {data['newVersion']}") # Subscribe to update events await nc.subscribe( "nova.v2.events.system.update.started", cb=event_handler ) await nc.subscribe( "nova.v2.events.system.update.completed", cb=event_handler ) print("Monitoring system events...") try: await asyncio.Event().wait() except KeyboardInterrupt: pass finally: await nc.close() if __name__ == '__main__': asyncio.run(main()) ``` ### Advanced: Connection with error handling It provides production-ready connection with reconnection logic, error handling, and authentication: ```python import asyncio import nats from nats.errors import ConnectionClosedError, TimeoutError import os async def main(): async def disconnected_cb(): print("Got disconnected from NATS!") async def reconnected_cb(): print("Reconnected to NATS") async def error_cb(e): print(f"NATS error: {e}") async def closed_cb(): print("Connection to NATS is closed") # Prepare connection parameters nats_url = "ws://your-nova-ip:80/api/nats" token = os.getenv("NATS_TOKEN") # Get token from environment variable connect_kwargs = { "servers": [nats_url], "reconnected_cb": reconnected_cb, "disconnected_cb": disconnected_cb, "error_cb": error_cb, "closed_cb": closed_cb, "max_reconnect_attempts": 10, "reconnect_time_wait": 2, } # Add token if available if token: connect_kwargs["token"] = token print("Connecting with token authentication...") # Connect with all settings nc = await nats.connect(**connect_kwargs) print(f"Connected to {nc.connected_url.netloc}") async def message_handler(msg): try: # Process message print(f"Received on {msg.subject}: {msg.data.decode()}") except Exception as e: print(f"Error processing message: {e}") # Subscribe with error handling cell_name = "cell" controller_name = "ur5-robot" # Your robot controller name from Setup subject = f"nova.v2.cells.{cell_name}.controllers.{controller_name}.state" try: await nc.subscribe(subject, cb=message_handler) # Keep connection alive await asyncio.Event().wait() except KeyboardInterrupt: print("Shutting down...") except ConnectionClosedError: print("Connection closed unexpectedly") finally: if nc.is_connected: await nc.close() if __name__ == '__main__': asyncio.run(main()) ``` #### Variable meanings - **`{cell}`**: Your cell identifier (typically `"cell"` unless you have multiple cells) - **`{controller}`**: The **robot controller name** you created in the NOVA Setup app (e.g., `"ur5-robot"`, `"abb-irb1600"`, `"fanuc-m10id12"`) - **`{motion-group}`**: The motion group name within the controller (e.g., `"manipulator"`, `"robot"`) > **Warning:** **Important**: The `{controller}` is the actual name you assigned to your robot in the NOVA Setup app. Find this name in Setup under your cell's robot controllers list. #### How to find identifiers 1. Open the Setup app. 2. Open the `Cells` tab. → The cell overview is displayed. 3. Look at the **robot controllers** section. 4. The name shown there is your `{controller}` value to use in subjects For example, if your Setup shows a robot named `fanuc-m10id12`, use: ```python subject = "nova.v2.cells.cell.controllers.fanuc-m10id12.state" ``` ### Next steps - Explore the [NATS Python SDK documentation](https://github.com/nats-io/nats.py) for advanced features - Check the [AsyncAPI specification](https://portal.wandelbots.io/docs/api/v2/async) for complete message schemas - Review the [NOVA API documentation](https://portal.wandelbots.io/docs/api/v2/ui) for REST API integration > **Info:** Need help? The NATS community provides extensive examples in the [nats.py repository](https://github.com/nats-io/nats.py/tree/main/nats/examples).