dal.scopes package

Submodules

dal.scopes.alert module

class dal.scopes.alert.Alert(name, version='latest', new=False, db='global')

Bases: Scope

activate(**kwargs)

Activate the alert, adding it to the active alerts list if validation passes.

Parameters:

**kwargs – Parameters to fill placeholders in Title, Info, and Action fields.

alert_metrics = None
classmethod clear_alerts(deactivation_type: str = 'requested')

Clear all active alerts.

Parameters:

deactivation_type (str, optional) – The type of deactivation. Defaults to DeactivationType.REQUESTED.

deactivate(deactivation_type: str = 'requested')

Deactivate the alert, removing it from the active alerts list.

Parameters:

deactivation_type (str, optional) – The type of deactivation. Defaults to DeactivationType.REQUESTED.

classmethod get_active() → List[str]

Get a list of active alert IDs.

Returns:

List of active alert IDs.

Return type:

List[str]

classmethod get_alert_metrics_handler()
classmethod is_active(alert_id: str) → bool

Check if a specific alert is active.

Parameters:

alert_id (str) – The ID of the alert to check.

Returns:

True if the alert is active, False otherwise.

Return type:

bool

scope = 'Alert'
validate_parameters(name: str, text: str, **kwargs)

Validate that all required placeholders in the text are provided in kwargs.

Parameters:

  • name (str) – The name of the field being validated (e.g., “Title”).

  • text (str) – The text containing placeholders to validate.

  • **kwargs – Parameters to fill the placeholders in the text.

dal.scopes.application module

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

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

class dal.scopes.application.Application(name, version='latest', new=False, db='global')

Bases: Scope

Application model

permissions = ['create', 'read', 'update', 'delete', 'execute']
scope = 'Application'

dal.scopes.callback module

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

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

Module that implements a Callback scope class https://github.com/python/cpython/blob/master/Lib/pydoc.py

class dal.scopes.callback.Callback(name, version='latest', new=False, db='global')

Bases: Scope

Callback class

static export_modules()

Get modules and save them to Layer1

this takes about 4 seconds to run

and prints a LOT of stuff to the terminal

currently (14-10 on spawner) uses about 19kb of memory (?)

static fetch_modules_api()

Retrieve saved modules from Layer1, called from REST API

static get_full_modules() → dict
static get_methods(module: str) → dict
static get_modules() → list
is_valid()
permissions = ['create', 'read', 'update', 'delete', 'execute']
remove(force=False)

Removes Scope

scope = 'Callback'
template_depends(force=False)
static user_can_execute(user, callback_name: str = '') → bool

Check if user has permission to execute a callback.

Parameters:

  • user (BaseUser) – The user object to check permissions for.

  • callback_name (str) – Optional specific callback name to check

Returns:

True if user can execute the callback, False otherwise

Return type:

bool

Note

TODO Remove after migration to endpoints - this exists because frontend apps execute callbacks directly.

dal.scopes.configuration module

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

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

Module that implements Configuration scope class

class dal.scopes.configuration.Config(*args: Any, default_box: bool = False, default_box_attr: Any | None = None, default_box_none_transform: bool = True, frozen_box: bool = False, camel_killer_box: bool = False, conversion_box: bool = True, modify_tuples_box: bool = False, box_safe_prefix: str = 'x', box_duplicates: str = 'ignore', box_intact_types: Tuple | List = (), box_recast: Dict | None = None, box_dots: bool = False, **kwargs: Any)

Bases: Box

Config with dot accessible elements

class dal.scopes.configuration.Configuration(name, version='latest', new=False, db='global')

Bases: Scope

Configuration class

get_param(param: str)

Returns the configuration value of a key in the format param.subparam.subsubparam

get_value() → dict

Returns a dictionary with the configuration values

scope = 'Configuration'

dal.scopes.fleetrobot module

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

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

Module that implements Robot namespace

class dal.scopes.fleetrobot.FleetRobot(name: str, version='latest', new=False, db='global')

Bases: Scope

Represent the Robot scope in the redis-master.

Parameter: Dict
async async_send_cmd(command: str, *, flow: str | None = None, node: str | None = None, port=None, data=None, response_required=False) → None

Send an action command to the Robot.

See flow-initiator/flow_initiator/spawner/spawner.py for possible commands.

async_spawner_client: AsyncMessageClient
classmethod get_manager() → str | None

Get the manager id.

Returns:

The manager robot id or None if not found.

Return type:

Optional[str]

classmethod get_members() → List[str]

Get a list with the members ids.

Returns:

List of member robot ids.

Return type:

List[str]

static get_robot_key_by_ip(ip_address: str, key_name: str) → bytes | None

Finds a key of a robot by the ip address.

Parameters:

  • ip_address (str) – The ip address of the desired robot.

  • key_name (str) – The name of required key.

Returns:

The public key.

Return type:

bytes

is_manager() → bool

Check if the Robot is a manager

Returns:

True if the Robot is a manager, False otherwise.

Return type:

bool

static list_all() → List[str]

List all the robots in the fleet

static name_to_id(robot_name: str) → str

Get the robot id by its name.

Parameters:

robot_name – Robot name.

Returns:

Robot id.

ping() → bool

Ping the robot

classmethod remove_entry(robot_id: str, force: bool = False)

Remove the robot from the registry

robot_id: str

the id of the robot to remove

force: bool

