[docs]classCallbackEnvironment:"""This callback should be called at the end of each episode. When multiprocessing is used, no global variables are available (as an own python instance is created). :param plot_interval: How many episodes to pass between each render call. """def__init__(self,plot_interval:int)->None:self.plot_interval=plot_interval
[docs]def__call__(self,env:BaseEnv)->None:""" This callback should be called at the end of each episode. When multiprocessing is used, no global variables are available (as an own python instance is created). :param env: Instance of the environment where the callback was triggered. """log.info("Environment callback triggered (env_id = {}, n_episodes = {}, run_name = {}.".format(env.env_id,env.n_episodes,env.run_name))# render first episodeifenv.n_episodes==1:env.render()# render progress over episodes (for each environment individually)elifenv.n_episodes%self.plot_interval==0:env.render()ifhasattr(env,"render_episodes"):env.render_episodes()
[docs]defmerge_callbacks(*args:MaybeCallback)->CallbackList:"""Take a number of arguments and merge them into a CallbackList object if they instantiate BaseCallback. :param args: List of callbacks. :return: CallbackList object which merges all callbacks. """cb_list=[]forcbinargs:ifisinstance(cb,BaseCallback):cb_list.append(cb)elifisinstance(cb,list):merge_callbacks(*cb)elifcbisNone:continueelse:raiseValueError(f"Invalid callback type: {type(cb)}.")returnCallbackList(cb_list)