ModelInterface

class rosa_kb.typedb_model_interface.ModelInterface(address, database_name, schema_path=None, data_path=None, force_database=False, force_data=False, infer=False)

Class to interact with the ROSA knowledge model in typeDB.

Connect to a typeDB server and interacts with it.

Connects TypeDBInterface to typeDB server, creating a database, loading a schema and a data file.

Parameters:

address (str) – TypeDB server address.
database_name (str) – database name.
schema_path (Union[list[str], str, None]) – list with paths to schema files (.tql).
data_path (Union[list[str], str, None]) – list with paths to data files (.tql).
force_database (Optional[bool]) – if database should override an existing database
force_data (Optional[bool]) – if the database data should be overriden.
infer (Optional[bool]) – if inference engine should be used.

Methods

`__init__`(address, database_name[, ...])	Connect to a typeDB server and interacts with it.
`activate_component`(c_name, value)	Activate a Component.
`add_measurement`(name, value)	Add new Quality Attribute or EnvironmentalAttribute measurement.
`cancel_action`(action_name)	Cancel Action.
`create_reconfiguration_plan`(c_activate, ...)	Create a reconfiguration plan.
`delete_component_status`(component_name)	Delete Component status.
`get_active_component_process`()	Get all attributes owned by a Component, and the Component type.
`get_adaptable_components`()	Get the name of adaptable Components.
`get_adaptable_functions`()	Get the name of adaptable Functions.
`get_adaptable_things_raw`(thing)	Get the name of adaptable individuals of a certain Thing.
`get_component_all_attributes`(component)	Get all attributes owned by a Component, and the Component type.
`get_component_configuration_priority`(cc_name)	Get component configuration priority value.
`get_component_parameters`(c_config)	Get ComponentParameters in a component configuration relationship.
`get_components_in_function_design`(fd_name)	Get components in relation with a function design.
`get_function_design_priority`(fd_name)	Get function design priority value.
`get_instances_of_thing_with_status`(thing, status)	Get name of instances of a certain Thing that have a certain status.
`get_instances_thing_always_improve`(thing)	Get name of instances of a certain Thing that have always-improve true.
`get_latest_completed_reconfiguration_plan_time`()	Get end-time of the most recent completed reconfiguration plan.
`get_latest_measurement`(name)	Get latest measurement value.
`get_latest_pending_reconfiguration_plan`()	Get latest pending reconfiguration plan.
`get_latest_pending_reconfiguration_plan_time`()	Get start-time of the most recent pending reconfiguration plan.
`get_latest_reconfiguration_plan`()	Get latest reconfiguration plan.
`get_latest_reconfiguration_plan_time`()	Get start-time of the most recent reconfiguration plan.
`get_measurement`(name, time)	Get measured value at a certain time.
`get_obsolete_component_configurations`()	Get selected component configurations that are not required anymore.
`get_obsolete_components`()	Get active components that are not required anymore.
`get_obsolete_fds`()	Get selected fds that are not required anymore.
`get_outdated_reconfiguration_plans`()	Get outdated reconfiguration plans.
`get_reconfiguration_plan`(start_time)	Get reconfiguration plan with start-time.
`get_reconfiguration_plan_result`(start_time)	Get result of recongiration plan with start_time.
`get_relationship_with_attribute`(entity, ...)	Get relationship name that has an attr and relates to a certain entity.
`get_selectable_actions`()	Get the name of selectable Actions.
`get_selectable_c_configs`(component_name)	Get the name of selectable component configurations for a Component.
`get_selectable_fds`(function_name)	Get the name of selectable funtion designs for a Function.
`get_selectable_thing_raw`(thing)	Get the name of selectable individuals of a specific Thing.
`get_solved_components`()	Get components with status solved.
`get_solved_functions`()	Get functions with status solved.
`get_unsolved_components`()	Get unsolved Components.
`get_unsolved_functions`()	Get unsolved Functions.
`get_unsolved_thing_raw`(thing)	Get unsolved individuals of a certain Thing type.
`has_action`(action_name)	Check whether model can an specific Action.
`insert_action`(action_name)	Add new Action.
`insert_component`(component_name[, ...])	Add new Component.
`insert_component_process`(component_name, pid)	Add new component-process.
`insert_function`(function_name)	Add new Function.
`insert_function_design`(function_design_name, ...)	Add new function-design.
`insert_functional_requirement`(action_name, ...)	Add new functional-requirement.
`insert_ros_node_component`(component_name, ...)	Add new ROSNode Component.
`is_action_feasible`(action_name)	Check if an Action only has status feasible.
`is_action_required`(action_name)	Check whether an Action is required.
`is_action_selectable`(action_name)	Check whether an Action can be selected, i.e., it can be performed.
`is_component_active`(name)	Check wheter a component is active or not.
`request_action`(action_name[, preference])	Request Action.
`select_component_configuration`(c_name, cc_name)	Select component-configuration 'cc_name' and unselect all other ccs.
`select_components_selected_config`(...)	Select component configuration and return them.
`select_configuration`(functions_selected_fd, ...)	Select configuration and create reconfiguration plan.
`select_fd_and_get_components`(...)	Select function design, and get components to activate and deactivate.
`select_function_design`(f_name, fd_name)	Select function-design 'fd_name' and unselect all other fds.
`select_relationship`(entity, entity_name, ...)	Select relationship individual, and unselect all other individuals.
`set_component_process_end_time`(start_time)	Set component process end time.
`unselect_obsolete_fds_cc`()	Unselect all obsolete function designs and component configurations.
`update_action_status`(action_name, action_status)	Update Action status.
`update_component_status`(component_name, ...)	Update Component status.
`update_function_design_priority`(fd_name, value)	Update function design priority.
`update_outdated_reconfiguration_plans_result`()	Set outdated reconfiguration plans result to 'abandoned'.
`update_reconfiguration_plan_result`(...)	Update reconfiguration plan result.

