Facilities in cube:its

cube:its provides a ROS-native implementation of ETSI C-ITS facilities on top of Vanetza. ROS 2 applications can send and receive CAM, DENM, CPM, VAM or low-level BTP packets by publishing and subscribing to topics. ASN.1, BTP, GeoNetworking, and the low-level radio APIs are handled internally.

Each facility is exposed through a set of ROS 2 topics and services, with message definitions from:

vanetza_msgs
etsi_its_cam_msgs, etsi_its_denm_msgs, etsi_its_cpm_ts_msgs, etsi_its_vam_ts_msgs
cube_den_msgs (for the DEN service)

The following sections show simple, minimal examples you can use as templates.

Running the Example Nodes

All code snippets in this chapter are a modifed version taken from the v2x_apps ROS 2 package in our example repository. To run any of these examples on your development machine, you need to:

Build and source the workspace (in your dev container or host):

1
2
3
host:~$ cd ~/cube-its/dev_ws
host:~/cube-its/dev_ws$ colcon build
host:~/cube-its/dev_ws$ source install/setup.bash

Make sure cube:its is running on the cube (and reachable via DDS), e.g.:

1
cube:~$ start-its --remote

Start an example node using ros2 run:

1
host:~/cube-its/dev_ws$ ros2 run v2x_apps <python_node>

Replace <python_node> with the actual node name from the examples, for instance:

1
2
3
host:~/cube-its/dev_ws$ ros2 run v2x_apps btp_listener
host:~/cube-its/dev_ws$ ros2 run v2x_apps btp_sender
host:~/cube-its/dev_ws$ ros2 run v2x_apps ...

Once the node is running, you should see log output in the terminal (e.g. received CAMs, DENMs, CPMs, etc.), and the corresponding /its/... or /vanetza/... topics will appear in ros2 topic list.

BTP – Basic Transport Protocol

BTP is the transport layer beneath CAM, DENM, CPM, VAM and the other ETSI ITS facilities. cube:its exposes BTP directly so that you can implement custom ETSI facilities or proprietary message types.

Listen for BTP Packets

Topic: /vanetza/btp_indication

Message: vanetza_msgs/msg/BtpDataIndication

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import rclpy
from rclpy.node import Node
from vanetza_msgs.msg import BtpDataIndication

class BtpListener(Node):
    def __init__(self):
        super().__init__("btp_listener")
        self.subscription = self.create_subscription(
            BtpDataIndication,
            "/vanetza/btp_indication",
            self.listener_callback,
            10,
        )

    def listener_callback(self, msg):
        self.get_logger().info(
            f"Received {len(msg.data)} bytes on BTP port {msg.destination_port}"
        )

def main(args=None):
    rclpy.init(args=args)
    rclpy.spin(BtpListener())
    rclpy.shutdown()

Send a BTP Packet

Service: /vanetza/btp_request

Service Type: vanetza_msgs/srv/BtpData

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import rclpy
from rclpy.node import Node
from vanetza_msgs.msg import TrafficClass
from vanetza_msgs.srv import BtpData

class BtpSender(Node):
    def __init__(self):
        super().__init__("btp_sender")
        self.client = self.create_client(BtpData, "/vanetza/btp_request")
        self.create_timer(1.0, self.trigger)

    def trigger(self):
        req = BtpData.Request()
        req.btp_type = BtpData.Request.BTP_TYPE_NON_INTERACTIVE
        req.transport_type = BtpData.Request.TRANSPORT_TYPE_SHB
        req.destination_port = 42
        req.traffic_class.id = TrafficClass.TC_OTHER
        req.data = b"Your payload"

        future = self.client.call_async(req)
        future.add_done_callback(self.done)

    def done(self, future):
        result = future.result()
        status = "accepted" if result.confirm == BtpData.Response.CONFIRM_ACCEPTED else "failed"
        self.get_logger().info(f"BTP request {status}")

