dal.helpers package

Subpackages

dal.helpers.flow package

Submodules

dal.helpers.cache module

# Size of each object in bytes, calculated using pympler asizeof

Type | Models |

│ Flow | 19.36 MB | │ Configuration | 21.86 MB | │ Ports | 23.18 MB | | Layout | 23.04 MB | | Annotation | 23.04 MB | | GraphicScene | 22.7 MB | ———————————

class dal.helpers.cache.ThreadSafeCache(maxsize=100, ttl=10800)

Bases: object

cache: TTLCache

get(key, default=None)

dal.helpers.helpers module

Developers: - Manuel Silva (manuel.silva@mov.ai) - 2020 - Tiago Paulino (tiago@mov.ai) - 2020

class dal.helpers.helpers.Helpers

Bases: object

Helper Methods

static find_by_key(data, target)

static get_args(prev_struct) → dict

static get_partial_dict(_input: dict, key: dict) → dict: Giving a full dict and a beggining return the full dict containing that beggining

static join_first(add_struct: dict, prev_struct: dict) → dict

static replace_dict_values(haystack: dict, needle: any, value: any) → None

static update_dict(_base_dict: dict, _add_dict: dict) → dict: Join a dictionary with another dictionary

dal.helpers.helpers.flatten(data: dict, output: list, path: str, join_char: str = '.') → list

Returns a list of paths

Parameters:

data (dict) – expects a dictionary to start
output (list) – variable to save the list of paths
path (str) – accumulative path
join_char (str) – character to join path entries

Returns:

a list of paths

Return type:

output (list)

dal.helpers.parsers module

Developers: - Manuel Silva (manuel.silva@mov.ai) - 2020

class dal.helpers.parsers.ObjectWithName(*args, **kwargs)

Bases: Protocol

property name: str

class dal.helpers.parsers.ParamParser(flow: Flow)

Bases: object

Parser for the node instance, container and flow parameters Supports configuration. parameters, var, flow and env variables

eval_config(_config: str, *__)

Returns the config expression evaluated: $(<contex> <configuration name>.<parameter reference>) ex.: $(config name.var1.var2)

Parameters:: _config (str) – <configuration name>.<parameter reference>
Returns:: the expression evaluated
Return type:: output (any)

eval_flow(param_name: str, default: str, instance: NodeInst | Container, node_name: str) → any

Returns the flow expression evaluated

ex.: $(flow myvar)

Parameters:: param_name (str): reference to a parameter default (str): default value with parsing instance (NodeInst || Container): an instance node_name (str): node instance name (may be in the context of a subflow)
Returns:: output (any): the expression evaluated

eval_param(param_name: str, default: str, instance: Flow | NodeInst | Container, node_name: str) → any

Returns the param expression evaluated or default: ex.: $(param name)

Parameters:

param_name (str) – reference to a parameter
default (str) – default value with parsing
instance (NodeInst || Container) – an instance
node_name (str) – node instance name (may be in the context of a subflow)

Returns:

the value of the parameter or the default

Return type:

output (any)

eval_reference(key: str, expression: str, instance: ObjectWithName, node_name: str) → str

Calls a specific function to evaluate the expression

Parameters:

key (str) – name of the requested parameter
expression (str) – the expression to be evaluated format: $(context reference)
instance (NodeInst || Container) – an instance
node_name (str) – node instance name (may be in the context of a subflow)

Returns:

the parameter value after evaluation

Return type:

output (str)

eval_var(reference: str, *__) → Any

Returns the var expression evaluated

ex.: $(var robot.name)

Parameters:

reference (str): reference to: a parameter <fleet or robot>.<parameter reference>

Returns:

output (any): the expression evaluated

logger = <Logger ParamParser.mov.ai (INFO)>

parse(key: str, expression: str, node_name: str, instance: ObjectWithName, context: str | None = None) → Any

Returns the parameter value. If the value is a valid expression, it is evaluated.

Parameters:

key (str) – name of the requested parameter
expression (str) – the expression to be evaluated format: $(context reference)
node_name (str) – the node name
instance (NodeInst || Container) – an instance
context (str) – the context of the evaluation (main flow)

Returns:

the parameter value after evaluation

Return type:

output (str)

dal.helpers.parsers.get_string_from_template(template: str, task_entry: object) → str: Applies a task entry into a template

Module contents

Developers: - Moawiya Mograbi (moawiya@mov.ai) - 2022

class dal.helpers.GFlow(flow)

Bases: object

Graph generator in use to calculate remaps

add_edge(key: str, _id: str, _type: str) → dict: Add a new edge to the vertex

calc_remaps() → dict: Returns the remaps after calculation

check_forced_links(port_type: str, link: Template) → str

check_ros_port(node_name: str, port_name: str)

function that verifies if a specific port name has a ROS message associated

Args:: node_name (str): simplified node name (e.g. publisher2) port_name (str): full port name (e.g. publisher2/bool_out/out)
Returns:: bool: True if port name has a ROS message associateds. False otherwise

forced_remap(link: Template) → str: Special remap rules

generate_graph() → dict: Generate the graph

generate_remaps() → dict: Generates the remaps

get_adjacent_unremappable_port(port_type, links: list)

get_port_template(node_inst: str, port_name: str) → str: Get the port template from the node template

get_remaps(force: bool = False) → dict: Returns the calculated remaps

get_vertex(key: str, _type: str = 'From') → dict: Get or create a new vertex

iterate_through_links(port_type, links: list) → list

logger = <Logger gflow.mov.ai (INFO)>

reverse_type(port_type: str) → str

set_remap(key: str, remap: str): Set the vertex remap value

should_skip(link: Template) → bool: Special rules for including vertex in the graph

sort_graph() → list: Sort the graph by the number of elements in links

class dal.helpers.Helpers

Bases: object

Helper Methods

static find_by_key(data, target)

static get_args(prev_struct) → dict

static get_partial_dict(_input: dict, key: dict) → dict: Giving a full dict and a beggining return the full dict containing that beggining

static join_first(add_struct: dict, prev_struct: dict) → dict

static replace_dict_values(haystack: dict, needle: any, value: any) → None

static update_dict(_base_dict: dict, _add_dict: dict) → dict: Join a dictionary with another dictionary

dal.helpers.flatten(data: dict, output: list, path: str, join_char: str = '.') → list

Returns a list of paths

Parameters:

data (dict) – expects a dictionary to start
output (list) – variable to save the list of paths
path (str) – accumulative path
join_char (str) – character to join path entries

Returns:

a list of paths

Return type:

output (list)