o 1 i @sddlZddlmZddlZddlZddlmZddlmZm Z m Z m Z ddlm Z m Z ddlZddlZddlmZddlmZmZmZddlmZmZmZdd lmZmZdd lmZm Z dd l!m"Z"dd l#m$Z$m%Z%m&Z&m'Z'm(Z(dd l)m*Z*e\Z+Z,Z-e \Z.Z/dZ0ede1fddZ2eGddde1Z3eGdddZ4ede e'de'fddZ5ede e'ddfddZ6ddde&fddZ7ede'de3fd d!Z8dS)"N)partial)Number)DictIteratorSetUnion)ListOptional)Columns) DeveloperAPIExperimentalAPI PublicAPI)packunpack is_compressed) Deprecateddeprecation_warning) try_import_tftry_import_torch)convert_to_torch_tensor)ModuleIDPolicyID TensorTypeSampleBatchTypeViewRequirementsDict)log_onceZdefault_policy tensor_dictc Cs|tj}|dur1trt|rt|dr1t|dkr1tr+t|r+t| Stt |S| D]Q\}}|tjkr?q5t |t sHJ||tjksW|dsW|drXq5t |ttfrdt|n|g}dd|D}zt|d}|r||WSWq5tyYq5wdS)aAttempt to count timesteps based on dimensions of individual elements. Returns the first successfully counted number of timesteps. We do not attempt to count on INFOS or any state_in_* and state_out_* keys. The number of timesteps we count in cases where we are unable to count is zero. Args: tensor_dict: A SampleBatch or another dict. Returns: count: The inferred number of timesteps >= 0. Nnumpyr state_in_Z state_out_cSs(g|]}t|ttfrt|n|qS) isinstancerlistnparray).0_vrri/home/app/PaddleOCR-VL-test/.venv_paddleocr/lib/python3.10/site-packages/ray/rllib/policy/sample_batch.py Usz+attempt_count_timesteps..)get SampleBatchSEQ_LENStf is_tensorhasattrlentorchintsumitemitemsr strINFOS startswithdicttupletreeflatten Exception)rseq_lenskvZv_list_lenrrr&attempt_count_timesteps"sF        r@c @seZdZdZejZejZejZejZej Z ej Z ej Z ej Z ej Z ejZejZejZejZejZejZdZdZdZdZdZdZdZd Zd Zd Zd Zed dZ ede!fddZ"ede!fddZ#ede!fddZ$e%ddZ&e%ddZ'e(de)fddZ*e(de)fddZ+e,ee-ddd d!d"Z.edxd$d%Z/edyd'e)ddfd(d)Z0ede1e2e3e4ffd*d+Z5ed,e6e3de6e7fd-d.Z8edzd/d0Z9ed{d2e:e3de6dfd3d4Z; 1d|d5e!d6e!ddfd7d8Ze:e!de6dfd?d@Z>e-dAdd d~dBdCZ?d~dDe!dEe)fdFdAZ@e( G & & 1ddHe3dIe)dJe)dKe:eAdLfdMdNZBede!fdOdPZCd{dQdRZDed{dSe:eEddTfdUdVZFed2eAe3errr&rQsR           zSampleBatch.__init__returncC|jS)z2Returns the amount of samples in the sample batch.r[r]rrr&__len__ zSampleBatch.__len__cCt|S)zReturns the same as len(self) (number of steps in this batch). To make this compatible with `MultiAgentBatch.agent_steps()`. r.rcrrr& agent_stepszSampleBatch.agent_stepscCrf)zReturns the same as len(self) (number of steps in this batch). To make this compatible with `MultiAgentBatch.env_steps()`. rgrcrrr& env_stepsrizSampleBatch.env_stepscC d|_dS)NTrRrcrrr&enable_slicing_by_batch_id! z&SampleBatch.enable_slicing_by_batch_idcCrk)NFrlrcrrr&disable_slicing_by_batch_id%rnz'SampleBatch.disable_slicing_by_batch_idcCs&|tjdptj|vo|tjdS)zCReturns True if `self` is either terminated or truncated at idx -1.)r) TERMINATEDS TRUNCATEDSrcrrr&is_terminated_or_truncated)sz&SampleBatch.is_terminated_or_truncatedcCs:t|tjdd otj|vpt|tjdd S)zReturns True if this SampleBatch only contains one trajectory. This is determined by checking all timesteps (except for the last) for being not terminated AND (if applicable) not truncated. Nrp)anyr)rqrrrcrrr&is_single_trajectory0s z SampleBatch.is_single_trajectory/concat_samples() from rllib.policy.sample_batchTnewerrorcCdSNrsamplesrrr&concat_samples<szSampleBatch.concat_samplesothercCs t||gS)agConcatenates `other` to this one and returns a new SampleBatch. Args: other: The other SampleBatch object to concat to this one. Returns: The new SampleBatch, resulting from concating `other` to `self`. .. testcode:: :skipif: True import numpy as np from ray.rllib.policy.sample_batch import SampleBatch b1 = SampleBatch({"a": np.array([1, 2])}) b2 = SampleBatch({"a": np.array([3, 4, 5])}) print(b1.concat(b2)) .. testoutput:: {"a": np.array([1, 2, 3, 4, 5])} )r~)r]rrrr&concatBs zSampleBatch.concatFshallowcs^t|}tfdd|}t||j|j|j|jd}||j |j |_ |j |_ |j |_ |S)zCreates a deep or shallow copy of this SampleBatch and returns it. Args: shallow: Whether the copying should be done shallowly. Returns: A deep or shallow copy of this SampleBatch object. cs t|tjrtj| dS|S)Ncopy)r r"ndarrayr#r>rrr&gs z"SampleBatch.copy..)rBrDrCrF) r7r9 map_structurer)rMrOrNrPset_get_interceptorrXrUrVrT)r]rZcopy_datarrr&r[s"   zSampleBatch.copyc#sTtjddur dndt}tjD]}t|ffdd |VqdS)aReturns an iterator over data rows, i.e. dicts with column values. Note that if `seq_lens` is set in self, we set it to 1 in the rows. Yields: The column values of the row in this iteration. .. testcode:: :skipif: True from ray.rllib.policy.sample_batch import SampleBatch batch = SampleBatch({ "a": [1, 2, 3], "b": [4, 5, 6], "seq_lens": [1, 2] }) for row in batch.rows(): print(row) .. testoutput:: {"a": 1, "b": 4, "seq_lens": 1} {"a": 2, "b": 5, "seq_lens": 1} {"a": 3, "b": 6, "seq_lens": 1} Ncs|djkr ||SSNr)r*)pr>ir]r<rr&rz"SampleBatch.rows..)r(r)r*r7ranger[r9map_structure_with_path)r] self_as_dictrrrr&rowsyszSampleBatch.rowskeyscCs g}|D] }|||q|S)aReturns a list of the batch-data in the specified columns. Args: keys: List of column names fo which to return the data. Returns: The list of data items ordered by the order of column names in `keys`. .. testcode:: :skipif: True from ray.rllib.policy.sample_batch import SampleBatch batch = SampleBatch({"a": [1], "b": [2], "c": [3]}) print(batch.columns(["a", "b"])) .. testoutput:: [[1], [2]] )append)r]routr=rrr&columnsszSampleBatch.columnscs|tjdu}|r|jstd|stj|jn tjt |tjt |}| t j dtfdd|}durKfddD|t j <||i|_|S)aShuffles the rows of this batch in-place. Returns: This very (now shuffled) SampleBatch. Raises: ValueError: If self[SampleBatch.SEQ_LENS] is defined. .. testcode:: :skipif: True from ray.rllib.policy.sample_batch import SampleBatch batch = SampleBatch({"a": [1, 2, 3, 4]}) print(batch.shuffle()) .. testoutput:: {"a": [4, 1, 3, 2]} NzbSampleBatch.shuffle not possible when your data has `seq_lens` defined AND is not zero-padded yet!cs|Sr{rr) permutationrr&rsz%SampleBatch.shuffle..csg|]}|qSrr)r$r)infosrr&r'z'SampleBatch.shuffle..)r(r)r*rO ValueErrorr"randomrr[r.r7rLr r5r9rupdaterW)r]Z has_time_rankrZshuffledr)rrr&shuffles    zSampleBatch.shuffleNkeycs|dus|tjtjfvsJd|dfdd}fdd}tj|tj|i}tjtjg}d}|durM|tjkrG|vrGtd|d ||}n#|D]}|tjksZ|vra||}nqO|durptd |d td d |DjksJdd|djd|S)aSplits by `eps_id` column and returns list of new batches. If `eps_id` is not present, splits by `dones` instead. Args: key: If specified, overwrite default and use key to split. Returns: List of batches, one per distinct episode. Raises: KeyError: If the `eps_id` AND `dones` columns are not present. .. testcode:: :skipif: True from ray.rllib.policy.sample_batch import SampleBatch # "eps_id" is present batch = SampleBatch( {"a": [1, 2, 3], "eps_id": [0, 0, 1]}) print(batch.split_by_episode()) # "eps_id" not present, split by "dones" instead batch = SampleBatch( {"a": [1, 2, 3, 4, 5], "dones": [0, 0, 1, 0, 1]}) print(batch.split_by_episode()) # The last episode is appended even if it does not end with done batch = SampleBatch( {"a": [1, 2, 3, 4, 5], "dones": [0, 0, 1, 0, 0]}) print(batch.split_by_episode()) batch = SampleBatch( {"a": [1, 2, 3, 4, 5], "dones": [0, 0, 0, 0, 0]}) print(batch.split_by_episode()) .. testoutput:: [{"a": [1, 2], "eps_id": [0, 0]}, {"a": [3], "eps_id": [1]}] [{"a": [1, 2, 3], "dones": [0, 0, 1]}, {"a": [4, 5], "dones": [0, 1]}] [{"a": [1, 2, 3], "dones": [0, 0, 1]}, {"a": [4, 5], "dones": [0, 0]}] [{"a": [1, 2, 3, 4, 5], "dones": [0, 0, 0, 0, 0]}] Nz"`SampleBatch.split_by_episode(key=z,)` invalid! Must be [None|'dones'|'eps_id'].csng}tjd}d}tjD]}tj|}||kr*||||}|}q||j|Sr)r)EPS_IDrr[r)slicesZ cur_eps_idoffsetrZ next_eps_idrcrr&slice_by_eps_id'sz5SampleBatch.split_by_episode..slice_by_eps_idcs|g}d}tjD]$}tj|stjvr-tj|r-|||d|d}q |jkr<||d|S)Nrr)rr[r)rqrrr)rrrrcrr&"slice_by_terminateds_or_truncateds6s zHSampleBatch.split_by_episode..slice_by_terminateds_or_truncatedsz does not have key `z`!z does not have keys !css|]}|jVqdSr{rbr$srrr& _sz/SampleBatch.split_by_episode..zCalling split_by_episode on z returns zwhich should in total have z timesteps!)r)rrJrKr1r[)r]rrrZ key_to_methodZkey_resolve_orderrrrcr&split_by_episodes40      zSampleBatch.split_by_episodestartendcs||tjdur*t|tjdkr*dkr$fdd|D}n fdd|D}|dur|dus:Jd}d|}||vr\||||||<|d7}d|}||vsEt|tj||}t|tt|} t || krt || ksJ| t |dd|d<nd} d}d}t |tjD]\} } | | 7} | krd}d|}|dur| }||vr|||| d||<|d7}d|}||vst|tj|| | | g}dkr|d 7<t |} | dkr|d| 8<t |ksJn |dur| kr| }qt|||j |j |j d Sttfd d ||j |j |j d S) a Returns a slice of the row data of this batch (w/o copying). Args: start: Starting index. If < 0, will left-zero-pad. end: Ending index. Returns: A new SampleBatch, which has a slice of this batch's data. Nrc sZi|])\}}|tjkr|ds|ttj f|jdd|jd|dgqS)rrN)shaperHr)r)r*r6r" concatenatezerosrrHr$r=r>rrrr& us  z%SampleBatch.slice..cs<i|]\}}|tjkr|ds|tfdd|qS)rc |Sr{rrrrr&r z.SampleBatch.slice...)r)r*r6r9rrrrr&rs z state_in_{}rrp)r<rErBrFcrr{rvaluerrr&rrz#SampleBatch.slice..rErBrF)r(r)r*r.r3formatr!nextiterr1 enumeraterIrMrPr9r)r]rrZ state_startZ state_endrZ state_idxZ state_keyr<data_lenr[rseq_lendiffrrr&sliceds            zSampleBatch.sliceslice_cs|jpd|jpt|tjt|krt|dkr&dkr&|jdvs(J|tjd}t fdd|}|durft t |tjd}t t |tj}||||tj<||t j<t||j |j|jdS)2Helper method to handle SampleBatch slicing using a slice object. The returned SampleBatch uses the same underlying data object as `self`, so changing the slice will also change `self`. Note that only zero or positive bounds are allowed for both start and stop values. The slice step must be 1 (or None, which is the same). Args: slice_: The python slice object to slice by. Returns: A new SampleBatch, however "linking" into the same data (sliced) as self. r)rNNc |Sr{rrrstoprr&rrz*SampleBatch._batch_slice..r)rrr.r)r*steprLr5r9rr0r1r rIrMrP)r]rrrZinfo_slice_startZinfo_slice_stoprrr& _batch_slices$   zSampleBatch._batch_slicesize num_slicesr=c Cs|dur|durtdd|dusJ|}|durIt|ts Jg}t|}d}|rG||t|}||}||||||8}|}|s*|St|tsPJg}t|}d}|ro||}||||||8}|}|sZ|S)a(Returns SampleBatches, each one representing a k-slice of this one. Will start from timestep 0 and produce slices of size=k. Args: size: The size (in timesteps) of each returned SampleBatch. num_slices: The number of slices to produce. k: Deprecated: Use size or num_slices instead. The size (in timesteps) of each returned SampleBatch. Returns: The list of `num_slices` (new) SampleBatches or n (new) SampleBatches each one of size `size`. Nr=zsize or num_slicesr)rr r0r.r) r]rrr=rleftrlen_rrrr& timeslicess8  zSampleBatch.timesliceszSampleBatch.right_zero_padcCrzr{r)r]rNexclude_statesrrr&zero_pad%szSampleBatch.zero_padrNrcsdtj}|durtdt|fdd}t}t||d__ S)a*Right (adding zeros at end) zero-pads this SampleBatch in-place. This will set the `self.zero_padded` flag to True and `self.max_seq_len` to the given `max_seq_len` value. Args: max_seq_len: The max (total) length to zero pad to. exclude_states: If False, also right-zero-pad all `state_in_x` data. If True, leave `state_in_x` keys as-is. Returns: This very (now right-zero-padded) SampleBatch. Raises: ValueError: If self[SampleBatch.SEQ_LENS] is None (not defined). .. testcode:: :skipif: True from ray.rllib.policy.sample_batch import SampleBatch batch = SampleBatch( {"a": [1, 2, 3], "seq_lens": [1, 2]}) print(batch.right_zero_pad(max_seq_len=4)) batch = SampleBatch({"a": [1, 2, 3], "state_in_0": [1.0, 3.0], "seq_lens": [1, 2]}) print(batch.right_zero_pad(max_seq_len=5)) .. testoutput:: {"a": [1, 0, 0, 0, 2, 3, 0, 0], "seq_lens": [1, 2]} {"a": [1, 0, 0, 0, 0, 2, 3, 0, 0, 0], "state_in_0": [1.0, 3.0], # <- all state-ins remain as-is "seq_lens": [1, 2]} NzNCannot right-zero-pad SampleBatch if no `seq_lens` field present! SampleBatch=c sdur |dds|dtjkrdS|jtks |jjtjur&dg}ntjft |dd|jd}d}}tjD]}||||||||<|7}||7}qA|t |ksdJ|}t |D]\}}|t |dkrz|||<||}qjdS)NTrrrrG) r6r)r*rHobjecttyper"Zstr_rrr.r) pathrZf_padZ f_pad_baseZf_basercurrrrrlengthrNr]rr&_zero_pad_in_placeYs* $  z6SampleBatch.right_zero_pad.._zero_pad_in_placeT) r(r)r*rr.r7r9rrOrN)r]rNrr<rrrrr&right_zero_pad)s '  r/ framework pin_memory use_streamstreamtorch.cuda.StreamrcCsD|dkr tdus J|D]\}}t|||||d||<q|St)9TODO: transfer batch to given device as framework tensor.r/N)rrr)r/r3rNotImplementedError)r]devicerrrrr=r>rrr& to_device}s   zSampleBatch.to_devicecCstddt|DS)aReturns sum over number of bytes of all data buffers. For numpy arrays, we use ``.nbytes``. For all other value types, we use sys.getsizeof(...). Returns: The overall size in bytes of the data buffer (all columns). css,|]}t|tjr |jnt|VqdSr{)r r"rnbytessys getsizeof)r$r>rrr&rs  z)SampleBatch.size_bytes..)r1r9r:rcrrr& size_bytess zSampleBatch.size_bytescCs$z||WSty|YSw)z=Returns one column (by key) from the data or a default value.) __getitem__rK)r]rdefaultrrr&r(s   zSampleBatch.get module_idMultiAgentBatchcCst|pt|i|jS)akReturns the respective MultiAgentBatch Note, if `module_id` is not provided uses `DEFAULT_POLICY`_ID`. Args; module_id: An optional module ID. If `None` the `DEFAULT_POLICY_ID` is used. Returns: The MultiAgentBatch (using DEFAULT_POLICY_ID) corresponding to this SampleBatch. )rDEFAULT_POLICY_IDr[)r]rrrr&as_multi_agentszSampleBatch.as_multi_agentcCst|tr ||S|tjkr|tjS|dkr&tdr#tdddd|jSt ||s5||vr5|j |t ||}|jdurR||jvrM|||j|<|j|}|S)a'Returns one column (by key) from the data or a sliced new batch. Args: key: The key (column name) to return or a slice object for slicing this SampleBatch. Returns: The data under the given key or a sliced version of this batch. rISampleBatch['is_training']SampleBatch.is_trainingFoldrxryN)r r_slicer)rJrqrrrIr-rTaddr7rrXrW)r]rrrrr&rs(        zSampleBatch.__getitem__cCs|tjkr tdt|dst|||dS|dkr+tdr&tdddd||_dS||vr5|j |t|||||j vrH||j |<dSdS) zInserts (overrides) an entire column (by key) in the data buffer. Args: key: The column name to set a value for. item: The data to insert. zCannot set `DONES` anymore in a SampleBatch! Instead, set the new TERMINATEDS and TRUNCATEDS keys. The values under DONES will then be automatically computed using terminated|truncated.rUNrIrrFr) r)rJrKr-r7 __setitem__rrrErUrrW)r]rr2rrr&rs,    zSampleBatch.__setitem__cCsB|jdurt|jtrd|jvr||j|jd<|jdS|jS)NrE)rXr rEboolrWrcrrr&rIs  rtrainingztf1.placeholdercCs||_|jdddS)z1Sets the `is_training` flag for this SampleBatch.rEN)rErWrL)r]rrrr& set_trainingszSampleBatch.set_trainingcCs|j|t||dSr{)rVrr7 __delitem__r]rrrr&rs zSampleBatch.__delitem__new_obsbulkrcs fdd}t|S)aCompresses the data buffers (by column) in place. Args: bulk: Whether to compress across the batch dimension (0) as well. If False will compress n separate list items, where n is the batch size. columns: The columns to compress. Default: Only compress the obs and new_obs columns. Returns: This very (now compressed) SampleBatch. csl|dvrdS}t|D]%\}}|t|dkr/r#t|||<n tdd|D||<||}qdS)NrrcSg|]}t|qSr)rr$orrr&r'6rzDSampleBatch.compress.._compress_in_place..)rr.rr"r#)rrrrrrrr]rr&_compress_in_place-s  z0SampleBatch.compress.._compress_in_placer9r)r]rrrrrr&compresss zSampleBatch.compresscsfdd}t|S)aDecompresses data buffers (per column if not compressed) in place. Args: columns: The columns to decompress. Default: Only decompress the obs and new_obs columns. Returns: This very (now uncompressed) SampleBatch. cs|dvrdS}|ddD]}||}qt|r%t|||d<dSt|dkrAt|drCtdd|D||d<dSdSdS)NrrpcSrr)rrrrr&r'VrzRSampleBatch.decompress_if_needed.._decompress_in_place..)rrr.r"r#)rrrrrr]rr&_decompress_in_placeKs   z>SampleBatch.decompress_if_needed.._decompress_in_placer)r]rrrrr&decompress_if_needed=s z SampleBatch.decompress_if_neededcCs||juri|_||_dS)z.Sets a function to be called on every getitem.N)rXrW)r]fnrrr&r\s  zSampleBatch.set_get_interceptorcCs^t|}|tjdurd|jd|dS|tjd|jdt|dd|dS)Nz SampleBatch(z: )z (seqs=r<z): )r!rr(r)r*r[remover.)r]rrrr&__repr__ds   zSampleBatch.__repr__c s|jr||S|jp d|jpt|t|krt||tjdurt|tjdkr|js_d}t t t |tjD]\}}|j ||fg|||}q>|j t|tj|f|j\}|j\}|||jr~|j|jfdd}|tjd}t||} |durt|ttjfr||tj<|||| tj<t| |j|j|j|jr|jnd|jdS|tjd}tfdd|} |durt|ttjfr||tj<|| tj<t| |j|j|jdS) rrNcs4|dtjkr|dds|S|S)Nrr)r)r*r6)rr) start_padded start_seq_len stop_padded stop_seq_lenrr&map_s   z SampleBatch._slice..map_)rErBrDrCrFcrr{rrrrr&rrz$SampleBatch._slice..r)rRrrrr.r(r)r*r\rmapr0extendrrOrNrLr5r9rr r!r"rrIrMrPr) r]rZsum_rlZstart_unpaddedZ stop_unpaddedr rrr)rrrrr r r&rns\          zSampleBatch._slice)ryc Cstg}g}|tjdurt|tjdkrt|tj|ks#Jdd}d}d}d}d}|t|tjkr|tj|} || 7}||jsG| n|j7}||kr|d} |jsv||||f||7}||kru||} || | 8}|d8}n |||f|}||| fd}|d}|d7}|t|tjks6||fSd} | |j kr|| | |f| |7} | |j ks||fS)NrzFERROR: `slice_size` must be larger than the max. seq-len in the batch!r) r(r)r*r.r"allrOrNrr[) r]Z slice_sizeZ data_slicesZdata_slices_states start_posZcurrent_slize_sizeZactual_slice_idxZ start_idxidxrZend_idxZoverheadrrrr&_get_slice_indicessT    zSampleBatch._get_slice_indiceslastview_requirementsindexc CsNtjtjtjtjtjtji}i}|D]\}}|jdurq|j p"|}|dkr| ||}|j durz||d}t ||} | |j } |tjtjfvrMdnd} |j | } |j| d} | dkrad} tt|||| dg| | g||<qtdd||||<q||||dkr|dnd||<qt|tjdgtjd d S) a Creates single ts SampleBatch at given index from `self`. For usage as input-dict for model (action or value function) calls. Args: view_requirements: A view requirements dict from the model for which to produce the input_dict. index: An integer index value indicating the position in the trajectory for which to generate the compute_actions input dict. Set to "last" to generate the dict at the very end of the trajectory (e.g. for value estimation). Note that "last" is different from -1, as "last" will use the final NEXT_OBS as observation input. Returns: The (single-timestep) input dict for ModelV2 calls. FrNrprrcSs |ddS)Nrprrrrr&r9rz8SampleBatch.get_single_step_input_dict..rG)r<)r)OBSNEXT_OBS PREV_ACTIONSACTIONS PREV_REWARDSREWARDSr3Zused_for_compute_actionsdata_colr(Z shift_fromr.Zbatch_repeat_valueZshift_tor"r#rr9rrY)r]rrZ last_mappingsZ input_dictZview_colZview_reqrrZtraj_lenZmissing_at_endZ obs_shiftZfrom_Zto_rrr&get_single_step_input_dictsF            z&SampleBatch.get_single_step_input_dict)rr)r`r))F)r`r)r{)NN)NNN)Tr/FFN)r`N)r)W__name__ __module__ __qualname____doc__r rrrrqrrr5r*TZACTION_DIST_INPUTSZ ACTION_PROBZ ACTION_LOGPZVF_PREDSZVALUES_BOOTSTRAPPEDrrZ ACTION_DISTrrZENV_ID AGENT_INDEXZ UNROLL_IDZ OBS_EMBEDSZ RETURNS_TO_GOZATTENTION_MASKSrJZCUR_OBSr rQr0rdrhrjr rmror rrsru staticmethodrr~rrrrr4rrrrtrrr rrrrrrrrrr(rrrrpropertyrIrr frozensetrrrrrrrrrrrrr&r)ds  m     % 5r [/ 6 T   % &         W 2 r)c @seZdZdZedeeefdefddZ edefddZ edefd d Z edefd d Z ed ede dfddZeedeeefdedeedffddZeeedddde dddfddZed6ddZe    d7dededeed fd!d"Zedefd#d$Zeded%d&gfd'ed(eeddfd)d*Zeed%d&gfd(eeddfd+d,Zed6d-d.Z d/edefd0d1Z!d2d3Z"d4d5Z#dS)8raA batch of experiences from multiple agents in the environment. Attributes: policy_batches (Dict[PolicyID, SampleBatch]): Dict mapping policy IDs to SampleBatches of experiences. count: The number of env steps in this batch. policy_batchesrjcCs,|D] }t|ts Jq||_||_dS)auInitialize a MultiAgentBatch instance. Args: policy_batches: Dict mapping policy IDs to SampleBatches of experiences. env_steps: The number of environment steps in the environment this batch contains. This will be less than the number of transitions this batch contains across all policies in total. N)valuesr r)r(r[)r]r(rjr>rrr&rQOs  zMultiAgentBatch.__init__r`cCra)zThe number of env steps (there are >= 1 agent steps per env step). Returns: The number of environment steps contained in this batch. rbrcrrr&rjbszMultiAgentBatch.env_stepscCra)zSame as `self.env_steps()`.rbrcrrr&rdkrezMultiAgentBatch.__len__cCs"d}|jD]}||j7}q|S)zThe number of agent steps (there are >= 1 agent steps per env step). Returns: The number of agent steps total in this batch. r)r(r)r[)r]ctbatchrrr&rhps zMultiAgentBatch.agent_stepsr=c s ddlm}g}|jD]\}}|D]}||tj|tj|tj ||fqq | gt |dfdd}t |ddD])\}} | D]\}}}}}|jd i|qOd7|krr|dksrJqIdkrz|tdksJS) aReturns k-step batches holding data for each agent at those steps. For examples, suppose we have agent1 observations [a1t1, a1t2, a1t3], for agent2, [a2t1, a2t3], and for agent3, [a3t3] only. Calling timeslices(1) would return three MultiAgentBatches containing [a1t1, a2t1], [a1t2], and [a1t3, a2t3, a3t3]. Calling timeslices(2) would return two MultiAgentBatches containing [a1t1, a1t2, a2t1], and [a1t3, a2t3, a3t3]. This method is used to implement "lockstep" replay mode. Note that this method does not guarantee each batch contains only data from a single unroll. Batches might contain data from multiple different envs. r)SampleBatchBuildercs>dksJtddD}d|dS)NrcSi|] \}}||qSr)Zbuild_and_resetrrrr&rrzDMultiAgentBatch.timeslices..finish_slice..)rr3clearr)r+Z cur_sliceZcur_slice_sizeZfinished_slicesrr& finish_slices z0MultiAgentBatch.timeslices..finish_slicecSs |ddS)Nr)xrrr&rrz,MultiAgentBatch.timeslices..rNr)Z)ray.rllib.evaluation.sample_batch_builderr,r(r3rrr)rr#r$sort collections defaultdict itertoolsgroupbyZ add_valuesr.) r]r=r,ZstepsZ policy_idr+rowr0_grouprr/r&r|s<      zMultiAgentBatch.timeslicescCs(t|dkrt|vr|tSt||dS)aReturns SampleBatch or MultiAgentBatch, depending on given policies. If policy_batches is empty (i.e. {}) it returns an empty MultiAgentBatch. Args: policy_batches: Mapping from policy ids to SampleBatch. env_steps: Number of env steps in the batch. Returns: The single default policy's SampleBatch or a MultiAgentBatch (more than one policy). rr(rj)r.rrr;rrr&wrap_as_neededs zMultiAgentBatch.wrap_as_neededrvTrwr}cCrfr{)concat_samples_into_ma_batchr|rrr&r~szMultiAgentBatch.concat_samplescCstdd|jD|jS)z{Deep-copies self into a new MultiAgentBatch. Returns: The copy of self with deep-copied data. cSr-rrrrrr&rrz(MultiAgentBatch.copy..)rr(r3r[rcrrr&rszMultiAgentBatch.copyr/FNrrrrcCsJ|dkr#tdus J|jD]\}}|j|||||d|j|<q|St)rr/N)rrrr)r/r(r3rr)r]rrrrrpidZ policy_batchrrr&rs  zMultiAgentBatch.to_devicecCstdd|jDS)ze Returns: The overall size in bytes of all policy batches (all columns). css|]}|VqdSr{)r)r$brrr&rsz-MultiAgentBatch.size_bytes..)r1r(r)rcrrr&rszMultiAgentBatch.size_bytesrArrrcCs"|jD] }|j||dqdS)a8Compresses each policy batch (per column) in place. Args: bulk: Whether to compress across the batch dimension (0) as well. If False will compress n separate list items, where n is the batch size. columns: Set of column names to compress. )rrN)r(r)r)r]rrr+rrr&rs zMultiAgentBatch.compresscCs|jD]}||q|S)zDecompresses each policy batch (per column), if already compressed. Args: columns: Set of column names to decompress. Returns: Self. )r(r)r)r]rr+rrr&rs z$MultiAgentBatch.decompress_if_neededcCs|S)zSimply returns `self` (already a MultiAgentBatch). Returns: This very instance of MultiAgentBatch. rrcrrr&r!szMultiAgentBatch.as_multi_agentrcCs |j|S)z0Returns the SampleBatch for the given policy id.)r(rrrr&r*rnzMultiAgentBatch.__getitem__cCdt|j|jSNz!MultiAgentBatch({}, env_steps={})rr4r(r[rcrrr&__str__. zMultiAgentBatch.__str__cCr@rArBrcrrr&r3rDzMultiAgentBatch.__repr__)r`rr)$rr r!r"r rrr)r0rQrjrdrhrrr%rr<rr~rr rr rrr r'rr4rrrrrCrrrrr&rEs| @             rr}r`c s,tdd|Dr t|Sg}ddg}g}d}}}|D]y}|jdkr%q|dur2|j}|j}|j}|j|ks<|j|kr@td|jdusI|durR|j|krRtd|r]|j|kr]td|durgt||j}|t j durw| |t j |j dur|d|j7<|d |j |j7<| |qt|dkrt Si}|dD]2t jkrtfd d |Dd |i|<qfd d |D} tt|d} tj| g| R|<q|gkrtrt|drt|}n|gkrtrt|drt|}t ||||||d |dpddS)aXConcatenates a list of SampleBatches or MultiAgentBatches. If all items in the list are or SampleBatch typ4, the output will be a SampleBatch type. Otherwise, the output will be a MultiAgentBatch type. If input is a mixture of SampleBatch and MultiAgentBatch types, it will treat SampleBatch objects as MultiAgentBatch types with 'default_policy' key and concatenate it with th rest of MultiAgentBatch objects. Empty samples are simply ignored. Args: samples: List of SampleBatches or MultiAgentBatches to be concatenated. Returns: A new (concatenated) SampleBatch or MultiAgentBatch. .. testcode:: :skipif: True import numpy as np from ray.rllib.policy.sample_batch import SampleBatch b1 = SampleBatch({"a": np.array([1, 2]), "b": np.array([10, 11])}) b2 = SampleBatch({"a": np.array([3]), "b": np.array([12])}) print(concat_samples([b1, b2])) c1 = MultiAgentBatch({'default_policy': { "a": np.array([1, 2]), "b": np.array([10, 11]) }}, env_steps=2) c2 = SampleBatch({"a": np.array([3]), "b": np.array([12])}) print(concat_samples([b1, b2])) .. testoutput:: {"a": np.array([1, 2, 3]), "b": np.array([10, 11, 12])} MultiAgentBatch = {'default_policy': {"a": np.array([1, 2, 3]), "b": np.array([10, 11, 12])}} css|]}t|tVqdSr{)r rrrrr&rgsz!concat_samples..rgNzNAll SampleBatches' `zero_padded` and `time_major` settings must be consistent!z?Samples must consistently either provide or omit `max_seq_len`!zPFor `zero_padded` SampleBatches, the values of `max_seq_len` must be consistent!rcg|]}|qSrrrr=rr&r'rz"concat_samples..rMcrErr)r$crFrr&r'rrMg?)r<rBrDrCrF)rtr=r[rOrNrMrrZr(r)r*r rPrr.rr5_concat_valuesrr9rr/r,ZTensorr+Zconvert_to_tensor) r}Zconcatd_seq_lensZconcatd_num_grad_updatesZconcated_samplesrOrNrMrZ concatd_dataZvalues_to_concatZ_concat_values_w_timerrFr&r~9s~.             r~cCstt}d}|D]9}t|trt|dkrq |}nt|ts+td t |j |j D] \}}|||q0||7}q i}| D] \}}t|||<qIt||S)aConcatenates a list of SampleBatchTypes to a single MultiAgentBatch type. This function, as opposed to concat_samples() forces the output to always be MultiAgentBatch which is more generic than SampleBatch. Args: samples: List of SampleBatches or MultiAgentBatches to be concatenated. Returns: A new (concatenated) MultiAgentBatch. .. testcode:: :skipif: True import numpy as np from ray.rllib.policy.sample_batch import SampleBatch b1 = MultiAgentBatch({'default_policy': { "a": np.array([1, 2]), "b": np.array([10, 11]) }}, env_steps=2) b2 = SampleBatch({"a": np.array([3]), "b": np.array([12])}) print(concat_samples([b1, b2])) .. testoutput:: {'default_policy': {"a": np.array([1, 2, 3]), "b": np.array([10, 11, 12])}} rz[`concat_samples_into_ma_batch` can only concat SampleBatch|MultiAgentBatch objects, not {}!)r4r5r!r r)r.rrrrrrr(r3rrjr~)r}r(rjrrr+rZbatchesrrr&r=s& "      r=rHcGstrt|drtj||rddSddSt|dtjr+tj||r'ddSddStrAt|drAtj||r=ddSddSt|dt rVg}|D]}| |qL|St dt |dd|d)zConcatenates a list of values. Args: values: The values to concatenate. time_major: Whether to concatenate along the first axis (time_major=False) or the second axis (time_major=True). rr)dim)Zaxisz$Unsupported type for concatenation: z first element: ) r/r,catr r"rrr+rr!r rr)rMr)Zconcatenated_listZsublistrrr&rIs  rIr+cCsBt|tr|j}t|dkrt|vr|jt}|Std|S)a:Converts a MultiAgentBatch to a SampleBatch if neccessary. Args: batch: The SampleBatchType to convert. Returns: batch: the converted SampleBatch Raises: ValueError if the MultiAgentBatch has more than one policy_id or if the policy_id is not `DEFAULT_POLICY_ID` raRLlib tried to convert a multi agent-batch with data from more than one policy to a single-agent batch. This is not supported and may be due to a number of issues. Here are two possible ones:1) Off-Policy Estimation is not implemented for multi-agent batches. You can set `off_policy_estimation_methods: {}` to resolve this.2) Loading multi-agent data for offline training is not implemented.Load single-agent data instead to resolve this.)r rr(rr.rr)r+Z policy_keysrrr& convert_ma_batch_to_sample_batchs     rL)9r4 functoolsrr6rnumbersrtypingrrrrrr rr"r9Zray.rllib.core.columnsr Zray.rllib.utils.annotationsr r r Zray.rllib.utils.compressionrrrZray._common.deprecationrrZray.rllib.utils.frameworkrrZray.rllib.utils.torch_utilsrZray.rllib.utils.typingrrrrrZray.utilrZtf1r+Ztfvr/r9rr7r@r)rr~r=rIrLrrrr&sX       Ait>