if True, the robot will be removed without checking if it is running if not, an exception will be raised if the robot is running

send_cmd(command: str, *, flow: str | None = None, node: str | None = None, port=None, data=None) → None

Send an action command to the Robot.

See flow-initiator/flow_initiator/spawner/spawner.py for possible commands.

set_robot_started(value: bool)
spawner_client: MessageClient
class dal.scopes.fleetrobot.Role(value)

Bases: Enum

Robot Role Enum.

MANAGER = 'manager'
MEMBER = 'member'

dal.scopes.flow module

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

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

class dal.scopes.flow.Flow(name, version='latest', new=False, db='global')

Bases: Scope

Flow class

Verifies if the links already exists if not add it to the Links hash

calc_remaps(new_link: dict | None = None) → dict

Calculate remaps data: {“port_name”: { “remap”: None, “links”: [“link_name”,…], “Type”: “From” or “To”}}

copy_node(copy_name: str, org_name: str, org_flow: str, org_type: str = 'NodeInst', options: dict | None = None)

copy a node instance

delete(key: str, name: str)

Delete object dependencies

delete_exposed_port(node_id: str, port_name: str = '*', node_inst_id: str | None = None) → bool

Delete exposed port :param node_id: Node template name :param port_name: port name to delete or * to delete all :param node_inst_id: delete only ports related w/ specific node instance

Info:
ExposedPorts is of type hash

  • key (str) is the name of the Node (Template)

  • value (dict) format is {“<NodeInst>”: [“Ports”]}

Returns:

True for success, False otherwise.

Find all links in all Flows with containers and delete links to the exposed port Also deletes exposed ports

Parameters:

  • node_inst_name (str) – The name of the node instance.

  • port_name (str) – The unexposed port name.

  • join_char (str) – ‘/’ when starting from a node instance or ‘__’ when starting from a subflow

  • prefix (str) – Node instance name when going to upper flows.

  • flows (dict) – List of flows with subflows.

Returns:

True for success, False otherwise.

Return type:

bool

Delete link

eval_config(_config: str, *__) → any
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(reference: str, param: str, node_name: str, full_params: dict, param_name: str, *__) → any

Evaluate the flow expression ex.: $(flow myvar)

Parameters:
reference (str): reference to a parameter

ex.: “var_A”

param (str): parameter expression $(<context> <parameter reference>)

ex.: $(flow var_A)

node_name (str): the name of the name full_params (dict): dictionary with all the parameters param_name (str): the parameter name

Returns:

output (any): the parameter value after evaluation

eval_param(param_ref: str, param: str, node_name: str, full_params: dict, *__) → any
Returns the param expression evaluated

ex.: $(param name)

Parameters:

  • param_ref (str) – reference to a parameter

  • param (str) – parameter expression $(<context> <parameter reference>) ex.: $(param var_A)

  • node_name (str) – node name (ignored)

  • full_params (dict) – dictionary with all the parameters

Returns:

the expression evaluated

Return type:

output (any)

eval_reference(param_value: str, node_name: str, full_params: dict, param_name: str)

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

Parameters:

  • param_value (str) – the parameter value/expression $(context reference)

  • node_name (str) – the node instance name

  • full_params (dict) – dictionary with all the node instance parameters

  • param_name (str) – the name of the parameter

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

expand_container(label: str, flow: str, prev_flows: list) → dict

Returns the Links and Nodes in a Flow container with namespace

static exposed_ports_diff(previous_ports: dict, current_ports: dict) → list

Returns deleted exposed ports by subtracting ports in current_ports from the previous_ports

Parameters:

  • previous_ports (dict) – dictionary with previous ports

  • current_ports (dict) – dictionary with current ports

Returns:

list of dictionaries with the ports to delete

Return type:

(list)

previous_ports and current_ports format:
{
<node_template_or_flow_name>:
{

<node_instance_name>: [<port_name1>, …, <port_nameN>]

}

}

get_container_params(node_name: str) → dict

Returns the container parameters after evaluating them

Parameters:

node_name (str): the container name

Returns:

params (dict): the container parameters after evaluating them

get_dict(prev_flows=None, recursive=False)

Get full structure of flow including containers expanded

get_lifecycle_nodes() → list

List of Ros2 Lifecycle Nodes

get_node(node_inst_name: str) → dict

Return the node dict from cache or from DB

get_node_attributes(node_name: str, attr: str, only_values: bool = False) → dict

“Generic method to get node attributes

get_node_dependencies(node_name: str, dependencies_collected: list | None = None, links_to_skip: list | None = None, skip_parent_node: str | None = None, first_level_only: bool = False) → list

Get node dependencies recursively

get_node_inst_params(node_name: str, attribute='Parameter') → dict

Returns the parameters changed in the node intance

get_node_namespace(node_name: str) → str

Return node namespace, ‘’ if not defined

get_node_params(node_name: str) → dict

Returns the final parameters of the node instance

get_node_path(node_name: str) → str

Return node type

get_node_persistent(node_name: str) → bool

Return node persistance

get_node_plugins(node_name: str) → list

Return NodeInst(s) of type ROS1_PLUGIN linked to node_name

get_node_port(node_name: str, port_inst: str) → Ports

Deprecated, still here because of tests/data/test_Callback_data.py

get_node_port_template(node_name: str, port_inst: str) → str

Returns the name of the template for that specific portsinst

get_node_ports_gen(node_name: str, port: str | None = None)

