o ciXo@szddlZddlZddlZddlZddlZddlZddlm Z m Z m Z m Z m Z mZmZmZddlmZddlmZmZddlmZddlmZddlmZmZddlmZdd lm Z dd l!m"Z"dd l#m$Z$m%Z%dd l&m'Z'ervdd l(m)Z)ej*ej*ej+ej+ej,ej,ej-ej-ej.ej.ej/ej/ej0ej0ej1ej1ej2ej2ej3ej3ej4ej4ddddddiZ5e6e7Z8e'ddGdddZ9dS)N)AnyDictListOptionalUnionSetTuple TYPE_CHECKING)Columns)MultiRLModuleSpec MultiRLModule)SingleAgentEpisode) flatten_dict)OverrideToImplementCustomLogic5OverrideToImplementCustomLogic_CallToSuperRecommended)unpack_if_needed) ReplayBuffer)from_jsonable_if_needed) 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_stateconfigrrrr kwargscKs||_|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!rrr r"r5r8r%r%X/home/ubuntu/.local/lib/python3.10/site-packages/ray/rllib/offline/offline_prelearner.py__init__Rs<      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_TrKsample_episodesto_numpy)rPschemainput_compress_columnsepisodesF)rQrPrRr+r,) rl_moduler=rS shared_datametrics)r&rG msgpack_numpy_validate_episodesr9addsampler!train_batch_size_per_learnerr)r4 model_configgetr'r_map_sample_batch_to_episoder2SCHEMAinput_read_schemarR_map_to_episodesr+r,r._should_module_be_updatedr)r:r=rS module_idr%rEr;__call__s|         & zOfflinePreLearner.__call__cCsddlm}|S)zSets the default replay buffer.r)EpisodeReplayBuffer)4ray.rllib.utils.replay_buffers.episode_replay_bufferre)r:rer%r%r;r6 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!r[)r:r%r%r;r7s z2OfflinePreLearner.default_prelearner_buffer_kwargsrScs8tdd|Ds tdtfdd|D}|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_donerCepsr%r%r; 7sz7OfflinePreLearner._validate_episodes..zWhen sampling from episodes (`input_read_episodes=True`) all recorded episodes must be done (i.e. either `terminated=True`) or `truncated=True`).cs8h|]}|jvr|js|jjjvr|qSr%)id_rYr9episode_id_to_indexkeysrlr:unique_episode_idsr%r; @s  z7OfflinePreLearner._validate_episodes..)all ValueErrorset)r:rSr%rrr;rX#s z$OfflinePreLearner._validate_episodescCs.|jsdSt|js|t|jvS|||S)z:Checks which modules in a MultiRLModule should be updated.T)r0callablerw)r:rcmulti_agent_batchr%r%r;rbIs   z+OfflinePreLearner._should_module_be_updatedFr1rQrPrRignore_final_observationr+r,c s6pg|r |r t} ndd} g} t|tjD]\} |r?tj|vr.|tjnd|vr<|dnd} nd} |rDqtjvrP| t| |n| | |} |rctddt | }ntjvrv| t|tj |n | |tj |}t tj |vrt |tj ntj| | |gitj|vr|tjnigtjvr| t|tj|n | |tj|g|tjg|tj|vrtjndtj|vr|tjndfd d |Dd d }|r|| |qd | iS)z!Maps a batch of data to episodes.cSs|Srjr%)rZspacer%r%r;convertlsz3OfflinePreLearner._map_to_episodes..convertrNcSsd|S)Nrr%)xr%r%r;sz4OfflinePreLearner._map_to_episodes..rFcsLi|]"\}}|vr$|vr$|dvr||vrt|n|gqS)rrtypevaluesrrCkvirRrQr%r; s   z6OfflinePreLearner._map_to_episodes..r roagent_id observationsinfosactionsrewards terminated truncatedextra_model_outputslen_lookback_bufferrS)r enumerater OBSAGENT_IDrtree map_structurecopydeepcopyNEXT_OBSr EPS_IDstruuiduuid4hexINFOSACTIONSREWARDS TERMINATEDS TRUNCATEDSitemsrPappend)r1r=rQrPrRrzr+r,r"r|rSobsr unpacked_obsunpacked_next_obsepisoder%rr;raSs          ; 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%)rCr)rr%r;rHszBOfflinePreLearner._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|qSrrrrr%r;rms  zBOfflinePreLearner._map_sample_batch_to_episode..rrS)rr r isinstancerrrangeshapenpndarraytolist TypeErrorrrrrrr rrrrrlenrrrrP) r1r=rQrPrRrSrrrrr%)rrRrrQr;r^s           . z.OfflinePreLearner._map_sample_batch_to_episoderj)$__name__ __module__ __qualname____doc__rrrgymSpacer rrrrr<rrrrdpropertyrr6r7rr rrXboolrb staticmethodr_rlistrrar^r%r%r%r;r6s  :(   &        r):r gymnasiumrloggingnumpyrrrtypingrrrrrrrr ray.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_bufferr"ray.rllib.utils.spaces.space_utilsrray.rllib.utils.typingrrray.util.annotationsr%ray.rllib.algorithms.algorithm_configrrr MODULE_IDrrrrrrrTr_ getLoggerrloggerrr%r%r%r;sJ(