API reference

Robothub framework

2.6.0
package

robothub

module
module
module
module
module
module
package
package
robothub_core_wrapper
Low-level python API to interact with RobotHub Agent.
module
module
variable
module

robothub.application

variable
constant
constant
constant
class
class
class
BaseSDKApplication
This class acts as the main entry point for the SDK user, managing a single device, creating pipelines, and polling the device for new data. Derived classes must implement the `setup_pipeline` method.
class

robothub.application.BaseApplication(robothub_core.RobotHubApplication, abc.ABC)

method
variable
property
method
method
method
on_stop(self)
Called when the application is stopped.
method
get_device(self) -> Optional[Union[OakCamera, depthai.Device]]: Optional[Union[OakCamera, depthai.Device]]
Get a device by its mxid. If the device is not running, this method returns None.

:return: The device or None if the device is not running.
method
restart_device(self)
Restart the device.
class

robothub.application.BaseDepthAIApplication(robothub.application.BaseApplication)

class

robothub.application.BaseSDKApplication(robothub.application.BaseApplication)

method
setup_pipeline(self, oak: OakCamera)
The entry point for the application. This method is called when a device is connected and ready to be used.

:param oak: The device that is ready to be used.
method
on_device_connected(self, oak: OakCamera)
Called when a camera is connected.

:param oak: The camera that was connected.
method
on_device_disconnected(self)
Called when a camera is disconnected. Opposite of on_device_connected.
module

robothub.events

variable
function
send_image_event(image: Union [ np.ndarray , bytes ], title: str, device_id: str = None, metadata: Optional [ dict ] = None, tags: List [ str ] = None, mjpeg_quality = 98, encode = False) -> Optional[str]: Optional[str]
Send a single image frame event to RH.

:param image: The image to send.
:param title: The title of the event.
:param device_id: The device ID to associate with the event.
:param metadata: A dictionary of metadata to associate with the event.
:param tags: A list of tags to associate with the event.
:param mjpeg_quality: The JPEG quality to use when encoding.
:param encode: Whether to encode the image as a JPEG before sending.
:return: The event ID. None if the event failed to send.
function
send_frame_event_with_zipped_images(cv_frame: np.ndarray, files: list, title: str, device_id: str, tags: List [ str ] = None, metadata: Optional [ dict ] = None, encode: bool = False, mjpeg_quality = 98) -> Optional[str]: Optional[str]
Send a collection of images as a single event to RH.

:param cv_frame: The main image frame to send.
:param files: A list of images to zip and send.
:param title: The title of the event.
:param device_id: The device ID to associate with the event.
:param tags: A list of tags to associate with the event.
:param metadata: A dictionary of metadata to associate with the event.
:param encode: Whether to encode the images as JPEGs before sending.
:param mjpeg_quality: The JPEG quality to use when encoding.
:return: The event ID. None if the event failed to send.
function
send_video_event(video: bytes | str, title: str, metadata: Optional [ dict ] = None) -> Optional[str]: Optional[str]
Send a video event to RH. The video can be a path to a video file or a bytes object.

:param video: Path to a video file or a bytes object.
:param title: Title of the video event.
:param metadata: Overlay metadata to be displayed on the video.
:return: The event ID. None if the event failed to send.
module

robothub.frame_buffer

variable
class
class
class

robothub.frame_buffer.PacketType(enum.Enum)

constant
constant
class

robothub.frame_buffer.FrameBuffer

method
__init__(self, maxlen: int = None)
A buffer for storing frames.

:param maxlen: The maximum number of frames to store in the buffer. If None, the buffer will be unbounded.
method
save_video_event(self, before_seconds: int, after_seconds: int, title: str, fps: int, frame_width: int, frame_height: int, on_complete: Optional [ Callable ] = None, delete_after_complete: bool = False) -> threading.Thread: threading.Thread
Saves a video event to the frame buffer, then calls `on_complete` when the video is ready.
When the video is ready, the `on_complete` function will be called with the path to the video file.
Note: When app is stopped, it is not guaranteed that the video will be saved.