[Generator] Returns ports node links ‘port’ must be PortInst/port_name

get_node_template_name(node_inst_name: str) → str

Returns the template of a node inst

get_node_transitions(node_name: str, port: str | None = None) → list

Returns node to transit to

get_node_type(node_name: str) → str

Return node type

get_nodelet_manager(node_name: str, port: str | None = None) → str

Returns nodelet manager name

get_port(port_template: str) → dict

Return ports dict from cache or from DB

get_start_nodes() → list

List of nodes with a start port

is_node_dummy(node_name: str) → bool

Return node dummy

is_node_remmapable(node_name: str) → bool

Return node remappable, True if not defined

is_node_to_launch(node_name: str) → bool

Return node launch, True if not defined

is_valid(new_link: dict | None = None) → tuple

Calculate remaps and validate links

static on_flow_delete(flow_id: str)

Actions to do when a flow is deleted - Delete all nodes of type Container with ContainerFlow = flow_id

parse_param(param: str, full_params: dict, node_name: str, key: str)

Check if the parameter has a reference to a config, param or env var

scope = 'Flow'

dal.scopes.form module

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

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

class dal.scopes.form.Form(name, version='latest', new=False, db='global')

Bases: Scope

Form model

scope = 'Form'

dal.scopes.message module

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

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

Module that implements a Message scope class For Reference: https://github.com/ros/ros_comm/blob/6a0672c0189024e38bfe51d917dea148f76fccf7/tools/rosmsg/src/rosmsg/__init__.py

class dal.scopes.message.Message(name, version='latest', new=False, db='global')

Bases: Scope

Message class

classmethod export_portdata(db='global') → dict

Gives all messages/services available organized by type. Possibly intended to replace get_all(), but kept for legacy reasons

static fetch_portdata_api() → dict

Retrieve data from database -> Called from a cloud function

static fetch_portdata_messages() → dict

Retrieve all the messages and services available for callbacks

classmethod get_all(db='global') → dict

Gives the all available messages of all types organized by package

classmethod get_msgs(package: str, msg_type='all', db='global') → list

Gives a list of all messages of some type in a package

classmethod get_packages(msg_type='all', db='global') → list

Gives a list of all packages containing messages of a type

static get_structure(message: str) → dict

Gives the full structure of a message given the ‘package/message’ input

is_valid()
scope = 'Message'
dal.scopes.message.iterate_message(context, spec, struct)

dal.scopes.node module

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

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

Module that implements a Node scope class

class dal.scopes.node.Node(name, version='latest', new=False, db='global')

Bases: Scope

Node class

delete(key, name, force=False)
get_params(attribute='Parameter')
is_persistent()
is_port_remappable(portinst)
is_state()
is_valid()
node_inst_depends() → list

Search Flows for NodeInstances

port_inst_depends(port_name: str) → list

Loop through NodeInst’s Links and return list with matching links dict_keys

remove(force=False) → bool

Removes Scope

rename(key, old_name, new_name)
scope = 'Node'
set_type()
template_depends(force=False)

dal.scopes.package module

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

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

Module for the Package namespace

class dal.scopes.package.FileStruct(name, struct_dict, prev_struct, db)

Bases: Struct

Helper class to provide callbacks on attributes set

Name: str
db: str
movaidb: MovaiDB
set_checksum(name, value)

Callback to calculate and save the checksum

class dal.scopes.package.Package(name, version='latest', new=False, db='global')

Bases: Scope

Package class deals with packages/files uploaded to the db

add(key, name, **kwargs)
classmethod dump(package, file_name, path_to)

Dump redis file to disk

dump_file(file_name, path_to)

Dump a file from redis. Uses the checksum to validate the dump.

file_exists(file_name, path_to)

check if file already exists and checksum is valid

get_checksum(file_name)

get the file checksum from the db

get_file_checksum(file_path)

Opens a file and calculates the checksum

static get_or_create(package_name)
is_checksum_valid(file_name, checksum)

Compares db checksum with file checksum

scope = 'Package'

dal.scopes.ports module

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

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

class dal.scopes.ports.Ports(name, version='latest', new=False, db='global')

Bases: Scope

Ports model

is_nodelet_client() → bool

Returns True if the port is a nodelet client

is_nodelet_server() → bool

Returns True if the port is a nodelet server

is_transition(port_type: str, port_name: str) → bool

Check if a port is of type transition

scope = 'Ports'

dal.scopes.robot module

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

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

Module that implements Robot namespace

class dal.scopes.robot.Robot

Bases: Scope

Robot class that deals with robot related stuff

You should also read the page Robot Parameters for a deep dive on the Parameters feature.

add_active_alert(alert_id: str, data: AlertActivationData)

Add an active alert to the Robot

async async_send_cmd(command, *, flow=None, node=None, port=None, data=None, wait_for_status=False) → dict | None

Send an action command to the Robot

if wait_for_status is True, we assume the Robot will return a message

async_spawner_client: AsyncMessageClient
clear_alerts(deactivation_type: str = 'requested') → List[AlertData]

Clear all active alerts from the Robot

classmethod cls_update_status(name: str, status: dict, db: str = 'all')

Class method to update the Robot status in the database This method reduces readings from the database compared to using Robot instance

get_states()

Gets the states of the robot from its own configuration. When Robot groups are implemented it should merge with the group configuration

pop_alert(alert_id: str, deactivation_type: str = 'requested') → AlertData | None

