&`i:UddlZddlZddlZddlZddlZddlZddlZddlm Z ddl m Z ddl m Z mZmZmZmZmZmZmZmZmZmZddlmZddlZddlmZmZddlmZm Z ddl!m"Z"dd l#m$Z$dd l%m&Z&m'Z'dd l(m)Z)dd l*m+Z+dd l,m-Z-ddl.m/Z/ddl0m1Z1m2Z2ddl3m4Z4ddl5m6Z6ddl7m8Z8ddl9m:Z:ddl;mZ>m?Z?m@Z@mAZAmBZBmCZCmDZDmEZEmFZFddlGmHZHmIZIddlJmKZKddlLmMZMmNZNmOZOmPZPddlQmRZRddlSmTZTddlUmVZVddlWmXZXmYZYddlZm[Z[m\Z\ddl]m^Z^m_Z_ddl`maZambZbddlcmdZdmeZedd lfmgZgdd!lhmiZidd"ljmkZkdd#llmmZmdd$lnmoZpmqZqdd%lrmsZsmtZtmuZumvZvmwZwmxZxmyZymzZzm{Z{m|Z|m}Z}dd&l~mZmZdd'lmZdd(lmZmZmZdd)lmZe r dd*lmZdd+lmZed\ZZZee\ZZejeZdaed,ed-<eXd6d/Zd0eud1ed2ed3efd4ZeXGd5d,ee+ZdS)7N) defaultdict) FunctionType) TYPE_CHECKINGAnyCallable CollectionDictListOptionalSetTupleTypeUnion)Space) ObjectRef cloudpickle)create_connectors_for_policymaybe_get_filters_for_syncing)validate_module_id) RLModuleSpec)BaseEnvconvert_to_base_env) EnvContext) EnvRunner)ExternalMultiAgentEnv) MultiAgentEnv)is_atari wrap_deepmind)RolloutMetrics) SyncSampler) ModelCatalog) Preprocessor) D4RLReader DatasetReader DatasetWriter InputReader IOContext JsonReader JsonWriter MixedInput NoopOutput OutputWriter ShuffledInput)Policy PolicySpec) PolicyMap)DEFAULT_POLICY_IDMultiAgentBatchconcat_samples convert_ma_batch_to_sample_batch) TorchPolicy) TorchPolicyV2) force_list) OldAPIStackoverride) summarizeupdate_global_seed_if_necessary)ERR_MSG_NO_GPUSHOWTO_CHANGE_CONFIG)FilterNoFilter) try_import_tftry_import_torch) from_config)create_policy_for_framework)do_minibatch_sgd) _TFRunBuilder)get_gpu_devicesget_tf_eager_cls_if_necessary) AgentID EnvCreatorEnvTypeModelGradients ModelWeightsMultiAgentPolicyConfigDictPartialAlgorithmConfigDictPolicyID PolicyStateSampleBatchTypeT)registry_contains_inputregistry_get_input) PublicAPI)disable_log_once_globallyenable_periodic_logginglog_once)ParallelIteratorWorkerAlgorithmConfig) RLlibCallback RolloutWorker_global_workerreturnctS)z>Returns a handle to the active rollout worker in this process.)r^w/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/rllib/evaluation/rollout_worker.pyget_global_workerrdrs rbenvseed worker_idx vector_idxcV|dSd}||ks Jd||z|z|z}t|ds1tdr td|ddSdS ||dS#t $r"td|d YdSwxYw) zzSet a deterministic random seed on environment. NOTE: this may not work with remote environments (issue #18154). Niz3Too many envs per worker. Random seeds may collide.resetenv_has_no_reset_methodzEnv z. doesn't have a `reset()` method. Cannot seed.)rfz doesn't support setting a seed via its `reset()` method! Implement this method as `reset(self, *, seed=None, options=None)` for it to abide to the correct API. Cannot seed.)hasattrrXloggerinforj Exception)rerfrgrhmax_num_envs_per_env_runner computed_seeds rc_update_env_seed_if_necessaryrrzs |(,0000< 100#&AAJNQUUM 3  - . . T KKRsRRR S S S S S T T  II=I ) ) ) ) )    KKRsRRR       s$A<<(B('B(ceZdZdZdddddddddd dedeeeegdfded d e d ee d e d ee dee e eeeffdeeedeeejjfdZeedZeedZeedefdZeede e eeefffdZejddeee ffdZdede fdZ de de de dee dee!e ff d Z" dlded!e dee#e!ffd"Z$d#e%e#e e e#ffddfd$Z&eedee'fd%Z(d&eege)fdee)fd'Z*d&eeege)fdee)fd(Z+e,fd)e deefd*Z- dmdddddddd+d)e d,eeed-eed.eed/eedee.d0ee/d1ee%e0e ee ege ffd2ee1defd3Z2e,ddd4d)e d5eee3ge fd1ee%e0e ee ege ffddfd6Z4 dld5eee3e5ge fddfd7Z6d8e%e0e ee eege ffddfd9Z7e8d:; dldZ;d&eee ee5ge)fdee)fd?ZdndCe de fdDZ?de!fdEZ@dFe!ddfdGZA dodHee0e dIe de e eBffdJZC dmdKe e eBfdLee dMee ddfdNZDde!fdOZE dldLe!dPeee ddfdQZFeedpdRZGdpdSZHdpdTZIdUe dVe dWe dXe ddf dYZJde!fdZZKde fd[ZLde fd\ZMde fd]ZNdddd^d_eOd-eed`ee e e/fdaee1ddf dbZPd_eOdeOfdcZQd_eOdeOfddZRddded_eOd-eed`ee e e/fddfdfZSd_eOddfdgZTdhZUdiZVdjZWdkZXdS)qr]a Common experience collection class. This class wraps a policy instance and an environment class to collect experiences from the environment. You can create many replicas of this class as Ray actors to scale RL training. This class supports vectorized and multi-agent policy evaluation (e.g., VectorEnv, MultiAgentEnv, etc.) .. testcode:: :skipif: True # Create a rollout worker and using it to collect experiences. import gymnasium as gym from ray.rllib.evaluation.rollout_worker import RolloutWorker from ray.rllib.algorithms.ppo.ppo_tf_policy import PPOTF1Policy worker = RolloutWorker( env_creator=lambda _: gym.make("CartPole-v1"), default_policy_class=PPOTF1Policy) print(worker.sample()) # Creating a multi-agent rollout worker from gymnasium.spaces import Discrete, Box import random MultiAgentTrafficGrid = ... worker = RolloutWorker( env_creator=lambda _: MultiAgentTrafficGrid(num_cars=25), config=AlgorithmConfig().multi_agent( policies={ # Use an ensemble of two policies for car agents "car_policy1": (PGTFPolicy, Box(...), Discrete(...), AlgorithmConfig.overrides(gamma=0.99)), "car_policy2": (PGTFPolicy, Box(...), Discrete(...), AlgorithmConfig.overrides(gamma=0.95)), # Use a single shared policy for all traffic lights "traffic_light_policy": (PGTFPolicy, Box(...), Discrete(...), {}), }, policy_mapping_fn=( lambda agent_id, episode, **kwargs: random.choice(["car_policy1", "car_policy2"]) if agent_id.startswith("car_") else "traffic_light_policy"), ), ) print(worker.sample()) .. testoutput:: SampleBatch({ "obs": [[...]], "actions": [[...]], "rewards": [[...]], "terminateds": [[...]], "truncateds": [[...]], "new_obs": [[...]]} ) MultiAgentBatch({ "car_policy1": SampleBatch(...), "car_policy2": SampleBatch(...), "traffic_light_policy": SampleBatch(...)} ) NrF) validate_envconfig worker_index num_workersrecreated_workerlog_dirspacesdefault_policy_classdataset_shards env_creatorrtrur[rvrwrxryrzr{r|c  @t_jd=addlm} |t |tr| |pi}| |j rB|dkr<|j D]!\} }t|tj| <"nH|jrA|dkr;|j D]!\} }t|tj| <"fd}t!j|dt%j|||n jj_| _|_t1j_t6rD|jdks|jr2t6st6jj r1tCj"d #jj jd krtInjj d krtKtMjj'jdjjj(| }|_)|_j*_+|_,d _-.jj-|_/j0j}|jj1z_2|j3 _4d_5dtmtnd_8jj9dn#jj9jzjj:dzz_9jdkrtwjjj9dx_<_=jdkrjdkr jj>r'|tj?j)_<jj@}j<B||j<j)t j<ttjCjDfrd}n{tj<rMjjFdkr=d_4jj@d}jjGHddufd}njjF d_4d}nd}|j<_<tj<j9jdj+Jj<j)K||||j9_=|_L| _MjNj<jLjM\_O_Pd_Qd_Rjdkr jjSn jjT}tjUrtjVjWXtjVjWjYkr|jZsg}jjdvrt}nHjjdkr8tttj_`}t||kr7ttjdt||tzn~tjUrktjVjWXtjVjWjYkr0|dkr*jjZstgd|dtmt_id_jkjOjQlD]}|jms|njjorVj<Ot j<ttttjCjDfstdj<d jdkr2tsd!djij<d_tn{d"jvr%juj<_tnMtj<j=jj1jj(jjwjjx#_t|}jjyd$krd}n#jjyd%ksJtd&}d}t|jj_|d}jj}durjdks jd krd}j<d_~n\tjt||jjj+|jjjjjjjj|' _~j|_j|_d_dS)(aInitializes a RolloutWorker instance. Args: env_creator: Function that returns a gym.Env given an EnvContext wrapped configuration. validate_env: Optional callable to validate the generated environment (only on worker=0). worker_index: For remote workers, this should be set to a non-zero and unique value. This index is passed to created envs through EnvContext so that envs can be configured per worker. recreated_worker: Whether this worker is a recreated one. Workers are recreated by an Algorithm (via EnvRunnerGroup) in case `restart_failed_env_runners=True` and one of the original workers (or an already recreated one) has failed. They don't differ from original workers other than the value of this flag (`self.recreated_worker`). log_dir: Directory where logs can be placed. spaces: An optional space dict mapping policy IDs to (obs_space, action_space)-tuples. This is used in case no Env is created on this RolloutWorker. selfrrZNc38K VN)samplersrc gen_rolloutsz,RolloutWorker.__init__..gen_rolloutss% $kkmm### $rbF)rutf2z ray.rllibDEBUG)rv vector_indexrwremoterxctSr)r1)agent_idepisodeworkerkws rcz(RolloutWorker.__init__..Rs4Erb)rv)timestepnum_grad_updates_per_policyi'c|Srrares rcwrapz$RolloutWorker.__init__..wrapJrbdeepmindT framestackc t|jjdjjdddk}|S)Ndim frameskiprr)rr noframeskip)rrumodelget env_config)reruse_framestacks rcrz$RolloutWorker.__init__..wrapsX' K-11%88#1$(K$:$>$>{A$N$NRS$S C Jrbc|Srrars rcrz$RolloutWorker.__init__..wraprrbc|Srrars rcrz$RolloutWorker.__init__..wraprrbrsub_environment env_context)rerzr{)rtftorchz@You are running ray with `local_mode=True`, but have configured zf GPUs to be used! In local mode, Policies are placed on the CPU and the `num_gpus` setting is ignored.) policy_dictz1You are running a multi-agent setup, but the env zT is not a subclass of BaseEnv, MultiAgentEnv, ActorHandle, or ExternalMultiAgentEnv!zBuilt filter map: {}custom_vector_env)make_envnum_envs remote_envsremote_env_batch_wait_msrrestart_failed_sub_environmentstruncate_episodescomplete_episodesinf) rre clip_rewardsrollout_fragment_lengthcount_steps_by callbacksmultiple_episodes_in_batchnormalize_actions clip_actionsobservation_fnsample_collector_classrender)localscopy_original_kwargsr^%ray.rllib.algorithms.algorithm_configr[ isinstancedictupdate_from_dictfreeze extra_python_environs_for_driveritemsstrosenviron extra_python_environs_for_workerrY__init__rrunum_env_runnersrw _ds_shardsrv threadingLock_locktf1 framework_strenable_tf1_exec_eagerlyexecuting_eagerlyenable_eager_execution log_levellogging getLoggersetLevelrVrWrrremote_worker_envsrcallbacks_classrrxpolicy_mapping_fnset_policy_mapping_fnr}get_rollout_fragment_lengthnum_envs_per_env_runnertotal_rollout_fragment_length_disable_preprocessor_apipreprocessing_enabled last_batchrint global_varsrf in_evaluationr;remake_sub_env_fncreate_local_env_runnerdeepcopyrrrayactor ActorHandlerpreprocessor_prefrrrron_sub_environment_created_get_make_sub_env_fnrzr{get_multi_agent_setupris_policy_to_train policy_map preprocessorsnum_gpusnum_gpus_per_env_runneris_initialized_privater_mode LOCAL_MODE _fake_gpusget_tf_gpu_deviceslistrangercuda device_countlen RuntimeErrorr<formatr=rmwarningr?filtersmulti_rl_module_spec_update_policy_mapvalues%_model_init_state_automatically_added/_update_model_view_requirements_from_init_stateis_multi_agentrr ValueErrorrn async_envrrrr batch_modefloatr' io_context render_envsamplerr rrrrsample_collector_get_input_creator_from_config input_reader_get_output_creator_from_config output_writerweights_seq_no)rr}rtrurvrwrxryrzr{r|kwargsr[keyvaluerr"configured_rollout_fragment_lengthrrrdevicespol#rollout_fragment_length_for_samplerpackrrs` @rcrzRolloutWorker.__init__s F'-hhmmoo  !& )IIIIII >Z55>$_&&77 "EEF   2 -|q7H7H$EKKMM - - U"%e** 3 -  4 -9I9I$EKKMM - - U"%e** 3 $ $ $ $ $ 'lEBBB4////'2KK 8S  )!-^%%   )%..&2P.))++/  & & ( ( ( ;  K  k * * 3 3DK4I J J J  q % ' ' ' ' [ "g - - # % % % K "*(;1-    ''- (, (C(C(E(E&6 F E  ""4;#@AAA'2-1[-T-T*.U. . * /1T T *06/O+O"59 ,7s+;+; " " {' D! k'%/0   q +DK,Ety Q Q Q+/.4'   " " 1$$K7%#{4=1A#B#BCCDH{/ 8 ' TXt'7888$(Wci.C$DEE% $(##  (E(S(S-2*;+3#'L"&!2!6!6|!D!D!L.6-2* !%TXDH *$(DIt?PRS T T T N 5 5 $ , 6    $(#<#<[,di$$D  $8!48K4U4U;!%!:5V5 5 1$1 04;?  A%% K 4      #))++s|/B/MMM%NG{(M99,..*g55uUZ%<%<%>%>??@@7||h&&"#*3w<<AADWW'     #))++s|/B/MMM1 K* NN&   0;8/D/D %)! D,<=== ?))++ F FC< FCCEEE K & $/ @UV% *DH***    ! ! KK.55dlCC D D D 8 !DNN DK / /![::48DDDNN ':-< K:)-)MK? ' ' 'DN/Q+ ; !%8 8 8DD;)-@@@@@27,, /D&/ T[$"3T& &  ; !T ) )   ! !T%6!%;%;F 8 DLL&N)(K#{9.+/"&+"?![5#{9'+{'C   DL*O)L)L)N)N O* * ,R4+O+O+Q+Q O, ,  .2rbctr)NotImplementedErrorrs rcrzRolloutWorker.make_envns "!rbcp|jo |jo|j}|sJd|d|jd|jddS)NzRolloutWorker z (idx=z; num_workers=z) not healthy!)rrrrvrw)r is_healthys rcassert_healthyzRolloutWorker.assert_healthyusq_Q):Qt?Q    .sC!4anErbc fi|].}|dt|dd|d|df/S)rroriginal_space)getattr).0es rc z,RolloutWorker.get_spaces..s<VVV!!A$1'71>>!EVVVrbc|j|jfSrr9rs rcrz*RolloutWorker.get_spaces..s.0@Arbr)INPUT_ENV_SPACES)foreach_policy foreach_env ray.rllib.envrF)rrz env_spacesrFs rc get_spaceszRolloutWorker.get_spacess~$$ E E  WVvVVV%% A A    5 6 6 6 6 6 6'1!}F# $ rbr@) num_returnsc<|}||jfS)aSame as sample() but returns the count as a separate value. Returns: A columnar batch of experiences (e.g., tensors) and the size of the collected batch. .. testcode:: :skipif: True import gymnasium as gym from ray.rllib.evaluation.rollout_worker import RolloutWorker from ray.rllib.algorithms.ppo.ppo_tf_policy import PPOTF1Policy worker = RolloutWorker( env_creator=lambda _: gym.make("CartPole-v1"), default_policy_class=PPOTFPolicy) print(worker.sample_with_count()) .. testoutput:: (SampleBatch({"obs": [...], "action": [...], ...}), 3) )rr+rr6s rcsample_with_countzRolloutWorker.sample_with_counts. ek!!rbr&c,tdr:tdt |i}t |t rii}|jD]\}}|j | ||s#| |j |}| }|rCt|dr3t|d|<|||||<||||<|fd|Dna|j | t$|r?|t$|j t$|itdr:tdt ||S)aHUpdate policies based on the given batch. This is the equivalent to apply_gradients(compute_gradients(samples)), but can be optimized to avoid pulling gradients into CPU memory. Args: samples: The SampleBatch or MultiAgentBatch to learn on. Returns: Dictionary of extra metadata from compute_gradients(). .. testcode:: :skipif: True import gymnasium as gym from ray.rllib.evaluation.rollout_worker import RolloutWorker from ray.rllib.algorithms.ppo.ppo_tf_policy import PPOTF1Policy worker = RolloutWorker( env_creator=lambda _: gym.make("CartPole-v1"), default_policy_class=PPOTF1Policy) batch = worker.sample() info = worker.learn_on_batch(samples) learn_on_batchz-Training on concatenated sample batches: {} N_build_learn_on_batchcNi|]!\}}|||"Srar)rBr=vbuilderss rcrDz0RolloutWorker.learn_on_batch..2s1VVV63S(3-"3"3A"6"6VVVrb learn_outzTraining out: {} )rXrmrnrr:rr2policy_batchesrrdecompress_if_neededr get_sessionrlrErRrQupdater1debug) rr&info_outto_fetchr=r6policy tf_sessionrVs @rcrQzRolloutWorker.learn_on_batchs$0 $ % %  KKBIIg&&     g / / HH%4::<< A A U*6t?V?V@@6**,,,-#//11 A'&2I"J"JA$1*>N$O$OHSM$*$@$@#PU$V$VHSMM$*$9$9%$@$@HSMM OOVVVVX^^EUEUVVV W W W W&.$2I2I!733.)4?-,(.11 K  N LL077 (8K8KLL M M Mrbexpected_batch_size num_sgd_itersgd_minibatch_sizestandardize_fieldsc|}|j|ksJd|d|jftd|j|t ||j||||}||jfS)aASample and batch and learn on it. This is typically used in combination with distributed allreduce. Args: expected_batch_size: Expected number of samples to learn on. num_sgd_iter: Number of SGD iterations. sgd_minibatch_size: SGD minibatch size. standardize_fields: List of sample fields to normalize. Returns: A tuple consisting of a dictionary of extra metadata returned from the policies' `learn_on_batch()` and the number of samples learned on. z:Batch size possibly out of sync between workers, expected:zgot:zIExecuting distributed minibatch SGD with epoch size {}, minibatch size {})rr+rmrnrrDr)rrarbrcrdr6rns rcsample_and_learnzRolloutWorker.sample_and_learnBs, {1111 H   K 4 111   44:F /55     O       U[  rb single_agentctdr:tdt ||durEt |}|jt|\}}|j |d<||fS| }ii}}|j j dkr|j D]q\}}|j|||s#|j|}t!|d||\||<||<rfd|D}fd|D}nf|j D]L\}}|j|||s#|j||\||<||<M|j |d<td r:td t |||fS) aFReturns a gradient computed w.r.t the specified samples. Uses the Policy's/ies' compute_gradients method(s) to perform the calculations. Skips policies that are not trainable as per `self.is_policy_to_train()`. Args: samples: The SampleBatch or MultiAgentBatch to compute gradients for using this worker's trainable policies. Returns: In the single-agent case, a tuple consisting of ModelGradients and info dict of the worker's policy. In the multi-agent case, a tuple consisting of a dict mapping PolicyID to ModelGradients and a dict mapping PolicyID to extra metadata info. Note that the first return value (grads) can be applied as is to a compatible worker using the worker's `apply_gradients()` method. .. testcode:: :skipif: True import gymnasium as gym from ray.rllib.evaluation.rollout_worker import RolloutWorker from ray.rllib.algorithms.ppo.ppo_tf_policy import PPOTF1Policy worker = RolloutWorker( env_creator=lambda _: gym.make("CartPole-v1"), default_policy_class=PPOTF1Policy) batch = worker.sample() grads, info = worker.compute_gradients(samples) compute_gradientszCompute gradients on: {} T batch_countrNcBi|]\}}||SrarTrBkrUbuilders rcrDz3RolloutWorker.compute_gradients..)GGGda7;;q>>GGGrbcBi|]\}}||SrarTrls rcrDz3RolloutWorker.compute_gradients..rorbgrad_outzCompute grad info: {} )rXrmrnrr:r4rr1rir+as_multi_agentrurrXrrrErZ_build_compute_gradients) rr&rgrqr]r=r6r_rns @rcrizRolloutWorker.compute_gradientsoszH ' ( ( T KK7>>y?Q?QRR S S S 4  6w??G!%1B!C!U!U"" Hh'.mH] #X% %((** ( ; $ , ,%4::<<   U*6t?V?V@@6-'(:(:(<(<>QRR/5/N/NU00, x}}HGGGhnn6F6FGGGHGGGGhnn6F6FGGGHH%4::<<   U*6t?V?V@@6/3s/C/U/U00, x}}#*- J   R KK4;;Ih >u E E E E E- ,rbcL|j|j}ng}|S)zReturns the thus-far collected metrics from this worker's rollouts. Returns: List of RolloutMetrics collected thus-far. )r get_metrics)routs rcryzRolloutWorker.get_metricss+ < #,**,,CCC rbfuncc|jgS|j}|s|jgSfd|DS)aCalls the given function with each sub-environment as arg. Args: func: The function to call for each underlying sub-environment (as only arg). Returns: The list of return values of all calls to `func([env])`. Nc&g|] }|Srara)rBrCr{s rc z-RolloutWorker.foreach_env..s!***DDGG***rb)r get_sub_environments)rr{envss ` rcrHzRolloutWorker.foreach_envs` > !I~2244 +D(() )+***T*** *rbc|jgS|j}|s||j|jgSg}t|D]?\}}|j|}||||@|S)a.Calls given function with each sub-env plus env_ctx as args. Args: func: The function to call for each underlying sub-environment and its EnvContext (as the args). Returns: The list of return values of all calls to `func([env, ctx])`. Nr)r rr enumeratecopy_with_overridesr.)rr{rretirCctxs rcforeach_env_with_contextz&RolloutWorker.foreach_env_with_contexts > !I~2244 D)9::; ;C!$ ) )1&:::JJ 443<<((((Jrb policy_idc6|j|S)aReturn policy for the specified id, or None. Args: policy_id: ID of the policy to return. None for DEFAULT_POLICY_ID (in the single agent case). Returns: The policy under the given ID (or None if not found). )rr)rrs rc get_policyzRolloutWorker.get_policy.s""9---rb)r:r;ru policy_staterpolicies_to_train module_spec policy_clsr_r:r;rrrc t|d| td||jvr9td|dt |j|du|dukrtd|C|j|t||||i|j |j |j \} } n0|tt||j |j|ji} |j| || |||i| ||| || |j|S) adAdds a new policy to this RolloutWorker. 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 algorithm. Note: Only one of `policy_cls` or `policy` must be provided. observation_space: The observation space of the policy to add. action_space: The action space of the policy to add. config: The config 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 collection of policy IDs to be trained or a callable taking PolicyID and - optionally - 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. Returns: The newly added policy. Raises: ValueError: If both `policy_cls` AND `policy` are provided. KeyError: If the given `policy_id` already exists in this worker's PolicyMap. FerrorNzOIf you pass in module_spec to the policy, the RLModule API needs to be enabled. 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: zTOnly one of `policy_cls` or `policy` must be provided to RolloutWorker.add_policy()!)policiesrerzr{)rr_ policy_statessingle_agent_rl_module_spec)rr rKeyErrorrkeysrurr/rerzr{typer:r;rr[rrset_is_policy_to_train) rrrr_r:r;rurrrrpolicy_dict_to_add_s rc add_policyzRolloutWorker.add_policy:sf 9E2222  "!   ' '2i22,,..//22  $ FdN 3 3.  >$(K$E$Ez"$5|V   H{%)%>%F % % ! :LL,'M "   2333 *$l3(3     ""#4555  (  ' '(9 : : :y))rb)rrrrc||jvrtd|d|j|=|j|=|||||dSdS)aURemoves a policy from this RolloutWorker. Args: policy_id: ID of the policy to be removed. None for DEFAULT_POLICY_ID. 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 collection of policy IDs to be trained or a callable taking PolicyID and - optionally - 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. rz' not in policy map!N)rr rrr)rrrrs rc remove_policyzRolloutWorker.remove_policys~2 DO + +J9JJJKK K OI &  y ) ""#4555  (  ' '(9 : : : : : ) (rbcb|*||_t|jstddSdS)zSets `self.policy_mapping_fn` to a new callable (if provided). Args: policy_mapping_fn: The new mapping function to use. If None, will keep the existing mapping function in place. Nz'`policy_mapping_fn` must be a callable!)rcallabler )rrs rcrz#RolloutWorker.set_policy_mapping_fnsI  (%6D "D233 L !JKKK ) ( L Lrbrct|sAt|tttfs Jdt|dfd }||_dS)aSets `self.is_policy_to_train()` to a new callable. Args: is_policy_to_train: A collection of policy IDs to be trained or a callable taking PolicyID and - optionally - 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. zERROR: `is_policy_to_train`must be a [list|set|tuple] or a callable taking PolicyID and SampleBatch and returning True|False (trainable or not?).Nc |vSrra)r=r6polss rcrz@RolloutWorker.set_is_policy_to_train..is_policy_to_trains d{"rbr)rrrsettupler)rrrs @rcrz$RolloutWorker.set_is_policy_to_trains$*++ #04e2DEE  2  E )**D # # # # # ##5rbalpha) stabilityr6cRfdjDS)aReturns all policies-to-train, given an optional batch. Loops through all policies currently in `self.policy_map` and checks the return value of `self.is_policy_to_train(pid, batch)`. Args: batch: An optional SampleBatchType for the `self.is_policy_to_train(pid, [batch]?)` check. Returns: The set of currently trainable policy IDs, given the optional `batch`. cNh|]!}j||"Sr)r)rBr=r6rs rc z6RolloutWorker.get_policies_to_train..sA   &.$2I2I#u2U2U. ...rbrrrNs``rcget_policies_to_trainz#RolloutWorker.get_policies_to_trainsB"     ++--    rbc *||j|fi|S)aCalls the given function with the specified policy as first arg. Args: func: The function to call with the policy as first arg. policy_id: The PolicyID of the policy to call the function with. Keyword Args: kwargs: Additional kwargs to be passed to the call. Returns: The return value of the function call. )r)rr{rrs rc for_policyzRolloutWorker.for_policy s#&tDOI.99&999rbc Rfd|jDS)arCalls the given function with each (policy, policy_id) tuple. Args: func: The function to call with each (policy, policy ID) tuple. Keyword Args: kwargs: Additional kwargs to be passed to the call. Returns: The list of return values of all calls to `func([policy, pid, **kwargs])`. c*g|]\}}||fiSrara)rBr=r_r{rs rcr~z0RolloutWorker.foreach_policy../s1WWW VVS++F++WWWrb)rrrr{rs ``rcrGzRolloutWorker.foreach_policy s2XWWWWt?T?T?V?VWWWWrbc VfdjDS)aK Calls the given function with each (policy, policy_id) tuple. Only those policies/IDs will be called on, for which `self.is_policy_to_train()` returns True. Args: func: The function to call with each (policy, policy ID) tuple, for only those policies that `self.is_policy_to_train` returns True. Keyword Args: kwargs: Additional kwargs to be passed to the call. Returns: The list of return values of all calls to `func([policy, pid, **kwargs])`. ctg|]4}j|dj||fi5Sr)rr)rBr=r{rrs rcr~z9RolloutWorker.foreach_policy_to_train..Fs^    &.$2I2I#t2T2T. D%s 5 5f 5 5...rbrrs```rcforeach_policy_to_trainz%RolloutWorker.foreach_policy_to_train1sJ*       ++--    rb new_filtersctfd|jDsJ|jD](}|j||)dS)zChanges self's filter to given and rebases any accumulated delta. Args: new_filters: Filters with new state to update local copy. c3 K|]}|vV dSrra)rBrmrs rc z-RolloutWorker.sync_filters..Xs(::1 #::::::rbN)allrsync)rrrms ` rc sync_filterszRolloutWorker.sync_filtersRsl ::::T\:::::::: 1 1A LO Q 0 0 0 0 1 1rb flush_afterci}|jD]2\}}|||<|r|3|S)zReturns a snapshot of filters. Args: flush_after: Clears the filter buffer state. Returns: Dict for serializable filters )rras_serializable reset_buffer)rrreturn_filtersrmfs rc get_filterszRolloutWorker.get_filters\s_L&&(( ! !DAq ! 1 1 3 3N1  !   rbcj|d}i}|jD]L}|jjr|j||r"|j|||<Mt|j||j|j|dS)NT)r) policy_idsrrrr) rrrru"checkpoint_trainable_policies_onlyr get_staterr)rrrr=s rcrzRolloutWorker.get_statels""t"44 ?'')) F FCKB F*2**3//3&*_S%9%C%C%E%E c"t335566+!%!7"&"9   rbstatec8t|trtj|}||dd|vr|dn|d}|D]\}}t |d||jvr|dd}|"t d|d|d nYt|trtj |n|}|||j|j|j|j ||jvr |j||d |vr||d |d ||d dSdS) NrrrFr policy_specz PolicyID 'z' was probably added on-the-fly (not part of the static `multagent.policies` config) and no PolicySpec objects found in the pickled policy state. Will not add `z`, but ignore it for now.)rrr:r;rurr)rbytespickleloadsrrrrrrmrrr/ deserializer policy_classr:r;ru set_staterr)rrrr=rspecrs rcrzRolloutWorker.set_states eU # # (L''E % *+++ '6&>&>E/ " "E'N "/!4!4!6!6 = = C s% 0 0 0 0$/))#'' t<<<NNOSOO14OOO9C48N8NX .t444TX OO"%#.#;*5*G%0%=*1 $do%%$..|<<< % ' '  & &u-@'A B B B 99) * * 6  ' '.B(C D D D D D 7 6rbrinference_onlyc&tjtfdjDS)aReturns each policies' model weights of this worker. Args: policies: List of PolicyIDs to get the weights from. Use None for all policies. inference_only: This argument is only added for interface consistency with the new api stack. Returns: Dict mapping PolicyIDs to ModelWeights. .. testcode:: :skipif: True from ray.rllib.evaluation.rollout_worker import RolloutWorker # Create a RolloutWorker. worker = ... weights = worker.get_weights() print(weights) .. testoutput:: {"default_policy": {"layer1": array(...), "layer2": ...}} NcXi|]&}|v|j|'Sra)r get_weights)rBr=rrs rcrDz-RolloutWorker.get_weights..sA    h %1133rb)rrrr7)rrrs`` rcrzRolloutWorker.get_weightsst:  DO002233Hh''      ++--    rbweightsrrc| ||jkr|rttt|t r`t jt|fdt| D}| D]\\}}||j vr!|j | |/tdrtd|d]||_|r||dSdS)aSets each policies' model weights of this worker. Args: weights: Dict mapping PolicyIDs to the new weights to be used. global_vars: An optional global vars dict to set this worker to. If None, do not update the global_vars. weights_seq_no: If needed, a sequence number for the weights version can be passed into this method. If not None, will store this seq no (in self.weights_seq_no) and in future calls - if the seq no did not change wrt. the last call - will ignore the call to save on performance. .. testcode:: :skipif: True from ray.rllib.evaluation.rollout_worker import RolloutWorker # Create a RolloutWorker. worker = ... weights = worker.get_weights() # Set `global_vars` (timestep) as well. worker.set_weights(weights, {"timestep": 42}) Nc(i|]\}}||Srara)rBrr=actual_weightss rcrDz-RolloutWorker.set_weights.. s1/5q#C*rb"set_weights_on_non_existent_policyz>`RolloutWorker.set_weights()` used with weights from policyID=z>, but this policy cannot be found on this worker! Skipping ...)rrr*iterrrrrrrrrr set_weightsrXrmrset_global_vars)rrrrr=wrs @rcrzRolloutWorker.set_weightssa:  !^t7J%J%J :d40@0@+A+A&B&BINN !$gnn.>.>)?)?!@!@9B7<<>>9R9R"--//  Q$/))OC(44Q7777BCCNN0$'000 -  .   - - - - - . .rbc|jS)aReturns the current `self.global_vars` dict of this RolloutWorker. Returns: The current `self.global_vars` dict of this RolloutWorker. .. testcode:: :skipif: True from ray.rllib.evaluation.rollout_worker import RolloutWorker # Create a RolloutWorker. worker = ... global_vars = worker.get_global_vars() print(global_vars) .. testoutput:: {"timestep": 424242} )rrs rcget_global_varszRolloutWorker.get_global_varss &rbrc |}|di}|jd|||n|jD]_}|j||dr@|j|t|fid| |i`|j|dS)a(Updates this worker's and all its policies' global vars. Updates are done using the dict's update method. Args: global_vars: The global_vars dict to update the `self.global_vars` dict from. policy_ids: Optional list of Policy IDs to update. If None, will update all policies on the to-be-updated workers. .. testcode:: :skipif: True worker = ... global_vars = worker.set_global_vars( ... {"timestep": 4242}) rNnum_grad_updates) rpoprr[rrron_global_var_updaterr)rrrglobal_vars_copygradient_updates_per_policyr=s rcrzRolloutWorker.set_global_vars1s.'++--&6&:&: )2' ' # 67>> '   ",!7::T_=Q=Q=S=S  C&.$2I2I#t2T2T.$99(./J/N/Ns/S/ST  011111rbc|j|j|jjD],}|}||-dS)z2Releases all resources used by this RolloutWorker.N)rer stoprcacherrZclose)rr_sesss rcrzRolloutWorker.stop`sq 8  N   ! ! !o+2244  F%%''D   rbc8|jdS)z4Locks this RolloutWorker via its own threading.Lock.N)racquirers rclockzRolloutWorker.locko rbc8|jdS)z6Unlocks this RolloutWorker via its own threading.Lock.N)rreleasers rcunlockzRolloutWorker.unlocksrrburl world_rank world_sizebackendcPtd||||tj|||||jD]8\}}t|ttfstd|||_ 9dS)z/Join a torch process group for distributed SGD.zGJoining process group, url={}, world_rank={}, world_size={}, backend={})r init_methodrankrz.This policy does not support torch distributedN) rmrnrr distributedinit_process_grouprrrr5r6r distributed_world_size)rrrrrr=r_s rcsetup_torch_data_parallelz'RolloutWorker.setup_torch_data_parallelws   ((.sJ G(T(T    ,,:* -    ?0022 7 7KCf{M&BCC  Df-7F ) )  7 7rbc|jS)z3Returns the kwargs dict used to create this worker.)rrs rc creation_argszRolloutWorker.creation_argss $$rbc(tjS)z;Returns the hostname of the process running this evaluator.)platformnoders rcget_hostzRolloutWorker.get_hosts}rbc>tjS)z  1 1 3 3 3   ! ! KK>T_>> ? ? ? KKG43EGG H H H H H " !rbcTddlm}tj|}|jpi|_t |D]\}}td |t|j |r|j }n7|j d}| |j pi|j |_ |j}d|j|<|jr%t!j||jd}||j}||_ ||_|S)aNProcesses the policy dict and creates a new copy with the processed attrs. This processes the observation_space and prepares them for passing to rl module construction. It also merges the policy configs with the algorithm config. During this processing, we will also construct the preprocessors dict. rrZzCreating policy for {}F) copy_frozenN)include_multi_binary)rr[rrrsortedrrmr\rrrurrvr:rr!get_preprocessor_for_spacer) rrr[rnamer merged_conf obs_space preprocessors rcrz-RolloutWorker._get_complete_policy_specs_dictsV JIIIII"mK88!/52!'(;(A(A(C(C!D!D! 6! 6 D+ LL188>> ? ? ?+,o>> G)0 261A1Ae1A1T1T ,,[-?-E2FFF(,'8K $$5I'+D t $) ? ,F%).     + , >I!,K ,5K ) )""rbcX|D]\}}|j|jd<|S)N__multi_rl_module_spec)rrru)rrrrs rc(_update_policy_dict_with_multi_rl_modulez6RolloutWorker._update_policy_dict_with_multi_rl_modules@"-!2!2!4!4 U U D+;?;TK 7 8 8rb)r_rc |jp$t|jj|jj|_t |D]\}}|Ht|t|j |j|j|j |j |j |j }n|}||j|<|pi|d}|r||dS)aAdds the given policy_dict to `self.policy_map`. Args: policy_dict: The MultiAgentPolicyConfigDict to be added to this worker's PolicyMap. policy: If the policy to add already exists, user can provide it here. policy_states: Optional dict from PolicyIDs to PolicyStates to restore the states of the policies being built. )capacitypolicy_states_are_swappableN)rr merged_configr:r;rvrf)rr0rupolicy_map_capacityr!rrrCrGrr:r;rvrfrr)rrr_rrr new_policyrestore_statess rcrzRolloutWorker._build_policy_maps$/ Y[4(, (O. . .  "( (9(9(;(;!#0+2D""#."4&1&C!,!9!%!2    $ $.DOD !+1r66tTBBN 5$$^444- 5 5rbct|D]E\}}|j|}|j|jt ||jt||FdS)z2Updates the filter dict for the given policy_dict.N)rrragent_connectorsaction_connectorsrrur)rrrrr$s rcrz!RolloutWorker._update_filter_dict>s~"( (9(9(;(;!.valid_moduleZs:s++ z22 :%%*4*;*;C*C*C' Z$>33K@@D'#t(+Z8.+..!+.. 5s!A77&B B rc*|Sr)default_sampler_inputioctxs rcrz>RolloutWorker._get_input_creator_from_config..qs!!>rbdatasetcDtjj|Sr)r$rrvr<rs rcrz>RolloutWorker._get_input_creator_from_config..ws  12E""rbchttjj|jjSr)r-r*ruinput_shuffle_buffer_sizer?s rcrz>RolloutWorker._get_input_creator_from_config..|*4;-u55t{7V""rbd4rlr-c$t|Sr)r#)r<env_names rcrz>RolloutWorker._get_input_creator_from_config..sHe!RolloutWorker._get_input_creator_from_config..s&DK.e<<<""rbchttjj|jjSr)r-r(rurArBr?s rcrz>RolloutWorker._get_input_creator_from_config..rCrb) rrurArrrrrSrTsplit)rr8rGs` @rcrz,RolloutWorker._get_input_creator_from_configYs   & dk(, 7 7$ ;% % [ 9 , ,>> > [ 9 , ,?...  *D 1 1   *C 0 0 5L K 6 6  &dk&899 9 t{) ) ){)//44R8H<<<< < \$+, - -    rbctjjtr jjSjjdSjjdkrfdSjjdkrfdSfdS)NctSr)r+r;s rcrz?RolloutWorker._get_output_creator_from_config..s rbr=c:t|jjS)N)compress_columns)r%ruoutput_compress_columnsr?s rcrz?RolloutWorker._get_output_creator_from_config..s  (K"""rblogdirc\t|j|jjjjSN) max_file_sizerN)r)ryruoutput_max_file_sizerOr?s rcrz?RolloutWorker._get_output_creator_from_config..s. "k>!%!D """rbcftjj|jjjjSrR)r)ruoutputrTrOr?s rcrz?RolloutWorker._get_output_creator_from_config..s1 ""k>!%!D """rb)rrurVrrs`rcrz-RolloutWorker._get_output_creator_from_configs dk(, 7 7 ;% % [  '-- - [ 9 , , [ 8 + +  rbcFfdjs fd}|SS)Nc|}|} |||}t|j||S)Nr)rrrrv)renv_ctxrerr} env_wrapperrfrts rc_make_sub_env_localz?RolloutWorker._get_make_sub_env_fn.._make_sub_env_localsx "55<5PPG+g&&C' S'***+c""C *T;3\   Jrbc |}j|j|d|S)NF)rvrrr)rrrrv)rsub_envr[rrs rc_make_sub_env_remotez@RolloutWorker._get_make_sub_env_fn.._make_sub_env_remotesd--l;;99$+ + ? ?%0%=%1$!@!!:rb)r)rr}rrtrZrfr^r[s`````` @rcrz"RolloutWorker._get_make_sub_env_fnsv         ,! '       ( '' &rbr)NN)F)NF)r_N)Y__name__ __module__ __qualname____doc__rIr rrJrrboolrr rOr rrr.r rdataDatasetrr9rrr"rQrrKmethodrOrQrrfrKrirrvrryrRrHrr1rrNrPrrrrHrrrrrUr rrrGrrrrrrLrrrrrrrrrrrr rMrrrrrrrrrrarbrcr]r]ss ==FIM.2%)!&!%@D7;;?N2N2N2 N2x*(=t(CDE N2 *+ N2  N2c]N2N2#N2heUl(;;<=N2'tF|4N2!ch&6!78N2N2N2N2` Xi""" Xi   XiU/UUUUnXi DeE5L&9!9:    SZA"5#)=#>""""2AoA$AAAAF+! +!+! +! !I +! tSy  +!+!+!+!`"O"O" O"O" ~t# $ O"O"O"O"b)F^T(N*B%CCD)F )F)F)F)FVXi T.1    +'A 6+47++++.gz2A56 a:0A . .H .XfEU . . . ..2#' i* .2(,7;.2 .2i*i*i*i*T&\*i* i* $E? i*uoi*34i*{+i*$ *X&(O1Ld1R(SS T i*l+i* i*i*i*i*\0EI ;;;;$HgY-@$AB ; $ *X&(O1Ld1R(SS T ; ;;;;FKO L L#Hgs^X-E$FG L  L L L L5! x (Hh6O+PRV+V"W W 5  5555>Y!!!15  o. X   "! 2)::: .12:H%: ::::*Xfh >ABX aXXXX" fh >AB a    B111111t  4    B 4I4I4I04I 4I  X{%: ;< 4I &.l%; 4I 4I4I4I4Il2#52# #2#2#2#2#h5 #$(?C .5.5.50.5 .5  X{%: ;< .5  .5.5.5.5`6/I6d6666,KKK 888t0+'+'+'+'+'rb)r_r])rimportlib.utilr1rrrr r collectionsrtypesrtypingrrrrr r r r r rrgymnasium.spacesrrrrrray.rllib.connectors.utilrrray.rllib.core.rl_moduler"ray.rllib.core.rl_module.rl_modulerray.rllib.env.base_envrrray.rllib.env.env_contextrray.rllib.env.env_runnerr&ray.rllib.env.external_multi_agent_envrray.rllib.env.multi_agent_envr%ray.rllib.env.wrappers.atari_wrappersrrray.rllib.evaluation.metricsrray.rllib.evaluation.samplerr ray.rllib.modelsr!ray.rllib.models.preprocessorsr"ray.rllib.offliner#r$r%r&r'r(r)r*r+r,r-ray.rllib.policy.policyr.r/ray.rllib.policy.policy_mapr0ray.rllib.policy.sample_batchr1r2r3r4ray.rllib.policy.torch_policyr5 ray.rllib.policy.torch_policy_v2r6ray.rllib.utilsr7ray.rllib.utils.annotationsr8r9ray.rllib.utils.debugr:r;ray.rllib.utils.errorr<r=ray.rllib.utils.filterr>r?ray.rllib.utils.frameworkr@rAray.rllib.utils.from_configrBray.rllib.utils.policyrCray.rllib.utils.sgdrDray.rllib.utils.tf_run_builderrEray.rllib.utils.tf_utilsrFrrGray.rllib.utils.typingrHrIrJrKrLrMrNrOrPrQrRray.tune.registryrSrTray.util.annotationsrUray.util.debugrVrWrX ray.util.iterrYrr[ray.rllib.callbacks.callbacksr\rrtfvrrrr_rmr^__annotations__rdrrrr]rarbrcrs8   ######                          #""""" 00000000877777;;;;;;????????000000......HHHHHH777777IIIIIIII777777444444))))))777777                          76666666111111 655555::::::&&&&&&========LLLLLLLLFFFFFFFF33333333EEEEEEEE333333>>>>>>000000888888                          JIIIIIII******WWWWWWWWWW000000<EEEEEE;;;;;;} R    q  8 $ $ -1)000      ), :=    F x'x'x'x'x'*Ix'x' x'x'x'rb