o $ie:@sddlZddlmZddlmZmZmZmZmZm Z m Z ddl Z ddl mZddlmZddlmZmZddlmZddlmZdd lmZdd lmZmZmZmZmZerbdd l m!Z!dd l"m#Z#eGd ddZ$dS)N) defaultdict) TYPE_CHECKINGAnyCallableDictListOptionalTuple)_DUMMY_AGENT_ID)AgentCollector)_PolicyCollector_PolicyCollectorGroup) PolicyMap) SampleBatch) OldAPIStack)AgentIDEnvID EnvInfoDictPolicyID TensorType) RLlibCallback) RolloutWorkerc @seZdZdZddddededeeddgefde dd e d f d d Z e d fdede defddZ deefddZdedefddZd7ddZdddededeeefdeddf ddZded eeefddfd!d"Z d8d#ed$e d%e ddfd&d'Zd9dede fd(d)Zdede fd*d+Zdede fd,d-Zdede fd.d/Zded0efd1d2Ze fdede efd3d4Z e!d5d6Z"dS): EpisodeV2z=Tracks the current state of a (possibly multi-agent) episode.N)worker callbacksenv_idpoliciespolicy_mapping_fnrrrrcCsttd|_||_d|_d|_d|_d|_d|_ i|_ i|_ i|_ i|_ ||_||_||_||_i|_i|_d|_i|_tt|_tt|_i|_i|_i|_i|_dS)aInitializes an Episode instance. Args: env_id: The environment's ID in which this episode runs. policies: The PolicyMap object (mapping PolicyIDs to Policy objects) to use for determining, which policy is used for which agent. policy_mapping_fn: The mapping function mapping AgentIDs to PolicyIDs. worker: The RolloutWorker instance, in which this episode runs. gNgmCgrN)random randrangeint episode_idr total_rewardactive_env_stepstotal_env_stepsactive_agent_stepstotal_agent_stepscustom_metrics user_data hist_datamediarr policy_mapr_agent_to_policy_agent_collectors_next_agent_index_agent_to_indexrfloat agent_rewardslist_agent_reward_history _has_init_obs_last_terminateds_last_truncateds _last_infos)selfrrrrrr:\/home/ubuntu/veenaModal/venv/lib/python3.10/site-packages/ray/rllib/evaluation/episode_v2.py__init__s4   zEpisodeV2.__init__Fagent_idrefreshreturncCsT||jvs|r|j|||jd}|j|<n|j|}||jvr(td|d|S)aReturns and stores the policy ID for the specified agent. If the agent is new, the policy mapping fn will be called to bind the agent to a policy for the duration of the entire episode (even if the policy_mapping_fn is changed in the meantime!). Args: agent_id: The agent ID to lookup the policy ID for. Returns: The policy ID for the specified agent. )rz.policy_mapping_fn returned invalid policy id 'z'!)r-rrr,KeyError)r9r=r> policy_idr:r:r; policy_fores   zEpisodeV2.policy_forcCst|jS)zReturns list of agent IDs that have appeared in this episode. Returns: The list of all agent IDs that have appeared so far in this episode. )r3r0keysr9r:r:r; get_agentsszEpisodeV2.get_agentscCs.||jvr|j|j|<|jd7_|j|S)zGet the index of an agent among its environment. A new index will be created if an agent is seen for the first time. Args: agent_id: ID of an agent. Returns: The index of this agent. )r0r/r9r=r:r:r; agent_indexs  zEpisodeV2.agent_indexcCs |jd7_|jd7_dS)z(Advance the episode forward by one step.rFN)r$r%rDr:r:r;stepszEpisodeV2.stepr)tinit_obs init_infosrJcCs|j||}||jvsJt|j|jdd|jdd||dd|j|<|j|j |j | ||j |||dd|j |<dS) zAdd initial env obs at the start of a new episode Args: agent_id: Agent ID. init_obs: Initial observations. init_infos: Initial infos dicts. t: timestamp. model max_seq_len_disable_action_flatteningF)rNdisable_action_flatteningis_policy_recurrent intial_states_enable_new_api_stack)r"rHrrKrLrJTN)r,rBr.r view_requirementsconfigget is_recurrentget_initial_state add_init_obsr"rHrr5)r9r=rKrLrJpolicyr:r:r;rYs*   zEpisodeV2.add_init_obsvaluescCs||jvsJ|jd7_|jd7_|tkr||d<|j|||tj}|j|7_|j|| |f|7<|j | |tj |vrT|tj |j |<tj|vra|tj|j|<tj|vrq|||tjdSdS)zAdd action, reward, info, and next_obs as a new step. Args: agent_id: Agent ID. values: Dict of action, reward, info, and next_obs. rFr=N)r.r&r'r add_action_reward_next_obsrREWARDSr#r2rBr4append TERMINATEDSr6 TRUNCATEDSr7INFOS set_last_info)r9r=r[rewardr:r:r;add_action_reward_done_next_obss"     z)EpisodeV2.add_action_reward_done_next_obs batch_builderis_done check_donesc Csi}|jD]!\}}|jdkrq||}|j|}||j} ||| f||<q|D]\}\}}} |rK|rK| sKtd |j |||d|j |i ddsWq-| rgtt| tjdkrltd| t|dkrz|} | |=ni} | } t|dd d ur|j|| || d || | |} dd lm} |jj| ||||j| |d ||jvrt||j|<|j|| |jq-|j|j 7_|j!|j"7_!d|_ d|_"d S) aBuild and return currently collected training samples by policies. Clear agent collector states if this episode is done. Args: batch_builder: _PolicyCollectorGroup for saving the collected per-agent sample batches. is_done: If this episode is done (terminated or truncated). check_dones: Whether to make sure per-agent trajectories are actually done. rzkEpisode {} terminated for all agents, but we still don't have a last observation for agent {} (policy {}). zkPlease ensure that you include the last observations of all live agents when setting done[__all__] to True.training_enabledTrFzPBatches sent to postprocessing must only contain steps from a single trajectory. explorationN)get_global_worker)repisoder=rArpostprocessed_batchoriginal_batches)#r.items agent_stepsrBr,build_for_trainingrTis_terminated_or_truncated ValueErrorformatr"r8rVis_single_trajectorylennpuniquerEPS_IDcopygetattrripostprocess_trajectory get_sessionset_get_interceptor#ray.rllib.evaluation.rollout_workerrjron_postprocess_trajectorypolicy_collectorsr $add_postprocessed_batch_for_trainingr& env_stepsr$) r9rerfrg pre_batchesr= collectorpidrZ pre_batch other_batches post_batchrjr:r:r;postprocess_episodesn            zEpisodeV2.postprocess_episodecCs.|dur||jvo |j|Stt|jS)zReturns whether this episode has initial obs for an agent. If agent_id is None, return whether we have received any initial obs, in other words, whether this episode is completely fresh. N)r5anyr3r[rGr:r:r; has_init_obs]szEpisodeV2.has_init_obscCs||p ||SN) is_terminated is_truncatedrGr:r:r;rfhszEpisodeV2.is_donecC|j|dSNF)r6rVrGr:r:r;rkzEpisodeV2.is_terminatedcCrr)r7rVrGr:r:r;rnrzEpisodeV2.is_truncatedinfocCs||j|<dSr)r8)r9r=rr:r:r;rbqrzEpisodeV2.set_last_infocCs |j|Sr)r8rVrGr:r:r; last_info_forts zEpisodeV2.last_info_forcCs|jSr)r%rDr:r:r;lengthyszEpisodeV2.length)r?N)FFr)#__name__ __module__ __qualname____doc__rrrrrrr<r boolrBrrEr!rHrIrrstrrYrdr rrrfrrrbrrpropertyrr:r:r:r;rs K #    *  , d  r)%r collectionsrtypingrrrrrrr numpyrvray.rllib.env.base_envr /ray.rllib.evaluation.collectors.agent_collectorr 5ray.rllib.evaluation.collectors.simple_list_collectorr r ray.rllib.policy.policy_maprray.rllib.policy.sample_batchrray.rllib.utils.annotationsrray.rllib.utils.typingrrrrrray.rllib.callbacks.callbacksrr~rrr:r:r:r;s  $