Remove an active alert from the Robot

scope = 'Robot'
send_cmd(command, *, flow=None, node=None, port=None, data=None) → None

Send an action command to the Robot

if wait_for_status is True, we assume the Robot will return a message

set_ip(ip_address: str)

Set the IP Adress of the Robot

set_model(model: str)

Set the Model of the Robot

set_name(name: str)

Set the Name of the Robot

set_param(param: str, value, db: str = 'all')

Sets or updates a parameter of the robots

set_role(role: Role)

Set the Role of the Robot

set_type(rType: str)

Set the Type of the Robot

spawner_client: MessageClient
update_status(status: dict, db: str = 'all')

Update the Robot status in the database

dal.scopes.role module

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

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

class dal.scopes.role.Role(name, version='latest', new=False, db='global')

Bases: Scope

classmethod create(*, name: str, resources: dict) → bool

Create a new role

create_permission(resource: str, permission: str) → bool

Create new role permission

delete_permission(resource: str, permission: str) → bool

Delete role permission

scope = 'Role'
classmethod update(*, name: str, resources: dict) → bool

Reset role data

dal.scopes.scope module

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

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

dal.scopes.scope.SCOPES_TO_VALIDATE

List of scopes that will be validated before writing into redis.

Type:

List[str]

class dal.scopes.scope.Scope(scope, name, version, new=False, db='global')

Bases: Struct

Scope main class.

validator

Validator for the scope.

Type:

JsonValidator

calc_scope_update(old_dict: dict, new_dict: dict)

Calc the objects differences and returns list with dict keys to delete/set.

Parameters:

  • old_dict (dict) – Old scope dictionary.

  • new_dict (dict) – New scope dictionary.

Raises:

  • SchemaTypeNotKnown – If the scope is not known to the validator.

  • ValueError – If the data does not conform to the schema.

classmethod get_all(db='global')
get_dict()

Returns the full dictionary of the scope from db

get_value(key: str, default: any = False) → any
has_scope_permission(user, permission) → bool
permissions = ['create', 'read', 'update', 'delete']
remove(force=True)

Removes Scope

remove_partial(dict_key)

Remove Scope key

transform_before_update(source_data: dict)

Transforms data into internal format

Parameters:

source_data (dict) – Data to transform.

Default: do nothing.

classmethod validate_format(scope, data: dict)

Check if the data is in a valid format for this scope.

Raises:

  • SchemaTypeNotKnown – If the scope is not known to the validator.

  • ValueError – If the data does not conform to the schema.

validator: Validator = <dal.validation.validator.JsonValidator object>

dal.scopes.statemachine module

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

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

Module to work with State Machines in GD_Nodes

class dal.scopes.statemachine.SMVars(_sm_name, _node_name='')

Bases: object

Class for user to write and read vars into a state machine

delete(name)

Same as delattr

get(name)

Same as getattr

get_dict()
set(name, value)

Same as setattr

class dal.scopes.statemachine.StateMachine(name, version='latest', new=False, db='global')

Bases: Scope

StateMachine class

Name: str

verifies if the links already exists if not add it to the Links hash

copy_node(copy_name: str, org_name: str, org_flow: str, org_type: str = 'State', options: dict | None = None)

copy a node instance

db: str
delete(key, name)

Delete object dependencies

delete link

is_valid()
movaidb: MovaiDB
scope = 'StateMachine'

dal.scopes.structures module

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

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

class dal.scopes.structures.Hash(name: str, init_value: dict, db: str, prev_struct: str)

Bases: dict

Custom dict that overrides methods

clear() → None.  Remove all items from D.
delete(var: str)

Deletes a hash field and returns True if successfully deleted

get(var: str, default=None)

Gets a hash field and returns it

pop(var: str)

Deletes a hash field and returns it

update(value: dict)

Updates the hash with desired dict

class dal.scopes.structures.List(name: str, init_value: list, db: str, prev_struct: str)

Bases: list

Custom list that overrides pop and append methods

append(value)

Append both to python list and redis list

pop()

Pop from python list and redis list

class dal.scopes.structures.Struct(name, struct_dict, prev_struct, db)

Bases: object

General structure… how to describe?

Name: str
add(key, name, **kwargs)
db: str
delete(key, name)
get_attributes(d: dict)
get_dict()
get_ref(value: str)

Receives a value and returns the value with refs if they exist

movaidb: MovaiDB
rename(key: str, old_name: str, new_name: str) → bool

dal.scopes.system module

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

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

class dal.scopes.system.System(name, version='latest', new=False, db='global')

Bases: Scope

System model

Name: str
db: str
movaidb: MovaiDB
scope = 'System'

dal.scopes.translation module

class dal.scopes.translation.Translation(name, version='latest', new=False, db='global')

Bases: Scope

Translation scope.

Name: str
db: str
movaidb: MovaiDB
scope = 'Translation'
validator: Validator = <dal.validation.validator.TranslationValidator object>

dal.scopes.user module

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

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

class dal.scopes.user.User(name, version='latest', new=False, db='global')

Bases: Scope

User class

Name: str
classmethod create(username: str, password: str) → bool

Create a new user

Parameters:

  • username (str) –

  • password (str) –

Returns:

obj (User)

create_id()

Create

db: str
get_permissions(resource_name: str) → list

Returns list of user permissions

get_refresh_token()

Generate refresh token

