&`iBddlZddlZddlZddlmZmZmZmZddlZ ddl m Z ddl m Z ddlmZmZddlmZmZmZmZmZerddlmZeGdd ejZeGd d ZeGd d e ZdS)N) TYPE_CHECKINGCallableOptionalTuple)deprecation_warning)BaseEnv) OldAPIStackoverride) EnvActionType EnvInfoDict EnvObsTypeEnvType MultiEnvDict) Preprocessorc *eZdZdZ d!dejdejdefdZdZ d"d e e d e d e fd Z d e d e d efdZd e d e ded dfdZ d!d e dede ed dfdZd e d e d dfdZd e d dfdZ d#de eegefdede dede d df d ZdS)$ ExternalEnvahAn environment that interfaces with external agents. Unlike simulator envs, control is inverted: The environment queries the policy to obtain actions and in return logs observations and rewards for training. This is in contrast to gym.Env, where the algorithm drives the simulation through env.step() calls. You can use ExternalEnv as the backend for policy serving (by serving HTTP requests in the run loop), for ingesting offline logs data (by reading offline transitions in the run loop), or other custom use cases not easily expressed through gym.Env. ExternalEnv supports both on-policy actions (through self.get_action()), and off-policy actions (through self.log_action()). This env is thread-safe, but individual episodes must be executed serially. .. testcode:: :skipif: True from ray.tune import register_env from ray.rllib.algorithms.dqn import DQN YourExternalEnv = ... register_env("my_env", lambda config: YourExternalEnv(config)) algo = DQN(env="my_env") while True: print(algo.train()) N action_spaceobservation_spacemax_concurrentctj|d|_||_||_i|_t|_tj |_ |tdddSdS)zInitializes an ExternalEnv instance. Args: action_space: Action space of the env. observation_space: Observation space of the env. TNa The `max_concurrent` argument has been deprecated. Please configurethe number of episodes using the `rollout_fragment_length` and`batch_mode` arguments. Please raise an issue on the Ray Github if these arguments do not support your expected use case for ExternalEnv)error) threadingThread__init__daemonrr _episodesset _finished Condition_results_avail_conditionr)selfrrrs n/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/rllib/env/external_env.pyrzExternalEnv.__init__6s !!$''' (!2(1(;(=(=%  % X        & %ct)aOverride this to implement the run loop. Your loop should continuously: 1. Call self.start_episode(episode_id) 2. Call self.[get|log]_action(episode_id, obs, [action]?) 3. Call self.log_returns(episode_id, reward) 4. Call self.end_episode(episode_id, obs) 5. Wait if nothing to do. Multiple episodes may be started at the same time. )NotImplementedErrorr!s r"runzExternalEnv.runTs "!r#T episode_idtraining_enabledreturnc"|tjj}||jvr"t d|||jvr"t d|t||j||j|<|S)abRecord the start of an episode. Args: episode_id: Unique string id for the episode or None for it to be auto-assigned and returned. training_enabled: Whether to use experiences for this episode to improve the policy. Returns: Unique string id for the episode. N!Episode {} has already completed.zEpisode {} is already started) uuiduuid4hexr ValueErrorformatr_ExternalEnvEpisoder )r!r(r)s r" start_episodezExternalEnv.start_episodebs  )J  ' '@GG SSTT T  ' '<CCJOOPP P%8 57G& & z"r# observationcV||}||S)aRecord an observation and get the on-policy action. Args: episode_id: Episode id returned from start_episode(). observation: Current environment observation. Returns: Action from the env action space. )_getwait_for_actionr!r(r4episodes r" get_actionzExternalEnv.get_actions)))J''&&{333r#actionc\||}|||dS)zRecord an observation and (off-policy) action taken. Args: episode_id: Episode id returned from start_episode(). observation: Current environment observation. action: Action for the observation. N)r6 log_action)r!r(r4r;r9s r"r=zExternalEnv.log_actions1))J'';/////r#rewardinfocj||}|xj|z c_|r |pi|_dSdS)aRecords returns (rewards and infos) from the environment. The reward will be attributed to the previous action taken by the episode. Rewards accumulate until the next action. If no reward is logged before the next action, a reward of 0.0 is assumed. Args: episode_id: Episode id returned from start_episode(). reward: Reward from the environment. info: Optional info dict. N)r6 cur_rewardcur_info)r!r(r>r?r9s r" log_returnszExternalEnv.log_returnssM))J''f$  *#zrG    * *r#c||}|j|j||dS)zRecords the end of an episode. Args: episode_id: Episode id returned from start_episode(). observation: Current environment observation. N)r6raddr(doner8s r" end_episodezExternalEnv.end_episodesF))J'' 7-... [!!!!!r#r2c||jvr"td|||jvr"td||j|S)z2Get a started episode by its ID or raise an error.r,zEpisode {} not found.)rr0r1r)r!r(s r"r6zExternalEnv._getse  ' '@GG SSTT T T^ + +4;;JGGHH H~j))r#Frmake_envnum_envs remote_envsremote_env_batch_wait_msrestart_failed_sub_environmentsrcN|dkrtdt|}|S)aConverts an RLlib MultiAgentEnv into a BaseEnv object. The resulting BaseEnv is always vectorized (contains n sub-environments) to support batched forward passes, where n may also be 1. BaseEnv also supports async execution via the `poll` and `send_actions` methods and thus supports external simulators. Args: make_env: A callable taking an int as input (which indicates the number of individual sub-environments within the final vectorized BaseEnv) and returning one individual sub-environment. num_envs: The number of sub-environments to create in the resulting (vectorized) BaseEnv. The already existing `env` will be one of the `num_envs`. remote_envs: Whether each sub-env should be a @ray.remote actor. You can set this behavior in your config via the `remote_worker_envs=True` option. remote_env_batch_wait_ms: The wait time (in ms) to poll remote sub-environments for, if applicable. Only used if `remote_envs` is True. Returns: The resulting BaseEnv object. rIzExternal(MultiAgent)Env does not currently support num_envs > 1. One way of solving this would be to treat your Env as a MultiAgentEnv hosting only one type of agent but with several copies.)r0ExternalEnvWrapper)r!rJrKrLrMrNenvs r" to_base_envzExternalEnv.to_base_envs9B q==9  !&& r#N)NT)NrIFrF)__name__ __module__ __qualname____doc__gymSpaceintrr'rstrboolr3r r r:r=floatr rCrGr6rrrRr#r"rrs B# i9 < " " "JN"3-BF < 4S 4z 4m 4 4 4 4 0 0,6 0@M 0  0 0 0 0MQ***',*4<[4I* ***** "c " "t " " " " *s *'< * * * *8<!()05 **8SE7N34** * #& * *. * ******r#rc VeZdZdZ ddedejdedefdZdZ d Z d Z d Z d Z d S)r2z&Tracked state for each active episode.Fr(results_avail_conditionr) multiagentcV||_||_||_||_t j|_t j|_|r0d|_d|_ i|_ ddi|_ ddi|_ i|_ dSd|_d|_d|_d|_d|_i|_dS)N__all__F)r(r`r)raqueueQueue data_queue action_queuenew_observation_dictnew_action_dictcur_reward_dictcur_terminated_dictcur_truncated_dict cur_info_dictnew_observation new_actionrAcur_terminated cur_truncatedrB)r!r(r`r)ras r"rz_ExternalEnvEpisode.__init__s%'>$ 0$+--!KMM  (,D %#'D #%D (15'9D $'0%&8D #!#D   #'D "DO!DO"'D !&D DMMMr#cj|jrdS|jSrS)rgempty get_nowaitr&s r"get_dataz_ExternalEnvEpisode.get_datas1 ? " " 4))+++r#c|jr||_||_n||_||_||jdddS)NTgN@timeout)rarirjrorp_sendrhget)r!r4r;s r"r=z_ExternalEnvEpisode.log_actions] ? %(3D %#)D #.D $DO  dD11111r#c|jr||_n||_||jddS)NTgr@rx)rarirorzrhr{r!r4s r"r7z#_ExternalEnvEpisode.wait_for_action$sH ? /(3D % %#.D   $$T5$999r#c|jr||_ddi|_ddi|_n||_d|_d|_|dS)NrcTF)rarirlrmrorqrrrzr}s r"rFz_ExternalEnvEpisode.done,s\ ? '(3D %(14'8D $(1%&8D # ##.D "&D !&D  r#c>|jri|js|jD]}d|j|d<|j|j|j|j|jd}|j |j|d<d|_d|_i|_nY|j|j |j |j |j d}|j |j |d<d|_d|_ d|_ |js d|dd<|j5|j||jddddS#1swxYwYdS)NFr))obsr> terminated truncatedr?off_policy_actionrdr?)rar)rnrirkrlrmrjrorArqrrrBrpr`rg put_nowaitnotify)r!agent_iditems r"rzz_ExternalEnvEpisode._send:s ? 9( M $ 2MMHGLD&x01CDD0."6!4* D#/,0,@()(,D %#'D #%D +/"1!/ D*,0O()#'D "DO!DO( 938V /0  ) 2 2 O & &t , , ,  ( / / 1 1 1 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2s4DDDN)F)rTrUrVrWr[rrr\rrvr=r7rFrzr^r#r"r2r2s00! "+!4   8,,, 222:::   #2#2#2#2#2r#r2cXeZdZdZ d ddZeedeeeeeeffd Z eed eddfd Z deeeeeeeffd Z e eede jjfd Ze eede jfdZdS)rPz+Internal adapter of ExternalEnv to BaseEnv.N external_envr preprocessorrcddlm}||_||_t t |||_|j|_|r |j |_ n |j |_ | dS)Nr)ExternalMultiAgentEnv) &ray.rllib.env.external_multi_agent_envrrprep issubclasstyperar _action_spacer_observation_spacestart)r!rrrs r"rzExternalEnvWrapper.__init__ds QPPPPP(  $T,%7%79NOO)6  E&2&DD # #&2&DD #r#r*c|jj5|}t|ddkrs|jj|}|jst dt|ddksdddn #1swxYwY|S)NrzServing thread has stopped.)rr _polllenwaitis_alive Exception)r!resultss r"pollzExternalEnvWrapper.pollss  7 C CjjllGgaj//Q&&!:??AAA**,,(1133C#$ABBB gaj//Q&& C C C C C C C C C C C C C C CsB!B::B>B> action_dictc@ddlm}|jrF|D]/\}}|jj|j|0dS|D]5\}}|jj|j||6dS)Nr)_DUMMY_AGENT_ID)ray.rllib.env.base_envrraitemsrrrhput)r!rrenv_idactionsr;s r" send_actionszExternalEnvWrapper.send_actionss:::::: ? #.#4#4#6#6 N N!+F3@DDWMMMM N N#."3"3"5"5  !+F3@DD?+  r#cddlm}iiiiif\}}}}}i}|jjD]\}|} |jr |jdn|j } |jr |j dn|j } | s| r |jj=| rq|j r$|j | d|<n | d|<| d|<| d|<| d|<| d|<d | vr | d |<|jro|D]R\} | D]8fd } | |d | |d | |d | |i9S||||||fS||||||d||d||||fS) Nr)with_dummy_agent_idrcrr>rrr?rc6|vr ||<dSdSrSr^)dzero_valreids r"fixz%ExternalEnvWrapper._poll..fixs.#1S611/7AcF8,,,21r#rdF)rrrrcopyrrvrarlrqrmrrr transformkeys)r!rall_obs all_rewardsall_terminatedsall_truncateds all_infosoff_policy_actionsr9datarqrreid_dictrrrs @@r"rzExternalEnvWrapper._polls ?>>>>>      L Ho~y  -7<<>>DDFF H HLC##%%D?,+I66+ ?+*955*   5 5%/4 H9/#'9#6#6tE{#C#CGCLL#';GCL#'> C '+L'9$&*;&7s#!%f #&$...23F.G&s+ ? ") ' ' X (  ' 'H888888C S)))C///C...C 2&&&& '"  $#G,,##K00##OY??##NI>>##I..##$677  r#c|jSrS)rr&s r"rz$ExternalEnvWrapper.observation_spaces &&r#c|jSrS)rr&s r"rzExternalEnvWrapper.action_spaces !!r#rS)rrrr)rTrUrVrWrr rrrrrrpropertyrXspacesDictrrYrr^r#r"rPrP`s}55KO ) 9G    Xg  |\<|S T    Xg      K    KKKKZ Xg'3:?'''X' Xg"ci"""X"""r#rP)rerr-typingrrrr gymnasiumrXray._common.deprecationrrrray.rllib.utils.annotationsr r ray.rllib.utils.typingr r r rrray.rllib.models.preprocessorsrrrr2rPr^r#r"rs  ;;;;;;;;;;;;777777******========<;;;;;; ZZZZZ)"ZZ Zz g2g2g2g2g2g2g2 g2T A"A"A"A"A"A"A" A"A"A"r#