ros_sugar.events

Contents

ros_sugar.events#

Available Events

Module Contents#

Classes#

OnAny

OnChange

OnChange Event is triggered when a given topic attribute changes in value from any initial value to any new value. The target attribute value is registered on the first recept of a message on the target topic, then the event is triggered on a change in that value. After a change the new value is registered and the event is triggered again on any new change, …etc.

OnChangeEqual

OnChangeEqual Event is a combination of OnChange and OnEqual events. OnChangeEqual is triggered when a given topic attribute changes in value from any initial value to given trigger goal value.

OnEqual

OnEqual Event is triggered when a given topic attribute value is equal to a given trigger value.

OnContainsAll

OnContainsAll Event is triggered when a given topic attribute value contains all of the given trigger list value.

OnContainsAny

OnContainsAny Event is triggered when a given topic attribute value contains one of the given trigger list value.

OnDifferent

OnDifferent Event is triggered when a given topic attribute value is different from a given trigger value.

OnGreater

OnGreater Event is triggered when a given topic attribute value is greater than a given trigger value.

OnLess

OnGreater Event is triggered when a given topic attribute value is less than a given trigger value.

Functions#

json_to_events_list

Loads a list of events from a JSON object

API#

ros_sugar.events.json_to_events_list(json_obj: Union[str, bytes, bytearray], topic_template: Optional[ros_sugar.io.topic.Topic] = None) List#

Loads a list of events from a JSON object

Parameters:

json_obj (str | bytes | bytearray) – JSON object containing a set of events

Raises:

ValueError – If the provided json object cannot be converted to an events list

Returns:

Events list

Return type:

List[Event]

class ros_sugar.events.OnAny(event_name: str, event_source: Union[ros_sugar.io.topic.Topic, str, Dict])#

Bases: ros_sugar.core.event.Event

callback(msg: Any) None#

Event topic listener callback

Parameters:

msg (Any) – Event trigger topic message

property under_processing: bool#
reset()#
property name: str#
clear() None#
trig() None#
property dictionary: Dict#
set_dictionary(dict_obj, topic_template: ros_sugar.io.topic.Topic)#
property json: str#
register_method(method_name: str, method: Callable[..., Any]) None#
register_actions(actions: Union[ros_sugar.core.action.Action, List[ros_sugar.core.action.Action]]) None#
clear_actions() None#
remove_method(method_name: str)#
class ros_sugar.events.OnChange(event_name: str, event_source: Union[ros_sugar.io.topic.Topic, str, Dict], nested_attributes: Union[str, List[str]], **kwargs)#

Bases: ros_sugar.core.event.Event

OnChange Event is triggered when a given topic attribute changes in value from any initial value to any new value. The target attribute value is registered on the first recept of a message on the target topic, then the event is triggered on a change in that value. After a change the new value is registered and the event is triggered again on any new change, …etc.

Example usage scenario:

  • Event on a change in the number of detected people of the robot by a vision system to play a friendly welcome message.

callback(msg) None#

Overrides Event callback to save previous event value for OnChange comparison

Parameters:

msg (Any) – Event topic message

property under_processing: bool#
reset()#
property name: str#
clear() None#
trig() None#
property dictionary: Dict#
set_dictionary(dict_obj, topic_template: ros_sugar.io.topic.Topic)#
property json: str#
register_method(method_name: str, method: Callable[..., Any]) None#
register_actions(actions: Union[ros_sugar.core.action.Action, List[ros_sugar.core.action.Action]]) None#
clear_actions() None#
remove_method(method_name: str)#
class ros_sugar.events.OnChangeEqual(event_name: str, event_source: Union[ros_sugar.io.topic.Topic, str, Dict], trigger_value: Union[float, int, bool, str, List], nested_attributes: Union[str, List[str]], **kwargs)#

Bases: ros_sugar.core.event.Event

OnChangeEqual Event is a combination of OnChange and OnEqual events. OnChangeEqual is triggered when a given topic attribute changes in value from any initial value to given trigger goal value.

Note

The difference between using OnChangeEqual as opposite to OnEqual or OnChange is that:

  • OnEqual will keep getting triggered every time a new message value is received that is equal to the trigger.

  • OnChange will keep getting triggered every time a new message value is received that is different from a previous value

  • OnChangeEqual will get triggered once when the topic message value reaches the trigger, making it convenient for many applications

Some example usage scenarios:

  • Event on the robot reaching a navigation goal point: reach_end Boolean topic OnChangeEqual to True (triggered once when reaching, does not trigger again if the robot is static and staying in ‘goal reaching’ state)

  • Event on an Enum value of a message attribute to detect reaching a given state.

  • Event of reaching 100% charge level of a robot to end charging.

callback(msg) None#

Overrides Event callback to save previous event value for OnChange comparison

Parameters:

msg (Any) – Event topic message

property under_processing: bool#
reset()#
property name: str#
clear() None#
trig() None#
property dictionary: Dict#
set_dictionary(dict_obj, topic_template: ros_sugar.io.topic.Topic)#
property json: str#
register_method(method_name: str, method: Callable[..., Any]) None#
register_actions(actions: Union[ros_sugar.core.action.Action, List[ros_sugar.core.action.Action]]) None#
clear_actions() None#
remove_method(method_name: str)#
class ros_sugar.events.OnEqual(event_name: str, event_source: Union[ros_sugar.io.topic.Topic, str, Dict], trigger_value: Union[float, int, bool, str, List], nested_attributes: Union[str, List[str]], **kwargs)#