__init__(address, database_name, schema_path=None, data_path=None, force_database=False, force_data=False, infer=False)

Connect to a typeDB server and interacts with it.

Connects TypeDBInterface to typeDB server, creating a database, loading a schema and a data file.

Parameters:

address (str) – TypeDB server address.
database_name (str) – database name.
schema_path (Union[list[str], str, None]) – list with paths to schema files (.tql).
data_path (Union[list[str], str, None]) – list with paths to data files (.tql).
force_database (Optional[bool]) – if database should override an existing database
force_data (Optional[bool]) – if the database data should be overriden.
infer (Optional[bool]) – if inference engine should be used.

activate_component(c_name, value)

Activate a Component.

Set ‘is-active’ attribute of a Component to true or false.

Parameters:

c_name (str) – component name
value (bool) – whether the component should be active or not

Return type:

Optional[Iterator[ConceptMap]]

Returns:

update query result

add_measurement(name, value)

Add new Quality Attribute or EnvironmentalAttribute measurement.

Add new Quality Attribute or EnvironmentalAttribute measurement. The new measurement ‘latest’ attribute is set to true, and the old measurements ‘latest’ attribute are set to false.

Parameters:

name (str) – QA/EA name
value (str) – measured value

Return type:

Optional[Iterator[ConceptMap]]

Returns:

query result

cancel_action(action_name)

Cancel Action.

Parameters:: action_name (str) – action name
Return type:: Optional[Iterator[ConceptMap]]
Returns:: insert query result

create_reconfiguration_plan(c_activate, c_deactivate, c_config)

Create a reconfiguration plan.

Parameters:

c_active – components to activate
c_deactive – components to deactivate
c_config (list[str]) – component configurations to select

Return type:

datetime | None

Returns:

the time the reconfiguration plan was created, or None in case there was a failure creating the reconfiguration plan

delete_component_status(component_name)

Delete Component status.

Parameters:: component_name (str) – component name
Return type:: Optional[Literal[True]]
Returns:: delete query result

get_active_component_process()

Get all attributes owned by a Component, and the Component type.

Parameters:: component – component name.
Return type:: ComponentProcessDict | None
Returns:: Dict with component type and all its attributes

get_adaptable_components()

Get the name of adaptable Components.

Get the name of adaptable Components. A component is adaptable when its ‘always-improve’ attribute is true, or when it status is ‘unsolved’ or ‘configuration error’

Return type:: list[str]
Returns:: name of adaptable components

get_adaptable_functions()

Get the name of adaptable Functions.

Get the name of adaptable Functions. A function is adaptable when its ‘always-improve’ attribute is true, or when it status is ‘unsolved’ or ‘configuration error’

Return type:: list[str]
Returns:: name of adaptable functions

get_adaptable_things_raw(thing)

Get the name of adaptable individuals of a certain Thing.

Get the name of adaptable individuals of a certain Thing. An individual is adaptable when it has an ‘always-improve’ attribute set to true, or when it has status ‘unsolved’ or ‘configuration error’

Parameters:: thing (str) – Thing type
Return type:: list[dict[str, MatchResultDict]]
Returns:: name of adaptable individuals

get_component_all_attributes(component)

