13.3. Utilities: utils

constants

decorators

rafcon.utils.decorators.avoid_parallel_execution(func)

A decorator to avoid the parallel execution of a function.

If the function is currently called, the second call is just skipped.

Parameters

func – The function to decorate

Returns

dict_operations

rafcon.utils.dict_operations.check_if_dict_contains_object_reference_in_values(object_to_check, dict_to_check)

Method to check if an object is inside the values of a dict. A simple object_to_check in dict_to_check.values() does not work as it uses the __eq__ function of the object and not the object reference.

Parameters

  • object_to_check – The target object.

  • dict_to_check – The dict to search in.

Returns

execution_log

rafcon.utils.execution_log.log_to_DataFrame(execution_history_items, data_in_columns=[], data_out_columns=[], scoped_in_columns=[], scoped_out_columns=[], semantic_data_columns=[], throw_on_pickle_error=True)

Returns all collapsed items in a table-like structure (pandas.DataFrame) with one row per executed state and a set of properties resp. columns (e.g. state_name, outcome, run_id) for this state. The data flow (data_in/out, scoped_data_in/out, semantic_data) is omitted from this table representation by default, as the different states have different data in-/out-port, scoped_data- ports and semantic_data defined. However, you can ask specific data-/scoped_data-ports and semantic data to be exported as table column, given they are primitive-valued, by including the port / key names in the {*}_selected-parameters. These table-columns will obviously only be well-defined for states having this kind of port-name-/semantic-key and otherwise will contain a None-like value, indicating missing data.

The available data per execution item (row in the table) can be printed using pandas.DataFrame.columns.

rafcon.utils.execution_log.log_to_collapsed_structure(execution_history_items, throw_on_pickle_error=True, include_erroneous_data_ports=False, full_next=False)

Collapsed structure means that all history items belonging to the same state execution are merged together into one object (e.g. CallItem and ReturnItem of an ExecutionState). This is based on the log structure in which all Items which belong together have the same run_id. The collapsed items hold input as well as output data (direct and scoped), and the outcome the state execution. :param dict execution_history_items: history items, in the simplest case directly the opened shelve log file :param bool throw_on_pickle_error: flag if an error is thrown if an object cannot be un-pickled :param bool include_erroneous_data_ports: flag if to include erroneous data ports :param bool full_next: flag to indicate if the next relationship has also to be created at the end of container states :return: - start_item, the StateMachineStartItem of the log file - next, a dict mapping run_id –> run_id of the next executed state on the same hierarchy level - concurrent, a dict mapping run_id –> []list of run_ids of the concurrent next executed states (if present) - hierarchy, a dict mapping run_id –> run_id of the next executed state on the deeper hierarchy level (the start state within that HierarchyState) - items, a dict mapping run_id –> collapsed representation of the execution of the state with that run_id :rtype: tuple

rafcon.utils.execution_log.log_to_ganttplot(execution_history_items)

Example how to use the DataFrame representation

rafcon.utils.execution_log.log_to_raw_structure(execution_history_items)

Logging with raw structure. :param dict execution_history_items: history items, in the simplest case directly the opened shelve log file :return: - start_item, the StateMachineStartItem of the log file - previous, a dict mapping history_item_id –> history_item_id of previous history item - next, a dict mapping history_item_id –> history_item_id of the next history item (except if next item is a concurrent execution branch) - concurrent, a dict mapping history_item_id –> []list of concurrent next history_item_ids (if present) - grouped, a dict mapping run_id –> []list of history items with this run_id :rtype: tuple

filesystem

rafcon.utils.filesystem.clean_file_system_paths_from_not_existing_paths(file_system_paths)

Cleans list of paths from elements that do not exist

If a path is no more valid/existing, it is removed from the list.

Parameters

file_system_paths (list[str]) – list of file system paths to be checked for existing

rafcon.utils.filesystem.copy_file_or_folder(src, dst)
rafcon.utils.filesystem.create_path(path)

Creates a absolute path in the file system.

Parameters

path – The path to be created

rafcon.utils.filesystem.get_default_config_path()
rafcon.utils.filesystem.get_default_log_path()
rafcon.utils.filesystem.make_file_executable(filename)
rafcon.utils.filesystem.make_tarfile(output_filename, source_dir)
rafcon.utils.filesystem.read_file(file_path, filename=None)

Open file by path and optional filename

If no file name is given the path is interpreted as direct path to the file to be read. If there is no file at location the return value will be None to offer a option for case handling.

Parameters

  • file_path (str) – Path string.

  • filename (str) – File name of the file to be read.

Returns

None or str

rafcon.utils.filesystem.separate_folder_path_and_file_name(path)
rafcon.utils.filesystem.write_file(file_path, content, create_full_path=False)

geometry

rafcon.utils.geometry.cal_dist_between_2_coord_frame_aligned_boxes(box1_pos, box1_size, box2_pos, box2_size)

Calculate Euclidean distance between two boxes those edges are parallel to the coordinate axis

The function decides condition based which corner to corner or edge to edge distance needs to be calculated.

