Basics

../../../../../../_images/MLPro-OA-Systems_OASystems.drawio.png

Ver. 0.3.0 (2025-06-06)

This module provides modules and template classes for adaptive systems and adaptive functions.

class mlpro.oa.systems.basics.PseudoTask(p_wrap_method: Callable[[Dict[int, Tuple[str, Instance]]], None], p_name='PseudoTask', p_range_max=0, p_duplicate_data=True, p_logging=True, p_visualize=False, **p_kwargs)

Bases: OAStreamTask

A template class PseudoTask, only to be used by the OASystem. This functions runs a wrapped method as it’s run method.

Parameters:

  • p_wrap_method

  • p_name

  • p_range_max

  • p_duplicate_data

  • p_logging

  • p_visualize

  • p_kwargs

_run(p_instances: Dict[int, Tuple[str, Instance]])

Custom method that is called by method run().

Parameters:

p_instances (InstDict) – Instances to be processed.

class mlpro.oa.systems.basics.OAFctSTrans(p_id=None, p_name: str = None, p_range_max: int = 2, p_autorun: int = 0, p_class_shared=None, p_ada: bool = True, p_afct_cls=None, p_state_space: ~mlpro.bf.math.basics.MSpace = None, p_action_space: ~mlpro.bf.math.basics.MSpace = None, p_input_space_cls=<class 'mlpro.bf.math.basics.ESpace'>, p_output_space_cls=<class 'mlpro.bf.math.basics.ESpace'>, p_output_elem_cls=<class 'mlpro.bf.systems.basics.State'>, p_threshold=0, p_buffer_size=0, p_wf_strans: ~mlpro.oa.streams.basics.OAStreamWorkflow = None, p_visualize: bool = False, p_logging=True, **p_kwargs)

Bases: FctSTrans, Model

This is a template class for Online Adaptive State Transition function. Please overwrite the _simulate_reaction() method or provide an adaptive class as a parameter with all the additional required parameters.

Parameters:

  • p_id

  • p_name

  • p_range_max

  • p_autorun

  • p_class_shared

  • p_ada

  • p_afct_cls

  • p_state_space

  • p_action_space

  • p_input_space_cls

  • p_output_space_cls

  • p_output_elem_cls

  • p_threshold

  • p_buffer_size

  • p_wf

  • p_visualize

  • p_logging

  • p_kwargs

simulate_reaction(p_state: State, p_action: Action, p_t_step: timedelta = None) → State

Simulates a state transition based on a state and action. Custom method _simulate_reaction() is called.

Parameters:

  • p_state (State) – State of the System.

  • p_action (Action) – External action provided for the action simulation

  • p_t_step (timedelta, optional.) – The timestep for which the system is to be simulated

Returns:

state – The new state of the System.

Return type:

State

_adapt(**p_kwargs) → bool

When called, the function and it’s components adapt based on the provided parameters.

Parameters:

p_kwargs – additional parameters for adaptation.

Returns:

adapted – Returns true if the Function has adapted

Return type:

bool

_adapt_on_event(p_event_id: str, p_event_object: Event) → bool
Parameters:

  • p_event_id

  • p_event_object

add_task_strans(p_task: OAStreamTask, p_pred_task=None)

Adds a task to the workflow.

Parameters:

  • p_task (OATask, StreamTask) – The task to be added to the workflow

  • p_pred_task (list[Task]) – Name of the predecessor tasks for the task to be added

_run_wf_strans(p_instances: Dict[int, Tuple[str, Instance]])

Runs the processing workflow, for state transition.

Parameters:

p_instances (InstDict) – Dictionary of instances to be processed by the workflow.

_setup_oafct_strans()

Adds a pseudo task to the processing workflow, with the method to be wrapped.

Returns:

False when successfully setup.

Return type:

bool

class mlpro.oa.systems.basics.OAFctSuccess(p_id=None, p_name: str = None, p_range_max: int = 2, p_autorun: int = 0, p_class_shared=None, p_ada: bool = True, p_afct_cls=None, p_state_space: ~mlpro.bf.math.basics.MSpace = None, p_action_space: ~mlpro.bf.math.basics.MSpace = None, p_input_space_cls=<class 'mlpro.bf.math.basics.ESpace'>, p_output_space_cls=<class 'mlpro.bf.math.basics.ESpace'>, p_output_elem_cls=<class 'mlpro.bf.systems.basics.State'>, p_threshold=0, p_buffer_size=0, p_wf_success: ~mlpro.oa.streams.basics.OAStreamWorkflow = None, p_visualize: bool = False, p_logging=True, **p_kwargs)

Bases: FctSuccess, Model

This is a template class for Online Adaptive Success Computation function. Please overwrite the _compute_success() method or provide an adaptive class as a parameter with all the additional required parameters.

