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. Accessible only when running as a Perception App.
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.decorators
function
variable
variable
variable
variable
function
measure_average_performance(func: Callable
[
...
,
Any
] = None, report_every_minutes: float
|
int = 5) -> Callable[..., Any]: Callable[..., Any]Report once every `report_every_minutes` minutes what the averages function duration is and what the max and min durations are.
class
variable
function
measure_call_frequency(func: Callable
[
...
,
Any
]) -> Callable[..., Any]: Callable[..., Any]Measure how often a given function is called.
function
with_sql_exception_handling(func: Callable
[
...
,
Any
]) -> Callable[..., Any]: Callable[..., Any]Try sql operation command, catch exception and finally close the cursor and the connection.
function
class
robothub.decorators.FuncDetail
variable
variable
variable
variable
variable
variable
constant
method
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
class
robothub.frame_buffer.PacketType(enum.Enum)
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.ThreadSaves 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
variable
class
constant
class
class
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
class
variable
variable
class
class
robothub.replay.capture_manager.PathType(enum.Enum)
class
robothub.replay.capture_manager.CaptureManager
method
method
method
method
method
method
method
method
method
method
module
robothub.replay.captures
class
class
ImageDirectoryCapture
Used for sending images from directory to the replay node instead of a video.
class
class
robothub.replay.captures.Capture(abc.ABC)
class
robothub.replay.captures.ImageDirectoryCapture(robothub.replay.captures.Capture)
method
variable
variable
method
method
method
method
method
class
robothub.replay.captures.VideoCapture(robothub.replay.captures.Capture)
module
robothub.replay.replay_builder
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
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
module
robothub.replay.utils
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. Set to C{True} at startup. Is set to C{False} when either App is stopped by Agent or it excepts. @return: True if App is running, False otherwise. 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 | intmethod
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: dictmethod
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: dictmodule
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|boolSends 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: FutureEventCreates 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
id
ID of the Event
variable
folder_path
Path to the folder of the Event in the App's container. Added frames, files & videos are stored in this folder.
method
variable
title
Title of the Event in the Cloud. Set to "Event" + UUID by default.
method
add_video(self, _bytes: bytes
|
bytearray, name: str
|
None = None, metadata: dict
|
None = None, filename: str
|
None = None, camera_serial: str
|
None = None)Adds a video to the Event. @param _bytes: Bytes of the encoded video. Must be H264 format. @type _bytes: bytes | bytearray @param name: Optional - Name of the video @type name: str | NoneType @param metadata: Optional - Metadata to be overlayed over the video. List length must be equal to number of frames in the video. @type metadata: list @param filename: Optional - can define name of the video file on Host. @type filename: str | NoneType @param camera_serial: Optional - Serial number (MxID) of camera that took the video @type camera_serial: str
method
add_frame(self, _bytes, camera_serial: str
|
None = None, name: str
|
None = None, metadata: dict
|
None = None, filename: str
|
None = None)Adds a frame to the Event. @param _bytes: Bytes of the encoded frame @type _bytes: bytes | bytearray @param camera_serial: Optional - Serial number (MxID) of camera that took the frame @type camera_serial: str @param name: Optional - Name of the frame @type name: str | NoneType @param metadata: Optional - Metadata to be overlayed over the frame @type metadata: list @param filename: Optional - can define name of the frame file on Host. @type filename: str | NoneType
method
add_file(self, _bytes, name: str
|
None = None, filename: str
|
None = None)Adds a file to the Event. @param _bytes: Bytes of the file. @type _bytes: bytes | bytearray @param name: Optional - Name of the file in the cloud @type name: str | NoneType @param filename: Optional - can define name of the file saved on Host. @type filename: str | NoneType
method
add_existing_file(self, filename: Path
|
str, copy: bool = True, name: str
|
None = None)Adds an existing file to the Event. @param filename: Path to the file to be added. @type filename: Path | str @param copy: If True, the file will be copied to the Event folder. If False, the file will be moved to the Event folder. @type copy: bool @param name: Optional - Name of the file in the cloud @type name: str | NoneType
method
set_title(self, title)Sets the title of the Event. @param title: Title for the Event @type title: str | NoneType
method
set_metadata(self, metadata: dict)Sets the metadata of the Event. @param metadata: Dictionary containing the metadata @type metadata: dict
method
add_tag(self, tag: str)Adds a tag to the Event. Cannot add more than 10 tags to one Event. @param tag: The tag string @type tag: str
method
method
property
keep_after_upload
Decides whether Event should be kept after upload, C{False} by default. Example usage: >>> self.keep_after_upload True return: boolmethod
keep_after_upload.setter(self, value: bool)Sets the keep after upload property. Example usage: >>> self.keep_after_upload = True
property
no_upload_by_default
If set to True, Event will not be uploaded. C{False} by default. Example usage: >>> self.no_upload_by_default True return: boolmethod
no_upload_by_default.setter(self, value: bool)Sets the keep after upload property. Example usage: >>> self.keep_after_upload = True
property
keep_when_space_low
Decides whether Event should be kept when storage space is low, C{False} by default. Example usage: >>> self.keep_after_upload True return: boolmethod
keep_when_space_low.setter(self, value: bool)Sets the keep when space low property. Example usage: >>> self.keep_when_space_low = True
class
robothub.robothub_core_wrapper.events.UploadedEventPublicAccess(typing.TypedDict)
variable
domain: str
Domain of the robot
variable
token: str
Token which allows limited time access to the Event
class
robothub.robothub_core_wrapper.events.UploadedEvent
variable
event_id: str
ID of the Event
variable
robothub_url: str
URL of the Event, accessible only with credentials
variable
public_access: Union[UploadedEventPublicAccess, None]
Object allowing public access
variable
module
robothub.robothub_core_wrapper.globals
constant
TEAM_ID
ID of RobotHub team Robot running the App belongs to.
constant
APP_VERSION
Version of the source the App was installed with.
constant
APP_INSTANCE_ID
ID of instance of the App on current Robot. Not the ID of the App.
constant
ROBOT_ID
ID of Robot running the App.
constant
constant
constant
STORAGE_DIR: str
Storage directory for Events and other files, can be used by the user
constant
PUBLIC_FILES_DIR: str
Storage directory intended for public assets
constant
constant
variable
variable
variable
module
robothub.robothub_core_wrapper.streams
class
Streams
Handles video streams
class
StreamHandle
Used for handling an open stream @param camera_serial: Serial number (MxID) of camera the stream is registered for @type camera_serial: str @param unique_key: Unique Key identifying the stream @type unique_key: str @param description: Name of the stream in the cloud @type description: str
constant
STREAMS
Instance of B{Streams} class, initialized at App startup.class
robothub.robothub_core_wrapper.streams.Streams
variable
method
method
create_video(self, camera_serial: str, unique_key: str, description: str) -> StreamHandle: StreamHandleCreates a stream and returns its L{StreamHandle} @param camera_serial: Serial number (MxID) of camera requesting the stream @type camera_serial: str @param unique_key: Key identifying the stream. Must be unique for each created stream, else an exception is thrown. @type unique_key: str @param description: Name of the stream in the cloud. Does not have to be unique. @type description: strmethod
destroy(self, stream: StreamHandle)Deletes a stream and its L{StreamHandle}. @param stream: key of stream to be destroyed @type stream: L{StreamHandle}method
destroy_streams_by_id(self, stream_ids: list)Destroys streams corresponding to {stream_ids}, throws if any of the corresponding streams does not exist. @param stream_ids: Keys of streams to be destroyed @type stream_ids: List[str]method
destroy_all_streams(self)Destroys all streams.
class
robothub.robothub_core_wrapper.streams.StreamHandle
method
variable
unique_key
Unique Key identifying the stream
variable
camera_serial
Serial number (MxID) of camera the stream is registered for
variable
description
Name of the stream in the cloud
method
publish_video_data(self, video_data: bytes
|
bytearray, timestamp: int, metadata: dict
|
None = None)Used to send a frame with a corresponding timestamp. Optionally, metadata to be rendered over the frame can be included. @param video_data: Bytes of the encoded frame. Encoding must be H264. @type video_data: bytes | bytearray @param timestamp: Timestamp of the frame. Does not have to correspond to when frame was taken on device, but must be increasing with each subsequent frame. @type timestamp: int
module
robothub.types
module
robothub.utils
function
setup_logger(name: str, level: int = logging.INFO)Initializes a logger with the given name and level. :param name: Logger name. :param level: Either a string or an integer. If a string, it must be one of the following: 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'. If an integer, it must be one of the following: log.DEBUG, log.INFO, log.WARNING, log.ERROR, log.CRITICAL.
function
get_device_performance_metrics(device: depthai.Device) -> Dict[str, Any]: Dict[str, Any]Returns a dictionary with statistics about the device.
function
get_device_details(device: depthai.Device, state: robothub_core.DeviceState) -> Dict[str, Any]: Dict[str, Any]Returns a dictionary with information about the device.
function
try_or_default(func, default = None)Tries to call the given function and returns its result. If an exception is raised, returns the default value.