&`i ddlZddlZddlZddlmZmZmZmZm Z m Z m Z m Z m Z mZmZddlZddlZddlmZmZddlmZddlmZddlmZmZmZmZddlm Z ddl!m"Z"dd l#m$Z$dd l%m&Z&dd l'm(Z(dd l)m*Z*dd l+m,Z,ddl-m.Z.ddl/m0Z0m1Z1ddl2m3Z3ddl4m5Z5ddl6m7Z7ddl8m9Z9m:Z:ddl;mZ>m?Z?m@Z@mAZAmBZBmCZCddlDmEZEerddlFmGZGe7\ZHZIZJejKeLZMedZNeEGddZOdS)N) TYPE_CHECKINGAnyCallable CollectionDictListOptionalTupleTypeTypeVarUnion)DEPRECATED_VALUEdeprecation_warning) ActorHandle) RayActorError)!COMPONENT_ENV_TO_MODULE_CONNECTORCOMPONENT_LEARNER!COMPONENT_MODULE_TO_ENV_CONNECTORCOMPONENT_RL_MODULE) LearnerGroup)validate_module_id) RLModuleSpec)BaseEnv) EnvContext) EnvRunner) RolloutWorker)get_dataset_and_shards)Policy PolicyState)FaultTolerantActorManager) OldAPIStack) try_import_tf)NUM_ENV_STEPS_SAMPLED_LIFETIMEWEIGHTS_SEQ_NO)AgentID EnvCreatorEnvType EpisodeIDPartialAlgorithmConfigDictPolicyIDSampleBatchType TensorType) DeveloperAPIAlgorithmConfigTc eZdZdZddddddddddeed deedeeegdfdee e d ed d e d ee d e dee de dee fdZddddddeeegdfd ed de d e fdZdZedefdZdee fdZdee fdZde fdZde fdZde fdZde fdZde fdZde fdZdadee de fd Zde fd!Zdddddddd"d d d#eed$ee d%eeee e fd&eee e fd'eee ddfd(Z! dbd+e"eege#feeege#fe ee fd,eee e fdee d-ee$d.e d/e d0e d1ee d2e dee"e%e e#fe#ffd3Z& dcd4eee'd5ee"ed6fd7eee d8eee e(fd-ee$d9ee ddfd:Z)ddd;e dee+ddfd?Z,ded@Z-dddddd*d*dAd+e"eege#feeege#fe ee fd e d0e d1ee d-ee$d.e d/e dee#fdBZ. daddddCd+e"eege#feeege#fe ee fdee d0e d1ee de f dDZ/dd)d*d*dEdFee"e ee e%e fd-ee$d.e d/e dee%e e#ff dGZ0e1d+eegee#fdeee#fdHZ2e1d+ee3e4gee#fdeee#fdIZ5dee fdJZ6e1d+ee e'ge#fdee#fdKZ7e1d+ee e'ge#fdee#fdLZ8e1 dadMe'dNee9de fdOZ:e1 dfdddddddedPdMe'dQee e dRee dSee;j<j=dTee;j<j=d ee"d e>fdUee?dVeee@eAge'fdWee"eBe'ee'ee9ge ffdXeeCdYeee"ee+fddfdZZDd*dd[dedeeegdfd\e d;e d]e d d d^eee'e%e;j<j=e;j<j=ffde"ee+ffd_ZEeFd`ZGdS)gEnvRunnerGroupzcSet of EnvRunners with n @ray.remote workers and zero or one local worker. Where: n >= 0. NTr) env_creator validate_envdefault_policy_classconfiglocal_env_runnerlogdir_setup tune_trial_id pg_offsetnum_env_runners num_workers local_workerr3r4r5r6r/r7r8r9r:r;r<c | tks | tkrtdddddlm} |s | }n%t |t r| j|}||_||_||_ tj |j |_ |j j |j j|j j|jr|jndd|_||_| |_|j|_|ja|jrN|jr$|jrt3d dd lm}||_n/|jrdd lm}||_ndd lm}||_n t@|_||_!|j"p|j|_#tI|j%d |_&|rX |'||| | n|j(|dS#tR$r&}|j*r|j+dj+d|d}~wwxYwdS)aInitializes a EnvRunnerGroup instance. Args: env_creator: Function that returns env given env config. validate_env: Optional callable to validate the generated environment (only on worker=0). This callable should raise an exception if the environment is invalid. default_policy_class: An optional default Policy class to use inside the (multi-agent) `policies` dict. In case the PolicySpecs in there have no class defined, use this `default_policy_class`. If None, PolicySpecs will be using the Algorithm's default Policy class. config: Optional AlgorithmConfig (or config dict). local_env_runner: Whether to create a local (non @ray.remote) EnvRunner in the returned set as well (default: True). If `num_env_runners` is 0, always create a local EnvRunner. logdir: Optional logging directory for workers. _setup: Whether to actually set up workers. This is only for testing. tune_trial_id: The Ray Tune trial ID, if this EnvRunnerGroup is part of an Algorithm run as a Tune trial. None, otherwise. z,WorkerSet(num_workers=..., local_worker=...)z9EnvRunnerGroup(num_env_runners=..., local_env_runner=...)T)oldnewerrorrr.)num_cpusnum_gpus resources max_restartsNz,Multi-agent recording is not supported, yet.)OfflineSingleAgentEnvRunner)MultiAgentEnvRunner)SingleAgentEnvRunner)'max_remote_requests_in_flight_per_actorinit_idr4r6r<r7),rr%ray.rllib.algorithms.algorithm_configr/ isinstancedict from_dict _env_creator _policy_class_remote_configrayput_remote_config_obj_refnum_cpus_per_env_runnernum_gpus_per_env_runnercustom_resources_per_env_runnerrestart_failed_env_runnersmax_num_env_runner_restarts _remote_args_tune_trial_id _pg_offsetenv_runner_cls"enable_env_runner_and_connector_v2outputis_multi_agent ValueError$ray.rllib.offline.offline_env_runnerrG$ray.rllib.env.multi_agent_env_runnerrH%ray.rllib.env.single_agent_env_runnerrIr_logdirignore_env_runner_failures!_ignore_ray_errors_on_env_runnersr %max_requests_in_flight_per_env_runner_worker_managerr9r<ractor_init_failedargs)selfr3r4r5r6r7r8r9r:r;r<r=r>r/rGrHrIes r/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/rllib/env/env_runner_group.py__init__zEnvRunnerGroup.__init__LsL * * *l>N.N.N BO     JIIIII 7$_&&FF  % % 7._.v66F'1$&)gd.A&B&B#+C+C,L422   ,#%3   &8 4=C, J()WXXX/J++, C/B++/C++&3#  - R1R . 9<        !-!+6(#3%5     !    & &).++G!   s"F** G4!GGrMcd|_|dkrd}|j}||j|d||_|jdkrt||\|_ |_ nd|_ d|_ | ||j |rG|j dkr*|js#|jr|js|}nd}|r,||j|d||j| |_dSdS) aASets up an EnvRunnerGroup instance. Args: validate_env: Optional callable to validate the generated environment (only on worker=0). config: Optional dict that extends the common config of the Algorithm class. num_env_runners: Number of remote EnvRunner workers to create. local_env_runner: Whether to create a local (non @ray.remote) EnvRunner in the returned set as well (default: True). If `num_env_runners` is 0, always create a local EnvRunner. NrTF) copy_frozen)tf_session_argsdataset)validate)r3r4 worker_indexr=r6spaces)_local_env_runnerrvcopyupdatelocal_tf_session_args framework _local_configinput_r_ds _ds_shards add_workers'validate_env_runners_after_constructionrm num_actorscreate_env_on_local_workerobservation_space action_space get_spaces _make_workerrS)rpr4r6r<r7r~rzs rrr9zEnvRunnerGroup._setups*"& a  #  !' 6 ; ; = =$$V%ABBB#[[U[;;EE1F   =I % %)?v(W(W %DHdooDH"DO  C      $//11A5556-65;5H6__&&FFF  %)%6%6 -)+) &7&&D " " "  c|jr |jdgng}|d|| d}td||S)zInfer observation and action spaces from one (local or remote) EnvRunner. Returns: A dict mapping from ModuleID to a 2-tuple containing obs- and action-space. rc*|SN)r) env_runners rrz+EnvRunnerGroup.get_spaces..:sz4466r)remote_worker_idsr7zQInferred observation/action spaces from remote worker (local worker has no env): )rm actor_idsforeach_env_runnerloggerinfo)rprrzs rrrzEnvRunnerGroup.get_spaces,s#--// T ! + + - -a 0 1 1  (( 6 6/!22)       :17 : :    rreturnc|jS)zReturns the local EnvRunner.)r{rps rrr7zEnvRunnerGroup.local_env_runnerFs %%rc4|jSz&Returns the list of remote worker IDs.)rmhealthy_actor_idsrs rrhealthy_env_runner_idsz%EnvRunnerGroup.healthy_env_runner_idsKs#55777rc*|Sr)rrs rrhealthy_worker_idsz!EnvRunnerGroup.healthy_worker_idsO**,,,rc4|jSz(Returns the number of remote EnvRunners.)rmrrs rrnum_remote_env_runnersz%EnvRunnerGroup.num_remote_env_runnersSs#..000rc*|Sr)rrs rrnum_remote_workersz!EnvRunnerGroup.num_remote_workersWrrc4|jSz-Returns the number of healthy remote workers.)rmnum_healthy_actorsrs rrnum_healthy_remote_env_runnersz-EnvRunnerGroup.num_healthy_remote_env_runners[#66888rc*|Sr)rrs rrnum_healthy_remote_workersz)EnvRunnerGroup.num_healthy_remote_workers_s22444rcntt|j|zSzFReturns the number of all healthy workers, including the local worker.)intboolr{rrs rrnum_healthy_env_runnersz&EnvRunnerGroup.num_healthy_env_runnerscs,4.//0043R3R3T3TTTrc*|Sr)rrs rrnum_healthy_workersz"EnvRunnerGroup.num_healthy_workersgs++---rtagc8|j|S)z/Returns the number of in-flight async requests.)r)rmnum_outstanding_async_reqs)rprs rrnum_in_flight_async_reqsz'EnvRunnerGroup.num_in_flight_async_reqsks#>>3>GGGrc4|jS)zATotal number of times managed remote workers have been restarted.)rmtotal_num_restartsrs rrnum_remote_worker_restartsz)EnvRunnerGroup.num_remote_worker_restartsorr) from_workerenv_steps_sampledconnector_statesrl_module_stateenv_runner_indices_to_update env_to_module module_to_envrrrrrcr|t|}|p|j}|j o|jp"|jo|jdup|jdko|j } |j o|jp |jo|j} |dkr4|jr-|j i| t|ini|pidS| s| sdS| r|gkri} nG|| dd|j }d|D} d |D} |jJt|jd r5t|jd r |J|jj}|J|jj}i} | r/| t"|| i| r/| t&|| inY|5t"|t&|i} n"|t"t&g } | || t<| s|j|j | n\| | t"i| | t&i| t"d| t&d| rd|j| r|j | |r| ||d d t1| |ddSdS)aSynchronizes the connectors of this EnvRunnerGroup's EnvRunners. The exact procedure works as follows: - If `from_worker` is None, set `from_worker=self.local_env_runner`. - If `config.use_worker_filter_stats` is True, gather all remote EnvRunners' ConnectorV2 states. Otherwise, only use the ConnectorV2 states of `from_worker`. - Merge all gathered states into one resulting state. - Broadcast the resulting state back to all remote EnvRunners AND the local EnvRunner. Args: config: The AlgorithmConfig object to use to determine, in which direction(s) we need to synch and what the timeouts are. from_worker: The EnvRunner from which to synch. If None, will use the local worker of this EnvRunnerGroup. env_steps_sampled: The total number of env steps taken thus far by all workers combined. Used to broadcast this number to all remote workers if `update_worker_filter_stats` is True in `config`. env_runner_indices_to_update: The indices of those EnvRunners to update with the merged state. Use None (default) to update all remote EnvRunners. NT training_onlyrcF|ttgS)N components) get_staterrws rrrz7EnvRunnerGroup.sync_env_runner_states..s$!++ A A(#.##rF)r7timeout_secondsc:g|]}t|v |tS)r.0ss rr z9EnvRunnerGroup.sync_env_runner_states..3(((8A==78===rc:g|]}t|v |tSr)rrs rrrz9EnvRunnerGroup.sync_env_runner_states..rr_env_to_module_module_to_envr set_statestatefuncrkwargsrr)rr7rbuse_worker_filter_statsmerge_env_runner_states in_evaluationupdate_worker_filter_statsbroadcast_env_runner_statesrrr#r)sync_filters_on_rollout_workers_timeout_shasattrrrr}r merge_statesrrgetpop$foreach_env_runner_async_fetch_readyrQ)rpr6rrrrrrrmerge broadcastenv_runner_statesenv_to_module_statesmodule_to_env_statess rrsync_env_runner_statesz%EnvRunnerGroup.sync_env_runner_statessszD  ( #$5 6 6 !:T%: 9 9 /.    5 .$62oE1"00 9 9 21   5 32   * * , , 1 1d6K 1  ! + +-889JKK  ',"     F Y  F E 2%%$&!!#+'+'>'> */"L(? ( ($((-((($ ((-((($)5 57GHH6 57GHH6)000$($9$HM(000$($9$HM$&!'%,,= - : :;O P P(%,,= - : :;O P P"5}7N7N7P7P5}7N7N7P7P%!! %0$9$999 %:%%!  (@Q < =  K$0%//0ABBBB''%))*KLLb''%))*KLLb  ! !"CT J J J  ! !"CT J J J  $0Y0%//0ABBB  :!((999  5 5 "3444"> # 6     !  rrFrrrreturn_obj_refs mark_healthy healthy_onlyrreturn_actor_idsc T|j|||||||||j|  S)zCalls the given function asynchronously and returns previous results if any. This is a convenience function that calls the underlying actor manager's `foreach_actor_async_fetch_ready()` method. ) rrrrrrrremote_actor_idsignore_ray_errorsr)rmforeach_actor_async_fetch_readyrk) rprrrrrrrrrs rrrz3EnvRunnerGroup.foreach_env_runner_async_fetch_ready8sF(#CC++%%."D-D   rpoliciesfrom_worker_or_learner_grouprto_worker_indices global_varsinference_onlycn |j|tdd}|s|||n|j}|td| d|Dntg} t |t r-|d| D|t}n|j j rft |tj j r/tj|j| |}n/|| |}n|||}|j j r[d|D}tj| |d d t) || n3tj|  fd } || d |||ja|A|j j r|j|n|j||jdSdSdS)aSyncs model weights from the given weight source to all remote workers. Weight source can be either a (local) rollout worker or a learner_group. It should just implement a `get_weights` method. Args: policies: Optional list of PolicyIDs to sync weights for. If None (default), sync weights to/from all policies. from_worker_or_learner_group: Optional (local) EnvRunner instance or LearnerGroup instance to sync from. If None (default), sync from this EnvRunnerGroup's local worker. to_worker_indices: Optional list of worker indices to sync the weights to. If None (default), sync to all remote workers. global_vars: An optional global vars dict to set this worker to. If None, do not update the global_vars. timeout_seconds: Timeout in seconds to wait for the sync weights calls to complete. Default is 0.0 (fire-and-forget, do not wait for any sync calls to finish). Setting this to 0.0 might significantly improve algorithm performance, depending on the algo's `training_step` logic. inference_only: Sync weights with workers that keep inference-only modules. This is needed for algorithms in the new stack that use inference-only modules. In this case only a part of the parameters are synced to the workers. Default is False. NzmNo `local_env_runner` in EnvRunnerGroup! Must provide `from_worker_or_learner_group` arg in `sync_weights()`!z`from_worker_or_trainer` is None. In this case, EnvRunnerGroup should have local_env_runner. But local_env_runner is also None.c(g|]}tdz|zS/)r)rps rrrz/EnvRunnerGroup.sync_weights..s#AAA1$s*Q.AAArc(g|]}tdz|zSr)r)rms rrrz/EnvRunnerGroup.sync_weights..s#MMM 1C 7! ;MMMr)rr)rrc:i|]\}}|ttfv||Sr)rr$)rkvs rr z/EnvRunnerGroup.sync_weights..s9###10.AAAqAAArrrrcX|tjdSr) set_weightsrVr)rrrl_module_state_refs rr _set_weightsz1EnvRunnerGroup.sync_weights.._set_weightss***373F+G+GUUUUUrF)rr7rr)r7 TypeErrorrrerrPrrrrUrbrVactorrrremote get_weightsitemsrWrrQrrrset_global_vars) rprrrrrrr weights_srcmodulesrrs ` @rr sync_weightszEnvRunnerGroup.sync_weightsYsD  (-I-QJ    " " $ $Z (D(P0;-,*  " W'BAAAAA)*  +|44 "-"7"7MMWMMM#1#8##$#% $G k39+@AA &)g#-44'.+95''OO'2&;&;#*'5'<''OO #."9"9%#1#:## "E% ## / 5 5 7 7###'*go&>&># 99$#&9:::&7$3 :'*go&>&>#VVVVVV''%%*&7$3 (  ,+7&IG)33ODDDD)55oFFF&%55kBBBBB - ,'&rr=rxcjfdtD}j||rmjdD]S}|jsH|}jr+t dt|O|RdSdS)aCreates and adds a number of remote workers to this worker set. Can be called several times on the same EnvRunnerGroup to add more EnvRunners to the set. Args: num_workers: The number of remote Workers to add to this EnvRunnerGroup. validate: Whether to validate remote workers after their construction process. Raises: RayError: If any of the constructed remote workers is not up and running properly. c jg|]/}jd|zdzzj0S)NrJ)r3r4ryr=r6)rrSrX)rir=old_num_workersrps rrrz.EnvRunnerGroup.add_workers..sf        -!,q014+k92       rc*|Sr)assert_healthyrs rrrz,EnvRunnerGroup.add_workers..s!**,,rz&Validation of EnvRunner failed! Error=N) rmrrange add_actors foreach_actorokrrkrrBstr)rpr=rx new_workersresultrqrs`` @rrrzEnvRunnerGroup.add_workerss  .99;;       ;''     '' 444  .<<,,  y  A=  %VcRSff%V%VWWWW   rnew_remote_workerscl|j|j|dS)zHard overrides the remote EnvRunners in this set with the provided ones. Args: new_remote_workers: A list of new EnvRunners (as `ActorHandles`) to use as new remote workers. N)rmclearr)rprs rrresetzEnvRunnerGroup.resets7 ""$$$ ''(:;;;;;rc |dddn*#t$rtdYnwxYw|jdS#|jwxYw)z9Calls `stop` on all EnvRunners (including the local one).c*|Sr)stoprs rrrz%EnvRunnerGroup.stop..,s!&&((rFT)rr7zFailed to stop workers!N)r Exceptionr exceptionrmr!rs rrr%zEnvRunnerGroup.stop%s )  # #"" $     8 8 8   6 7 7 7 7 7 8  & & ( ( ( ( (D & & ( ( ( (s&A!$AA!AA!!A<)rr7rrrrrc |r |r Jdg} |rH|jA|Jt|trt|j|g} n||jg} |js| S|j|||||||} tj| |j d| D} | | zS)aCalls the given function with each EnvRunner as its argument. Args: func: The function to call for each EnvRunners. The only call argument is the respective EnvRunner instance. local_env_runner: Whether to apply `func` to local EnvRunner, too. Default is True. healthy_only: Apply `func` on known-to-be healthy EnvRunners only. remote_worker_ids: Apply `func` on a selected set of remote EnvRunners. Use None (default) for all remote EnvRunners. timeout_seconds: Time to wait (in seconds) for results. Set this to 0.0 for fire-and-forget. Set this to None (default) to wait infinitely (i.e. for synchronous execution). return_obj_refs: Whether to return ObjectRef instead of actual results. Note, for fault tolerance reasons, these returned ObjectRefs should never be resolved with ray.get() outside of this EnvRunnerGroup. mark_healthy: Whether to mark all those EnvRunners healthy again that are currently marked unhealthy AND that returned results from the remote call (within the given `timeout_seconds`). Note that EnvRunners are NOT set unhealthy, if they simply time out (only if they return a RayActorError). Also note that this setting is ignored if `healthy_only=True` (b/c `mark_healthy` only affects EnvRunners that are currently tagged as unhealthy). Returns: The list of return values of all calls to `func([worker])`. z+Can not return ObjectRef from local worker.N)rrrrrrrc6g|]}|Sr)rrrs rrrz5EnvRunnerGroup.foreach_env_runner..{s JJJa!%%''JJJr) r7rPrgetattrrmrrr handle_remote_call_result_errorsrk ignore_errors) rprrr7rrrrr local_resultremote_resultss rrrz!EnvRunnerGroup.foreach_env_runner3s(V  9'7 9 9 8 9 9 7  = 5 A>>>$$$ = '(=t D DE $T%: ; ;< #--//  -;; %.++%<   "B d.T    KJ>+G+G+I+IJJJn,,r)rrrc@|j|||||S)aCalls the given function asynchronously with each EnvRunner as the argument. Does not return results directly. Instead, `fetch_ready_async_reqs()` can be used to pull results in an async manner whenever they are available. Args: func: The function to call for each EnvRunners. The only call argument is the respective EnvRunner instance. tag: A tag to identify the results from this async call when fetching with `fetch_ready_async_reqs()`. kwargs: An optional kwargs dict to be passed to the remote function calls. healthy_only: Apply `func` on known-to-be healthy EnvRunners only. remote_worker_ids: Apply `func` on a selected set of remote EnvRunners. Returns: The number of async requests that have actually been made. This is the length of `remote_worker_ids` (or self.num_remote_workers()` if `remote_worker_ids` is None) minus the number of requests that were NOT made b/c a remote EnvRunner already had its `max_remote_requests_in_flight_per_actor` counter reached for this tag. )rrrr)rmforeach_actor_async)rprrrrrs rrforeach_env_runner_asyncz'EnvRunnerGroup.foreach_env_runner_asyncs4B#77 %. 8   rtagsrrrr6c|j||||}tj||jd|DS)aGet results from outstanding asynchronous requests that are ready. Args: tags: Tags to identify the results from a specific async call. If None (default), returns results from all ready async requests. If a single string, returns results from all ready async requests with that tag. timeout_seconds: Time to wait for results. Default is 0, meaning those requests that are already ready. return_obj_refs: Whether to return ObjectRef instead of actual results. mark_healthy: Whether to mark all those workers healthy again that are currently marked unhealthy AND that returned results from the remote call (within the given `timeout_seconds`). Note that workers are NOT set unhealthy, if they simply time out (only if they return a RayActorError). Also note that this setting is ignored if `healthy_only=True` (b/c `mark_healthy` only affects workers that are currently tagged as unhealthy). Returns: A list of results successfully returned from outstanding remote calls, paired with the indices of the callee workers. r5r)cDg|]}|j|fSr)actor_idrr+s rrrz9EnvRunnerGroup.fetch_ready_async_reqs..s'NNN!QUUWW%NNNr)rmfetch_ready_async_reqsr r.rkr/)rpr6rrrr1s rrr:z%EnvRunnerGroup.fetch_ready_async_reqssv>-DD++% E   "B "D    ON~/K/K/M/MNNNNrcRt|fddS)aUCalls `func` with all workers' sub-environments as args. An "underlying sub environment" is a single clone of an env within a vectorized environment. `func` takes a single underlying sub environment as arg, e.g. a gym.Env object. Args: func: A function - taking an EnvType (normally a gym.Env object) as arg and returning a list of lists of return values, one value per underlying sub-environment per each worker. Returns: The list (workers) of lists (sub environments) of results. c.|Sr) foreach_envrrs rrrz,EnvRunnerGroup.foreach_env..s!----rTr7listrrprs `rrr=zEnvRunnerGroup.foreach_envs>"  # #----!% $     rcRt|fddS)a`Calls `func` with all workers' sub-environments and env_ctx as args. An "underlying sub environment" is a single clone of an env within a vectorized environment. `func` takes a single underlying sub environment and the env_context as args. Args: func: A function - taking a BaseEnv object and an EnvContext as arg - and returning a list of lists of return values over envs of the worker. Returns: The list (1 item per workers) of lists (1 item per sub-environment) of results. c.|Sr)foreach_env_with_contextr>s rrrz9EnvRunnerGroup.foreach_env_with_context..s!44T::rTr?r@rBs `rrrEz'EnvRunnerGroup.foreach_env_with_contexts>(  # #::::!% $     rcN|j|jjdS)zChecks for unhealthy workers and tries restoring their states. Returns: List of IDs of the workers that were restored. T)rr)rmprobe_unhealthy_actorsrU!env_runner_health_probe_timeout_srs rrprobe_unhealthy_env_runnersz*EnvRunnerGroup.probe_unhealthy_env_runnerss0 #:: /Q;   rcpg}|fddD]}|||S)aCalls `func` with each worker's (policy, PolicyID) tuple. Note that in the multi-agent case, each worker may have more than one policy. Args: func: A function - taking a Policy and its ID - that is called on all workers' Policies. Returns: The list of return values of func over all workers' policies. The length of this list is: (num_workers + 1 (local-worker)) * [num policies in the multi-agent config dict]. The local workers' results are first, followed by all remote workers' results c.|Sr)foreach_policyr>s rrrz/EnvRunnerGroup.foreach_policy..(sa&&t,,rTr?rextendrprresultsr,s ` rrrLzEnvRunnerGroup.foreach_policysX&(( , , , ,t)    A NN1    rcpg}|fddD]}|||S)aApply `func` to all workers' Policies iff in `policies_to_train`. Args: func: A function - taking a Policy and its ID - that is called on all workers' Policies, for which `worker.is_policy_to_train()` returns True. Returns: List[any]: The list of n return values of all `func([trainable policy], [ID])`-calls. c.|Sr)foreach_policy_to_trainr>s rrrz8EnvRunnerGroup.foreach_policy_to_train..<sa//55rTr?rMrOs ` rrrSz&EnvRunnerGroup.foreach_policy_to_train-sX(( 5 5 5 5)    A NN1    r policy_idbatchcp|jr)|jjdS|j||St)zCWhether given PolicyID (optionally inside some batch) is trainable.NT)r7is_policy_to_trainNotImplementedError)rprTrUs rrrWz!EnvRunnerGroup.is_policy_to_trainAs?   &$7?t(;;IuMM M% %r)rrr6 policy_statepolicy_mapping_fnpolicies_to_train module_specworkers policy_clspolicyrrrYrZr[r\r]c  |jrL||jjvr>td|dt|jj| t urt ddd|du|dukrtdt|d |*t|||||||| rt| nd| nWt|t||j |j |j ||| rt| nd| fd } |j4| |j|||| | n|jjdi || d dS)a Adds a policy to this EnvRunnerGroup's workers or a specific list of workers. Args: policy_id: ID of the policy to add. policy_cls: The Policy class to use for constructing the new Policy. Note: Only one of `policy_cls` or `policy` must be provided. policy: The Policy instance to add to this EnvRunnerGroup. If not None, the given Policy object will be directly inserted into the local worker and clones of that Policy will be created on all remote workers. Note: Only one of `policy_cls` or `policy` must be provided. observation_space: The observation space of the policy to add. If None, try to infer this space from the environment. action_space: The action space of the policy to add. If None, try to infer this space from the environment. config: The config object or overrides for the policy to add. policy_state: Optional state dict to apply to the new policy instance, right after its construction. policy_mapping_fn: An optional (updated) policy mapping function to use from here on. Note that already ongoing episodes will not change their mapping but will use the old mapping till the end of the episode. policies_to_train: An optional list of policy IDs to be trained or a callable taking PolicyID and SampleBatchType and returning a bool (trainable or not?). If None, will keep the existing setup in place. Policies, whose IDs are not in the list (or for which the callable returns False) will not be updated. module_spec: In the new RLModule API we need to pass in the module_spec for the new module that is supposed to be added. Knowing the policy spec is not sufficient. workers: A list of EnvRunner/ActorHandles (remote EnvRunners) to add this policy to. If defined, will only add the given policy to these workers. Raises: KeyError: If the given `policy_id` already exists in this EnvRunnerGroup. z Policy ID 'z' already exists in policy map! Make sure you use a Policy ID that has not been taken yet. Policy IDs that are already in your policy map: z)EnvRunnerGroup.add_policy(.., workers=..)zdThe `workers` argument to `EnvRunnerGroup.add_policy()` is deprecated! Please do not use it anymore.T)r@helprBNzeOnly one of `policy_cls` or `policy` must be provided to staticmethod: `EnvRunnerGroup.add_policy()`!F)rB) rTr^rrr6rYrZr[r\c"|jdidS)Nr) add_policy)workernew_policy_instance_kwargss rr_create_new_policy_fnz8EnvRunnerGroup.add_policy.._create_new_policy_fns% F  ; ; : ; ; ; ; ;r)rTr_rZr[r\r?r)r7 policy_mapKeyErrorrAkeysrrrerrQtyperrr6rrcr)rprTr^r_rrr6rYrZr[r\r]rfres @rrrczEnvRunnerGroup.add_policyMs9x   Y$2G2R%R%RCiCC-8==??@@CC  * * * ?@      $ FdN 3 3?  9E2222  !)-#%"3))"3$#$'8"9"9"9' * * * & &"*.#<<"(":#0}#--//"3$#$'8"9"9"9' * * * & < < < < <  ,!%00'!&7&7 + 11%0NN3MNNN  5NNNNNr)recreated_workerrzryrkrzc ^t|||j|||||j||j|j }|dkr |jdi|St jdn |j |z} t j di|j |j | j di|S)N) r3r4r5r6ryr=rklog_dirrzdataset_shardsr:r)placement_group_bundle_indexr) rQrTrirr_rarVutilget_current_placement_groupr`r r^options) rpr3r4ryr=rkr6rzr pg_bundle_idxs rrrzEnvRunnerGroup._make_workers#%!%!3%#-L?-     1  &4&0000 0x3355= B</    +CJ + +* + +D,? @ @ W-W @ @      rc tt|trtj|sd|vr|dd\}} t j|}|dSnA#ttf$r-}t d|d|d|Yd}~nd}~wwxYwdS)N.rJTzmodule z not found using input z with error: F) rProspathisfilersplit importlibrq find_specModuleNotFoundErrorrerwarning) class_path module_path class_namespecrqs rr _valid_modulezEnvRunnerGroup._valid_modules z3 ' ' GNN:.. z!!&0&7&7Q&?&? #K  ~// <<#4$'4   ^k^^*^^[\^^ us!A77B5#B00B5r)NNrFFTNF)NNNNrF)F)rN)NN)H__name__ __module__ __qualname____doc__rr r&rr'r rrrrrsr9rpropertyrr7rrrrrrrrrrrrrrr r0floatr rr*r,rrrr"r%rr4r:r!r=rrrErIrLrSr+rWgymrzSpacer)rr%r(rrrcr staticmethodrrrrrr2r2Es -1<@7;.2!% $'+)-$%VVVj)Vx 489 V 'tF|4 V *+ VV VV }VV"#VVVVv=A.2 !% FFFx 489F*+ F  F  FFFFP4&)&&&X&8S 8888-DI----11111-C----999995C5555UUUUU.S....HHHSMHSHHHH9C9999,0+/;?48<@CCC"Ci( C $C= C #4S#X#78 C"$sCx.1C'/tCy&9C CCCCT,0!+. %"!'+!&   i[!^ $d8YKN+C&Dc4PS9 T  c3h(  c]  "%     9   eE#q&M1$% &    F.2SW157;+.).QCQC4>*QC'/uY5N/O&PQC$DI. QC d3 ?34 QC "% QC!QC QCQCQCQCf- - s- d- t- - - - ^<[(9,A#BtAw[2Hfh5G5J,KPTUVPW[&FJ & &! &*2?*C &  & & &[ &.2#' IO 9=37QU.2PT .2AQ'IOIOIOIOT&\*IO IO $CJ$45 IOsz/0IO02LLMNIO{+IO$Hgy-A8-K$LMIO$ 8$(H_$=>DEG  IO"l+#IO&$uY %;<=>'IO( )IOIOIO[IOd"' ' ' '  ' x 489 '  '  ' ' "'  5!13:3C!CDD E '  y+% &' ' ' ' R\rr2)Pimportlib.utilr{loggingrwtypingrrrrrrr r r r r gymnasiumrrVray._common.deprecationrr ray.actorrray.exceptionsrray.rllib.corerrrrray.rllib.core.learnerrray.rllib.core.rl_moduler"ray.rllib.core.rl_module.rl_modulerray.rllib.env.base_envrray.rllib.env.env_contextrray.rllib.env.env_runnerr#ray.rllib.evaluation.rollout_workerrray.rllib.offlinerray.rllib.policy.policyrrray.rllib.utils.actor_managerr ray.rllib.utils.annotationsr!ray.rllib.utils.frameworkr"ray.rllib.utils.metricsr#r$ray.rllib.utils.typingr%r&r'r(r)r*r+r,ray.util.annotationsr-rOr/tf1tftfv getLoggerrrr0r2rrrrrsK                            "!!!!!(((((( 0/////777777;;;;;;******000000......======44444477777777CCCCCC333333333333RRRRRRRR                    .-----FEEEEEE} R  8 $ $ GCLLLLLLLLLLLLr