COMPAS EVE

Event-based communication for the COMPAS framework.

>>> import compas_eve as eve
>>> pub = eve.Publisher("/hello_world")
>>> sub = eve.EchoSubscriber("/hello_world")
>>> sub.subscribe()
>>> for i in range(10):
...    pub.publish(dict(text=f"Hello World {i}"))

It is extremely easy to send messages around. COMPAS EVE supports different transport mechanisms to send messages between different threads, processes, computers, etc.

Installation

Install using pip:

    pip install compas_eve

Or using conda:

    conda install compas_eve

Supported features

• Publisher/subscriber communication model (N-to-N communication)
• In-process events
• MQTT support
• ROS support through rosbridge
• Zenoh support
• Extensible codec system for message serialization (JSON, Protocol Buffers)

Examples

In-process events

The simplest option is to use in-process events. This works for simple applications and allows to communicate between threads.

import compas_eve as eve

pub = eve.Publisher("/hello_world")
sub = eve.EchoSubscriber("/hello_world")
sub.subscribe()

for i in range(10):
    pub.publish(dict(text=f"Hello World {i}"))

MQTT

MQTT is a protocol that allows to send messages between different systems/computers. Using MQTT is very simple as well:

import compas_eve as eve
from compas_eve.mqtt import MqttTransport

tx = MqttTransport("broker.hivemq.com")
eve.set_default_transport(tx)

pub = eve.Publisher("/hello_world")
sub = eve.EchoSubscriber("/hello_world")
sub.subscribe()

for i in range(10):
    pub.publish(dict(text=f"Hello World {i}"))

This example shows how to send and receive from a single script, but running publishers and subscribers on different scripts, different processes, or even different computers will work the exact same way.

ROS

The optional ROS backend uses native ROS message types through a rosbridge server:

pip install compas_eve[ros]

import time

import compas_eve as eve
from compas_eve.ros import RosTransport

tx = RosTransport("localhost", 9090)
topic = eve.Topic("/chatter", "std_msgs/String", queue_size=10)

sub = eve.EchoSubscriber(topic, transport=tx)
sub.subscribe()
eve.Publisher(topic, transport=tx).publish({"data": "Hello ROS"})
time.sleep(1)
tx.close()

ROS messages are passed as JSON-compatible dictionaries. Topic options supported by roslibpy—including compression, latch, throttle_rate, queue_size, queue_length, and reconnect_on_close—can be set on eve.Topic.

Zenoh

Apache Zenoh is a pub/sub/query protocol. In many ways, it is similar to MQTT but with some additional features and optimizations. COMPAS EVE also supports Zenoh as a transport protocol with an identical API to MQTT:

import compas_eve as eve
from compas_eve.zenoh import ZenohTransport

tx = ZenohTransport()
eve.set_default_transport(tx)

pub = eve.Publisher("/hello_world")
sub = eve.EchoSubscriber("/hello_world")
sub.subscribe()

for i in range(10):
    pub.publish(dict(text=f"Hello World {i}"))

Using different codecs

By default, COMPAS EVE uses JSON for message serialization. However, you can use different codecs for more efficient serialization:

import compas_eve as eve
from compas_eve import JsonMessageCodec
from compas_eve.codecs import ProtobufMessageCodec
from compas_eve.mqtt import MqttTransport

# Use JSON codec (default)
json_codec = JsonMessageCodec()
tx = MqttTransport("broker.hivemq.com", codec=json_codec)

# Or use Protocol Buffers for binary serialization (requires compas_pb)
pb_codec = ProtobufMessageCodec()
tx = MqttTransport("broker.hivemq.com", codec=pb_codec)

Usage from Rhinoceros 3D

It is possible to use the same code from within Rhino/Grasshopper.

To install compas_eve, use the the syntax # r: compas_eve at the top of any Python 3.x script in Rhino/Grasshopper.