get_supports()
get_token()

Generate authentication token

has_permission(resource_name: str, permission_name: str, skip_superuser: bool = False) → bool

Returns permission check

hash_password(password)

Hash a password for storing

movaidb: MovaiDB
classmethod reset(*, username: str, current_pass: str | None = None, new_pass: str, confirm_pass: str, validate_current_pass: bool = True) → bool

Reset user password

Parameters:

  • current_pass (-) – old password

  • new_pass (-) – new password

  • confirm_pass (-) – confirm password

  • validate_current_password (-) – confirm current password before setting the new password

scope = 'User'
user_permissions() → dict
verify_password(provided_password)

Verify a stored password against one provided by user

classmethod verify_token(token)

Validate if token is valid

dal.scopes.widget module

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

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

class dal.scopes.widget.Widget(name, version='latest', new=False, db='global')

Bases: Scope

Layout model

Name: str
db: str
movaidb: MovaiDB
scope = 'Widget'

Module contents

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

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

class dal.scopes.Alert(name, version='latest', new=False, db='global')

Bases: Scope

Name: str
activate(**kwargs)

Activate the alert, adding it to the active alerts list if validation passes.

Parameters:

**kwargs – Parameters to fill placeholders in Title, Info, and Action fields.

alert_metrics = None
classmethod clear_alerts(deactivation_type: str = 'requested')

Clear all active alerts.

Parameters:

deactivation_type (str, optional) – The type of deactivation. Defaults to DeactivationType.REQUESTED.

db: str
deactivate(deactivation_type: str = 'requested')

Deactivate the alert, removing it from the active alerts list.

Parameters:

deactivation_type (str, optional) – The type of deactivation. Defaults to DeactivationType.REQUESTED.

classmethod get_active() → List[str]

Get a list of active alert IDs.

Returns:

List of active alert IDs.

Return type:

List[str]

classmethod get_alert_metrics_handler()
classmethod is_active(alert_id: str) → bool

Check if a specific alert is active.

Parameters:

alert_id (str) – The ID of the alert to check.

Returns:

True if the alert is active, False otherwise.

Return type:

bool

movaidb: MovaiDB
scope = 'Alert'
validate_parameters(name: str, text: str, **kwargs)

Validate that all required placeholders in the text are provided in kwargs.

Parameters:

  • name (str) – The name of the field being validated (e.g., “Title”).

  • text (str) – The text containing placeholders to validate.

  • **kwargs – Parameters to fill the placeholders in the text.

class dal.scopes.Application(name, version='latest', new=False, db='global')

Bases: Scope

Application model

Name: str
db: str
movaidb: MovaiDB
permissions = ['create', 'read', 'update', 'delete', 'execute']
scope = 'Application'
class dal.scopes.Callback(name, version='latest', new=False, db='global')

Bases: Scope

Callback class

Name: str
db: str
static export_modules()

Get modules and save them to Layer1

this takes about 4 seconds to run

and prints a LOT of stuff to the terminal

currently (14-10 on spawner) uses about 19kb of memory (?)

static fetch_modules_api()

Retrieve saved modules from Layer1, called from REST API

static get_full_modules() → dict
static get_methods(module: str) → dict
static get_modules() → list
is_valid()
movaidb: MovaiDB
permissions = ['create', 'read', 'update', 'delete', 'execute']
remove(force=False)

Removes Scope

scope = 'Callback'
template_depends(force=False)
static user_can_execute(user, callback_name: str = '') → bool

Check if user has permission to execute a callback.

Parameters:

  • user (BaseUser) – The user object to check permissions for.

  • callback_name (str) – Optional specific callback name to check

Returns:

True if user can execute the callback, False otherwise

Return type:

bool

Note

TODO Remove after migration to endpoints - this exists because frontend apps execute callbacks directly.

class dal.scopes.Config(*args: Any, default_box: bool = False, default_box_attr: Any | None = None, default_box_none_transform: bool = True, frozen_box: bool = False, camel_killer_box: bool = False, conversion_box: bool = True, modify_tuples_box: bool = False, box_safe_prefix: str = 'x', box_duplicates: str = 'ignore', box_intact_types: Tuple | List = (), box_recast: Dict | None = None, box_dots: bool = False, **kwargs: Any)

Bases: Box

Config with dot accessible elements

class dal.scopes.Configuration(name, version='latest', new=False, db='global')

Bases: Scope

Configuration class

Name: str
db: str
get_param(param: str)

Returns the configuration value of a key in the format param.subparam.subsubparam

get_value() → dict

Returns a dictionary with the configuration values

movaidb: MovaiDB
scope = 'Configuration'
class dal.scopes.FleetRobot(name: str, version='latest', new=False, db='global')

Bases: Scope

Represent the Robot scope in the redis-master.

Name: str
Parameter: Dict
async async_send_cmd(command: str, *, flow: str | None = None, node: str | None = None, port=None, data=None, response_required=False) → None

Send an action command to the Robot.

See flow-initiator/flow_initiator/spawner/spawner.py for possible commands.

async_spawner_client: AsyncMessageClient
db: str
classmethod get_manager() → str | None

Get the manager id.

Returns:

The manager robot id or None if not found.

Return type:

Optional[str]

classmethod get_members() → List[str]

Get a list with the members ids.

Returns:

List of member robot ids.

Return type:

List[str]