Get all attributes owned by a Component, and the Component type.

Parameters:: component (str) – component name.
Return type:: ComponentDict | None
Returns:: Dict with component type and all its attributes

get_component_configuration_priority(cc_name)

Get component configuration priority value.

Parameters:: cc_name (str) – component configuration name
Return type:: list[str]
Returns:: component configuration priority value

get_component_parameters(c_config)

Get ComponentParameters in a component configuration relationship.

Parameters:: c_config (str) – component configuration name.
Return type:: ComponentConfigurationDict | None
Returns:: Dict with component-name, parameter-key, parameter-value

get_components_in_function_design(fd_name)

Get components in relation with a function design.

Parameters:: fd_name (str) – name of the function design.
Return type:: list[str]
Returns:: component names in relation with fd_name.

get_function_design_priority(fd_name)

Get function design priority value.

Parameters:: fd_name (str) – function design name
Return type:: list[str]
Returns:: function design priority value

get_instances_of_thing_with_status(thing, status)

Get name of instances of a certain Thing that have a certain status.

Parameters:

thing (str) – thing type to query.
status (str) – status.

Return type:

list[str]

Returns:

name of instances with status.

get_instances_thing_always_improve(thing)

Get name of instances of a certain Thing that have always-improve true.

Parameters:: thing (str) – Thing type to query.
Return type:: list[str]
Returns:: name of instances with always-improve true.

get_latest_completed_reconfiguration_plan_time()

Get end-time of the most recent completed reconfiguration plan.

Return type:: datetime | None
Returns:: end-time of the most recent completed reconfiguration plan.

get_latest_measurement(name)

Get latest measurement value.

Parameters:: name (str) – QA/EA name
Return type:: float | None
Returns:: measurement value, or None if there is no measurement

get_latest_pending_reconfiguration_plan()

Get latest pending reconfiguration plan.

Return type:: ReconfigPlanDict | None
Returns:: dict representing the reconfiguration plan, its keys are: start_time, c_activate, c_deactivate, c_config

get_latest_pending_reconfiguration_plan_time()

Get start-time of the most recent pending reconfiguration plan.

Return type:: datetime | None
Returns:: start-time of the most recent pending reconfiguration plan.

get_latest_reconfiguration_plan()

Get latest reconfiguration plan.

Return type:: ReconfigPlanDict | None
Returns:: dict representing the reconfiguration plan, its keys are: start_time, c_activate, c_deactivate, c_config

get_latest_reconfiguration_plan_time()

Get start-time of the most recent reconfiguration plan.

Return type:: datetime | None
Returns:: start-time of the most recent reconfiguration plan.

get_measurement(name, time)

Get measured value at a certain time.

Parameters:

name (str) – QA/EA name
time (datetime) – measurement time

Return type:

float | None

Returns:

measurement value, or None if there is no measurement

get_obsolete_component_configurations()

Get selected component configurations that are not required anymore.

Return type:: list[str]
Returns:: List with selected configs that are not required anymore

get_obsolete_components()

Get active components that are not required anymore.

Return type:: list[str]
Returns:: List with active components that are not required anymore

get_obsolete_fds()

Get selected fds that are not required anymore.

Return type:: list[str]
Returns:: List with selected fds that are not required anymore

get_outdated_reconfiguration_plans()

Get outdated reconfiguration plans.

A reconfiguration plan is considered outdated when it doesn’t have an end-time and its start-time is lower than the end-time of a reconfiguration plan with result ‘completed’.

Return type:: list[datetime]
Returns:: start-time of all outdated reconfiguration plans

get_reconfiguration_plan(start_time)

Get reconfiguration plan with start-time.

Parameters:: start_time (datetime) – start-time of the desired reconfiguration plan.
Return type:: ReconfigPlanDict
Returns:: dict representing the reconfiguration plan, its keys are: start_time, c_activate, c_deactivate, c_config

get_reconfiguration_plan_result(start_time)

Get result of recongiration plan with start_time.

Parameters:: start_time (datetime) – reconfiguration plan start-time
Return type:: str | None
Returns:: reconfiguration plan result

get_relationship_with_attribute(entity, entity_name, relation, r_attribute, r_value)

Get relationship name that has an attr and relates to a certain entity.

Get relationship name that has an attr and relates to a certain entity. For example, a ‘function-design’ that has the ‘is-selected’ attribute set to true and is related to a ‘Function’ with ‘function-name’ ‘my_function’.

Parameters:

entity (str) – entity type
entity_name (str) – entity name
relation (str) – relation type
r_attribute (str) – attribute name
r_value (str | int | float | bool | datetime) – attribute value

