Skip to content

Events

RoSys provides an integrated event bus to chain otherwise separated parts of the system together. This allows loosely coupled actors where each one has its own dedicated role in a more complex workflow. For example the Lizard actor repeatedly reads the machine data like the current velocity, writes them into the world and finally emits the event NEW_MACHINE_DATA. The Odometer actor has registered itself for this event to compute the new position of the robot. Other actors may also register for the same event to monitor battery level or bump events.

Example

Implement a geofence actor pausing the automation when the robot position exceeds a certain threshold:

#!/usr/bin/env python3
import rosys.ui
from nicegui import ui
from rosys import Runtime, event
from rosys.actors import Actor
from rosys.automations import drive_path
from rosys.world import PathSegment, Pose, Spline


class GeoFenceGuard(Actor):

    def __init__(self) -> None:
        super().__init__()
        event.register(event.Id.NEW_MACHINE_DATA, self.check_position)

    def check_position(self) -> None:
        if abs(self.world.robot.prediction.x) > 3 or abs(self.world.robot.prediction.y) > 3:
            event.emit(event.Id.PAUSE_AUTOMATION, 'robot left the area')


runtime = Runtime().with_actors(GeoFenceGuard())
rosys.ui.configure(ui, runtime)
label = ui.label()
ui.timer(0.1, lambda: label.set_text(f'pose: {runtime.world.robot.prediction}'))
path = [PathSegment(spline=Spline.from_poses(Pose(), Pose(x=5, y=1)))]


async def automation():
    await drive_path(runtime.world, runtime.hardware, path)
rosys.ui.automation_controls(default_automation=automation)

ui.run(title='RoSys')

Geofence