releso.spor.SPORObject

class releso.spor.SPORObject(*, 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)

Bases: BaseModel

Base class SPORObject can not be instantiated.

Base class for all possible SPOR object classes. Theses objects represent the possible steps that can be/are executed after the FFD finished.

The following arguments need to be set for SPOR objects, further parameters are use-case specific.

__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_observations`()	Return number of observation the step generates.
`run`(step_information, environment_id[, ...])	Performs the defined step.
`set_logger_name_recursively`(logger_name)	Set the logger_name variable for all child elements.

Attributes

`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.

additional_observations: ObservationDefinition | ObservationDefinitionMulti | List[ObservationDefinition | ObservationDefinitionMulti] | None: How many additional observations does this object yield. Please define this correctly and not just write in a single number. Be responsible, the single number functionality is only in here due to backwards compatibility reasons.

get_default_observation(observations: ndarray | Dict[str, ndarray] | Tuple[ndarray, ...]) → ndarray | Dict[str, ndarray] | Tuple[ndarray, ...]

Generate default observations if no were returned or step failed.

Generates default observations for all additional observations defined in this object.

Parameters:

observations (ObservationType) – Already defined observations to
to. (which the additional observations are added)

Returns:

Observations with now the additional observations of this object added (values used are the default values)

Return type:

ObservationType

get_observations() → ObservationDefinition | ObservationDefinitionMulti | List[ObservationDefinition | ObservationDefinitionMulti] | None

Return number of observation the step generates.

Returns the number of observations which are generated by this SPORStep.

Returns:

Union[: ObservationDefinition, ObservationDefinitionMulti, List[ObservationDefinition]]: Additional observation definitions

name: str: name of the SPOR step. Makes so that it is possible to distinguish different steps more easily.

reward_on_completion: float | None: Reward which is returned if the task completed with out an error. If this is set this reward will overwrite any other reward this function will generate. (Implementation specific might differ for child classes see documentation). If None no default reward is returned when task is finished. Defaults to None.

reward_on_error: float: Reward which is returned if any kind of error is thrown during completing the defined task.

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]

Performs the defined step.

This function is called to complete the defined step of the current SPORObject. This functions needs to be overloaded by all child classes.

Note

Please use these default values for the return values: observation = None, reward = 0, done = False, info = dict()

Parameters:

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

Returns:

The full compliment of the step information are returned. The include observation, reward, done, info.

Return type:

StepReturnType

run_on_reset: bool: This boolean can disable the running of the task when a reset is performed.

stop_after_error: bool: Whether to stop after this step if this step has thrown an error. This is important if