&`i ddlZddlZddlmZddlmZmZmZmZm Z m Z m Z m Z ddl ZddlZddlmZmZddlmZddlmZmZddlmZddlmZdd lmZdd lmZdd l m!Z!dd l"m#Z#m$Z$m%Z%dd l&m'Z'ddl(m)Z)ddl*m+Z+ddl,m-Z-m.Z.ddl/m0Z0m1Z1m2Z2m3Z3m4Z4m5Z5m6Z6m7Z7m8Z8m9Z9m:Z:m;Z;mm?Z?erddl@mAZAddlBmCZCddlDmEZEejFeGZHdZIdZJdZKe'GddZLe'GddeZMe'deNdedeNd eOd!e#f d"ZPe'd#ee$d!e$fd$ZQe'Gd%d&ZRd'ed!eefd(ZSd)ee9e e!ee)ffd*e9d!e e!ee)ffd+ZTdS),N) defaultdict) TYPE_CHECKINGDictIteratorListOptionalSetTupleUnion)ASYNC_RESET_RETURNBaseEnv)ExternalEnvWrapper) MonitorEnvget_wrapper_by_cls)_PolicyCollectorGroup) EpisodeV2)RolloutMetrics) Preprocessor)Policy)MultiAgentBatch SampleBatchconcat_samples) OldAPIStack)Filter)convert_to_numpy)get_original_spaceunbatch)ActionConnectorDataTypeAgentConnectorDataTypeAgentID EnvActionTypeEnvID EnvInfoDict EnvObsTypeMultiAgentDict MultiEnvDictPolicyIDPolicyOutputTypeSampleBatchType StateBatchesTensorStructType)log_onceSimpleImageViewer) RLlibCallback) RolloutWorkeriig@@c^eZdZdZd deefdZdedee effdZ dZ d Z d Z dS) _PerfStatsz"&"6"B   r?c`|j|S|SN)r3rNrQrPs r=getz_PerfStats.getxs& = ==?? "==?? "r?rS)__name__ __module__ __qualname____doc__rfloatr>strr intrDrNrQrTr?r=r2r27sFF # #% # # # # &# &eCJ&7 & & & &      #####r?r2ceZdZdZdS)_NewDefaultDictc:||x}||<|SrS)default_factory)r<env_idrets r= __missing__z_NewDefaultDict.__missing__s#!11&999d6l r?N)rUrVrWrcr\r?r=r^r^s#r?r^ episode_id batch_builderlarge_batch_thresholdmultiple_episodes_in_batchreturncti}|jD]\}}|jdkr|j|krRtdrCtd|j|j|dz|sdndz|}|||<t||jS)aBuild MultiAgentBatch from a dict of _PolicyCollectors. Args: env_steps: total env steps. policy_collectors: collected training SampleBatchs by policy. Returns: Always returns a sample batch in MultiAgentBatch format. rlarge_batch_warningz9More than {} observations in {} env steps for episode {} avare buffered in the sampler. If this is more than you expected, check that that you set a horizon on your environment correctly and that it terminates at some point. Note: In multi-agent environments, `rollout_fragment_length` sets the batch size based on (across-agents) environment steps, not the steps of individual agents, which can result in unexpectedly large batches.zuAlso, you may be waiting for your Env to terminate (batch_mode=`complete_episodes`). Make sure it does at some point.)policy_batches env_steps) policy_collectorsitems agent_stepsr,loggerwarningformatrmbuildr)rdrerfrgma_batchpid collectorbatchs r=_build_multi_agent_batchrys H'9??AAY  A % %   $'< < < !B B < NN$f!-}/F ! ! 6--#   ,!!  (m>U V V VVr? eval_datact|}d|vrzEnvRunnerV2.__init__s@ h 2 3 3 R "+E(#%(?%-  ) ) + + ! 9;CR  #D D  ,e << )4+H2+M N N N. ###r?c|jsdS ddlm}|S#ttf$r$d|_t dYnwxYwdS)zCMaybe construct a SimpleImageViewer instance for episode rendering.Nrr-Fz\Could not import gymnasium.envs.classic_control.rendering! Try `pip install gymnasium[all]`.)r(gymnasium.envs.classic_control.renderingr. ImportErrorModuleNotFoundErrorrqrr)r<r.s r=rz$EnvRunnerV2._get_simple_image_viewers| 4  R R R R R R$$&& &01    DL NN?      ts2AAcD|jjjD]H}t |dd5|j||j||I|j |j|j|jj||dS)N explorationpolicy environmentepisodetf_sessrrpolicies env_indexr) r policy_mapcachevaluesgetattrron_episode_startr get_sessionr)r<rraps r=_call_on_episode_startz"EnvRunnerV2._call_on_episode_start"s(.5577  Aq-..: .. $#MMOO / ((<^\, )     r?rhc4t|jjS)zCreate a new batch builder. We create a _PolicyCollectorGroup based on the full policy_map as the batch builder. )rrr)r<_s r=rzEnvRunnerV2._new_batch_builder:s %T\%<===r?c#DK |}|D]}|V)zSamples and yields training episodes continuously. Yields: Object containing state, action, reward, terminal condition, and other fields as dictated by `policy`. )step)r<outputsos r=runzEnvRunnerV2.runBs7 iikkG   r?c|jddtj}|j\}}}}}}tj|z }tj} ||||||\} } } |jdtj| z tj} || }|jdtj| z tj}|| | ||}|jdtj|z tj}|j||jd |tjz|z | | S) z;Samples training episodes by stepping through environments.r6)unfiltered_obsrewards terminateds truncatedsinfosr7)to_evalr8) active_envsr eval_resultsoff_policy_actionsr9r:) rrDtimerpoll_process_observations_do_policy_eval_process_policy_eval_results send_actions _maybe_render)r<t0rrrrrr env_poll_timet1rrrt2rt3actions_to_sendt4s r=rzEnvRunnerV2.stepNs gq))) Y[[ N   ! !       b( Y[[)-(B(B)#! )C) ) % Wg 7r9IJJJY[[++G+<<  . b0@AAAY[[  - -#%1 .    6 b8HIIIY[[ ##O444 o}ty{{/JR/OPPP r?rrc t|j}|"|D]}||j|Si}|}|D];}||}|j|} | ||<.R.R  $" B B '*@ AAAA **,, " J JE**511I' 2F+1+G+G+I+I i ( ( &~&3"7#899&5!+m"3     r?rrrrrct}tt}g}|D]\} t | t r:|ddusJdd|| d|||U|jvr |} | |j<n |j} | s| | |ds|drd} nd} | | d| ditt} i} i}| D]\}}|dksJ| |}t|dp| |}|| |<t|dp| |d}|||<| |s|s|rt j| jt jt j| |t j| |dt j|t j|t j| |it j|i}| |||f| r| D]{}| |ds+| |ds| |rE| |}|jj|}tA|j!}| |d}| |i}t j| jt jt j| |t j|t jdt j| |dt j|t j|"i}| |||f}| D]\}}|jj|}|j#s Jd fd |D}|#|}|D]<}| |j$sm| %|j$|j&j't j|j&j't j|j&j't j n%| (|j$|j&j'| s| |j$dsp| |j$dsU| |j$s;tS|j*|j$|j&}|||>| +| jd kr3|j,-|j|j.|jj|  | r4|| |dp |d||||j/r@|0|j1| }|r|||j1=|||fS)aProcess raw obs from env. Group data for active agents by policy. Reset environments that are done. Args: unfiltered_obs: The unfiltered, raw observations from the BaseEnv (vectorized, possibly multi-agent). Dict of dict: By env index, then agent ID, then mapped to actual obs. rewards: The rewards MultiEnvDict of the BaseEnv. terminateds: The `terminated` flags MultiEnvDict of the BaseEnv. truncateds: The `truncated` flags MultiEnvDict of the BaseEnv. infos: The MultiEnvDict of infos dicts of the BaseEnv. Returns: A tuple of: A list of envs that were active during this step. AgentConnectorDataType for active agents for policy evaluation. SampleBatches and RolloutMetrics for completed agents for output. __all__Tz&ERROR: When a sub-environment (env-id zZ) returns an error as observation, the terminateds[__all__] flag must also be set to True!)raenv_obs_or_exceptionis_donerrrF __common__r5z.EnvRunnerV2 requires agent connectors to work.c8g|]\}}t||Sr\)r).0agent_iddataras r= z5EnvRunnerV2._process_observations..hs9:::&$+68TBB:::r?rinit_obs init_infostrrrrrr)2setrlistror Exception_handle_done_episodercreate_episode has_init_obsradd set_last_inforTrboolrTrENV_ID AGENT_INDEX agent_indexREWARDS TERMINATEDS TRUNCATEDSINFOSNEXT_OBSappendrrrrrobservation_spacesampleagent_connectorsr add_init_obsrraw_dictadd_action_reward_done_next_obsrrarron_episode_steprr._try_build_truncated_episode_multi_agent_batchr) r<rrrrrrrrenv_obsrall_agents_donesample_batches_by_policyagent_terminatedsagent_truncatedsrobsragent_terminatedagent_truncated values_dictr obs_spacerewardinfobatchesacd_list processedditem sample_batchras @r=rz!EnvRunnerV2._process_observationssN#&%% @KD@Q@Q@B .3355_ 5_ 5OFG '9-- "6*95===V>==))!)0 +## *T222%)%8%8%@%@07%f--%)%:6%B'')) =++GV<<<6"9- (F1CI1N ("&"''''  ! !,f 0A0A,PR0S0S T T T(34'8'8 $ " ! !() T) T #9,,,,&-&8&8&B&B #'' 2Wk&6I6M6Mh6W6W$$ /?!(+"&v&y1?!&)--h>>##.= *++H55$(7 M7> &+W-@-@-J-J ')<)>!+//%@@!#??844! !*1*<*.>.@.@ #K-Y7>>+?VWWWW'?&D&D&F&F) 8) 8" 7!%!8!C+DDCDD+::::*1:::#33H== "88A"// ;; ,,%&Z%&V_[5I%J'(v{7H'Ifokm< - ??J ( 8,00UCC 8,// EBB 8#??1:66 8 6ah AFSS *11$777186 LLNNN ~!!//<!^!\4#$ 0 ))' 2Sj6H6S / 5#RR(0'    5NN<000,V4GW,,r?rarrc|j|}|j|}|||||jsBt |j||j|j}|r|||j|=dSdS)zBuilds a MultiAgentSampleBatch from the episode and adds it to outputs. Args: env_id: The env id. is_done: Whether the env is done. outputs: The list of outputs to add the )rer check_donesN)rrpostprocess_episoderryrdrr)r<rarrrrema_sample_batchs r=_build_done_episodezEnvRunnerV2._build_done_episodes"26:,V4 ##' $   / -6"+0 O  0///$V,,, - -r?r!rc tt}|D]7\}}|} || ||f8|D]\} } |jj| } fd| D} | | } | D]}|j |j j tj |j j tj|j j tj|| |dS)zProcess resetted obs through agent connectors for policy eval. Args: env_id: The env id. obs: The Resetted obs. episode: New episode. to_eval: List of agent connector data for policy eval. cg|]^\}}t|tj|tjtjjtj|i_Sr\)rrrrr rr r )rrr!rarrs r=rz?EnvRunnerV2.__process_resetted_obs_for_eval..ss 6 6 6"Hc'#,c#)5# w~#/1D1DX1N1N    6 6 6r?rN)rrrorrrrrrrrrrrrr )r<rar!rrrper_policy_resetted_obsrraw_obsr agents_obsrr)r*r+s ` `` r=__process_resetted_obs_for_evalz+EnvRunnerV2.__process_resetted_obs_for_evalsr 9DD8I8I!$V!2!2!4!4 K K Hg")"4"4X">">I #I . 5 5x6I J J J J%<%B%B%D%D - - !Iz\,Y7F 6 6 6 6 6 6&0 6 6 6H//99I - -$$ZV_[-AB v{/@Afokm4 %  "))!,,,, -% - -r?rrct|tr&|}|tdnX|j|}||||jj| |||| ||| |} |j |\} } | &| tkst| |tsn$|tdj|jjjD]} | j|| o| tkrf||j|<|||||| | |||||dSdSdS)aHandle an all-finished episode. Add collected SampleBatch to batch builder. Reset corresponding env, etc. Args: env_id: Environment ID. env_obs_or_exception: Last per-environment observation or Exception. env_infos: Last per-environment infos. is_done: If all agents are done. active_envs: Set of active env ids. to_eval: Output container for policy eval data. outputs: Output container for collected sample batches. T)episode_faulty)rN)rrrrrextendrrrr2 end_episoderr try_resetr rrrresetr+_EnvRunnerV2__process_resetted_obs_for_evalrr) r<rarrrrrepisode_or_exception new_episode resetted_obsresetted_infosrs r=rz EnvRunnerV2._handle_done_episodes, *I 6 6 ?.B NN>>>> ? ? ? ?.2.CF.K NN))(T\5L*     $ $VWg > > > !5666!%!4!4V!+C+CF+K+K (L.$#555!,v"6 BB6~TBBBCCC D(.5577 - -A  $ $V , , , ,  # 8J(J(J,7D !& )  ' ' V < < <  0 0           OOF # # # # # $ #(J(Jr?c||jvsJt||jj|jj|j|j}|j|j|j|jj|||S)aCreates a new EpisodeV2 instance and returns it. Calls `on_episode_created` callbacks, but does NOT reset the respective sub-environment yet. Args: env_id: Env ID. Returns: The newly created EpisodeV2 instance. )rrr)rrrrpolicy_mapping_fnron_episode_createdr)r<rarAs r=rzEnvRunnerV2.create_episodelsT22222  L # L *<o     **<^\, +   r?r@c|j|j|j|jj|||jjjD]H}t|dd5|j||j|| It|tr'|}|j dkrd|j d}t|||jvr |j|=dSdS)zCleans up an episode that has finished. Args: env_id: Env ID. episode_or_exception: Instance of an episode if it finished successfully. Otherwise, the exception that was thrown, rrNrrzData from episode z does not show any agent interactions. Hint: Make sure for at least one timestep in the episode, env.step() returns non-empty values.)ron_episode_endrrrrrrrrrrtotal_agent_stepsrdrr)r<rar@rrmsgs r=r<zEnvRunnerV2.end_episodes) &&<^\,( '   (.5577  Aq-..: ,, $0MMOO - *I 6 6 &*G(A--E);EEE !oo% T* * *%f--- + *r?rec|jdkr|j}|j}n|j}|j}||z|jkr|jdkr'||z|jksJd|d|d|jd||jkr||d|jd kr!t|j||j |j Std rt d dS) Nrmrpz built_steps (z) + ongoing_steps (z) != rollout_fragment_length (z).F)rerrno_agent_stepszoYour environment seems to be stepping w/o ever emitting agent observations (agents are never requested to act)!)rrmactive_env_stepsrpactive_agent_stepsrr0ryrdrrr,rqrr)r<rer built_steps ongoing_stepss r=rz:EnvRunnerV2._try_build_truncated_episode_multi_agent_batchsK  ; . .'1K#4MM(3K#6M  &$*G G G#}44"]2d6SSSSRKRRMRR040MRRRTSST:::++-QV+WWW(1,,/&!/4 *++ ) tr?cVjj}dtffd }i}|D]w\}} t ||}n#t $r||}YnwxYwt d|D}|||jfd|D||<x|S)aCall compute_actions on collected episode data to get next action. Args: to_eval: Mapping of policy IDs to lists of AgentConnectorDataType objects (items in these lists will be the batch's items for the model forward pass). Returns: Dict mapping PolicyIDs to compute_actions_from_input_dict() outputs. rzcd}|D]M}j|j}||jd}|||krt d|d|}Nt jj|S)NT)refreshzePolicy map changed. The list of eval data that was handled by a same policy is now handled by policy zC and {policy_id}. Please don't do this in the middle of an episode.)rrarrr _get_or_raiserr)rzrr+rrvr<s r=_try_find_policy_againz;EnvRunnerV2._do_policy_eval.._try_find_policy_againsI /9((T(BB(SI-=-=$LEHLLL   !8)DD Dr?c&g|]}|jjSr\)rr-)rr+s r=rz/EnvRunnerV2._do_policy_eval.."s888$888r?c4g|]}j|jSr\)rra)rrr<s r=rz/EnvRunnerV2._do_policy_eval..(s#MMMa$/9MMMr?)timestepepisodes) rrrrorTrrcompute_actions_from_input_dictglobal_timestep) r<rrrUrrrzr input_dicts ` r=rzEnvRunnerV2._do_policy_evals<*  E.D E E E E E E :< $+MMOO   Iy C!.x!C!C C C C"8!7 !B!B C 988i888J'-&L&L/MMMM9MMM'M''L # # sAA A rrc tt}|D]}i||<|D]\}}||d} t| } ||d} ||d} t | t rt j| } t| } t|j j |} | j r| j s Jdt| D]\} }|| j}|| j}|| jj}t'j| fd| }t'j| fd| }t+||||||f}| |j\}}}||vs |||vr|n |||}t+||||||f}| j ||||vsJ||||<|S)aProcess the output of policy neural network evaluation. Records policy evaluation results into agent connectors and returns replies to send back to agents in the env. Args: active_envs: Set of env IDs that are still active. to_eval: Mapping of policy IDs to lists of AgentConnectorDataType objects. eval_results: Mapping of policy IDs to list of actions, rnn-out states, extra-action-fetches dicts. off_policy_actions: Doubly keyed dict of env-ids -> agent ids -> off-policy-action, returned by a `BaseEnv.poll()` call. Returns: Nested dict of env id -> agent id -> actions to be sent to Env (np.ndarrays). rrz/EnvRunnerV2 requires action connectors to work.c||SrSr\xis r=z:EnvRunnerV2._process_policy_eval_results..e 1Q4r?c||SrSr\r`s r=rcz:EnvRunnerV2._process_policy_eval_results..jrdr?)rrrorrrrarrayrrTrrraction_connectors enumeraterarrrtree map_structureroutputon_policy_output)r<rrrrrrarrzactionsrnn_outextra_action_outrrbactionrr\ rnn_statesfetchesac_dataaction_to_sendaction_to_buffers r=rz(EnvRunnerV2._process_policy_eval_results-s0FQQUEVEV! ) )F&(OF # #%,MMOOA CA C Iy(4Y(?(BG&w//G$0$;A$>G%1)%>>>>4B'11[- C^r?c|jr|jsdStj}|j}t |t jr3t|j dkr|j |n|dvrtd|jd|j dtj|z dS)zVisualize environment.N)TFNz The env's (z) `try_render()` method returned an unsupported value! Make sure you either return a uint8/w x h x 3 (RGB) image or handle rendering in a window and then return `True`.r;)rrrr try_renderrrndarrayr~shapeimshowrrrD)r<t5rendereds r=rzEnvRunnerV2._maybe_renders| 4#<  F Y[[>,,.. h + + HN0C0Cq0H0H  % , ,X 6 6 6 6 0 0 01dn111  /r1ABBBBBr?)rrmN)1rUrVrWrXr r r2r[rZr>rrrrrr)rrrrrrrrr&r r r"r'rr rr2r r$r#r?r%rrrr<rrrr(rrrr\r?r=rrsEEE(+)< < < < %) < # <  < "%< < < < < < |&   0>'<>>>> Xo.    9d?+9999v   .23;.? n     @N-$N-N-" N- ! N-  N-  E  Xt23 34 U>?2 34 6 N-N-N-N-`'-'-'-o& '-'-'-'-R/-/-%gz122 3/-E4 4556 /-  /- h%; <<= /-/-/-/-bW$W$-W$ W$ Z W$ h%; <<= W$o&W$ W$W$W$W$r!U!y!!!!F2.2.38I9M3N2.2.2.2.h-2-=F- t[/1 2----^9h%; <<=9 h(( )9999vaZah%; <<=a8%556 a ) aaaaFCCCCCr?rrc|}|sdSg}|D]Y}t|t}|sdS|D](\}}|t ||)Z|S)zAtari games have multiple logical episodes, one per life. However, for metrics reporting we count full episodes, all lives included. N)get_sub_environmentsrrnext_episode_resultsrr)rsub_environments atari_outsub_envmonitoreps_reweps_lens r=rrs  4466 tI#??$Wj99 44 ' < < > > ? ? GW   ^GW== > > > > ? r?mappingrc||vr5td||||S)aReturns an object under key `policy_id` in `mapping`. Args: mapping (Dict[PolicyID, Union[Policy, Preprocessor, Filter]]): The mapping dict from policy id (str) to actual object (Policy, Preprocessor, etc.). policy_id: The policy ID to lookup. Returns: Union[Policy, Preprocessor, Filter]: The found object. Raises: ValueError: If `policy_id` cannot be found in `mapping`. z\Could not find policy for agent: PolicyID `{}` not found in policy map, whose keys are `{}`.)rrskeys)rrs r=rTrTsJ" 228&GLLNN2S2S    9 r?)Uloggingr collectionsrtypingrrrrrr r r numpyrriray.rllib.env.base_envr r ray.rllib.env.external_envr%ray.rllib.env.wrappers.atari_wrappersrr5ray.rllib.evaluation.collectors.simple_list_collectorrray.rllib.evaluation.episode_v2rray.rllib.evaluation.metricsrray.rllib.models.preprocessorsrray.rllib.policy.policyrray.rllib.policy.sample_batchrrrray.rllib.utils.annotationsrray.rllib.utils.filterrray.rllib.utils.numpyr"ray.rllib.utils.spaces.space_utilsrrray.rllib.utils.typingrrr r!r"r#r$r%r&r'r(r)r*r+ray.util.debugr,rr.ray.rllib.callbacks.callbacksr/#ray.rllib.evaluation.rollout_workerr0 getLoggerrUrqrrrLr2r^r[r ryrrrrTr\r?r=rs^ ######SSSSSSSSSSSSSSSSSSSS >>>>>>>>999999PPPPPPPPWWWWWW555555777777777777******VVVVVVVVVV333333))))))222222JJJJJJJJ $#####BJJJJJJ;;;;;;AAAAAA  8 $ $! $   D#D#D#D#D#D#D# D#N k   2W2W(2W2W!% 2W  2W2W2W 2Wj  tK/@ [       XCXCXCXCXCXCXC XCv7tN/C$ (E&,">?? @MU 6< '(r?