Parameters:

  • p_id

  • p_name

  • p_range_max

  • p_autorun

  • p_class_shared

  • p_ada

  • p_afct_cls

  • p_state_space

  • p_action_space

  • p_input_space_cls

  • p_output_space_cls

  • p_output_elem_cls

  • p_threshold

  • p_buffer_size

  • p_wf_success

  • p_visualize

  • p_logging

  • p_kwargs

compute_success(p_state: State) → bool

Assesses the given state regarding success criteria. Custom method _compute_success() is called.

Parameters:

p_state (State) – System state.

Returns:

success – True, if given state is a success state. False otherwise.

Return type:

bool

_adapt_on_event(p_event_id: str, p_event_object: Event) → bool
Parameters:

  • p_event_id

  • p_event_object

add_task_success(p_task: StreamTask, p_pred_tasks: list = None)

Adds a task to the workflow.

Parameters:

  • p_task (OATask, StreamTask) – The task to be added to the workflow

  • p_pred_task (list[Task]) – Name of the predecessor tasks for the task to be added

_run_wf_success(p_instances: Dict[int, Tuple[str, Instance]])

Runs the success computation workflow of the system.

Parameters:

p_instances (InstDict) – Dictionary of instances to be processed by the workflow.

_adapt(**p_kwargs) → bool

When called, the function and it’s components adapt based on the provided parameters.

Parameters:

p_kwargs – additional parameters for adaptation.

Returns:

adapted – Returns true if the Function has adapted

Return type:

bool

_setup_oafct_success()

Adds a pseudo task to the success computation workflow, with the method to be wrapped.

Returns:

False when successfully setup.

Return type:

bool

class mlpro.oa.systems.basics.OAFctBroken(p_id=None, p_name: str = None, p_range_max: int = 2, p_autorun: int = 0, p_class_shared=None, p_ada: bool = True, p_afct_cls=None, p_state_space: ~mlpro.bf.math.basics.MSpace = None, p_action_space: ~mlpro.bf.math.basics.MSpace = None, p_input_space_cls=<class 'mlpro.bf.math.basics.ESpace'>, p_output_space_cls=<class 'mlpro.bf.math.basics.ESpace'>, p_output_elem_cls=<class 'mlpro.bf.systems.basics.State'>, p_threshold=0, p_buffer_size=0, p_wf_broken: ~mlpro.oa.streams.basics.OAStreamWorkflow = None, p_visualize: bool = False, p_logging=True, **p_kwargs)

Bases: FctBroken, Model

This is a template class for Online Adaptive Broken Computation function. Please overwrite the _compute_broken() method or provide an adaptive class as a parameter with all the additional required parameters.

Parameters:

  • p_id

  • p_name

  • p_range_max

  • p_autorun

  • p_class_shared

  • p_ada

  • p_afct_cls

  • p_state_space

  • p_action_space

  • p_input_space_cls

  • p_output_space_cls

  • p_output_elem_cls

  • p_threshold

  • p_buffer_size

  • p_wf_broken

  • p_visualize

  • p_logging

  • p_kwargs

compute_broken(p_state: State) → bool

Assesses the given state regarding breakdown criteria. Custom method _compute_success() is called.

Parameters:

p_state (State) – System state.

Returns:

broken – True, if given state is a breakdown state. False otherwise.

Return type:

bool

add_task_broken(p_task: StreamTask, p_pred_tasks: list = None)

Adds a task to the workflow.

Parameters:

  • p_task (OATask, StreamTask) – The task to be added to the workflow

  • p_pred_task (list[Task]) – Name of the predecessor tasks for the task to be added

_run_wf_broken(p_instances: Dict[int, Tuple[str, Instance]])

Runs the success computation workflow of the system.

Parameters:

p_instances (InstDict) – Dictionary of instances to be processed by the workflow.

_adapt(**p_kwargs) → bool

Custom method to adapt all the internal adaptive elements of the OAFct Broken.

Parameters:

p_kwargs – Parameters to be used for adaptation.

Returns:

True if any of the element has adapted.

Return type:

bool

_adapt_on_event(p_event_id: str, p_event_object: Event) → bool
Parameters:

  • p_event_id

  • p_event_object

_setup_oafct_broken()

Adds the PseudoTask with broken computation function as the host task to the workflow

Returns:

True after the setup is completed.

Return type:

bool

