o $izl@snddlZddlZddlZddlmZmZmZmZmZm Z m Z m Z ddl Z ddlZddlZddlmZddlmZmZddlmZddlmZddlmZmZddlmZdd lm Z dd l!m"Z"m#Z#dd l$m%Z%erpdd l&m'Z'ej(ej(ej)ej)ej*ej*ej+ej+ej,ej,ej-ej-ej.ej.ej/ej/ej0ej0ej1ej1ej2ej2d d ddddiZ3e4e5Z6e%ddGdddZ7dS)N) TYPE_CHECKINGAnyDictListOptionalSetTupleUnion)Columns) MultiRLModuleMultiRLModuleSpec)SingleAgentEpisode) flatten_dict)OverrideToImplementCustomLogic5OverrideToImplementCustomLogic_CallToSuperRecommended)unpack_if_needed) ReplayBuffer) EpisodeTypeModuleID) PublicAPI)AlgorithmConfig agent_indexdones unroll_idalpha) stabilityc@seZdZdZedddddddeeejejfdee dee e e fd e e e ff d d Zed e e ejfd e e ejffddZed efddZed e e e ffddZdeed eefddZd%d efddZeeedddddfded e e ee ejffde e e fdedeee deedejd ejd e e e fd e e ee!ffd!d"Z"eeeddfded e e ee ejffde e e fdedeee d e e ee!ff d#d$Z#dS)&OfflinePreLearnera?Class that coordinates data transformation from dataset to learner. This class is an essential part of the new `Offline RL API` of `RLlib`. It is a callable class that is run in `ray.data.Dataset.map_batches` when iterating over batches for training. It's basic function is to convert data in batch from rows to episodes (`SingleAGentEpisode`s for now) and to then run the learner connector pipeline to convert further to trainable batches. These batches are used directly in the `Learner`'s `update` method. The main reason to run these transformations inside of `map_batches` is for better performance. Batches can be pre-fetched in `ray.data` and therefore batch trransformation can be run highly parallelized to the `Learner''s `update`. This class can be overridden to implement custom logic for transforming batches and make them 'Learner'-ready. When deriving from this class the `__call__` method and `_map_to_episodes` can be overridden to induce custom logic for the complete transformation pipeline (`__call__`) or for converting to episodes only ('_map_to_episodes`). Custom `OfflinePreLearner` classes can be passed into `AlgorithmConfig.offline`'s `prelearner_class`. The `OfflineData` class will then use the custom class in its data pipeline. N)spaces module_spec module_stateconfigrrrrkwargscKs||_|jj|_|jj|_||_|j||pd\|_|_|jj|j|jd|_ |jj |_ |j |_ d|_|jsB|jsB|jrZ|jjpH|j}|j|jjB}|di||_dSdS)N)NN)input_observation_spaceinput_action_spacer)r input_read_episodesinput_read_sample_batchesbuild_module set_stateobservation_space action_spacebuild_learner_connector_learner_connectorpolicies_to_train_policies_to_trainis_multi_agent_is_multi_agentiter_since_last_module_update is_statefulprelearner_buffer_classdefault_prelearner_buffer_class default_prelearner_buffer_kwargsprelearner_buffer_kwargsepisode_buffer)selfr rrrr!r4r7r$r$a/home/ubuntu/veenaModal/venv/lib/python3.10/site-packages/ray/rllib/offline/offline_prelearner.py__init__Qs<      zOfflinePreLearner.__init__batchreturnc sv|jrDddlddlfdd|dD}||}|j||jj|jj|j r5|jj ddnd|j ddp>dd d d }nZ|j rtj|j|d t|jjB|jjd d }||}|j||jj|jj|j rx|jj ddnd|j ddpdd d d }n|j|j|t|jjBd |jj|j|jdd }|j|j i|idd}|D] }|||s||=qt|S)aqPrepares plain data batches for training with `Learner`'s. Args: batch: A dictionary of numpy arrays containing either column data with `self.config.input_read_schema`, `EpisodeType` data, or `BatchType` data. Returns: A `MultiAgentBatch` that can be passed to `Learner.update` methods. rNcs"g|] }tj|jdqS)) object_hook)r from_stateunpackbdecode).0statemnpmsgpackr$r: s z.OfflinePreLearner.__call__..item max_seq_lenn_stepT) num_itemsbatch_length_TrJsample_episodesto_numpy)rOschemainput_compress_columnsepisodesF)rPrOrQr*r+) rl_moduler<rR shared_datametrics)r%rF msgpack_numpy_validate_episodesr8addsampler train_batch_size_per_learnerr(r3 model_configgetr&r_map_sample_batch_to_episoder1SCHEMAinput_read_schemarQ_map_to_episodesr*r+r-_should_module_be_updatedr)r9r<rR module_idr$rDr:__call__s|         & zOfflinePreLearner.__call__cCsddlm}|S)zSets the default replay buffer.r)EpisodeReplayBuffer)4ray.rllib.utils.replay_buffers.episode_replay_bufferrd)r9rdr$r$r:r5 s z1OfflinePreLearner.default_prelearner_buffer_classcCs|jjd|jjdS)zSets the default arguments for the replay buffer. Note, the `capacity` might vary with the size of the episodes or sample batches in the offline dataset. )capacity batch_size_B)r rZ)r9r$r$r:r6s z2OfflinePreLearner.default_prelearner_buffer_kwargsrRcCsbtdd|Ds tdt}t}|D]}|j|vr.|j|jjvr.||j||q|S)aValidate episodes sampled from the dataset. Note, our episode buffers cannot handle either duplicates nor non-ordered fragmentations, i.e. fragments from episodes that do not arrive in timestep order. Args: episodes: A list of `SingleAgentEpisode` instances sampled from a dataset. Returns: A set of `SingleAgentEpisode` instances. Raises: ValueError: If not all episodes are `done`. css|]}|jVqdSN)is_done)rBepsr$r$r: 6sz7OfflinePreLearner._validate_episodes..zWhen sampling from episodes (`input_read_episodes=True`) all recorded episodes must be done (i.e. either `terminated=True`) or `truncated=True`).)all ValueErrorsetid_r8episode_id_to_indexrX)r9rRunique_episode_idscleaned_episodesrkr$r$r:rW"s   z$OfflinePreLearner._validate_episodescCs.|jsdSt|js|t|jvS|||S)z:Checks which modules in a MultiRLModule should be updated.T)r/callablero)r9rbmulti_agent_batchr$r$r:raIs   z+OfflinePreLearner._should_module_be_updatedFr0rPrOrQignore_final_observationr*r+c spgg} t|tjD]\} |r4tj|vr#|tjnd|vr1|dnd} nd} |r9qtjvrBt| n| } |rRtddt| } ntjvrbt|tj n|tj } t tj |vr~t |tj nt j| | | gitj|vr|tjnigtjvrt|tjn|tjg|tjg|tj|vrȈtjndtj|vr|tjndfdd|Dd d }|r|| |qd | iS) z!Maps a batch of data to episodes.rNcSsd|S)Nrr$)xr$r$r:sz4OfflinePreLearner._map_to_episodes..rFcsLi|]"\}}|vr$|vr$|dvr||vrt|n|gqS)rrtypevaluesrrBkvirQrPr$r: s   z6OfflinePreLearner._map_to_episodes..r rpagent_id observationsinfosactionsrewards terminated truncatedextra_model_outputslen_lookback_bufferrR) enumerater OBSAGENT_IDrtree map_structurecopydeepcopyNEXT_OBSr EPS_IDstruuiduuid4hexINFOSACTIONSREWARDS TERMINATEDS TRUNCATEDSitemsrOappend)r0r<rPrOrQrvr*r+r!rRobsr unpacked_obsunpacked_next_obsepisoder$rr:r`Ssx         8 z"OfflinePreLearner._map_to_episodesc spgg}t|tjD]l\|r)d|vr&|ddnd}nd}|r.qttrFtfddtjdDn"ttj r^tjvrYt n n t dt dtj |vrtjvrt|tj dn |tj dndtj|vrtj|vr|tjd}|tjd}nLtj|vrڈtj|vr|tjd}d }n0tj|vrtj|vr|tjd}d }nd |vr|d d}d }nd }d }ttj|vr t|tjdntj|tj|vr7|tjnigttjvrNt|tjn|tj|tj||fd d |Ddd } |rw| || qd|iS)z6Maps an old stack `SampleBatch` to new stack episodes.rrNcsg|]}|dfqS).r$)rBr)rr$r:rGszBOfflinePreLearner._map_sample_batch_to_episode..zUnknown observation type: z. When mapping from old recorded `SampleBatches` batched observations should be either of type `np.array` or - if the column is compressed - of `str` type.FdoneTcsJi|]!\}}|vr#|vr#|dvr||vrt|n|qSryr{r}rr$r:rZs  zBOfflinePreLearner._map_sample_batch_to_episode..rrR)rr r isinstancerrrangeshapenpndarraytolist TypeErrorrzrrrrr rrrrrlenrrrrO) r0r<rPrOrQrRrrrrr$)rrQrrPr:r]s           . z.OfflinePreLearner._map_sample_batch_to_episoderi)$__name__ __module__ __qualname____doc__rrrgymSpacer rrrrr;rrrrcpropertyrr5r6rr rrWboolra staticmethodr^r listrr`r]r$r$r$r:r5s  :(   '     x  r)8rloggingrtypingrrrrrrrr gymnasiumrnumpyrrray.rllib.core.columnsr (ray.rllib.core.rl_module.multi_rl_moduler r "ray.rllib.env.single_agent_episoder ray.rllib.utilsrray.rllib.utils.annotationsrrray.rllib.utils.compressionr,ray.rllib.utils.replay_buffers.replay_bufferrray.rllib.utils.typingrrray.util.annotationsr%ray.rllib.algorithms.algorithm_configrrr MODULE_IDrrrrrrrTr^ getLoggerrloggerrr$r$r$r:sH(