static get_robot_key_by_ip(ip_address: str, key_name: str) → bytes | None

Finds a key of a robot by the ip address.

Parameters:

  • ip_address (str) – The ip address of the desired robot.

  • key_name (str) – The name of required key.

Returns:

The public key.

Return type:

bytes

is_manager() → bool

Check if the Robot is a manager

Returns:

True if the Robot is a manager, False otherwise.

Return type:

bool

static list_all() → List[str]

List all the robots in the fleet

movaidb: MovaiDB
static name_to_id(robot_name: str) → str

Get the robot id by its name.

Parameters:

robot_name – Robot name.

Returns:

Robot id.

ping() → bool

Ping the robot

classmethod remove_entry(robot_id: str, force: bool = False)

Remove the robot from the registry

robot_id: str

the id of the robot to remove

force: bool

if True, the robot will be removed without checking if it is running if not, an exception will be raised if the robot is running

send_cmd(command: str, *, flow: str | None = None, node: str | None = None, port=None, data=None) → None

Send an action command to the Robot.

See flow-initiator/flow_initiator/spawner/spawner.py for possible commands.

set_robot_started(value: bool)
spawner_client: MessageClient
class dal.scopes.Flow(name, version='latest', new=False, db='global')

Bases: Scope

Flow class

Name: str

Verifies if the links already exists if not add it to the Links hash

calc_remaps(new_link: dict | None = None) → dict

Calculate remaps data: {“port_name”: { “remap”: None, “links”: [“link_name”,…], “Type”: “From” or “To”}}

copy_node(copy_name: str, org_name: str, org_flow: str, org_type: str = 'NodeInst', options: dict | None = None)

copy a node instance

db: str
delete(key: str, name: str)

Delete object dependencies

delete_exposed_port(node_id: str, port_name: str = '*', node_inst_id: str | None = None) → bool

Delete exposed port :param node_id: Node template name :param port_name: port name to delete or * to delete all :param node_inst_id: delete only ports related w/ specific node instance

Info:
ExposedPorts is of type hash

  • key (str) is the name of the Node (Template)

  • value (dict) format is {“<NodeInst>”: [“Ports”]}

Returns:

True for success, False otherwise.

Find all links in all Flows with containers and delete links to the exposed port Also deletes exposed ports

Parameters:

  • node_inst_name (str) – The name of the node instance.

  • port_name (str) – The unexposed port name.

  • join_char (str) – ‘/’ when starting from a node instance or ‘__’ when starting from a subflow

  • prefix (str) – Node instance name when going to upper flows.

  • flows (dict) – List of flows with subflows.

Returns:

True for success, False otherwise.

Return type:

bool

Delete link

eval_config(_config: str, *__) → any
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(reference: str, param: str, node_name: str, full_params: dict, param_name: str, *__) → any

Evaluate the flow expression ex.: $(flow myvar)

Parameters:
reference (str): reference to a parameter

ex.: “var_A”

param (str): parameter expression $(<context> <parameter reference>)

ex.: $(flow var_A)

node_name (str): the name of the name full_params (dict): dictionary with all the parameters param_name (str): the parameter name

Returns:

output (any): the parameter value after evaluation

eval_param(param_ref: str, param: str, node_name: str, full_params: dict, *__) → any
Returns the param expression evaluated

ex.: $(param name)

Parameters:

  • param_ref (str) – reference to a parameter

  • param (str) – parameter expression $(<context> <parameter reference>) ex.: $(param var_A)

  • node_name (str) – node name (ignored)

  • full_params (dict) – dictionary with all the parameters

Returns:

the expression evaluated

Return type:

output (any)

eval_reference(param_value: str, node_name: str, full_params: dict, param_name: str)

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

Parameters:

  • param_value (str) – the parameter value/expression $(context reference)

  • node_name (str) – the node instance name

  • full_params (dict) – dictionary with all the node instance parameters

  • param_name (str) – the name of the parameter

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

expand_container(label: str, flow: str, prev_flows: list) → dict

Returns the Links and Nodes in a Flow container with namespace

static exposed_ports_diff(previous_ports: dict, current_ports: dict) → list

Returns deleted exposed ports by subtracting ports in current_ports from the previous_ports

Parameters:

  • previous_ports (dict) – dictionary with previous ports

  • current_ports (dict) – dictionary with current ports

Returns:

list of dictionaries with the ports to delete

Return type:

(list)

previous_ports and current_ports format:
{
<node_template_or_flow_name>:
{

<node_instance_name>: [<port_name1>, …, <port_nameN>]

}

}

get_container_params(node_name: str) → dict

Returns the container parameters after evaluating them

Parameters:

node_name (str): the container name

Returns:

params (dict): the container parameters after evaluating them

get_dict(prev_flows=None, recursive=False)

Get full structure of flow including containers expanded

get_lifecycle_nodes() → list

List of Ros2 Lifecycle Nodes

get_node(node_inst_name: str) → dict

Return the node dict from cache or from DB

get_node_attributes(node_name: str, attr: str, only_values: bool = False) → dict

“Generic method to get node attributes

get_node_dependencies(node_name: str, dependencies_collected: list | None = None, links_to_skip: list | None = None, skip_parent_node: str | None = None, first_level_only: bool = False) → list

Get node dependencies recursively

get_node_inst_params(node_name: str, attribute='Parameter') → dict

Returns the parameters changed in the node intance

get_node_namespace(node_name: str) → str