Parameters

  • box1_pos (tuple) – x and y position of box 1

  • box1_size (tuple) – x and y size of box 1

  • box2_pos (tuple) – x and y position of box 2

  • box2_size (tuple) – x and y size of box 2

Returns

distance

rafcon.utils.geometry.deg2rad(degree)

Converts angle given in degrees into radian

Parameters

degree (float) – Angle to be converted

Returns

Angle in radian

Return type

float

rafcon.utils.geometry.dist(p1, p2)

Calculates the distance between two points

The function calculates the Euclidean distance between the two 2D points p1 and p2 :param p1: Tuple with x and y coordinate of the first point :param p2: Tuple with x and y coordinate of the second point :return: The Euclidean distance

rafcon.utils.geometry.equal(t1, t2, digit=None)

Compare two iterators and its elements on specific digit precision

The method assume that t1 and t2 are are list or tuple.

Parameters

  • t1 – First element to compare

  • t2 – Second element to compare

  • digit (int) – Number of digits to compare

Return type

bool

Returns

True if equal and False if different

rafcon.utils.geometry.point_left_of_line(point, line_start, line_end)

Determines whether a point is left of a line

Parameters

  • point – Point to be checked (tuple with x any y coordinate)

  • line_start – Starting point of the line (tuple with x any y coordinate)

  • line_end – End point of the line (tuple with x any y coordinate)

Returns

True if point is left of line, else False

hashable

class rafcon.utils.hashable.Hashable

Bases: object

static get_object_hash_string(object_)
mutable_hash(obj_hash=None)

Creates a hash with the (im)mutable data fields of the object

Example:
>>> my_obj = type("MyDerivedClass", (Hashable,), { "update_hash": lambda self, h: h.update("RAFCON") })()
>>> my_obj_hash = my_obj.mutable_hash()
>>> print('Hash: ' + my_obj_hash.hexdigest())
Hash: c8b2e32dcb31c5282e4b9dbc6a9975b65bf59cd80a7cee66d195e320484df5c6
Parameters

obj_hash – The hash object (see Python hashlib)

Returns

The updated hash object

update_hash(obj_hash)

Should be implemented by derived classes to update the hash with their data fields

Parameters

obj_hash – The hash object (see Python hashlib)

static update_hash_from_dict(obj_hash, object_)

Updates an existing hash object with another Hashable, list, set, tuple, dict or stringifyable object

Parameters

  • obj_hash – The hash object (see Python hashlib documentation)

  • object – The value that should be added to the hash (can be another Hashable or a dictionary)

i18n

rafcon.utils.i18n.create_mo_files(logger)
rafcon.utils.i18n.setup_l10n(logger=None)

Setup RAFCON for localization

Specify the directory, where the translation files ({*}.mo) can be found (rafcon/locale/) and set localization domain (“rafcon”).

Parameters

logger – which logger to use for printing (either logging.log or distutils.log)

installation

rafcon.utils.installation.install_fonts(restart=False)
rafcon.utils.installation.install_locally_required_files()
rafcon.utils.installation.installed_font_faces_for_font(font_name)
rafcon.utils.installation.started_in_virtualenv()
rafcon.utils.installation.started_without_installation()
rafcon.utils.installation.update_font_cache(path)

log

exception rafcon.utils.log.RAFCONDeprecationWarning

Bases: DeprecationWarning

rafcon.utils.log.add_logging_level(level_name, level_num, method_name=None)

Add new logging level

Comprehensively adds a new logging level to the logging module and the currently configured logging class.

method_name becomes a convenience method for both logging itself and the class returned by logging.getLoggerClass() (usually just logging.Logger). If method_name is not specified, level_name.lower() is used.

Parameters

  • level_name (str) – the level name

  • level_num (int) – the level number/value

Raises

AttributeError – if the level name is already an attribute of the logging module or if the method name is already present

>>> add_logging_level('TRACE', logging.DEBUG - 5)
>>> logging.getLogger(__name__).setLevel("TRACE")
>>> logging.getLogger(__name__).trace('that worked')
>>> logging.trace('so did this')
>>> logging.TRACE
5
rafcon.utils.log.get_logger(name)

Returns a logger for the given name

The function is basically a wrapper for logging.getLogger and only ensures that the namespace is within “rafcon.” and that the propagation is enabled.

Parameters

name (str) – The namespace of the new logger

Returns

Logger object with given namespace

Return type

logging.Logger

class rafcon.utils.log.log_exceptions(logger=None)

Bases: object

Decorator to catch all exceptions and log them

log_helpers

class rafcon.utils.log_helpers.LoggingViewHandler

Bases: logging.Handler

A LoggingHandler for Gtk.TextViews

The LoggingViewHandler prints log messages in special Gtk.TextView`s that provide a `print_message method. The views must register themselves to the handler. There can be multiple views registered for one handler.

classmethod add_logging_view(name, text_view)
emit(record)

Logs a new record

If a logging view is given, it is used to log the new record to. The code is partially copied from the StreamHandler class.

Parameters

record –

Returns

classmethod remove_logging_view(name)
class rafcon.utils.log_helpers.NoHigherLevelFilter(level)

