he SrSSKJr SSKJr SSKJr SSKJrJ r J r SSK r SSK rSSKJr SSKJrJrJrJr SSKJrJrJr SS KJrJrJr SS KJrJr /S Qr "S S \RB\\\4\RDRF5r$"SS\RB\\\4\RDRF5r%"SS\RL\\\\4\RDRF5r'"SS\RB\\\4\RDRF5r("SS\RL\\\\4\RDRF5r)g)aA collection of stateful observation wrappers. * ``DelayObservation`` - A wrapper for delaying the returned observation * ``TimeAwareObservation`` - A wrapper for adding time aware observations to environment observation * ``FrameStackObservation`` - Frame stack the observations * ``NormalizeObservation`` - Normalized the observations to have unit variance with a moving mean * ``MaxAndSkipObservation`` - Return only every ``skip``-th frame (frameskipping) and return the max between the two last frames. ) annotations)deque)deepcopy)AnyFinal SupportsFloatN)ActTypeObsTypeWrapperActTypeWrapperObsType)BoxDictTuple) batch_space concatenatecreate_empty_array)RunningMeanStdcreate_zero_array)DelayObservationTimeAwareObservationFrameStackObservationNormalizeObservationMaxAndSkipObservationcX^\rSrSrSrS SjrSSS.S U4SjjjrS SjrSrU=r $) r#aAdds a delay to the returned observation from the environment. Before reaching the :attr:`delay` number of timesteps, returned observations is an array of zeros with the same shape as the observation space. No vector version of the wrapper exists. Note: This does not support random delay values, if users are interested, please raise an issue or pull request to add this feature. Example: >>> import gymnasium as gym >>> env = gym.make("CartPole-v1") >>> env.reset(seed=123) (array([ 0.01823519, -0.0446179 , -0.02796401, -0.03156282], dtype=float32), {}) >>> env = DelayObservation(env, delay=2) >>> env.reset(seed=123) (array([0., 0., 0., 0.], dtype=float32), {}) >>> env.step(env.action_space.sample()) (array([0., 0., 0., 0.], dtype=float32), 1.0, False, False, {}) >>> env.step(env.action_space.sample()) (array([ 0.01823519, -0.0446179 , -0.02796401, -0.03156282], dtype=float32), 1.0, False, False, {}) Change logs: * v1.0.0 - Initially added c[R"[U5[R5(d[ S[U535eSU::d[ SU35e[ RRRXS9 [ RRX5 [U5Ul [5Ulg)zInitialises the DelayObservation wrapper with an integer. Args: env: The environment to wrap delay: The number of timesteps to delay observations z5The delay is expected to be an integer, actual type: rz7The delay needs to be greater than zero, actual value: )delayN)np issubdtypetypeinteger TypeError ValueErrorgymutilsRecordConstructorArgs__init__ObservationWrapperintrrobservation_queue)selfenvrs a/home/james-whalen/.local/lib/python3.13/site-packages/gymnasium/wrappers/stateful_observation.pyr'DelayObservation.__init__Bs}}T%["**55GU }U EzI%Q  ''000C ''2!$U /4wNseedoptionscR>URR5 [TU] XS9$)z7Resets the environment, clearing the observation queue.r0)r*clearsuperresetr+r1r2 __class__s r-r6DelayObservation.resetXs( $$&w}$}88r/cURRU5 [UR5UR:aURR 5$[ UR 5$)zReturn the delayed observation.)r*appendlenrpopleftrobservation_spacer+ observations r-r@DelayObservation.observation`sS %%k2 t%% & 3))113 3$T%;%;< >> import gymnasium as gym >>> from gymnasium.wrappers import TimeAwareObservation >>> env = gym.make("CartPole-v1") >>> env = TimeAwareObservation(env) >>> env.observation_space Box([-4.80000019 -inf -0.41887903 -inf 0. ], [4.80000019e+00 inf 4.18879032e-01 inf 5.00000000e+02], (5,), float64) >>> env.reset(seed=42)[0] array([ 0.0273956 , -0.00611216, 0.03585979, 0.0197368 , 0. ]) >>> _ = env.action_space.seed(42) >>> env.step(env.action_space.sample())[0] array([ 0.02727336, -0.20172954, 0.03625453, 0.32351476, 1. ]) Normalize time observation space example: >>> env = gym.make('CartPole-v1') >>> env = TimeAwareObservation(env, normalize_time=True) >>> env.observation_space Box([-4.8 -inf -0.41887903 -inf 0. ], [4.8 inf 0.41887903 inf 1. ], (5,), float32) >>> env.reset(seed=42)[0] array([ 0.0273956 , -0.00611216, 0.03585979, 0.0197368 , 0. ], dtype=float32) >>> _ = env.action_space.seed(42) >>> env.step(env.action_space.sample())[0] array([ 0.02727336, -0.20172954, 0.03625453, 0.32351476, 0.002 ], dtype=float32) Flatten observation space example: >>> env = gym.make("CartPole-v1") >>> env = TimeAwareObservation(env, flatten=False) >>> env.observation_space Dict('obs': Box([-4.8 -inf -0.41887903 -inf], [4.8 inf 0.41887903 inf], (4,), float32), 'time': Box(0, 500, (1,), int32)) >>> env.reset(seed=42)[0] {'obs': array([ 0.0273956 , -0.00611216, 0.03585979, 0.0197368 ], dtype=float32), 'time': array([0], dtype=int32)} >>> _ = env.action_space.seed(42) >>> env.step(env.action_space.sample())[0] {'obs': array([ 0.02727336, -0.20172954, 0.03625453, 0.32351476], dtype=float32), 'time': array([1], dtype=int32)} Change logs: * v0.18.0 - Initially added * v1.0.0 - Remove vector environment support, add ``flatten`` and ``normalize_time`` parameters time) dict_time_keyc^^^[RRRTUUTS9 [RRTU5 UTlUTlURb3URRbURRTl OUn[U[R5(ah[U[RR5(aURTl O-URn[U[R5(aMh[U[RR5(d [!S5eSTlTR (aU4SjTl['SS5nO+STl['STR[(R*S 9n[UR,[.5(aOTUR,R15;de[/TU0UR,R2E5mU4S jTlOg[UR,[65(a,[7UR,R2U4-5mS TlO[/UR,US 9mS TlTR (a'[2R8"T5TlU4SjTlgTTlSTlg)aInitialize :class:`TimeAwareObservation`. Args: env: The environment to apply the wrapper flatten: Flatten the observation to a `Box` of a single dimension normalize_time: if `True` return time in the range [0,1] otherwise return time as remaining timesteps before truncation dict_time_key: For environment with a ``Dict`` observation space, the key for the time space. By default, `"time"`. )flattennormalize_timerQNzaThe environment must be wrapped by a TimeLimit wrapper or the spec specify a `max_episode_steps`.rcf>[R"UTR- /[RS9$Ndtype)rarray max_timestepsfloat32)rPr+s r-/TimeAwareObservation.__init__..s%bhh***+2::7r/g?cJ[R"U/[RS9$rV)rrYint32)rPs r-r\r]sbhhvRXX6Vr/rWc>TU0UE$N)obsrPrQs r-r\r]s t7Ss7Sr/c X4-$rbrcrdrPs r-r\r]ssW}r/rfc XS.$)Nrfrcrfs r-r\r]ss7Qr/c2>[R"TU5$rb)spacesrS)rdr>s r-r\r]sV^^!36r/cU$rbrc)rds r-r\r]sSr/)r$r%r&r'r(rSrTspecmax_episode_stepsrZ isinstanceWrapperwrappers TimeLimit_max_episode_stepsr,r# timesteps_time_preprocess_funcr rr`r>rkeysri_append_data_funcr flatten_space_obs_postprocess_func)r+r,rSrTrQ wrapped_env time_spacer>s` ` @r-r'TimeAwareObservation.__init__sE" ''00 )' 1 ''c2$+ +9 88 CHH$>$>$J!$!;!;D K[#++66k3<<+A+ABB)4)G)GD&)oo [#++66 k3<<+A+ABB w    *D &S#J)VD &Q 2 2"((CJ c++T 2 2 (=(=(B(B(DD DD $ Kc.C.C.J.JK! &TD " --u 5 5 %c&;&;&B&Bj]&R S %DD " $)>)>Z P %QD " <<@F@T@T!AD "*D &ARD ")8D &r/ctURURXRUR555$)zAdds to the observation with the current time information. Args: observation: The observation to add the time step to Returns: The observation with the time information appended to it )rwrursrrr?s r-r@ TimeAwareObservation.observations7))  " "77G   r/cL>U=RS- sl[TU] U5$)zSteps through the environment, incrementing the time step. Args: action: The action to take Returns: The environment's step using the action with the next observation containing the timestep info )rrr5step)r+actionr8s r-rTimeAwareObservation.step s" !w|F##r/Nr0c,>SUl[TU] XS9$)zReset the environment setting the time to zero. Args: seed: The seed to reset the environment options: The options used to reset the environment Returns: Resets the environment with the initial timestep info added the observation rr0)rrr5r6r7s r-r6TimeAwareObservation.resetsw}$}88r/)rurwrsrSrZrTr>rr)TF)r,rBrSboolrTrrQstrr@r rEr )rr rE@tuple[WrapperObsType, SupportsFloat, bool, bool, dict[str, Any]]r1rCr2rDrEz%tuple[WrapperObsType, dict[str, Any]]) rFrGrHrIrJr'r@rr6rKrLrMs@r-rrjs9|$ Q9 $ Q9 &Q9Q9 Q9  Q9f   $ $ I $ %)49!93H9 .99r/rcd\rSrSrSrSS.S SjjrS SjrSSS.S S jjrS rg)ri+a~ Stacks the observations from the last ``N`` time steps in a rolling manner. For example, if the number of stacks is 4, then the returned observation contains the most recent 4 observations. For environment 'Pendulum-v1', the original observation is an array with shape [3], so if we stack 4 observations, the processed observation has shape [4, 3]. Users have options for the padded observation used: * "reset" (default) - The reset value is repeated * "zero" - A "zero"-like instance of the observation space * custom - An instance of the observation space No vector version of the wrapper exists. Example: >>> import gymnasium as gym >>> from gymnasium.wrappers import FrameStackObservation >>> env = gym.make("CarRacing-v3") >>> env = FrameStackObservation(env, stack_size=4) >>> env.observation_space Box(0, 255, (4, 96, 96, 3), uint8) >>> obs, _ = env.reset() >>> obs.shape (4, 96, 96, 3) Example with different padding observations: >>> env = gym.make("CartPole-v1") >>> env.reset(seed=123) (array([ 0.01823519, -0.0446179 , -0.02796401, -0.03156282], dtype=float32), {}) >>> stacked_env = FrameStackObservation(env, 3) # the default is padding_type="reset" >>> stacked_env.reset(seed=123) (array([[ 0.01823519, -0.0446179 , -0.02796401, -0.03156282], [ 0.01823519, -0.0446179 , -0.02796401, -0.03156282], [ 0.01823519, -0.0446179 , -0.02796401, -0.03156282]], dtype=float32), {}) >>> stacked_env = FrameStackObservation(env, 3, padding_type="zero") >>> stacked_env.reset(seed=123) (array([[ 0. , 0. , 0. , 0. ], [ 0. , 0. , 0. , 0. ], [ 0.01823519, -0.0446179 , -0.02796401, -0.03156282]], dtype=float32), {}) >>> stacked_env = FrameStackObservation(env, 3, padding_type=np.array([1, -1, 0, 2], dtype=np.float32)) >>> stacked_env.reset(seed=123) (array([[ 1. , -1. , 0. , 2. ], [ 1. , -1. , 0. , 2. ], [ 0.01823519, -0.0446179 , -0.02796401, -0.03156282]], dtype=float32), {}) Change logs: * v0.15.0 - Initially add as ``FrameStack`` with support for lz4 * v1.0.0 - Rename to ``FrameStackObservation`` and remove lz4 and ``LazyFrame`` support along with adding the ``padding_type`` parameter r6) padding_typec[RRRXUS9 [RRX5 [ R "[U5[ R5(d[S[U535eSU:d[SU35e[U[5(a'US:XdUS:Xa[UR5UlOYX1R;a X0lSnOA[U[5(a[SU<35e[SU<S URS 35e[!URUS 9UlX lX0l['[)UR"5Vs/sHo@RPM snUR"S 9Ul[-URUR"S 9Ulg s snf)aObservation wrapper that stacks the observations in a rolling manner. Args: env: The environment to apply the wrapper stack_size: The number of frames to stack. padding_type: The padding type to use when stacking the observations, options: "reset", "zero", custom obs ) stack_sizerz:The stack_size is expected to be an integer, actual type: rz padding_valuerrrrrange obs_queuer stacked_obs)r+r,rr_s r-r'FrameStackObservation.__init__is ''00 l 1  T'}}T*-rzz::LTR\M]L^_ :~Nzl[  lC ( ( G #|v'=*;Cr)r+rrdreward terminated truncatedinfo updated_obss r-rFrameStackObservation.stepsh4888==3H0ZD c" 22DNNDDTDT U J4??r/Nr0cURRXS9up4URS:XaX0l[ UR S- 5H(nUR RUR5 M* UR RU5 [[URRUR UR55nXd4$)zReset the environment, returning the stacked observation and info. Args: seed: The environment seed options: The reset options Returns: The stacked observations and info r0r6r~) r,r6rrrrrr;rrr>r)r+r1r2rdrrrs r-r6FrameStackObservation.resetsHHNNN>     '!$ t*+A NN ! !$"4"4 5, c" 22DNNDDTDT U   r/)rr>rrrr)r,rBrr)rz str | ObsTyperr rErr) rFrGrHrIrJr'rr6rKrcr/r-rr+st8~'. 3X &3X3X $ 3Xj@$@ I@(%)4!!!3H! .!!r/rcl\rSrSrSrS S Sjjr\S Sj5r\RS Sj5rS Sjr Sr g)riaNormalizes observations to be centered at the mean with unit variance. The property :attr:`update_running_mean` allows to freeze/continue the running mean calculation of the observation statistics. If ``True`` (default), the ``RunningMeanStd`` will get updated every time ``step`` or ``reset`` is called. If ``False``, the calculated statistics are used but not updated anymore; this may be used during evaluation. A vector version of the wrapper exists :class:`gymnasium.wrappers.vector.NormalizeObservation`. Note: The normalization depends on past trajectories and observations will not be normalized correctly if the wrapper was newly instantiated or the policy was changed recently. Example: >>> import numpy as np >>> import gymnasium as gym >>> env = gym.make("CartPole-v1") >>> obs, info = env.reset(seed=123) >>> term, trunc = False, False >>> while not (term or trunc): ... obs, _, term, trunc, _ = env.step(1) ... >>> obs array([ 0.1511158 , 1.7183299 , -0.25533703, -2.8914354 ], dtype=float32) >>> env = gym.make("CartPole-v1") >>> env = NormalizeObservation(env) >>> obs, info = env.reset(seed=123) >>> term, trunc = False, False >>> while not (term or trunc): ... obs, _, term, trunc, _ = env.step(1) >>> obs array([ 2.0059888, 1.5676788, -1.9944268, -1.6120394], dtype=float32) Change logs: * v0.21.0 - Initially add * v1.0.0 - Add `update_running_mean` attribute to allow disabling of updating the running mean / standard, particularly useful for evaluation time. Casts all observations to `np.float32` and sets the observation space with low/high of `-np.inf` and `np.inf` and dtype as `np.float32` c[RRRXS9 [RRX5 UR R ce[RR[R*[RUR R [RS9Ul[UR R UR RS9UlX lSUlg)zThis wrapper will normalize observations such that each observation is centered with unit variance. Args: env (Env): The environment to apply the wrapper epsilon: A stability parameter that is used when scaling the observations. )epsilonN)lowhighshaperX)rrXT)r$r%r&r'r(r>rrir rinfr[rrXobs_rmsr_update_running_mean)r+r,rs r-r'NormalizeObservation.__init__s ''000G ''2$$**666!$''--** "0" &((..d6L6L6R6R   $(!r/cUR$)zWProperty to freeze/continue the running mean calculation of the observation statistics.r)r+s r-update_running_mean(NormalizeObservation.update_running_mean s(((r/cXlg)z`Sets the property to freeze/continue the running mean calculation of the observation statistics.Nr)r+settings r-rrs %,!r/cLUR(a0URR[R"U/55 [R "XRR - [R"URRUR-5- 5$)zSNormalises the observation using the running mean and variance of the observations.) rrupdaterrYr[meansqrtvarrr?s r-r@ NormalizeObservation.observationsj  $ $ LL  + 7 8zz <<,, , 8H8H4<<8W0X X  r/)rrrr>N)g:0yE>)r,rBrfloat)rEr)rrr) rFrGrHrIrJr'propertyrsetterr@rKrcr/r-rrs@$L)0)), , r/rc8\rSrSrSrSSSjjrS SjrSrg) ri aSkips the N-th frame (observation) and return the max values between the two last observations. No vector version of the wrapper exists. Note: This wrapper is based on the wrapper from [stable-baselines3](https://stable-baselines3.readthedocs.io/en/master/_modules/stable_baselines3/common/atari_wrappers.html#MaxAndSkipEnv) Example: >>> import gymnasium as gym >>> env = gym.make("CartPole-v1") >>> obs0, *_ = env.reset(seed=123) >>> obs1, *_ = env.step(1) >>> obs2, *_ = env.step(1) >>> obs3, *_ = env.step(1) >>> obs4, *_ = env.step(1) >>> skip_and_max_obs = np.max(np.stack([obs3, obs4], axis=0), axis=0) >>> env = gym.make("CartPole-v1") >>> wrapped_env = MaxAndSkipObservation(env) >>> wrapped_obs0, *_ = wrapped_env.reset(seed=123) >>> wrapped_obs1, *_ = wrapped_env.step(1) >>> np.all(obs0 == wrapped_obs0) np.True_ >>> np.all(wrapped_obs1 == skip_and_max_obs) np.True_ Change logs: * v1.0.0 - Initially add c([RRRXS9 [RRX5 [ R "[U5[ R5(d[S[U535eUS:a[SU35eURRc [S5eX l [ R"S/URRQ7URRS9Ulg)zThis wrapper will return only every ``skip``-th frame (frameskipping) and return the max between the two last frames. Args: env (Env): The environment to apply the wrapper skip: The number of frames to skip )skipz4The skip is expected to be an integer, actual type: zDThe skip value needs to be equal or greater than two, actual value: Nz4The observation space must have the shape attribute.rW)r$r%r&r'rnrrr r!r"r#r>r_skipzerosrX _obs_buffer)r+r,rs r-r'MaxAndSkipObservation.__init__As ''000A T'}}T$Z44FtDzlS  !8VW[V\]   & & .ST T 88  -&&,, -S5J5J5P5P r/cSnS=p40n[UR5HnURRU5upxp4nX`RS- :XaXpRS'X`RS- :XaXpRS'U[ U5- nU(d U(dM O [ R"URSS9n XX4U4$)a*Step the environment with the given action for ``skip`` steps. Repeat action, sum reward, and max over last observations. Args: action: The action to step through the environment with Returns: Max of the last two observations, reward, terminated, truncated, and info from the environment r^Frrr~)axis)rrr,rrrrmax) r+r total_rewardrrrirdr max_frames r-rMaxAndSkipObservation.step[s !&& tzz"A7;xx}}V7L 4CJJN"&)  #JJN"&)  # E&M )LYY#FF4++!4  tCCr/)rrN))r,rBrr)r)rFrGrHrIrJr'rrKrcr/r-rr s$: 4D$D IDr/r)*rJ __future__r collectionsrcopyrtypingrrrnumpyr gymnasiumr$gymnasium.spacesrigymnasium.corer r r r r rrgymnasium.vector.utilsrrrgymnasium.wrappers.utilsrr__all__r(r%r&rrrnrrrrcr/r-rs+#,,!KK--OOF D=7GW45syy7V7VD=N~9>7G;<II##~9B]!KK'9:II##]!@R >7G;<II##R jUDKK'9:II##UDr/