:param before_seconds: Number of seconds to save before the event occurred.
:param after_seconds: Number of seconds to save after the event occurred.
:param title: Title of the video event.
:param fps: The FPS of the video.
:param frame_width: Video frame width.
:param frame_height: Video frame height.
:param on_complete: Callback function to call when the video is ready. Path to the video file will be passed as the first argument.
:param delete_after_complete: If True, delete the video file after the callback function is called. Default: False.
:return: The 'threading.Thread' where the video is processed.
method
add_frame(self, packet: Union [ FramePacket , dai.ImgFrame ])
Default callback for the frame buffer. It will append the packet to the buffer and put it in all temporary queues.
property
module

robothub.live_view

class

robothub.live_view.LiveView(abc.ABC)

method
variable
property
property
method
publish(self, h264_frame: Union [ np.array , List ])
Publishes a frame to the RobotHub Live View.

:param h264_frame: H264 frame to publish.
method
add_rectangle(self, rectangle: BoundingBox, label: str)
Adds a rectangle (bounding box) to the live view.

:param rectangle: Tuple (x1, y1, x2, y2) where (x1, y1) is the top left corner and (x2, y2) is the bottom right corner.
:param label: Label to display on the rectangle.
method
add_text(self, text: str, coords: Tuple [ int , int ], size: int = None, color: Tuple [ int , int , int ] = None, thickness: int = None, outline: bool = True, background_color: Tuple [ int , int , int ] = None, background_transparency: float = 0.5)
Adds text to the live view.

:param text: Text to display.
:param coords: Tuple (x, y) where (x, y) is the top left corner of the text.
:param size: Size of the text.
:param color: Color of the text. E.g., (255, 0, 0) for red.
:param thickness: Thickness of the text.
:param outline: True to display an outline around the text, False otherwise.
:param background_color: Color of the background. E.g., (0, 0, 0) for black.
:param background_transparency: Transparency of the background. 0.0 is fully transparent, 1.0 is fully opaque.
method
add_line(self, pt1: Tuple [ int , int ], pt2: Tuple [ int , int ], color: Tuple [ int , int , int ] = None, thickness: int = None)
Adds a line to the live view.

:param pt1: (x, y) coordinates of the start point of the line.
:param pt2: (x, y) coordinates of the end point of the line.
:param color: Color of the line. E.g., (255, 0, 0) for red.
:param thickness: Thickness of the line.
static method
LiveView.get(unique_key: str = None, name: str = None) -> Optional[LiveView]: Optional[LiveView]
Gets a Live View by its unique key or name. Name takes precedence over unique key.

:param unique_key: Unique key of the Live View.
:param name: Name of the Live View.
:return: Live View with the given unique key or name.
:raises ValueError: If neither name nor unique_key is specified.
static method
LiveView.get_by_name(name: str) -> Optional[LiveView]: Optional[LiveView]
Gets a Live View by its name.

:param name: Name of the Live View.
:return: Live View with the given name. None if a Live View with the given name does not exist.
static method
LiveView.get_by_unique_key(unique_key: str) -> Optional[LiveView]: Optional[LiveView]
Gets a Live View by its unique key.

:param unique_key: Unique key of the Live View.
:return: Live View with the given unique key.
:raises ValueError: If a Live View with the given unique key does not exist.
class

robothub.live_view.DepthaiLiveView(robothub.live_view.LiveView)

class

robothub.live_view.SdkLiveView(robothub.live_view.LiveView)

method
__init__(self, name: str, unique_key: str, device_mxid: str, fps: int, frame_width: int, frame_height: int, max_buffer_size: int)
Class for creating and publishing Live Views.

:param name: Name of the Live View.
:param unique_key: Live View identifier.
:param device_mxid: MXID of the device that is streaming the Live View.
:param frame_width: Frame width.
:param frame_height: Frame height.
:param max_buffer_size: Maximum number of seconds to buffer.
variable
variable
property
property
CLASS_METHOD
method
save_video_event(self, before_seconds: int, after_seconds: int, title: str)
Saves a video event to the frame buffer, then calls `on_complete` when the video is ready.
When the video is ready, the `on_complete` function will be called with the path to the video file.
Note: When app is stopped, it is not guaranteed that the video will be saved.

:param before_seconds: Number of seconds to save before the event occurred.
:param after_seconds: Number of seconds to save after the event occurred.
:param title: Title of the video event.
module