Return node namespace, ‘’ if not defined

get_node_params(node_name: str) → dict

Returns the final parameters of the node instance

get_node_path(node_name: str) → str

Return node type

get_node_persistent(node_name: str) → bool

Return node persistance

get_node_plugins(node_name: str) → list

Return NodeInst(s) of type ROS1_PLUGIN linked to node_name

get_node_port(node_name: str, port_inst: str) → Ports

Deprecated, still here because of tests/data/test_Callback_data.py

get_node_port_template(node_name: str, port_inst: str) → str

Returns the name of the template for that specific portsinst

get_node_ports_gen(node_name: str, port: str | None = None)

[Generator] Returns ports node links ‘port’ must be PortInst/port_name

get_node_template_name(node_inst_name: str) → str

Returns the template of a node inst

get_node_transitions(node_name: str, port: str | None = None) → list

Returns node to transit to

get_node_type(node_name: str) → str

Return node type

get_nodelet_manager(node_name: str, port: str | None = None) → str

Returns nodelet manager name

get_port(port_template: str) → dict

Return ports dict from cache or from DB

get_start_nodes() → list

List of nodes with a start port

is_node_dummy(node_name: str) → bool

Return node dummy

is_node_remmapable(node_name: str) → bool

Return node remappable, True if not defined

is_node_to_launch(node_name: str) → bool

Return node launch, True if not defined

is_valid(new_link: dict | None = None) → tuple

Calculate remaps and validate links

movaidb: MovaiDB
static on_flow_delete(flow_id: str)

Actions to do when a flow is deleted - Delete all nodes of type Container with ContainerFlow = flow_id

parse_param(param: str, full_params: dict, node_name: str, key: str)

Check if the parameter has a reference to a config, param or env var

scope = 'Flow'
class dal.scopes.Form(name, version='latest', new=False, db='global')

Bases: Scope

Form model

Name: str
db: str
movaidb: MovaiDB
scope = 'Form'
class dal.scopes.Message(name, version='latest', new=False, db='global')

Bases: Scope

Message class

Name: str
db: str
classmethod export_portdata(db='global') → dict

Gives all messages/services available organized by type. Possibly intended to replace get_all(), but kept for legacy reasons

static fetch_portdata_api() → dict

Retrieve data from database -> Called from a cloud function

static fetch_portdata_messages() → dict

Retrieve all the messages and services available for callbacks

classmethod get_all(db='global') → dict

Gives the all available messages of all types organized by package

classmethod get_msgs(package: str, msg_type='all', db='global') → list

Gives a list of all messages of some type in a package

classmethod get_packages(msg_type='all', db='global') → list

Gives a list of all packages containing messages of a type

static get_structure(message: str) → dict

Gives the full structure of a message given the ‘package/message’ input

is_valid()
movaidb: MovaiDB
scope = 'Message'
class dal.scopes.Node(name, version='latest', new=False, db='global')

Bases: Scope

Node class

Name: str
db: str
delete(key, name, force=False)
get_params(attribute='Parameter')
is_persistent()
is_port_remappable(portinst)
is_state()
is_valid()
movaidb: MovaiDB
node_inst_depends() → list

Search Flows for NodeInstances

port_inst_depends(port_name: str) → list

Loop through NodeInst’s Links and return list with matching links dict_keys

remove(force=False) → bool

Removes Scope

rename(key, old_name, new_name)
scope = 'Node'
set_type()
template_depends(force=False)
class dal.scopes.Package(name, version='latest', new=False, db='global')

Bases: Scope

Package class deals with packages/files uploaded to the db

Name: str
add(key, name, **kwargs)
db: str
classmethod dump(package, file_name, path_to)

Dump redis file to disk

dump_file(file_name, path_to)

Dump a file from redis. Uses the checksum to validate the dump.

file_exists(file_name, path_to)

check if file already exists and checksum is valid

get_checksum(file_name)

get the file checksum from the db

get_file_checksum(file_path)

Opens a file and calculates the checksum

static get_or_create(package_name)
is_checksum_valid(file_name, checksum)

Compares db checksum with file checksum

movaidb: MovaiDB
scope = 'Package'
class dal.scopes.Ports(name, version='latest', new=False, db='global')

Bases: Scope

Ports model

Name: str
db: str
is_nodelet_client() → bool

Returns True if the port is a nodelet client

is_nodelet_server() → bool

Returns True if the port is a nodelet server

is_transition(port_type: str, port_name: str) → bool

Check if a port is of type transition

movaidb: MovaiDB
scope = 'Ports'
class dal.scopes.Robot

Bases: Scope

Robot class that deals with robot related stuff

You should also read the page Robot Parameters for a deep dive on the Parameters feature.

Name: str
add_active_alert(alert_id: str, data: AlertActivationData)

Add an active alert to the Robot

async async_send_cmd(command, *, flow=None, node=None, port=None, data=None, wait_for_status=False) → dict | None

Send an action command to the Robot

if wait_for_status is True, we assume the Robot will return a message

async_spawner_client: AsyncMessageClient
clear_alerts(deactivation_type: str = 'requested') → List[AlertData]

Clear all active alerts from the Robot

classmethod cls_update_status(name: str, status: dict, db: str = 'all')

Class method to update the Robot status in the database This method reduces readings from the database compared to using Robot instance

db: str
get_states()

Gets the states of the robot from its own configuration. When Robot groups are implemented it should merge with the group configuration

