h`t SrSSKJr SSKrSSKrSSKJr SSKJr SSK J r J r J r SSK rSSKrSSKJrJr SSKJrJrJr SS KJrJr /S Qr"S S \R6\\\\4\ \\\4\R8R:5r"S S\R6\\\\4\ \\\4\R8R:5r"SS\R6\\\\4\R8R:5r "SS\R6\\\\4\R8R:5r!"SS\R6\\\\4\R8R:5r"g)aA collections of rendering-based wrappers. * ``RenderCollection`` - Collects rendered frames into a list * ``RecordVideo`` - Records a video of the environments * ``HumanRendering`` - Provides human rendering of environments with ``"rgb_array"`` * ``AddWhiteNoise`` - Randomly replaces pixels with white noise * ``ObstructView`` - Randomly places patches of white noise to obstruct the pixel rendering ) annotationsN)Callable)deepcopy)AnyGeneric SupportsFloat)errorlogger)ActTypeObsType RenderFrame)DependencyNotInstalledInvalidProbability)RenderCollection RecordVideoHumanRendering AddWhiteNoise ObstructViewc^\rSrSrSrS S Sjjr\S5rS U4SjjrSSS.SU4Sjjjr SS jr S r U=r $)r#aCollect rendered frames of an environment such ``render`` returns a ``list[RenderedFrame]``. No vector version of the wrapper exists. Example - Return the list of frames for the number of steps ``render`` wasn't called. >>> import gymnasium as gym >>> env = gym.make("LunarLander-v3", render_mode="rgb_array") >>> env = RenderCollection(env) >>> _ = env.reset(seed=123) >>> for _ in range(5): ... _ = env.step(env.action_space.sample()) ... >>> frames = env.render() >>> len(frames) 6 >>> frames = env.render() >>> len(frames) 0 Example - Return the list of frames for the number of steps the episode was running. >>> import gymnasium as gym >>> env = gym.make("LunarLander-v3", render_mode="rgb_array") >>> env = RenderCollection(env, pop_frames=False) >>> _ = env.reset(seed=123) >>> for _ in range(5): ... _ = env.step(env.action_space.sample()) ... >>> frames = env.render() >>> len(frames) 6 >>> frames = env.render() >>> len(frames) 6 Example - Collect all frames for all episodes, without clearing them when render is called >>> import gymnasium as gym >>> env = gym.make("LunarLander-v3", render_mode="rgb_array") >>> env = RenderCollection(env, pop_frames=False, reset_clean=False) >>> _ = env.reset(seed=123) >>> for _ in range(5): ... _ = env.step(env.action_space.sample()) ... >>> _ = env.reset(seed=123) >>> for _ in range(5): ... _ = env.step(env.action_space.sample()) ... >>> frames = env.render() >>> len(frames) 12 >>> frames = env.render() >>> len(frames) 12 Change logs: * v0.26.2 - Initially added c"[RRRXUS9 [RRX5 UR ceUR R S5(ae/UlX lX0l [URR5Ul URR S3URS;a6URSRURR S35 gg)aVInitialize a :class:`RenderCollection` instance. Args: env: The environment that is being wrapped pop_frames (bool): If true, clear the collection frames after ``meth:render`` is called. Default value is ``True``. reset_clean (bool): If true, clear the collection frames when ``meth:reset`` is called. Default value is ``True``. ) pop_frames reset_cleanN_list render_modes)gymutilsRecordConstructorArgs__init__Wrapper render_modeendswith frame_listrrrenvmetadataappend)selfr$rrs V/home/james-whalen/.local/lib/python3.13/site-packages/gymnasium/wrappers/rendering.pyrRenderCollection.__init__ds ''00 [ 1  T'***??++G4444-/$& !2!23 hh"" #5 )~1N N MM. ) 0 0DHH4H4H3I1O P Oc4URRS3$)z(Returns the collection render_mode name.r)r$r!r's r(r!RenderCollection.render_modes((&&'u--r*ct>[TU]U5nURR[TU]55 U$)z;Perform a step in the base environment and collect a frame.)superstepr#r&render)r'actionoutput __class__s r(r0RenderCollection.steps1f% uw~/0 r*Nseedoptionsc>[TU]XS9nUR(a/UlURR [TU]55 U$)zQReset the base environment, eventually clear the frame_list, and collect a frame.r6)r/resetrr#r&r1)r'r7r8r3r4s r(r:RenderCollection.resetsCD:    DO uw~/0 r*cNURnUR(a/UlU$)zFReturns the collection of frames and, if pop_frames = True, clears it.)r#r)r'framess r(r1RenderCollection.renders ?? DO r*)r#r%rr)TT)r$gym.Env[ObsType, ActType]rboolrr@r2r returnz9tuple[ObsType, SupportsFloat, bool, bool, dict[str, Any]]r7 int | Noner8zdict[str, Any] | NonerBztuple[ObsType, dict[str, Any]])rBzlist[RenderFrame]) __name__ __module__ __qualname____firstlineno____doc__rpropertyr!r0r:r1__static_attributes__ __classcell__r4s@r(rr#s :~  Q &QQ Q:.. B%)4 ! 3H  '  r*rc^\rSrSrSrSSSSSSS4SSjjrS rSSS .SU4S jjjrSS jrSU4S jjr U4Sjr SSjr Sr Sr SrU=r$)raT Records videos of environment episodes using the environment's render function. .. py:currentmodule:: gymnasium.utils.save_video Usually, you only want to record episodes intermittently, say every hundredth episode or at every thousandth environment step. To do this, you can specify ``episode_trigger`` or ``step_trigger``. They should be functions returning a boolean that indicates whether a recording should be started at the current episode or step, respectively. The ``episode_trigger`` should return ``True`` on the episode when recording should start. The ``step_trigger`` should return ``True`` on the n-th environment step that the recording should be started, where n sums over all previous episodes. If neither :attr:`episode_trigger` nor ``step_trigger`` is passed, a default ``episode_trigger`` will be employed, i.e. :func:`capped_cubic_video_schedule`. This function starts a video at every episode that is a power of 3 until 1000 and then every 1000 episodes. By default, the recording will be stopped once reset is called. However, you can also create recordings of fixed length (possibly spanning several episodes) by passing a strictly positive value for ``video_length``. Examples - Run the environment for 50 episodes, and save the video every 10 episodes starting from the 0th: >>> import os >>> import gymnasium as gym >>> env = gym.make("LunarLander-v3", render_mode="rgb_array") >>> trigger = lambda t: t % 10 == 0 >>> env = RecordVideo(env, video_folder="./save_videos1", episode_trigger=trigger, disable_logger=True) >>> for i in range(50): ... termination, truncation = False, False ... _ = env.reset(seed=123) ... while not (termination or truncation): ... obs, rew, termination, truncation, info = env.step(env.action_space.sample()) ... >>> env.close() >>> len(os.listdir("./save_videos1")) 5 Examples - Run the environment for 5 episodes, start a recording every 200th step, making sure each video is 100 frames long: >>> import os >>> import gymnasium as gym >>> env = gym.make("LunarLander-v3", render_mode="rgb_array") >>> trigger = lambda t: t % 200 == 0 >>> env = RecordVideo(env, video_folder="./save_videos2", step_trigger=trigger, video_length=100, disable_logger=True) >>> for i in range(5): ... termination, truncation = False, False ... _ = env.reset(seed=123) ... _ = env.action_space.seed(123) ... while not (termination or truncation): ... obs, rew, termination, truncation, info = env.step(env.action_space.sample()) ... >>> env.close() >>> len(os.listdir("./save_videos2")) 2 Examples - Run 3 episodes, record everything, but in chunks of 1000 frames: >>> import os >>> import gymnasium as gym >>> env = gym.make("LunarLander-v3", render_mode="rgb_array") >>> env = RecordVideo(env, video_folder="./save_videos3", video_length=1000, disable_logger=True) >>> for i in range(3): ... termination, truncation = False, False ... _ = env.reset(seed=123) ... while not (termination or truncation): ... obs, rew, termination, truncation, info = env.step(env.action_space.sample()) ... >>> env.close() >>> len(os.listdir("./save_videos3")) 2 Change logs: * v0.25.0 - Initially added to replace ``wrappers.monitoring.VideoRecorder`` Nrzrl-videoTcg)NT)episodes r(RecordVideo.s4r*c [RRRUUUUUUUS9 [RRX5 UR S;a[ SUR S3S5eUc UcSSKJn U nX0l X@l Xl Xl [RRU5Ul[RR#UR 5(a$[$R&"S UR S 35 [R("UR S S 9 UcUR*R-S S5nXplX`lSUlUS:waUO [5S5UlSUl/Ul/UlSUlSUl SSK!n g![Dan [FRH"S5U eSn A ff=f)a@Wrapper records videos of rollouts. Args: env: The environment that will be wrapped video_folder (str): The folder where the recordings will be stored episode_trigger: Function that accepts an integer and returns ``True`` iff a recording should be started at this episode step_trigger: Function that accepts an integer and returns ``True`` iff a recording should be started at this step video_length (int): The length of recorded episodes. If 0, entire episodes are recorded. Otherwise, snippets of the specified length are captured name_prefix (str): Will be prepended to the filename of the recordings fps (int): The frame per second in the video. Provides a custom video fps for environment, if ``None`` then the environment metadata ``render_fps`` key is used if it exists, otherwise a default value of 30 is used. disable_logger (bool): Whether to disable moviepy logger or not, default it is disabled gc_trigger: Function that accepts an integer and returns ``True`` iff garbage collection should be performed after this episode ) video_folderepisode_trigger step_trigger video_length name_prefixdisable_logger>NansihumanzRender mode is z), which is incompatible with RecordVideo.zXInitialize your environment with a render_mode that returns an image, such as rgb_array.Nr)capped_cubic_video_schedulezOverwriting existing videos at zh folder (try specifying a different `video_folder` for the `RecordVideo` wrapper if this is not desired)T)exist_ok render_fpsinfF>MoviePy is not installed, run `pip install "gymnasium[other]"`)%rrrrr r! ValueErrorgymnasium.utils.save_videor^rWrXr[ gc_triggerospathabspathrVisdirr warnmakedirsr%getframes_per_secrZ _video_namefloatrY recordingrecorded_framesrender_historystep_id episode_idmoviepy ImportErrorr r) r'r$rVrWrXrYrZfpsr[rgr^rwes r(rRecordVideo.__init__s6 ''00 %+%%#) 1  T' ??5 5!#//!22[\j   "|'; N9O.(,$GGOOL9 77==** + + KK1$2C2C1DEst  D%%5 ;--##L"5C#& +'+1=1Be $2413    ..P  s F%% G /GG cUR(dS5eURR5n[U[5(a@[ U5S:Xa[ R"S5 gU=RU- slUSn[U[R5(aURRU5 gUR5 [ R"S[U5S35 g)Nz1Cannot capture a frame, recording wasn't started.rzeTrying to capture render frame but 'env.render()' has just been called. The frame cannot be captured.rcz^Recording stopped: expected type of frame returned by render to be a numpy array, got instead .)rrr$r1 isinstancelistlenr rlrtnpndarrayrsr&stop_recordingtype)r'frames r(_capture_frameRecordVideo._capture_frame?s~~RRR~! eT " "5zQ {   5 ( "IE eRZZ ( (  ' ' .    ! KKpquv{q|p}}~ r*r6c$>[TU]XS9up4U=RS- slUR(a)UR[ S5:XaUR 5 UR(aJURUR5(a*URURSUR35 UR(aCUR5 [UR5UR:aUR 5 X44$)z>d//5<?    !   D$8$8$I$I  D$4$4#5Yt>O!P Q >>    !4''(4+<+<<##%yr*cURRU5up#pEnU=RS- slUR(aJURUR5(a*UR UR SUR35 UR (aCUR5 [UR5UR:aUR5 X#XEU4$)z]Steps through the environment using action, recording observations if :attr:`self.recording`.rz-step-) r$r0rurXrrZrrrrrsrYr)r'r2rrew terminated truncatedrs r(r0RecordVideo.stepgs15 f0E-*    !2!24<!J K >>    !4''(4+<+<<##%44r*c>[TU]5nUR(a*[U[5(aU=R U- sl[ UR5S:aURn/UlX!-$U$)ziCompute the render frames as specified by render_mode attribute during initialization of the environment.r)r/r1rrr~rrsrrt)r' render_out tmp_historyr4s r(r1RecordVideo.renderxshW^% >>jT::  J . t"" #a '--K"$D + + r*cf>[TU]5 UR(aUR5 gg)z+Closes the wrapper then the video recorder.N)r/closerrr)r'r4s r(rRecordVideo.closes$   >>    ! r*c`UR(aUR5 SUlXlg)zkStart a new recording. If it is already recording, stops the current recording before starting the new one.TN)rrrrp)r' video_names r(rRecordVideo.start_recordings" >>    !%r*cUR(dS5e[UR5S:Xa[R"S5 O~SSKJn U"URURS9nUR(aSOSn[RRURUR S 35nUR#XTS 9 AU?/UlS UlSUlUR$(a7UR%UR&5(a[(R*"5 ggg![an[R"S5UeSnAff=f) z+Stop current recording and saves the video.z7stop_recording was called, but no recording was startedrz9Ignored saving a video as there were zero frames to save.)ImageSequenceCliprdN)rybarz.mp4)r F)rrrrsr rl"moviepy.video.io.ImageSequenceCliprrxr rror[rhrijoinrVrpwrite_videofilergrvgccollect)r'rrzclipmoviepy_loggerris r(rRecordVideo.stop_recordings~~XXX~ t## $ ) KKS T P %T%9%9t?R?RSD%)%8%8TeN77<< 1 1d6F6F5Gt3LMD   =  ! ??tt?? JJL @?! 22T s D'' E 1EE cd[UR5S:a[R"S5 gg)z.Warn the user in case last video wasn't saved.rz0Unable to save last video! Did you call close()?N)rrsr rlr,s r(__del__RecordVideo.__del__s' t## $q ( KKJ K )r*)rpr[rvrWrorgrZrsrrrtrurXrVrY)r$r?rVstrrWCallable[[int], bool] | NonerXrrYintrZrryrDr[r@rgrrCrA)rBzRenderFrame | list[RenderFrame])rr)rErFrGrHrIrrr:r0r1rrrrrKrLrMs@r(rrs CR9=59%#3GP &PP6 P 3 P  PPPP1Pd,%)4!3H '&55 B5" " &8LLr*rc^\rSrSrSr/SQrSSjr\S5rSU4Sjjr SSS.SU4S jjjr SS jr S r U4S jr S rU=r$)riaAllows human like rendering for environments that support "rgb_array" rendering. This wrapper is particularly useful when you have implemented an environment that can produce RGB images but haven't implemented any code to render the images to the screen. If you want to use this wrapper with your environments, remember to specify ``"render_fps"`` in the metadata of your environment. The ``render_mode`` of the wrapped environment must be either ``'rgb_array'`` or ``'rgb_array_list'``. No vector version of the wrapper exists. Example: >>> import gymnasium as gym >>> from gymnasium.wrappers import HumanRendering >>> env = gym.make("LunarLander-v3", render_mode="rgb_array") >>> wrapped = HumanRendering(env) >>> obs, _ = wrapped.reset() # This will start rendering to the screen The wrapper can also be applied directly when the environment is instantiated, simply by passing ``render_mode="human"`` to ``make``. The wrapper will only be applied if the environment does not implement human-rendering natively (i.e. ``render_mode`` does not contain ``"human"``). >>> env = gym.make("phys2d/CartPole-v1", render_mode="human") # CartPoleJax-v1 doesn't implement human-rendering natively >>> obs, _ = env.reset() # This will start rendering to the screen Warning: If the base environment uses ``render_mode="rgb_array_list"``, its (i.e. the *base environment's*) render method will always return an empty list: >>> env = gym.make("LunarLander-v3", render_mode="rgb_array_list") >>> wrapped = HumanRendering(env) >>> obs, _ = wrapped.reset() >>> env.render() # env.render() will always return an empty list! [] Change logs: * v0.25.0 - Initially added ) rgb_arrayrgb_array_list depth_arraydepth_array_listc8[RRRU5 [RRX5 SUlSUlSUlURRUR;d"SURSURS35eSURR;dS5eSURS;aC[URR5Ul URSRS5 gg) zeInitialize a :class:`HumanRendering` instance. Args: env: The environment that is being wrapped Nz&Expected env.render_mode to be one of z but got ''r`zYThe base environment must specify 'render_fps' to be used with the HumanRendering wrapperr]r)rrrrr screen_sizewindowclockr$r!ACCEPTED_RENDER_MODESr%rr&)r'r$s r(rHumanRendering.__init__s ''006 T'   HH D$>$> > m 3D4N4N3OzZ]ZiZiYjjk l m > DHH-- - g f g - $--7 7$TXX%6%67DM MM. ) 0 0 9 8r*cg)zAlways returns ``'human'``.r]rQr,s r(r!HumanRendering.render_modesr*cF>[TU]U5nUR5 U$)zHPerform a step in the base environment and render a frame to the screen.)r/r0 _render_frame)r'r2resultr4s r(r0HumanRendering.steps"f%  r*Nr6cB>[TU]XS9nUR5 U$)zURb.SSKnURR5 UR5 [TU]5 g)zClose the rendering window.Nr)rrrquitr/r)r'rr4s r(rHumanRendering.close?s2 ;; "  NN   ! KKM  r*)rr%rr)r$r?)r2r rBz/tuple[ObsType, SupportsFloat, bool, bool, dict]rC)rBNone)rErFrGrHrIrrrJr!r0r:r1rrrKrLrMs@r(rrsm$L:0%)4!3H ')Vr*rcJ^\rSrSrSrSSSjjrSU4SjjrSrU=r$) riIa9Randomly replaces pixels with white noise. If used with ``render_mode="rgb_array"`` and ``AddRenderObservation``, it will make observations noisy. The environment may also become partially-observable, turning the MDP into a POMDP. Example - Every pixel will be replaced by white noise with probability 0.5: >>> env = gym.make("LunarLander-v3", render_mode="rgb_array") >>> env = AddWhiteNoise(env, probability_of_noise_per_pixel=0.5) >>> env = HumanRendering(env) >>> obs, _ = env.reset(seed=123) >>> obs, *_ = env.step(env.action_space.sample()) cSUs=::aS:dO [SU35e[RRR UUUS9 [R R X5 X lX0lg)zWrapper replaces random pixels with white noise. Args: env: The environment that is being wrapped probability_of_noise_per_pixel: the probability that a pixel is white noise is_noise_grayscale: if True, RGB noise is converted to grayscale rrzIprobability_of_noise_per_pixel should be in the interval [0,1). Received )probability_of_noise_per_pixelis_noise_grayscaleN)rrrrrr rr)r'r$rrs r(rAddWhiteNoise.__init__Zss26Q6$[\z[{|  ''00 +I1 1 T'.L+"4r*c >[TU]5nUR(alURR SS[ R "/SQ5-UR[ RS9RSSS9RSS5nO4URR S SUR[ RS9nURRURS S 5UR:n[ R"US X!5$) zCompute the render frames as specified by render_mode attribute during initialization of the environment, then add white noise.rrrgŏ1w-!?gbX9?gv/?sizedtypercTkeepdimsrr.N)r/r1r np_randomintegersrarrayruint8sumrepeatrandomrwhere)r'rnoisemaskr4s r(r1AddWhiteNoise.rendervsW^%  " "''"((#;<<#))(( ( R$'2 NN++%%hh ,E NN ! !*"2"21Q"7 811 2 xxY;;r*)rrF)r$r?rrqrr@rBr rErFrGrHrIrr1rKrLrMs@r(rrIs8 $$) 5 &5).5! 58<>> env = gym.make("LunarLander-v3", render_mode="rgb_array") >>> env = ObstructView(env, obstructed_pixels_ratio=0.5, obstruction_width=50) >>> env = HumanRendering(env) >>> obs, _ = env.reset(seed=123) >>> obs, *_ = env.step(env.action_space.sample()) c(SUs=::aS:dO [SU35eUS:a[SU35e[RRR UUUUS9 [R R X5 X#S-- UlX0lX@lg)a<Wrapper obstructs pixels with white noise patches. Args: env: The environment that is being wrapped obstructed_pixels_ratio: the percentage of pixels obstructed with white noise obstruction_width: the width of the obstruction patches is_noise_grayscale: if True, RGB noise is converted to grayscale rrzBobstructed_pixels_ratio should be in the interval [0,1). Received z=obstruction_width should be larger or equal than 1. Received )obstructed_pixels_ratioobstruction_widthrrN) rerrrrr obstruction_centers_ratiorr)r'r$rrrs r(rObstructView.__init__s+/a/TUlTmn  q OPaObc  ''00 $;/1 1 T')@VWCW)W&!2"4r*c >[T U]5nURnUSUS-n[X0R-5nUR R SX45n[R"XRSUS45n[R"USUS4[S9nURS-nURU- n[U6HFupSU[X- S5[X-US52[X- S5[X-US524'MH UR(alUR R SS[R "/SQ5-UR[R"S 9R%S SS 9R'S S 5n O4UR R SSUR[R"S 9n [R("US X5$)zCompute the render frames as specified by render_mode attribute during initialization of the environment, then add white noise patches.rr)rrTrrrrrcrrr)r/r1rrrrrr unravel_indexzerosr@rzipmaxminrrrrrr) r'r render_shapen_pixelsn_obstructionscentersrlowhighxyrr4s r(r1ObstructView.rendersW^% !'' ?\!_4X(F(FFG..))!XF""7!_l1o,NOxxa,q/:$G$$)%%+MDA AGQ#ah Q"@@AGQ#ah Q"@@B "  " "''"((#;<<#))(( ( R$'2 NN++%%hh ,ExxY;;r*)rrrr)r$r?rrqrrrr@rrrMs@r(rrsC*$) #5 &#5"'#5 #5 ! #5J%<%rsA# $..#88F }KK'723 GWk )*II##}@OLKK'723 GWk )*II##OLdQKK'723SYY5T5TQhI<KK'723SYY5T5TI