Bases: logging.Filter

Filter high log levels

A logging filter that filters out all logging records, whose level are smaller than the level specified in the constructor.

Variables

level – the highest level the filter will pass

filter(record)

Filter high log levels

Filters all records, whose logging level is smaller than the level specified in the constructor

Parameters

record –

Returns

multi_event

rafcon.utils.multi_event.create(*events)

Creates a new multi_event

The multi_event listens to all events passed in the “events” parameter.

Parameters

events – a list of threading.Events

Returns

The multi_event

Return type

threading.Event

rafcon.utils.multi_event.or_clear(self)

A function to overwrite the default clear function of the threading.Event

Parameters

self – Reference to the event

rafcon.utils.multi_event.or_set(self)

A function to overwrite the default set function of threading.Events

Parameters

self – Reference to the event

rafcon.utils.multi_event.orify(e, changed_callback)

Add another event to the multi_event

Parameters

  • e – the event to be added to the multi_event

  • changed_callback – a method to call if the event status changes, this method has access to the multi_event

Returns

plugins (utils)

rafcon.utils.plugins.load_plugins()

Loads all plugins specified in the RAFCON_PLUGIN_PATH environment variable

rafcon.utils.plugins.run_hook(hook_name, *args, **kwargs)

Runs the passed hook on all registered plugins

The function checks, whether the hook is available in the plugin.

Parameters

  • hook_name – Name of the hook, corresponds to the function name being called

  • args – Arguments

  • kwargs – Keyword arguments

rafcon.utils.plugins.run_on_state_machine_execution_finished()

Runs the on_state_machine_execution_finished methods of all registered plugins

rafcon.utils.plugins.run_post_inits(setup_config)

Runs the post_init methods of all registered plugins

Parameters

setup_config –

rafcon.utils.plugins.run_pre_inits()

Runs the pre_init methods of all registered plugins

profiler

rafcon.utils.profiler.start(name)
rafcon.utils.profiler.stop(name, result_path='/tmp/rafcon-docs/2410', view=False)

resources

rafcon.utils.resources.get_data_file_path(*rel_path)

Get a file installed as data_file (located in share folder)

rafcon.utils.resources.get_repository_share_path()

Get the share folder from the repository

If started from repository, the path to the share folder within the repository is returned, otherwise, None.

rafcon.utils.resources.search_in_share_folders(*rel_path)

Search all possible share folders for a given resource

storage_utils

rafcon.utils.storage_utils.get_current_time_string()
rafcon.utils.storage_utils.load_dict_from_yaml(path)

Loads a dictionary from a yaml file :param path: the absolute path of the target yaml file :return:

rafcon.utils.storage_utils.load_objects_from_json(path, as_dict=False)

Loads a dictionary from a json file.

Parameters

path – The relative path of the json file.

Returns

The dictionary specified in the json file

rafcon.utils.storage_utils.write_dict_to_json(dictionary, path, **kwargs)

Write a dictionary to a json file. :param path: The relative path to save the dictionary to :param dictionary: The dictionary to get saved :param kwargs: optional additional parameters for dumper

timer

class rafcon.utils.timer.Timer(logger, name='', continues=True)

Bases: object

property duration
reset()
start()
stop(key)
rafcon.utils.timer.measure_time(func)

type_helpers

rafcon.utils.type_helpers.convert_string_to_type(string_value)

Converts a string into a type or class

Parameters

string_value – the string to be converted, e.g. “int”

Returns

The type derived from string_value, e.g. int

rafcon.utils.type_helpers.convert_string_value_to_type_value(string_value, data_type)

Helper function to convert a given string to a given data type

Parameters

  • string_value (str) – the string to convert

  • data_type (type) – the target data type

Returns

the converted value

rafcon.utils.type_helpers.type_inherits_of_type(inheriting_type, base_type)

Checks whether inheriting_type inherits from base_type

Parameters

  • inheriting_type (str) –

  • base_type (str) –

Returns

True is base_type is base of inheriting_type

vividict

class rafcon.utils.vividict.Vividict(dictionary=None)

Bases: dict, yaml.YAMLObject, jsonconversion.jsonobject.JSONObject

Extended dictionary for arbitrary nested access

A class which inherits from dict and can store an element for an arbitrary nested key. The single elements of the key do not have to exist beforehand.

classmethod from_dict(dictionary)

Creates a Vividict from a (possibly nested) python dict

Parameters

dictionary (dict) – Python dict to be converted to a Vividict

Returns

A Vividict with all key-values pairs from the given dict

Return type

Vividict

set_dict(new_dict)

Sets the dictionary of the Vividict

The method is able to handle nested dictionaries, by calling the method recursively.

Parameters

new_dict – The dict that will be added to the own dict

to_dict(*args, **kwargs)

Converts the Vividict to a common Python dict

Returns

A Python dict with all key-values pairs from this Vividict

Return type

dict

static vividict_to_dict(vividict, native_strings=False)

Helper method to create Python dicts from arbitrary Vividict objects

Parameters

vividict (Vividict) – A Vividict to be converted

Returns

A Python dict

Return type

dict