Return type:

list[str]

Returns:

name of individuals of the relation

get_selectable_actions()

Get the name of selectable Actions.

Return type:: list[str]
Returns:: name of selectable actions

get_selectable_c_configs(component_name)

Get the name of selectable component configurations for a Component.

Parameters:: component_name (str) – component name
Return type:: list[str]
Returns:: name of selectable component configurations

get_selectable_fds(function_name)

Get the name of selectable funtion designs for a Function.

Parameters:: function_name (str) – function name
Return type:: list[str]
Returns:: name of selectable function designs

get_selectable_thing_raw(thing)

Get the name of selectable individuals of a specific Thing.

Parameters:: thing (str) – Thing type
Return type:: list[dict[str, MatchResultDict]]
Returns:: name of selectable individuals

get_solved_components()

Get components with status solved.

Return type:: list[str]

get_solved_functions()

Get functions with status solved.

Return type:: list[str]

get_unsolved_components()

Get unsolved Components.

Get unsolved Components. A component is unsolved when its ‘is-required’ attribute is True and its status is ‘unsolved’.

Return type:: list[str]
Returns:: name of unsolved components

get_unsolved_functions()

Get unsolved Functions.

Get unsolved Functions. A function is unsolved when its ‘is-required’ attribute is True and its status is ‘unsolved’.

Return type:: list[str]
Returns:: name of unsolved functions

get_unsolved_thing_raw(thing)

Get unsolved individuals of a certain Thing type.

Get unsolved individuals of a certain Thing type. An individual is unsolved when its ‘is-required’ attribute is True and its status is ‘unsolved’.

Return type:: list[dict[str, MatchResultDict]]
Returns:: name of unsolved individuals

has_action(action_name)

Check whether model can an specific Action.

Parameters:: action_name (str) – Action name
Return type:: bool
Returns:: whether the action exists in the model or not

insert_action(action_name)

Add new Action.

Parameters:: action_name (str) – action name
Return type:: Optional[Iterator[ConceptMap]]
Returns:: insert query result

insert_component(component_name, always_improve=False)

Add new Component.

Parameters:

component_name (str) – component name
always_improve (Optional[bool]) – if it should always try to select the best

component configuration :rtype: Optional[Iterator[ConceptMap]] :return: insert query result

insert_component_process(component_name, pid)

Add new component-process.

Parameters:

component_name (str) – component name
pid (int) – pid

Return type:

Optional[Iterator[ConceptMap]]

Returns:

insert query result

insert_function(function_name)

Add new Function.

Parameters:: function_name (str) – function name
Return type:: Optional[Iterator[ConceptMap]]
Returns:: insert query result

insert_function_design(function_design_name, function_name, components_names, priority=None)

Add new function-design.

Parameters:

function_design_name (str) – function design name
function_name (str) – function name
components_names (list[str]) – list of required components
priority (Optional[float]) – function design priority

Return type:

Optional[Iterator[ConceptMap]]

Returns:

insert query result

insert_functional_requirement(action_name, functions_names)

Add new functional-requirement.

Parameters:

action_name (str) – action name
functions_names (list[str]) – list of required functions

Return type:

Optional[Iterator[ConceptMap]]

Returns:

insert query result

insert_ros_node_component(component_name, package, executable, always_improve=False, lifecycle_node=False)

Add new ROSNode Component.

Parameters:

component_name (str) – component name
package (str) – ros package name
executable (str) – ros executable name
always_improve (Optional[bool]) – if it should always try to select the best

component configuration :type lifecycle_node: Optional[bool] :param lifecycle_node: if the ros node is a lifecycle node :rtype: Optional[Iterator[ConceptMap]] :return: insert query result

is_action_feasible(action_name)

Check if an Action only has status feasible.

Parameters:: action_name (str) – Action name
Return type:: bool
Returns:: whether the action is feasible or not

is_action_required(action_name)

Check whether an Action is required.

Parameters:: action_name (str) – Action name
Return type:: bool
Returns:: whether the action is required or not

is_action_selectable(action_name)

Check whether an Action can be selected, i.e., it can be performed.

Parameters:: action_name (str) – Action name
Return type:: bool
Returns:: whether the action is feasible or not

is_component_active(name)

Check wheter a component is active or not.

Parameters:: name (str) – component name
Return type:: bool
Returns:: whether the component is active or not

request_action(action_name, preference='')

Request Action.

Parameters:

action_name (str) – action name
preference (Optional[str]) – QA/EA that has preference when selecting a config