Bases: ros_sugar.core.event.Event

OnEqual Event is triggered when a given topic attribute value is equal to a given trigger value.

Example usage scenario:

  • Event when the detection id (object type) in an object detection topic is equal to a specific object (to raise an event on detecting another robot, a human, etc.)

property under_processing: bool#
reset()#
property name: str#
clear() None#
trig() None#
property dictionary: Dict#
set_dictionary(dict_obj, topic_template: ros_sugar.io.topic.Topic)#
property json: str#
callback(msg: Any) None#
register_method(method_name: str, method: Callable[..., Any]) None#
register_actions(actions: Union[ros_sugar.core.action.Action, List[ros_sugar.core.action.Action]]) None#
clear_actions() None#
remove_method(method_name: str)#
class ros_sugar.events.OnContainsAll(event_name: str, event_source: Union[ros_sugar.io.topic.Topic, str, Dict], trigger_value: List, nested_attributes: Union[str, List[str]], **kwargs)#

Bases: ros_sugar.core.event.Event

OnContainsAll Event is triggered when a given topic attribute value contains all of the given trigger list value.

property under_processing: bool#
reset()#
property name: str#
clear() None#
trig() None#
property dictionary: Dict#
set_dictionary(dict_obj, topic_template: ros_sugar.io.topic.Topic)#
property json: str#
callback(msg: Any) None#
register_method(method_name: str, method: Callable[..., Any]) None#
register_actions(actions: Union[ros_sugar.core.action.Action, List[ros_sugar.core.action.Action]]) None#
clear_actions() None#
remove_method(method_name: str)#
class ros_sugar.events.OnContainsAny(event_name: str, event_source: Union[ros_sugar.io.topic.Topic, str, Dict], trigger_value: List, nested_attributes: Union[str, List[str]], **kwargs)#

Bases: ros_sugar.core.event.Event

OnContainsAny Event is triggered when a given topic attribute value contains one of the given trigger list value.

property under_processing: bool#
reset()#
property name: str#
clear() None#
trig() None#
property dictionary: Dict#
set_dictionary(dict_obj, topic_template: ros_sugar.io.topic.Topic)#
property json: str#
callback(msg: Any) None#
register_method(method_name: str, method: Callable[..., Any]) None#
register_actions(actions: Union[ros_sugar.core.action.Action, List[ros_sugar.core.action.Action]]) None#
clear_actions() None#
remove_method(method_name: str)#
class ros_sugar.events.OnDifferent(event_name: str, event_source: Union[ros_sugar.io.topic.Topic, str, Dict], trigger_value: Union[float, int, bool, str, List], nested_attributes: Union[str, List[str]], **kwargs)#

Bases: ros_sugar.core.event.Event

OnDifferent Event is triggered when a given topic attribute value is different from a given trigger value.

property under_processing: bool#
reset()#
property name: str#
clear() None#
trig() None#
property dictionary: Dict#
set_dictionary(dict_obj, topic_template: ros_sugar.io.topic.Topic)#
property json: str#
callback(msg: Any) None#
register_method(method_name: str, method: Callable[..., Any]) None#
register_actions(actions: Union[ros_sugar.core.action.Action, List[ros_sugar.core.action.Action]]) None#
clear_actions() None#
remove_method(method_name: str)#
class ros_sugar.events.OnGreater(event_name: str, event_source: Union[ros_sugar.io.topic.Topic, str, Dict], trigger_value: Union[float, int, bool, str], nested_attributes: Union[str, List[str]], or_equal: bool = False, **kwargs)#

Bases: ros_sugar.core.event.Event

OnGreater Event is triggered when a given topic attribute value is greater than a given trigger value.

Example usage scenario:

  • Event when a drone is higher than a certain allowed elevation (location z coordinate > elevation level), to bring the drone down into allowed limits.

property under_processing: bool#
reset()#
property name: str#
clear() None#
trig() None#
property dictionary: Dict#
set_dictionary(dict_obj, topic_template: ros_sugar.io.topic.Topic)#
property json: str#
callback(msg: Any) None#
register_method(method_name: str, method: Callable[..., Any]) None#
register_actions(actions: Union[ros_sugar.core.action.Action, List[ros_sugar.core.action.Action]]) None#
clear_actions() None#
remove_method(method_name: str)#
class ros_sugar.events.OnLess(event_name: str, event_source: Union[ros_sugar.io.topic.Topic, str, Dict], trigger_value: Union[float, int, bool, str], nested_attributes: Union[str, List[str]], or_equal: bool = False, **kwargs)#

Bases: ros_sugar.core.event.Event

OnGreater Event is triggered when a given topic attribute value is less than a given trigger value.

Example usage scenario:

  • Event when the robot battery level falls under a certain low limit, to go back to the charging station, for example.

property under_processing: bool#
reset()#
property name: str#
clear() None#
trig() None#
property dictionary: Dict#
set_dictionary(dict_obj, topic_template: ros_sugar.io.topic.Topic)#
property json: str#
callback(msg: Any) None#
register_method(method_name: str, method: Callable[..., Any]) None#
register_actions(actions: Union[ros_sugar.core.action.Action, List[ros_sugar.core.action.Action]]) None#
clear_actions() None#
remove_method(method_name: str)#