h8EzSrSSKJr SSKJrJr SSKJr SSKJ r SSK r SSK J r SSKJrJrJr SS KJr SS KJrJr SS KJrJrJrJr SS KJrJr SS KJ r "SS\5r!"SS\5r""SS\"5r#"SS\"5r$"SS\"5r%"SS\"5r&"SS\"5r'"SS\"5r("SS\"5r)g) z9Vectorizes observation wrappers to works for `VectorEnv`.) annotations)CallableSequence)deepcopy)AnyN)Space)ActTypeEnvObsType)warn) VectorEnvVectorObservationWrapper) batch_space concatenatecreate_empty_arrayiterate) ArrayType AutoresetMode)transform_observationcP^\rSrSrSrSSU4SjjjrSSjrSrU=r$) TransformObservationaTransforms an observation via a function provided to the wrapper. This function allows the manual specification of the vector-observation function as well as the single-observation function. This is desirable when, for example, it is possible to process vector observations in parallel or via other more optimized methods. Otherwise, the ``VectorizeTransformObservation`` should be used instead, where only ``single_func`` needs to be defined. Example - Without observation transformation: >>> import gymnasium as gym >>> envs = gym.make_vec("CartPole-v1", num_envs=3, vectorization_mode="sync") >>> obs, info = envs.reset(seed=123) >>> obs array([[ 0.01823519, -0.0446179 , -0.02796401, -0.03156282], [ 0.02852531, 0.02858594, 0.0469136 , 0.02480598], [ 0.03517495, -0.000635 , -0.01098382, -0.03203924]], dtype=float32) >>> envs.close() Example - With observation transformation: >>> import gymnasium as gym >>> from gymnasium.spaces import Box >>> def scale_and_shift(obs): ... return (obs - 1.0) * 2.0 ... >>> import gymnasium as gym >>> envs = gym.make_vec("CartPole-v1", num_envs=3, vectorization_mode="sync") >>> new_obs_space = Box(low=envs.observation_space.low, high=envs.observation_space.high) >>> envs = TransformObservation(envs, func=scale_and_shift, observation_space=new_obs_space) >>> obs, info = envs.reset(seed=123) >>> obs array([[-1.9635296, -2.0892358, -2.055928 , -2.0631256], [-1.9429494, -1.9428282, -1.9061728, -1.9503881], [-1.9296501, -2.00127 , -2.0219676, -2.0640786]], dtype=float32) >>> envs.close() c p>[TU]U5 Uc$Ub X@l[X@R5UlOX0lUbX@lUR [URUR5:wa<[SUSUR S[URUR535 X lg)aConstructor for the transform observation wrapper. Args: env: The vector environment to wrap func: A function that will transform the vector observation. If this transformed observation is outside the observation space of ``env.observation_space`` then provide an ``observation_space``. observation_space: The observation spaces of the wrapper. If None, then it is computed from ``single_observation_space``. If ``single_observation_space`` is not provided either, then it is assumed to be the same as ``env.observation_space``. single_observation_space: The observation space of the non-vectorized environment. If None, then it is assumed the same as ``env.single_observation_space``. NzFor zl, the observation space and the batched single observation space don't match as expected, observation_space=z#, batched single_observation_space=) super__init__single_observation_spacernum_envsobservation_space_single_observation_spacer func)selfenvr rr __class__s i/home/james-whalen/.local/lib/python3.13/site-packages/gymnasium/wrappers/vector/vectorize_observation.pyrTransformObservation.__init__8s   $'30H-)4,mm*&&7 "'31I.  ! ![  ) )4==&   seHILI^I^H__BCNOSOlOlnrn{n{C|B}~  c$URU5$)z)Apply function to the vector observation.)r r! observationss r$r)!TransformObservation.observations]syy&&r&)rr rr)NN)r"r r zCallable[[ObsType], Any]r Space | Nonerr+r)r returnr ) __name__ __module__ __qualname____firstlineno____doc__rr)__static_attributes__ __classcell__r#s@r$rrsL!N+/15 # #'#( # #/ ##J''r&rcn^\rSrSrSr"SS\5rS U4SjjrS SjrS Sjr Sr U=r $) VectorizeTransformObservationbaVectorizes a single-agent transform observation wrapper for vector environments. Most of the lambda observation wrappers for single agent environments have vectorized implementations, it is advised that users simply use those instead via importing from `gymnasium.wrappers.vector...`. The following example illustrate use-cases where a custom lambda observation wrapper is required. Example - The normal observation: >>> import gymnasium as gym >>> envs = gym.make_vec("CartPole-v1", num_envs=3, vectorization_mode="sync") >>> obs, info = envs.reset(seed=123) >>> envs.close() >>> obs array([[ 0.01823519, -0.0446179 , -0.02796401, -0.03156282], [ 0.02852531, 0.02858594, 0.0469136 , 0.02480598], [ 0.03517495, -0.000635 , -0.01098382, -0.03203924]], dtype=float32) Example - Applying a custom lambda observation wrapper that duplicates the observation from the environment >>> import numpy as np >>> import gymnasium as gym >>> from gymnasium.spaces import Box >>> from gymnasium.wrappers import TransformObservation >>> envs = gym.make_vec("CartPole-v1", num_envs=3, vectorization_mode="sync") >>> old_space = envs.single_observation_space >>> new_space = Box(low=np.array([old_space.low, old_space.low]), high=np.array([old_space.high, old_space.high])) >>> envs = VectorizeTransformObservation(envs, wrapper=TransformObservation, func=lambda x: np.array([x, x]), observation_space=new_space) >>> obs, info = envs.reset(seed=123) >>> envs.close() >>> obs array([[[ 0.01823519, -0.0446179 , -0.02796401, -0.03156282], [ 0.01823519, -0.0446179 , -0.02796401, -0.03156282]], [[ 0.02852531, 0.02858594, 0.0469136 , 0.02480598], [ 0.02852531, 0.02858594, 0.0469136 , 0.02480598]], [[ 0.03517495, -0.000635 , -0.01098382, -0.03203924], [ 0.03517495, -0.000635 , -0.01098382, -0.03203924]]], dtype=float32) c"\rSrSrSrSSjrSrg)(VectorizeTransformObservation._SingleEnvz@Fake single-agent environment used for the single-agent wrapper.cXlg)z%Constructor for the fake environment.Nr)r!rs r$r1VectorizeTransformObservation._SingleEnv.__init__s%6 "r&r=N)rr)r.r/r0r1r2rr3r&r$ _SingleEnvr:s N 7r&r@c z>[TU]U5 SUR;a%[SUS35 [R UlO8[URS[5(deURSUlU"URURR540UD6Ul URRUl [URUR5Ul URURR:HUl[!URUR5Ulg)zConstructor for the vectorized transform observation wrapper. Args: env: The vector environment to wrap. wrapper: The wrapper to vectorize **kwargs: Keyword argument for the wrapper autoreset_modezVector environment (z+) is missing `autoreset_mode` metadata key.N)rrmetadatar r NEXT_STEPrB isinstancer@r"rwrapperrrrsame_outrout)r!r"rFkwargsr#s r$r&VectorizeTransformObservation.__init__s  3<< / &se+VW #0"9"9D cll+; BH  )- (F(F%!,  ) )4==" ..$((2L2LL %d&C&CT]]Sr&cPURRU5up#pEnURU5nUR[R :XaTSU;aNUSn[ [XvS55H.unupU (dMURRU 5Xx'M0 X#XEU4$)zoSteps through the vector environments, transforming the observation and for final obs individually transformed. final_obs _final_obs) r"stepr)rBr SAME_STEP enumerateziprF observation) r!actionsobsrewards terminations truncationsinfosrLisub_obs has_final_obss r$rN"VectorizeTransformObservation.steps:>w9O6l$   -"9"9 9kU>Rk*I/8I\230++G!=#'<<#;#;G#DIL 0 \==r&c X^TR(a>[TR[U4Sj[ TR U555U5$[ [TR[U4Sj[ TRR U555TR55$)zIterates over the vector observations applying the single-agent wrapper ``observation`` then concatenates the observations together again.c3Z># UH nTRRU5v M" g7fNrFr .0rTr!s r$ =VectorizeTransformObservation.observations..s*LLL%%c**L(+c3Z># UH nTRRU5v M" g7fr_r`ras r$rcrds*#TC ))#..#Tre) rGrrtuplerrrr"rHr(s` r$r)*VectorizeTransformObservation.observationss ==--&t'='=|L  11#*488+E+E|#THH   r&)rBrrHrGrrF)r"r rFz0type[transform_observation.TransformObservation]rIr)rSr r-z?tuple[ObsType, ArrayType, ArrayType, ArrayType, dict[str, Any]]r,) r.r/r0r1r2r r@rrNr)r3r4r5s@r$r7r7bs[&P7S7!T !TB!T !TF>> H>$r&r7c0^\rSrSrSrSU4SjjrSrU=r$)FilterObservationa|Vector wrapper for filtering dict or tuple observation spaces. Example - Create a vectorized environment with a Dict space to demonstrate how to filter keys: >>> import numpy as np >>> import gymnasium as gym >>> from gymnasium.spaces import Dict, Box >>> from gymnasium.wrappers import TransformObservation >>> from gymnasium.wrappers.vector import VectorizeTransformObservation, FilterObservation >>> envs = gym.make_vec("CartPole-v1", num_envs=3, vectorization_mode="sync") >>> make_dict = lambda x: {"obs": x, "junk": np.array([0.0])} >>> new_space = Dict({"obs": envs.single_observation_space, "junk": Box(low=-1.0, high=1.0)}) >>> envs = VectorizeTransformObservation(env=envs, wrapper=TransformObservation, func=make_dict, observation_space=new_space) >>> envs = FilterObservation(envs, ["obs"]) >>> obs, info = envs.reset(seed=123) >>> envs.close() >>> obs {'obs': array([[ 0.01823519, -0.0446179 , -0.02796401, -0.03156282], [ 0.02852531, 0.02858594, 0.0469136 , 0.02480598], [ 0.03517495, -0.000635 , -0.01098382, -0.03203924]], dtype=float32)} c@>[TU]U[RUS9 g)zConstructor for the filter observation wrapper. Args: env: The vector environment to wrap filter_keys: The subspaces to be included, use a list of strings or integers for ``Dict`` and ``Tuple`` spaces respectivesly ) filter_keysN)rrrrj)r!r"rmr#s r$rFilterObservation.__init__s%  &88k  r&r?)r"r rmzSequence[str | int]r.r/r0r1r2rr3r4r5s@r$rjrjs,   r&rjc0^\rSrSrSrSU4SjjrSrU=r$)FlattenObservationiaObservation wrapper that flattens the observation. Example: >>> import gymnasium as gym >>> envs = gym.make_vec("CarRacing-v3", num_envs=3, vectorization_mode="sync") >>> obs, info = envs.reset(seed=123) >>> obs.shape (3, 96, 96, 3) >>> envs = FlattenObservation(envs) >>> obs, info = envs.reset(seed=123) >>> obs.shape (3, 27648) >>> envs.close() cB>[TU]U[R5 g)zConstructor for any environment's observation space that implements ``spaces.utils.flatten_space`` and ``spaces.utils.flatten``. Args: env: The vector environment to wrap N)rrrrq)r!r"r#s r$rFlattenObservation.__init__s 3FFGr&r?)r"r ror5s@r$rqrqs HHr&rqc4^\rSrSrSrSSU4SjjjrSrU=r$)GrayscaleObservationiaObservation wrapper that converts an RGB image to grayscale. Example: >>> import gymnasium as gym >>> envs = gym.make_vec("CarRacing-v3", num_envs=3, vectorization_mode="sync") >>> obs, info = envs.reset(seed=123) >>> obs.shape (3, 96, 96, 3) >>> envs = GrayscaleObservation(envs) >>> obs, info = envs.reset(seed=123) >>> obs.shape (3, 96, 96) >>> envs.close() c@>[TU]U[RUS9 g)zConstructor for an RGB image based environments to make the image grayscale. Args: env: The vector environment to wrap keep_dim: If to keep the channel in the observation, if ``True``, ``obs.shape == 3`` else ``obs.shape == 2`` )keep_dimN)rrrru)r!r"rwr#s r$rGrayscaleObservation.__init__+s%  &;;h  r&r?)F)r"r rwboolror5s@r$rurus    r&ruc0^\rSrSrSrSU4SjjrSrU=r$)ResizeObservationi7aResizes image observations using OpenCV to shape. Example: >>> import gymnasium as gym >>> envs = gym.make_vec("CarRacing-v3", num_envs=3, vectorization_mode="sync") >>> obs, info = envs.reset(seed=123) >>> obs.shape (3, 96, 96, 3) >>> envs = ResizeObservation(envs, shape=(28, 28)) >>> obs, info = envs.reset(seed=123) >>> obs.shape (3, 28, 28, 3) >>> envs.close() c@>[TU]U[RUS9 g)zConstructor that requires an image environment observation space with a shape. Args: env: The vector environment to wrap shape: The resized observation shape shapeN)rrrr{r!r"r~r#s r$rResizeObservation.__init__Gs  3EEUSr&r?)r"r r~ztuple[int, ...]ror5s@r$r{r{7s TTr&r{c0^\rSrSrSrSU4SjjrSrU=r$)ReshapeObservationiQaReshapes array based observations to shapes. Example: >>> import gymnasium as gym >>> envs = gym.make_vec("CarRacing-v3", num_envs=3, vectorization_mode="sync") >>> obs, info = envs.reset(seed=123) >>> obs.shape (3, 96, 96, 3) >>> envs = ReshapeObservation(envs, shape=(9216, 3)) >>> obs, info = envs.reset(seed=123) >>> obs.shape (3, 9216, 3) >>> envs.close() c@>[TU]U[RUS9 g)zConstructor for env with Box observation space that has a shape product equal to the new shape product. Args: env: The vector environment to wrap shape: The reshaped observation space r}N)rrrrrs r$rReshapeObservation.__init__as  3FFeTr&r?)r"r r~zint | tuple[int, ...]ror5s@r$rrQs UUr&rc<^\rSrSrSrSU4SjjrSrU=r$)RescaleObservationikaLinearly rescales observation to between a minimum and maximum value. Example: >>> import gymnasium as gym >>> envs = gym.make_vec("MountainCar-v0", num_envs=3, vectorization_mode="sync") >>> obs, info = envs.reset(seed=123) >>> obs.min() np.float32(-0.46352962) >>> obs.max() np.float32(0.0) >>> envs = RescaleObservation(envs, min_obs=-5.0, max_obs=5.0) >>> obs, info = envs.reset(seed=123) >>> obs.min() np.float32(-0.90849805) >>> obs.max() np.float32(0.0) >>> envs.close() cB>[TU]U[RUUS9 g)zConstructor that requires the env observation spaces to be a :class:`Box`. Args: env: The vector environment to wrap min_obs: The new minimum observation bound max_obs: The new maximum observation bound )min_obsmax_obsN)rrrr)r!r"rrr#s r$rRescaleObservation.__init__s)   ! 4 4  r&r?)r"r r%np.floating | np.integer | np.ndarrayrrror5s@r$rrks.&  7 7   r&rc0^\rSrSrSrSU4SjjrSrU=r$)DtypeObservationiaObservation wrapper for transforming the dtype of an observation. Example: >>> import numpy as np >>> import gymnasium as gym >>> envs = gym.make_vec("CartPole-v1", num_envs=3, vectorization_mode="sync") >>> obs, info = envs.reset(seed=123) >>> obs.dtype dtype('float32') >>> envs = DtypeObservation(envs, dtype=np.float64) >>> obs, info = envs.reset(seed=123) >>> obs.dtype dtype('float64') >>> envs.close() c@>[TU]U[RUS9 g)zConstructor for Dtype observation wrapper. Args: env: The vector environment to wrap dtype: The new dtype of the observation )dtypeN)rrrr)r!r"rr#s r$rDtypeObservation.__init__s  3DDERr&r?)r"r rrror5s@r$rrs SSr&r)*r2 __future__rcollections.abcrrcopyrtypingrnumpynp gymnasiumrgymnasium.corer r r gymnasium.loggerr gymnasium.vectorr rgymnasium.vector.utilsrrrrgymnasium.vector.vector_envrrgymnasium.wrappersrrr7rjrqrur{rrrr?r&r$rs?".00!@XX@4K'3K'\z$<zz  5  FH6H2 8 8T5T4U6U4& 6& RS4Sr&