def main(args=None):
    rclpy.init(args=args)
    rclpy.spin(BtpSender())
    rclpy.shutdown()

CAM – Cooperative Awareness Messages

CAMs are continuous situational awareness broadcasts that describe a station's dynamic state (position, speed, heading, vehicle type, and so on). cube:its handles the CAM pipeline:

Generates CAMs from GNSS/IMU and motion data
Encodes CAMs using ASN.1 UPER

CAM Dissemination

cube:its includes a Cooperative Awareness (CA) service with a two-stage dissemination process for CAMs.

First, the CAM provider is a node that prepares CAMs at a fixed rate, including optional containers such as the LowFrequencyContainer. These CAMs are published on the /its/cam_provided topic.

Second, the CA transmission node evaluates the provided CAMs and applies filtering, which affects the effective transmission rate and the presence of optional message data. For example, the default vehicle rules adapt the rate between 1 and 10 Hz, and the LowFrequencyContainer is included only every 500 ms. Transmitted CAMs are published on the /its/cam_transmitted topic.

CAM Dissemination Rules (Plugins)

The default dissemination rules in the cube_ca_plugins package follow the ETSI standard.

VehicleDisseminationRules is the default profile, except for RSUs. It adapts the transmission rate between 0.1 and 1.0 s based on the station's recent changes in heading, speed, and position. It includes the LowFrequencyContainer and SpecialVehicleContainer only once every 0.5 s.

If the station type is set to "road side unit", RsuDisseminationRules applies. It limits the transmission rate to 1 Hz.

Custom CAM triggering logic can be implemented as a plugin, while still using the Vanetza and cube:its stack.

Typical use cases:

Researching new CAM adaptation strategies
Reducing channel load in dense traffic
Domain-specific trigger rules for VRUs, robots, or special vehicles

The cube_ca_plugins package contains the ETSI-conformant CAM rate adaptation plugins, plus sample code for a custom dissemination rules plugin:

https://github.com/cubesys-GmbH/ros_cube_ca_plugins

The reference implementation can be used as a template for custom logic.

Receive CAMs

Topic: /its/cam_received

Message: etsi_its_cam_msgs/msg/CAM

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import rclpy
from rclpy.node import Node
from etsi_its_cam_msgs.msg import CAM

class CamListener(Node):
    def __init__(self):
        super().__init__("cam_listener")
        self.subscription = self.create_subscription(
            CAM, "/its/cam_received", self.callback, 10
        )

    def callback(self, msg):
        self.get_logger().info(f"Received CAM from StationId={msg.header.station_id.value}")

def main(args=None):
    rclpy.init(args=args)
    rclpy.spin(CamListener())
    rclpy.shutdown()

CPM – Collective Perception Messages

CPM shares detected objects between cooperative participants (vehicles, RSUs, robots). cube:its handles ASN.1 encoding, headers, and GeoNetworking.

You only need to publish CPM objects on ROS topics.

Provide CPMs

Publisher: /its/cpm_provided

Message: etsi_its_cpm_ts_msgs/msg/CollectivePerceptionMessage

Helper: /its/position_vector for ego GNSS

Example: Publish one perceived object every second.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
import rclpy
import etsi_its_cpm_ts_msgs.msg as cpm_msg
from rclpy.node import Node
from vanetza_msgs.msg import PositionVector

class CpmProvider(Node):
    def __init__(self):
        super().__init__("cpm_provider")
        self.position_vector = None
        self.create_subscription(PositionVector, "/its/position_vector", self.position_update, 1)

        self.pub = self.create_publisher(cpm_msg.CollectivePerceptionMessage, "/its/cpm_provided", 1)
        self.create_timer(1.0, self.publish)

    def position_update(self, msg):
        self.position_vector = msg

    def generate_cpm(self):
        # minimal example with one object
        cpm = cpm_msg.CollectivePerceptionMessage()
        cpm.payload.management_container.reference_position.latitude.value = int(self.position_vector.latitude * 1e7)
        cpm.payload.management_container.reference_position.longitude.value = int(self.position_vector.longitude * 1e7)
        return cpm

    def publish(self):
        if self.position_vector:
            self.pub.publish(self.generate_cpm())
            self.get_logger().info("CPM provided")

