releso.spor.SPORObjectExecutor

class releso.spor.SPORObjectExecutor(*, save_location: Path, logger_name: str | None = None, name: str, stop_after_error: bool = True, reward_on_error: float, reward_on_completion: float | None = None, run_on_reset: bool = True, additional_observations: ObservationDefinition | ObservationDefinitionMulti | List[ObservationDefinition | ObservationDefinitionMulti] | None = None, multi_processor: MultiProcessor | MPIClusterMultiProcessor | None = None, use_communication_interface: bool = False, add_step_information: bool = False, working_directory: str)

Bases: SPORObject

Base definition of a SPORCommandLineObject. Can be instantiated.

Base class for all possible SPOR object classes which use the command line. This class is meant to add some default functionality, so that users do not need to add them themselves. These are:

access to the (MPI) Multi-processors via
ReLeSO.spor.MultiProcessor
command line command
additional flags
working directory
whether or not to use the
SPOR Communication Interface

__init__(**data: Any) → None: Constructor for the ReLeSO basemodel object.

Methods

`convert_to_pathlib_add_datetime`(v)	Add timestamp to save_location, of applicable.
`get_default_observation`(observations)	Generate default observations if no were returned or step failed.
`get_logger`()	Gets the currently defined environment logger.
`get_multiprocessing_prefix`(core_count)	Add commandline prefix for mpi multiprocessor.
`get_observations`()	Return number of observation the step generates.
`run`(step_information, environment_id[, ...])	Run method for the Execution types of SPORSteps.
`set_logger_name_recursively`(logger_name)	Set the logger_name variable for all child elements.
`setup_working_directory`(environment_id)	Set up the working directory for the SPORStep.
`spor_com_interface`(reset, environment_id, ...)	Add spor com interface command line options.
`spor_com_interface_add`(returned_step_dict, ...)	Add returned step information of the spor com interface.
`spor_com_interface_read`(output, step_dict)	Read in return values of the spor com interface.

Attributes

`multi_processor`	Definition of the multi-processor.
`use_communication_interface`	whether or not to use the SPOR communication interface.
`add_step_information`	interface commandline options including the observations, info, done, reward
`working_directory`	Path to the directory in which the program should run, helpful for programs using relative paths.
`name`	name of the SPOR step.
`stop_after_error`	Whether to stop after this step if this step has thrown an error.
`reward_on_error`	Reward which is returned if any kind of error is thrown during completing the defined task.
`reward_on_completion`	Reward which is returned if the task completed with out an error.
`run_on_reset`	This boolean can disable the running of the task when a reset is performed.
`additional_observations`	How many additional observations does this object yield.
`save_location`	Definition of the save location of the logs and validation results.
`logger_name`	name of the logger.

add_step_information: bool: interface commandline options including the observations, info, done, reward

get_multiprocessing_prefix(core_count: int) → str

Add commandline prefix for mpi multiprocessor.

Function which tries to get the correct multiprocessing command if available else an empty string is returned.

Parameters:

core_count (int) – Maximal number of cores the multiprocessor can
use.

Returns:

command string for the multi processor

Return type:

str

multi_processor: MultiProcessor | MPIClusterMultiProcessor | None: Definition of the multi-processor. Defaults to None.

run(step_information: Tuple[ndarray | Dict[str, ndarray] | Tuple[ndarray, ...], float, bool, Dict], environment_id: UUID4, validation_id: int | None = None, core_count: int = 1, reset: bool = False) → Tuple[ndarray | Dict[str, ndarray] | Tuple[ndarray, ...], float, bool, Dict]

Run method for the Execution types of SPORSteps.

Abstract method.

Parameters:

step_information (StepReturnType) – _description_
environment_id (UUID4) – _description_
validation_id (Optional[int], optional) – _description_. Defaults to None.
core_count (int, optional) – _description_. Defaults to 1.
reset (bool, optional) – _description_. Defaults to False.

Raises:

NotImplementedError – _description_

Returns:

_description_

Return type:

StepReturnType

setup_working_directory(environment_id: UUID4)

Set up the working directory for the SPORStep.

Parameters:: environment_id (UUID4) – Id of the environment. This is the working directory.

spor_com_interface(reset: bool, environment_id: UUID4, validation_id: int | None, step_information: Tuple[ndarray | Dict[str, ndarray] | Tuple[ndarray, ...], float, bool, Dict]) → List[str]

Add spor com interface command line options.

Generates the additional command line argument options used for the spor com interface.

Parameters:

reset (bool) – Boolean on whether or not this object is called because of a reset.
validation_id (Optional[int]) – During validation the validation id signifies the validation episode. If none the current episode is not a validation episode.
step_information (StepReturnType) – Previously collected step values. Should be up-to-date.
environment_id (UUID4) – Environment ID which is used to distinguish different environments which run in parallel.

Returns:

List of additional command line options.

Return type:

List[str]

spor_com_interface_add(returned_step_dict: Dict[str, Any], step_dict: Dict[str, Any])

Add returned step information of the spor com interface.

Add the newly returned values to the observation, reward, done and info return values.

Parameters:

returned_step_dict (Dict) – Newly received values.
step_dict (Dict) – Already known values of the step return this dict
place. (will be updated in)

spor_com_interface_read(output: bytes, step_dict: Dict)

Read in return values of the spor com interface.

Interprets in the printed values from the spor com interface and will add them to the observation, reward, done and info return values.

Parameters:

output (bytes) – Output of the command line call. To be interpreted.
step_dict (Dict) – Already known values of the step return this dict
place. (will be updated in)

use_communication_interface: bool: whether or not to use the SPOR communication interface. Defaults to False.

working_directory: str: Path to the directory in which the program should run, helpful for programs using relative paths. If {} are the first characters of the path, they will be replaced by the save_location of the current run. If {} are present in the path, the current environment id will be used to replace the placeholder. This is necessary for multiprocessing.