o `۷i8@sddlZddlmZddlmZddlZddlmZddlm Z ddl m Z ddl m Z ddlmZdd lmZmZmZdd lmZdd lmZdd lmZdd lmZeeZeddGddde ZdS)N)Path)List)AlgorithmConfig)Columns) EnvRunner)SingleAgentEnvRunner)SingleAgentEpisode)OverrideToImplementCustomLogic5OverrideToImplementCustomLogic_CallToSuperRecommendedoverride)pack_if_needed) EpisodeType) PublicAPI)log_oncealpha) stabilitycseZdZdZeeedeffdd Zde fddZ eee dddd d d d e d e d e de de deef fddZeee dddZe deeddfddZZS)OfflineSingleAgentEnvRunnerz7The environment runner to record the single agent case.configc stjdd|i|d|_tjj}|jjj |j d|j_|j j |_ |j j |_ |j j|_|j j|_d|_|j j|_|jrD||_n|jsY|j jrP|jdksU|jdkrYtdd|jd|_|jdkrwddl}|jdi|j|_n8|jd krdd lm }|j!di|j|_n"|jd krddl"}|j#di|j|_n|jdurtd |jd |j $d|ji|j j%|_%|j j&|_&|j j'|_'|j'rd|_(nd|_(|j j)|_*d|_+g|_,dS)NrF)cpurzrTo set up the output path, the environment `env` must be provided when creating the `OfflineSingleAgentEnvRunner`.zrun-gcss3)fsabszUnknown filesystem: z9. Filesystems can be 'gcs' for GCS, 's3' for S3, or 'abs' filesystemT)-super__init__episodes_to_numpyraydata DataContext get_currentexecution_optionsresource_limitscopynum_cpus_per_env_runnerroutput_write_methodoutput_write_method_kwargsoutput_filesystemroutput_filesystem_kwargsfilesystem_kwargsfilesystem_objectoutput output_pathenv_get_subdir_path subdir_pathcreate_env_on_local_worker worker_index ValueErrorzfill worker_pathgcsfs GCSFileSystempyarrowr S3FileSystemadlfsAzureBlobFileSystemupdateoutput_write_episodesoutput_compress_columnsoutput_max_rows_per_filewrite_data_this_iteroutput_write_remaining_datawrite_remaining_data_sample_counter_samples)selfrkwargs data_contextr7rr; __class__rZ/home/ubuntu/vllm_env/lib/python3.10/site-packages/ray/rllib/offline/offline_env_runner.pyrsd                    z$OfflineSingleAgentEnvRunner.__init__returncCst|jtr |jS|jjdkr|jjjdjjj S|jjdkr0|jj ddjj S|jjdkr>|jjjj St d|jj)zmReturns the subdir path for storing data. Returns: The subdir path as a string. syncrasync unwrappedvector_entry_pointz"Unknown `gym_env_vectorize_mode`: ) isinstancer/strlowerrgym_env_vectorize_moderOenvsrJ__name__get_attrr4)rFrrrKr0s"      z,OfflineSingleAgentEnvRunner._get_subdir_pathNF num_timesteps num_episodesexplorerandom_actions force_resetrYrZr[r\r]c stj|||||d}|jd7_|jr@ddlddltdr&tdt dd|Ds1J|j fd d |Dn| ||j rSt|j |j krSd |_|jr|j rd |_t|j |j kr|j d|j }|j |j d|_ tj|}t|j |j ksdntj|j }z0t|j|j|jd |jd} t||j| fi|jtd| dWnty} z t| WYd} ~ nd} ~ ww|jj dt|j d|S)z2Samples from environments and writes data to disk.rXrNmsgpackzPacking episodes with `msgpack` and encode array with `msgpack_numpy` for serialization. This is needed for recording episodes.css|]}|jduVqdS)FN)is_numpy.0epsrrrK sz5OfflineSingleAgentEnvRunner.sample..cs g|] }j|jdqS))default)packb get_stateencoderamnpr_rrK s z6OfflineSingleAgentEnvRunner.sample..TF-rzWrote samples to storage at .recording_buffer_size)keyvalue)!rsamplerDr>r_ msgpack_numpyrloggerinfoallrEextend_map_episodes_to_datar@lenrArr from_itemsrr.joinpathr1r6r5getattrr'as_posixr( Exceptionerrormetrics log_value) rFrYrZr[r\r]samplessamples_to_write samples_dspatherIrirKrqsj    z"OfflineSingleAgentEnvRunner.samplec Cs|jr\|jr\tj|j}|jd7_z0t|j|j |j d|j d}t ||j |fi|jtd|dWnty[}z t|WYd}~nd}~wwtdt|jdS)zWrites the reamining samples to disk Note, if the user defined `max_rows_per_file` the number of rows for the remaining samples could be less than the defined maximum row number by the user. r^rlrz"Wrote final samples to storage at zY. Note Note, final samples could be smaller in size than `max_rows_per_file`, if defined.NzExperience buffer length: )rErCrr ryrDrr.rzr1r6r5r{r'r|r(rsrtr}r~debugrx)rFrrrrrrKstops*   z OfflineSingleAgentEnvRunner.stoprcs"|D]ttD]tjjtjjtjjtj tj j vr't  n tj tj j vr:t ntjtjtj j vrUt  dn dtjtdkrgdnjtjtdkrudnji fddjD}j|q qdS)zConverts list of episodes to list of single dict experiences. Note, this method also appends all sampled experiences to the buffer. Args: samples: List of episodes to be converted. r^Fcs4i|]}||jvrt|n|qSr)r?r get_extra_model_outputs)rbkirqrFrrK Ms  zEOfflineSingleAgentEnvRunner._map_episodes_to_data..N)rangerxrEPS_IDid_AGENT_IDagent_id MODULE_ID module_idOBSr?r get_observationsACTIONS get_actionsREWARDS get_rewardsNEXT_OBS TERMINATEDS is_terminated TRUNCATEDS is_truncatedextra_model_outputskeysrEappend)rFr sample_datarrrKrw(sD       z1OfflineSingleAgentEnvRunner._map_episodes_to_data)rLN)rV __module__ __qualname____doc__r rr rrrRr0r intboolrrrqrrr rw __classcell__rrrIrKrs>e_% r) loggingpathlibrtypingrr%ray.rllib.algorithms.algorithm_configrray.rllib.core.columnsrray.rllib.env.env_runnerr%ray.rllib.env.single_agent_env_runnerr"ray.rllib.env.single_agent_episoderray.rllib.utils.annotationsr r r ray.rllib.utils.compressionr ray.rllib.utils.typingr ray.util.annotationsrray.util.debugrLogger__file__rsrrrrrKs"