def main(args=None):
    rclpy.init(args=args)
    rclpy.spin(CpmProvider())
    rclpy.shutdown()

DENM – Decentralized Environmental Notification Messages

DENMs are event-driven hazard and warning messages (accidents, road works, obstacles). cube:its provides a DEN service (/its/den_request) that:

fills ITS PDU headers automatically
assigns Action IDs
handles updates & terminations
performs all ASN.1 encoding

You only provide the DENM payload.

Send & Receive DENMs

Receive Topic: /its/denm_received

Transmit Service: /its/den_request

Service Type: cube_den_msgs/srv/Transmission

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
import rclpy
import etsi_its_denm_msgs.msg as denm_msg
from cube_den_msgs.srv import Transmission
from vanetza_msgs.msg import PositionVector, GeoNetArea
from rclpy.node import Node

class DenmNode(Node):
    def __init__(self):
        super().__init__("denm_node")
        self.position_vector = None
        self.create_subscription(PositionVector, "/its/position_vector", self.position_update, 1)
        self.create_subscription(denm_msg.DENM, "/its/denm_received", self.receive_callback, 1)
        self.client = self.create_client(Transmission, "/its/den_request")
        self.create_timer(1.0, self.transmit)

    def receive_callback(self, msg):
        self.get_logger().info(f"Received DENM from StationId={msg.header.station_id.value}")

    def position_update(self, msg):
        self.position_vector = msg

    def transmit(self):
        if not self.position_vector:
            return

        req = Transmission.Request()
        req.request = Transmission.Request.REQUEST_TRIGGER
        req.message = denm_msg.DENM()
        req.destination_area.type = GeoNetArea.TYPE_CIRCLE
        req.destination_area.latitude = self.position_vector.latitude
        req.destination_area.longitude = self.position_vector.longitude
        req.destination_area.distance_a = 500.0

        future = self.client.call_async(req)
        future.add_done_callback(lambda f: self.get_logger().info("DENM sent"))

def main(args=None):
    rclpy.init(args=args)
    rclpy.spin(DenmNode())
    rclpy.shutdown()

VAM – Vulnerable Road User Awareness

VAMs are messages for pedestrians, cyclists, e-scooters, and other VRUs. cube:its exposes a topic on which VAMs can be provided, and transmits them according to ETSI rules.

Provide VAMs

Publisher: /its/vam_provided

Message: etsi_its_vam_ts_msgs/msg/VAM

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import rclpy
import etsi_its_vam_ts_msgs.msg as vam_msg
from rclpy.node import Node
from vanetza_msgs.msg import PositionVector

class VamProvider(Node):
    def __init__(self):
        super().__init__("vam_provider")
        self.pos = None
        self.create_subscription(PositionVector, "/its/position_vector", self.update_pos, 1)

        self.pub = self.create_publisher(vam_msg.VAM, "/its/vam_provided", 1)
        self.create_timer(1.0, self.publish)

    def update_pos(self, msg):
        self.pos = msg

    def publish(self):
        if not self.pos:
            return
        vam = vam_msg.VAM()
        vam.vam.vam_parameters.basic_container.station_type.value = vam_msg.TrafficParticipantType.PEDESTRIAN
        vam.vam.vam_parameters.basic_container.reference_position.latitude.value = int(self.pos.latitude * 1e7)
        vam.vam.vam_parameters.basic_container.reference_position.longitude.value = int(self.pos.longitude * 1e7)
        self.pub.publish(vam)
        self.get_logger().info("VAM provided")

def main(args=None):
    rclpy.init(args=args)
    rclpy.spin(VamProvider())
    rclpy.shutdown()