wrapper

Source code: armscan_env/wrapper.py

---

class AddObservationsWrapper(env: ArmscanEnv | Env)

When implementing it, make sure that additional_obs_space is available before super().__init__(env) is called.

abstract property additional_obs_space: <module 'gymnasium.spaces' from '/home/runner/work/armscan_env/armscan_env/gymnasium/gymnasium/spaces/__init__.py'>

abstract get_additional_obs_array() → ndarray

observation(observation: ndarray) → ndarray

class ArmscanEnvFactory(name2volume: dict[str, ~SimpleITK.SimpleITK.Image], observation: ~armscan_env.envs.base.Observation[~armscan_env.envs.state_action.LabelmapStateAction, ~typing.Any], reward_metric: ~armscan_env.envs.base.RewardMetric[~armscan_env.envs.state_action.LabelmapStateAction] = <armscan_env.envs.rewards.LabelmapClusteringBasedReward object>, termination_criterion: ~armscan_env.envs.base.TerminationCriterion | None = <armscan_env.envs.labelmaps_navigation.LabelmapEnvTerminationCriterion object>, slice_shape: tuple[int, int] | None = None, max_episode_len: int | None = None, rotation_bounds: tuple[float, float] = (180, 180), translation_bounds: tuple[float | None, float | None] = (None, None), render_mode_train: ~typing.Literal['plt', 'animation'] | None = None, render_mode_test: ~typing.Literal['plt', 'animation'] | None = None, render_mode_watch: ~typing.Literal['plt', 'animation'] | None = 'animation', venv_type: ~tianshou.highlevel.env.VectorEnvType = VectorEnvType.SUBPROC_SHARED_MEM_AUTO, seed: int | None = None, n_stack: int = 1, project_actions_to: ~typing.Literal['x', 'y', 'xy', 'zy'] | None = None, apply_volume_transformation: bool = False, best_reward_memory: int = 0, exclude_keys_from_framestack: ~collections.abc.Sequence[str] = (), **make_kwargs: ~typing.Any)

Factory for creating ArmscanEnv environments, making use of various wrappers.

create_env(mode: EnvMode) → ArmscanEnv

Creates a single environment for the given mode.

Returns:

ArmscanEnv environment

class BestActionRewardMemory(env: ArmscanEnv, n_best: int)

Adds the action that would lead to the highest image variance to the observation. In focus-stigmation agents, this helps in the initial exploratory phase of episodes, as it allows wandering around the state space without worrying about losing track of the best image found so far.

Parameters:

• env -- The environment to wrap.

• n_best -- Number of best states to keep track of.

property additional_obs_space: <module 'gymnasium.spaces' from '/home/runner/work/armscan_env/armscan_env/gymnasium/gymnasium/spaces/__init__.py'>

get_additional_obs_array() → ndarray

reset(**kwargs: Any) → tuple[ObsType, dict[str, Any]]

Uses the reset() of the env that can be overwritten to change the returned data.

reset_wrapper() → None

step(action: ActType) → tuple[ObsType, SupportsFloat, bool, bool, dict[str, Any]]

Uses the step() of the env that can be overwritten to change the returned data.

class ObsHeap(max_size: int, padding_value: float | None = None, padding_item: ObsType | None = None)

Heap of the best rewards and their corresponding observations. The size of the heap is fixed and the best n rewards are kept. When the first reward-observation pair is inserted, the heap is initialized with the padding value and item. By default, padding is done with the first inserted value and item.

Parameters:

• max_size -- the maximum size of the heap

• padding_value -- the value to use for padding

• padding_item -- the item to use for padding

get_all() → list[ObsType]

get_n_best(n: int = 1) → list[ObsType]

push(value: float, item: ObsType) → None

class ObsRewardHeapItem(obs: ObsType, reward: float)

Heap of the best rewards and their corresponding observations.

get_obs() → ObsType

to_tuple() → tuple[float, ObsType]

class PatchedActionWrapper(env: Env[ObsType, ActType])

abstract action(action: ActType) → ndarray

step(action: ActType) → tuple[ObsType, SupportsFloat, bool, bool, dict[str, Any]]

Uses the step() of the env that can be overwritten to change the returned data.

class PatchedFlattenObservation(env: Env[ObsType, ActType])

Flattens the environment's observation space and each observation from reset and step functions. Had to copy-paste and adjust.

observation(observation: ObsType) → Any

Apply function to the observation.

reset(**kwargs: Any) → tuple[ObsType, dict[str, Any]]

Uses the reset() of the env that can be overwritten to change the returned data.

step(action: ActType) → tuple[ObsType, SupportsFloat, bool, bool, dict[str, Any]]

Uses the step() of the env that can be overwritten to change the returned data.

class PatchedFrameStackObservation(env: Env[ObsType, ActType], stack_size: int, *, padding_type: str | ObsType = 'reset', excluded_observation_keys: Sequence[str] = ())

Had to copy-paste and adjust.

The inheriting from RecordConstructorArgs in original FrameStack is not possible together with overridden getattr, which we however need in order to not become crazy. So there is no way of fixing this after inheriting from original FrameStackObservation, wherefore we copy-paste the whole class and adjust it.

Adjustments:

1. reset takes **kwargs

2. render takes **kwargs

3. __getattr__ is overridden to pass the attribute to the wrapped environment (like in pre-1.0 wrappers)

4. No inheritance from RecordConstructorArgs

5. Observation space is converted to MultiBoxSpace if it is a DictSpace

6. Excluded observation keys are supported for Dict observation spaces

Parameters:

• env -- the environment to wrap

• stack_size -- the number of observations to stack

• padding_type -- the type of padding to use

• excluded_observation_keys -- the keys of the observations to exclude from stacking. The observations with these keys will be passed through without stacking. Can only be used with Dict observation spaces.

reset(**kwargs: Any) → tuple[ObsType, dict[str, Any]]

Uses the reset() of the env that can be overwritten to change the returned data.

step(action: ActType) → tuple[ObsType, SupportsFloat, bool, bool, dict[str, Any]]

Uses the step() of the env that can be overwritten to change the returned data.

class PatchedWrapper(env: ArmscanEnv | Env)

render(**kwargs: Any) → Any

Uses the render() of the env that can be overwritten to change the returned data.

reset(**kwargs: Any) → tuple[ObsType, dict[str, Any]]

Uses the reset() of the env that can be overwritten to change the returned data.