robothub.live_view_utils

variable
function
is_h264_frame(data)
Check if the given data (numpy array) starts with an H.264 NAL unit.
This function checks for the presence of H.264 start codes and NAL types.
function
create_stream_handle(camera_serial: str, unique_key: str, name: str)
Create or get existing stream handle.

:param camera_serial: Device MXID.
:param unique_key: Unique key for the stream.
:param name: Name of the stream.
:return: robothub_core.StreamHandle object.
package

robothub.replay

module

robothub.replay.capture_manager

class

robothub.replay.capture_manager.PathType(enum.Enum)

constant
constant
constant
class

robothub.replay.capture_manager.ReadFrameType(enum.Enum)

constant
constant
constant
constant
module

robothub.replay.replay_builder

class

robothub.replay.replay_builder.CameraType(enum.Enum)

constant
constant
class

robothub.replay.replay_builder.ReplayBuilder

method
method
method
method
method
method
method
method
method
change_mp4_video_frame_rate(self, new_fps: float)
Replay FPS is usually capped between 5 - 15 FPS depending on Replay configuration.
When the source .mp4 file(s) are at much higher FPS then that, this allows to convert them to lower FPS.
Videos are converted and stored at the same location as the source video.
method
method
module

robothub.replay.replay_camera

class

robothub.replay.replay_camera.StreamName

variable
variable
variable
variable
variable
variable
variable
variable
variable
method
__post_init__(self)
class

robothub.replay.replay_camera.ReplayCamera(abc.ABC)

class

robothub.replay.replay_camera.ColorReplayCamera(robothub.replay.replay_camera.ReplayCamera)

method
method
method
method
property
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
property
property
property
property
property
property
property
property
property
class

robothub.replay.replay_camera.MonoReplayCamera(robothub.replay.replay_camera.ReplayCamera)

method
method
method
property
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
method
property
property
property
property
property
package

robothub.robothub_core_wrapper

module
app
Contains the base class for your application.
module
client
Mediates communication between App and Agent. Contains mainly internal methods.
module
communicator
Handles communication with App's frontend server
module
device
Defines base classes for handling of devices and their states.
module
events
Defines methods for sending Events to the cloud.
module
globals
Defines various global variables.
module
streams
Contains classes and methods for streaming to App's Frontend server and the Cloud.
module

robothub.robothub_core_wrapper.app

class
RobotHubApplication
Base class for your RobotHub App.
class

robothub.robothub_core_wrapper.app.RobotHubApplication(abc.ABC)

method
__init__(self)
Constructor
variable
stop_event
Deprecated, will be removed. Use C{app_is_running} or C{wait} functions instead.
variable
variable
property
running
Deprecated, will be removed. Use C{app_is_running} instead.
method
wait(self, time: float | int | None = None)
Deprecated, will be removed. Use C{wait} instead.

Invoking this function will sleep current thread for B{time} seconds.

If C{self.running -> False}, returns immediately. If App is stopped midway through C{self.wait}, likewise returns immediately.

Should be preferred over C{time.sleep} in all use cases, as C{time.sleep} will not be interrupted if App is stopped, often causing unexpected behaviour.

@param time: Time to sleep for, given in seconds. Negative values are interpreted as 0.
@type time: float | int
method
on_start(self)
Entrypoint of the App. Must be defined by the user.
method
start_execution(self)
Optional method, is executed after L{on_start()} returns.

Can be used as a second entrypoint to the App.
method
on_stop(self)
Method for cleanup of threads/processes, is called when:
    - App is stopped by Agent (e.g. due to a request to Stop from the Cloud)
    - App stops on its own - for example by calling C{exit()}
    - App throws an uncaught exception

Should be used to quickly and gracefully stop all execution. Not defining this method properly, such as not joining all threads that were started in C{self.on_start()} often results in undefined behaviour.
Not called when C{os._exit()} is called.
method
on_event_uploaded(self, event: UploadedEvent)
Optional. Is called when an Event is uploaded to cloud. UploadedEvent argument contains information about the uploaded Event.

Is not called if Event is rejected by Agent