class mlpro.oa.systems.basics.OASystem(p_id=None, p_name: str = None, p_range_max: int = 0, p_autorun: int = 0, p_class_shared: Shared = None, p_ada: bool = True, p_mode=0, p_latency=None, p_t_step: timedelta = None, p_fct_strans: FctSTrans = None, p_fct_success: FctSuccess = None, p_fct_broken: FctBroken = None, p_wf_strans: OAStreamWorkflow = None, p_wf_success: OAStreamWorkflow = None, p_wf_broken: OAStreamWorkflow = None, p_mujoco_file=None, p_frame_skip: int = 1, p_state_mapping=None, p_action_mapping=None, p_camera_conf: tuple = (None, None, None), p_visualize: bool = False, p_logging: bool = True, **p_kwargs)

Bases: OAFctBroken, OAFctSTrans, OAFctSuccess, ASystem

This ist a template class for Adaptive State Based System.

Parameters:

  • p_id – Id of the system.

  • p_name (str) – Name of the system.

  • p_range_max – Range of the system.

  • p_autorun – Whether the system should autorun as a Task.

  • p_class_shared – The shared class for multisystem.

  • p_mode – Mode of the System. Simulation or real.

  • p_ada (bool) – The adaptability of the system.

  • p_latency (timedelta) – Latency of the system.

  • p_t_step (timedelta) – Simulation timestep of the system.

  • p_fct_strans (FctSTrans | AFctSTrans | OAFctSTrans) – External state transition function.

  • p_fct_success (FctSuccess | AFctSuccess | OAFctSuccess) – External success computation function.

  • p_fct_broken (FctBroken | AFctBroken | OAFctBroken) – External broken computation function.

  • p_wf_strans (OAWorkflow) – State transition workflow. Optional.

  • p_wf_success (OAWorkflow) – Success computation workflow. Optional.

  • p_wf_broken (OAWorkflow) – Broken computation workflow. Optional

  • p_mujoco_file – Mujoco file for simulation using mujoco engine.

  • p_frame_skip – Number of frames to be skipped during visualization.

  • p_state_mapping – State mapping for Mujoco.

  • p_action_mapping – Action Mapping for Mujoco.

  • p_camera_conf – Camera Configuration for Mujoco.

  • p_visualize – Visualization switch.

  • p_logging – Logging level for the system.

  • p_kwargs – Additional Parameters

_adapt(**p_kwargs) → bool

Adapts the system based on the parameters provided. Calls the adapt methods of all the adaptive elements of the system.

Parameters:

p_kwargs – Additional parameters for the adaptation of the system.

Returns:

True if any of the element has adapted.

Return type:

bool

switch_adaptivity(p_ada: bool)

Switches the adaptivity of all the internal adaptive elements of OASystem.

Parameters:

p_ada (bool) – The boolean flag indicating if the adaptivity shall be switched on or off.

_adapt_on_event(p_event_id: str, p_event_object: Event) → bool
Parameters:

  • p_event_id

  • p_event_object

simulate_reaction(p_state: State, p_action: Action, p_t_step: timedelta = None) → State

Simulates a state transition based on a state and action. Custom method _simulate_reaction() is called.

Parameters:

  • p_state (State) – State of the System.

  • p_action (Action) – External action provided for the action simulation

  • p_t_step (timedelta, optional.) – The timestep for which the system is to be simulated

Returns:

state – The new state of the System.

Return type:

State

compute_success(p_state: State) → bool

Assesses the given state regarding success criteria. Custom method _compute_success() is called.

Parameters:

p_state (State) – System state.

Returns:

success – True, if given state is a success state. False otherwise.

Return type:

bool

compute_broken(p_state: State) → bool

Assesses the given state regarding breakdown criteria. Custom method _compute_success() is called.

Parameters:

p_state (State) – System state.

Returns:

broken – True, if given state is a breakdown state. False otherwise.

Return type:

bool

init_plot(p_figure: Figure = None, p_plot_settings: PlotSettings = None, **p_kwargs)

Initializes the plot for all the internal elements of OASystem.

Parameters:

  • p_figure (Figure, optional) – Matplotlib figure, if one exists already

  • p_plot_settings (PlotSettings) – Additional plot settings

  • p_kwargs – Additional plot parameters

update_plot(**p_kwargs)

Updates the all the internal plots.

Parameters:

p_kwargs (additional plot parameters)

switch_visualization(p_object=None, p_visualize=None)

Method to switch the visualization of an object in an online adaptive system.

Parameters:

  • p_object (object) – The object whose visualization is to be switched off. Although, a valid object is any object with visualization property in MLPro, within this runtime. It is suggested to use only on the Functions, Workflows, Tasks and System itself.

  • p_visualize (bool) – The bool value for setting the visualization of an object.

Notes

Please do not turn off the visualization by getting functions of the system (e.g. self.get_fctstrans()), if the functions are not provided externally to the system, since this will refer to the system itself.

Examples

myOASystem.switch_visualization(p_visualize = False, p_object = self.get_fct_workflow())

get_workflow_strans()
get_workflow_success()
get_workflow_broken()