releso.spor.SPORList

class releso.spor.SPORList(*, save_location: Path, logger_name: str | None = None, steps: List[SPORObjectCommandLine | SPORObjectExternalPythonFunction | SPORObjectInternalPythonFunction], reward_aggregation: Literal['sum', 'min', 'max', 'mean', 'absmax'])

Bases: BaseModel

The SPORList defines the custom defined steps.

In the current version these steps take place after the FFD is done. These can include the fluid solver, post-processing steps, reward generation, additional logging, etc. In a future version the FFD might be incorporated into a SPOR step to stream line the internal functionality.

If a step sends a stop signal all consecutive steps will be ignored and the Each step can generate observations, rewards and info for the current episode step. These are aggregated as follows.

__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_logger()

Gets the currently defined environment logger.

get_observations()

Aggregate the observations of all steps.

run(step_information, environment_id[, ...])

Run through all steps of the list and handles each output.

set_logger_name_recursively(logger_name)

Set the logger_name variable for all child elements.

Attributes

steps

List of SPOR objects.

reward_aggregation

Definition on how the reward should be aggregated.

save_location

Definition of the save location of the logs and validation results.

logger_name

name of the logger.
get_observations() → List[ObservationDefinition | ObservationDefinitionMulti] | None

Aggregate the observations of all steps.

Returns the aggregated observations as a flattened list of observations spaces coming from the SPORSteps.

Returns:

Number of observations.

Return type:

int

reward_aggregation: Literal['sum', 'min', 'max', 'mean', 'absmax']

Definition on how the reward should be aggregated. Definition for each option is given Solver Postprocessing Observation Reward.

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 through all steps of the list and handles each output.

When done all received observations and infos are jointly returned with the reward and the indicator whether or not the episode should be terminated.

The values are added in-place to the step_information variable but is also returned.

Parameters:

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

  • validation_id (Optional[int], optional) – If currently in validation the validation needs to be supplied for steps which need this. If none given no validation is currently performed. Defaults to None.

  • core_count (int, optional) – Wanted core_count of the task to run. Defaults to 1.

  • reset (bool, optional) – Boolean on whether or not this object is called because of a reset. Defaults to False.

Returns:

Results of all steps.

Return type:

StepReturnType

steps: List[SPORObjectCommandLine | SPORObjectExternalPythonFunction | SPORObjectInternalPythonFunction]

List of SPOR objects. Each SPOR object is a step in the environment.