@param event: Object containing information about the uploaded event.
@type event: L{UploadedEvent}
method
on_configuration_changed(self, configuration_changes: dict)
Is called when the Agent receives a new configuration. New configuration is loaded into the CONFIGURATION variable.
A dictionary containing configuration changes [:param: configuration_changes] is provided so that user can decide proper behavior when this
method is overriden.
Default behavior is app restart. Override this method if you want to change the behavior.

@param configuration_changes: Dictionary containing configuration changes.
@type configuration_changes: dict
method
on_assigned_devices_changed(self)
Is called when Devices assigned to the app have changed.
method
stop(self)
Stops the App.
method
restart(self)
Restarts the App. This blocks execution until Agent stops the App. App is then started.
method
restart_host(self)
Requests the Agent to restart Host. Blocks execution until request is resolved.
method
shutdown_host(self)
Requests the Agent to shutdown Host. Blocks execution until request is resolved.
method
run(self)
Launch script uses this method to start the App.
module

robothub.robothub_core_wrapper.client

class
AgentClient
Internally used to communicate with agent, send / receive messages over (FIFO) / STDIN (JSON serialized objects)
constant
class

robothub.robothub_core_wrapper.client.AgentClient

method
method
classify_event(self, event_id: str, classification: str, blocking: bool = False)
Sends a classification for an Event with id equal to {event_id}.
method
publish_device_info(self, device_info: dict)
Publishes device info. C{device_info} needs to be a properly formatted dictionary.

@param device_info: Dictionary containing device info
@type device_info: dict
method
publish_device_stats(self, device_stats: dict)
Publishes device stats. C{device_stats} needs to be a properly formatted dictionary.

@param device_stats: Dictionary containing device stats
@type device_stats: dict
module

robothub.robothub_core_wrapper.communicator

class
CommunicatorResponse
Responses to requests from App are instances of this class
class
Communicator
Handles communication with App's frontend server
constant
COMMUNICATOR
Instance of Communicator class initialized at startup.
class

robothub.robothub_core_wrapper.communicator.CommunicatorResponse

variable
sessionId: str|None
ID of specific session, can be None
variable
payload: Any
Response payload, structure defined by user
class

robothub.robothub_core_wrapper.communicator.Communicator(abc.ABC)

method
method
notify(self, key: str, payload: str | list | dict | None, target: str | None = None)
Sends a notification to the FE server.

@param key: Used to define different types of notifications
@type key: str
@param payload: Content of the notification
@type payload: str | list | dict | None
@param target: If target is a valid session ID, notification is sent to the specific session. If None, notification is broadcast.
@type target: str | None
method
request(self, key: str, payload: str | list | dict | None, target: str | None = None, timeoutSeconds: float | int = 30) -> CommunicatorResponse|bool: CommunicatorResponse|bool
Sends a request to the FE server and blocks until a response is received or until timeout. 

@param key: Used to define different types of requests
@type key: str
@param payload: Content of the request
@type payload: str | list | dict | None
@param target: If target is a valid session ID, request is sent to the specific session. If None, request is broadcast.
@type target: str | None
@param timeoutSeconds: Timeout length in seconds.
@type timeoutSeconds: float | int
method
requestAsync(self, key: str, payload: Any, target: str | None = None, timeoutSeconds = 30, on_response: Callable [ [ Any ] , None ] | None = None)
Sends an asynchronous request to the FE server, calls a callback on the response if a response is received before timeout.

@param key: Used to define different types of requests
@type key: str
@param payload: Content of the request
@type payload: str | list | dict | None
@param target: If target is a valid session ID, request is sent to the specific session. If None, request is broadcast.
@type target: str | None
@param timeoutSeconds: Timeout length in seconds.
@type timeoutSeconds: float | int
@param on_response: Callback to be invoked on the response. 
@type on_response: Callable[[Any], None]
method
on_frontend(self, session_start: Callable | None = None, session_end: Callable | None = None, notification: Callable | None = None, request: Callable | None = None)
Sets callbacks for session start & end, notifications and requests from the FE server.

@param session_start: Callback for "session started" messages from FE
@type session_start: Callable
@param session_end: Callback for "session ended" messages from FE
@type session_end: Callable
@param notification: Callback for notifications from FE
@type notification: Callable
@param request: Callback for requests from FE
@type request: Callable
method
set_devices_changed_cb(self, devices_changed_cb: Callable)
Sets the device change callback. 