movaidb: MovaiDB
pop_alert(alert_id: str, deactivation_type: str = 'requested') → AlertData | None

Remove an active alert from the Robot

scope = 'Robot'
send_cmd(command, *, flow=None, node=None, port=None, data=None) → None

Send an action command to the Robot

if wait_for_status is True, we assume the Robot will return a message

set_ip(ip_address: str)

Set the IP Adress of the Robot

set_model(model: str)

Set the Model of the Robot

set_name(name: str)

Set the Name of the Robot

set_param(param: str, value, db: str = 'all')

Sets or updates a parameter of the robots

set_role(role: Role)

Set the Role of the Robot

set_type(rType: str)

Set the Type of the Robot

spawner_client: MessageClient
update_status(status: dict, db: str = 'all')

Update the Robot status in the database

class dal.scopes.Role(name, version='latest', new=False, db='global')

Bases: Scope

Name: str
classmethod create(*, name: str, resources: dict) → bool

Create a new role

create_permission(resource: str, permission: str) → bool

Create new role permission

db: str
delete_permission(resource: str, permission: str) → bool

Delete role permission

movaidb: MovaiDB
scope = 'Role'
classmethod update(*, name: str, resources: dict) → bool

Reset role data

class dal.scopes.SMVars(_sm_name, _node_name='')

Bases: object

Class for user to write and read vars into a state machine

delete(name)

Same as delattr

get(name)

Same as getattr

get_dict()
set(name, value)

Same as setattr

class dal.scopes.Scope(scope, name, version, new=False, db='global')

Bases: Struct

Scope main class.

validator

Validator for the scope.

Type:

JsonValidator

Name: str
calc_scope_update(old_dict: dict, new_dict: dict)

Calc the objects differences and returns list with dict keys to delete/set.

Parameters:

  • old_dict (dict) – Old scope dictionary.

  • new_dict (dict) – New scope dictionary.

Raises:

  • SchemaTypeNotKnown – If the scope is not known to the validator.

  • ValueError – If the data does not conform to the schema.

db: str
classmethod get_all(db='global')
get_dict()

Returns the full dictionary of the scope from db

get_value(key: str, default: any = False) → any
has_scope_permission(user, permission) → bool
movaidb: MovaiDB
permissions = ['create', 'read', 'update', 'delete']
remove(force=True)

Removes Scope

remove_partial(dict_key)

Remove Scope key

transform_before_update(source_data: dict)

Transforms data into internal format

Parameters:

source_data (dict) – Data to transform.

Default: do nothing.

classmethod validate_format(scope, data: dict)

Check if the data is in a valid format for this scope.

Raises:

  • SchemaTypeNotKnown – If the scope is not known to the validator.

  • ValueError – If the data does not conform to the schema.

validator: Validator = <dal.validation.validator.JsonValidator object>
class dal.scopes.StateMachine(name, version='latest', new=False, db='global')

Bases: Scope

StateMachine class

Name: str

verifies if the links already exists if not add it to the Links hash

copy_node(copy_name: str, org_name: str, org_flow: str, org_type: str = 'State', options: dict | None = None)

copy a node instance

db: str
delete(key, name)

Delete object dependencies

delete link

is_valid()
movaidb: MovaiDB
scope = 'StateMachine'
class dal.scopes.Struct(name, struct_dict, prev_struct, db)

Bases: object

General structure… how to describe?

Name: str
add(key, name, **kwargs)
db: str
delete(key, name)
get_attributes(d: dict)
get_dict()
get_ref(value: str)

Receives a value and returns the value with refs if they exist

movaidb: MovaiDB
rename(key: str, old_name: str, new_name: str) → bool
class dal.scopes.System(name, version='latest', new=False, db='global')

Bases: Scope

System model

Name: str
db: str
movaidb: MovaiDB
scope = 'System'
class dal.scopes.Translation(name, version='latest', new=False, db='global')

Bases: Scope

Translation scope.

Name: str
db: str
movaidb: MovaiDB
scope = 'Translation'
validator: Validator = <dal.validation.validator.TranslationValidator object>
class dal.scopes.User(name, version='latest', new=False, db='global')

Bases: Scope

User class

Name: str
classmethod create(username: str, password: str) → bool

Create a new user

Parameters:

  • username (str) –

  • password (str) –

Returns:

obj (User)

create_id()

Create

db: str
get_permissions(resource_name: str) → list

Returns list of user permissions

get_refresh_token()

Generate refresh token

get_supports()
get_token()

Generate authentication token

has_permission(resource_name: str, permission_name: str, skip_superuser: bool = False) → bool

Returns permission check

hash_password(password)

Hash a password for storing

movaidb: MovaiDB
classmethod reset(*, username: str, current_pass: str | None = None, new_pass: str, confirm_pass: str, validate_current_pass: bool = True) → bool

Reset user password

Parameters:

  • current_pass (-) – old password

  • new_pass (-) – new password

  • confirm_pass (-) – confirm password

  • validate_current_password (-) – confirm current password before setting the new password

scope = 'User'
user_permissions() → dict
verify_password(provided_password)

Verify a stored password against one provided by user

classmethod verify_token(token)

Validate if token is valid

class dal.scopes.Widget(name, version='latest', new=False, db='global')

Bases: Scope

Layout model

Name: str
db: str
movaidb: MovaiDB
scope = 'Widget'