Return type:

Optional[Iterator[ConceptMap]]

Returns:

insert query result or None when the action was already required

select_component_configuration(c_name, cc_name)

Select component-configuration ‘cc_name’ and unselect all other ccs.

Set the ‘is-selected’ attribute of the ‘cc_name’ component configuration to true, and set ‘is-selected’ to false for all other component configurations related to ‘c_name’.

Parameters:

c_name (str) – component name
cc_name (str) – component configuration name

Return type:

Optional[Iterator[ConceptMap]]

Returns:

update query result

select_components_selected_config(components_selected_config)

Select component configuration and return them.

Set the ‘is-selected’ attribute of the component configurations in components_selected_config to true, and returns a list with the name of the component configurations that need to be updated.

Parameters:: components_selected_config (list[Tuple[str, str]]) – a list of tuples with the form (COMPONENT_NAME, CC_NAME) representing the component configurations that were selected for which components
Return type:: list[str]
Returns:: list with the name of the component configurations that need to be updated

select_configuration(functions_selected_fd, components_selected_config)

Select configuration and create reconfiguration plan.

Parameters:

functions_selected_fd (list[Tuple[str, str]]) – a list of tuples with the form (FUNCTION_NAME, FD_NAME) representing the function designs that were selected for which function
components_selected_config (list[Tuple[str, str]]) – a list of tuples with the form (COMPONENT_NAME, CC_NAME) representing the component configurations that were selected for which components

Return type:

datetime | None

Returns:

reconfig plan creation time.

select_fd_and_get_components(functions_selected_fd)

Select function design, and get components to activate and deactivate.

Set the ‘is-selected’ attribute of the function desings in functions_selected_fd to true, and returns a tuple with a list with the components that need to be activated and a list with the components that need to be deactivated.

Parameters:: functions_selected_fd (list[Tuple[str, str]]) – a list of tuples with the form (FUNCTION_NAME, FD_NAME) representing the function designs that were selected for which function
Return type:: Tuple[list[str], list[str]]
Returns:: tuple with the form (c_activate, c_deactivate), indicating which components should be activated and deactivated

select_function_design(f_name, fd_name)

Select function-design ‘fd_name’ and unselect all other fds.

Set the ‘is-selected’ attribute of the ‘fd_name’ function design to true, and set ‘is-selected’ to false for all other function designs related to ‘f_name’.

Parameters:

f_name (str) – function name
fd_name (str) – function design name

Return type:

Optional[Iterator[ConceptMap]]

Returns:

update query result

select_relationship(entity, entity_name, relation, r_name)

Select relationship individual, and unselect all other individuals.

Set the ‘is-selected’ attribute of the ‘r_name’ individual to true, and set ‘is-selected’ to false for all other ‘relation’ individuals related to ‘entity_name’.

Parameters:

entity (str) – entity type
entity_name (str) – entity name
relation (str) – relation type
r_name (str) – relationship name

Return type:

Optional[Iterator[ConceptMap]]

Returns:

update query result

set_component_process_end_time(start_time)

Set component process end time.

Parameters:: start_time (datetime) – component-process start-time.
Return type:: Optional[Iterator[ConceptMap]]
Returns:: update query result

unselect_obsolete_fds_cc()

Unselect all obsolete function designs and component configurations.

An obsolete function design or component configuration is the one that has ‘is-selected’ attribute set to true, but it is no required anymore

Return type:: None

update_action_status(action_name, action_status)

Update Action status.

Parameters:

action_name (str) – action name
action_status (str) – action_status

Return type:

Optional[Iterator[ConceptMap]]

Returns:

insert query result

update_component_status(component_name, component_status)

Update Component status.

Parameters:

component_name (str) – component name
component_status (str) – component status

Return type:

Optional[Iterator[ConceptMap]]

Returns:

delete query result

update_function_design_priority(fd_name, value)

Update function design priority.

Parameters:

fd_name (str) – function design name
value (float) – new priority value

Return type:

Optional[Iterator[ConceptMap]]

Returns:

query result

update_outdated_reconfiguration_plans_result()

Set outdated reconfiguration plans result to ‘abandoned’.

Return type:: Optional[Iterator[ConceptMap]]
Returns:: update query result

update_reconfiguration_plan_result(start_time, result_value)

Update reconfiguration plan result.

Parameters:

start_time (str | datetime) – reconfiguration plan start_time
result_value (Literal['completed', 'failed', 'abandoned']) – reconfig plan result

Return type:

Optional[Iterator[ConceptMap]]

Returns:

update query result