This function will be called each time configuration of any assigned device changes.
@param devices_changed_cb: Callback for requests from FE
@type devices_changed_cb: Callable
module

robothub.robothub_core_wrapper.device

class
RobotHubDevice
RobotHub device base class.
class
DeviceState
Possible states of a device connected to the Agent.
class

robothub.robothub_core_wrapper.device.RobotHubDevice(abc.ABC)

variable
type: str
String denoting type of the device, allowed values: ["oak"]
variable
oak: Dict[str, Any]
Dictionary containing various information about the device
class

robothub.robothub_core_wrapper.device.DeviceState(enum.Enum)

constant
CONNECTED: str
Agent is actively communicating with the device. In the case of DepthAI devices, this means a pipeline is running on the device.
constant
DISCONNECTED: str
Agent is not actively communicating with the device. Either device is disconnected or no App is running on it.
constant
CONNECTING: str
Agent is connecting to the device. Occurs when an App is initializing on the device, but full communication is not yet going on.
constant
UNKNOWN: str
Agent is trying to obtain info about device state. Occurs when Agent is starting up.
module

robothub.robothub_core_wrapper.events

class
Events
Used to prepare Events for agent to consume and inform agent via AgentClient
class
FutureEvent
Object describing an Event
class
UploadedEventPublicAccess
Object that describes public access (without credentials) to an uploaded Event.
class
UploadedEvent
Contains information about an uploaded Event, has no methods. Automatically sent as input to App's "on_event_uploaded" method.
constant
class

robothub.robothub_core_wrapper.events.Events

method
method
prepare(self) -> FutureEvent: FutureEvent
Creates a new empty Event.

@return: An empty Event of type L{FutureEvent}
method
upload(self, event: FutureEvent)
Uploads an Event.

@param event: Event to be uploaded.
@type event: L{FutureEvent}
method
send_frame_event(self, imagedata: bytes | bytearray, camera_serial: str, title: str | None = None, frame_name: str | None = None, frame_metadata: dict | None = None)
Creates and immediately uploads an Event with a single frame. Does not allow for full customization of the Event.

@param imagedata: The encoded frame bytes
@type imagedata: bytes | bytearray
@param camera_serial: Serial number (MxID) of camera that took the picture
@type camera_serial: str
@param title: Optional title for the Event
@type title: str | NoneType
@param frame_name: Optional name for the frame, does not define the filename
@type frame_name: str | NoneType
@param frame_metadata: metadata of valid format will be visualized over the frame in the cloud. Optional.
@type frame_metadata: dict | NoneType
method
send_video_event(self, video: bytes | bytearray, title: str | None = None, video_name: str | None = None, video_metadata: dict | None = None)
Creates and immediately uploads an Event with a single video. Does not allow for full customization of the Event.

@param video: The encoded video bytes. Encoding must be H264.
@type video: bytes | bytearray
@param title: Optional title for the Event
@type title: str | NoneType
@param video_name: Optional name for the video, does not define the filename
@type video_name: str | NoneType
@param video_metadata: Metadata of valid format will be visualized over the video in the cloud. Optional.
@type video_metadata: dict | NoneType
method
send_binary_file_event(self, binary_data: bytes | bytearray, title: str | None = None, file_name: str | None = None)
Creates and immediately uploads an Event with a file of arbitrary type. Does not allow for full customization of the Event.

@param binary_data: The file in bytes
@type binary_data: bytes | bytearray
@param title: Optional title for the Event
@type title: str | NoneType
@param file_name: Optional name for the file, does not define the filename
@type file_name: str | NoneType
method
send_text_file_event(self, text: str, title: str | None = None, text_name: str | None = None)
Creates and immediately uploads an Event with a single string. Does not allow for full customization of the Event.

@param text: The text in string form
@type text: str
@param text: The text in string form
@type text: str
@param title: Optional title for the Event
@type title: str | NoneType
@param text_name: Optional name for the text
@type text_name: str | NoneType
class

robothub.robothub_core_wrapper.events.FutureEvent

variable