

Module easyagents.callbacks.duration

View Source

import math

import easyagents.core as core

class Fast(core.AgentCallback):

    """Train for small number of episodes / steps in order to do a dry run of the algorithms or callbacks."""

    def __init__(self, num_iterations=None):

        self._num_iterations = num_iterations

        self._num_episodes_per_iteration = 3

        self._max_steps_per_episode = 50

    def on_play_begin(self, agent_context: core.AgentContext):

        agent_context.play.max_steps_per_episode = self._max_steps_per_episode

        if isinstance(agent_context, core.EpisodesTrainContext):

            agent_context.num_episodes_per_iteration = self._num_episodes_per_iteration

    def on_train_begin(self, agent_context: core.AgentContext):

        tc = agent_context.train

        if self._num_iterations is None:

            self._num_iterations = 10

        if isinstance(tc, core.EpisodesTrainContext):

            ec: core.EpisodesTrainContext = tc

            ec.num_episodes_per_iteration = self._num_episodes_per_iteration

            ec.num_epochs_per_iteration = 1

        if isinstance(tc, core.StepsTrainContext):

            dc: core.StepsTrainContext = tc

            if self._num_iterations is None:

                self._num_iterations = 5 * self._num_episodes_per_iteration * self._max_steps_per_episode

            dc.num_steps_buffer_preload = math.ceil(self._num_iterations / 10)

        tc.num_iterations = self._num_iterations

        tc.num_iterations_between_eval = math.ceil(tc.num_iterations/3)

        tc.num_episodes_per_eval = self._num_episodes_per_iteration

        tc.max_steps_per_episode = self._max_steps_per_episode

class _SingleEpisode(Fast):

    """Train / Play only for 1 episode (no evaluation in training, max. 10 steps)."""

    def __init__(self):

        super().__init__(num_iterations=1)

        self._num_episodes_per_iteration = 1

        self._max_steps_per_episode = 10

    def on_train_begin(self, agent_context: core.AgentContext):

        super().on_train_begin(agent_context)

        tc = agent_context.train

        if isinstance(tc, core.StepsTrainContext):

            tc.num_iterations = self._max_steps_per_episode

        tc.num_iterations_between_eval = 0

        tc.num_episodes_per_eval = 0

class _SingleIteration(Fast):

    """Train / play for a single iteration with 3 episodes, and evaluation of 2 episodes"""

    def __init__(self):

        super().__init__(num_iterations=1)

Classes

Fast

class Fast(
    num_iterations=None
)

Train for small number of episodes / steps in order to do a dry run of the algorithms or callbacks.

View Source

class Fast(core.AgentCallback):

    """Train for small number of episodes / steps in order to do a dry run of the algorithms or callbacks."""

    def __init__(self, num_iterations=None):

        self._num_iterations = num_iterations

        self._num_episodes_per_iteration = 3

        self._max_steps_per_episode = 50

    def on_play_begin(self, agent_context: core.AgentContext):

        agent_context.play.max_steps_per_episode = self._max_steps_per_episode

        if isinstance(agent_context, core.EpisodesTrainContext):

            agent_context.num_episodes_per_iteration = self._num_episodes_per_iteration

    def on_train_begin(self, agent_context: core.AgentContext):

        tc = agent_context.train

        if self._num_iterations is None:

            self._num_iterations = 10

        if isinstance(tc, core.EpisodesTrainContext):

            ec: core.EpisodesTrainContext = tc

            ec.num_episodes_per_iteration = self._num_episodes_per_iteration

            ec.num_epochs_per_iteration = 1

        if isinstance(tc, core.StepsTrainContext):

            dc: core.StepsTrainContext = tc

            if self._num_iterations is None:

                self._num_iterations = 5 * self._num_episodes_per_iteration * self._max_steps_per_episode

            dc.num_steps_buffer_preload = math.ceil(self._num_iterations / 10)

        tc.num_iterations = self._num_iterations

        tc.num_iterations_between_eval = math.ceil(tc.num_iterations/3)

        tc.num_episodes_per_eval = self._num_episodes_per_iteration

        tc.max_steps_per_episode = self._max_steps_per_episode

---

Ancestors (in MRO)

• easyagents.core.AgentCallback
• abc.ABC

Descendants

• easyagents.callbacks.duration._SingleEpisode
• easyagents.callbacks.duration._SingleIteration

Methods

on_api_log

def on_api_log(
    self,
    agent_context: easyagents.core.AgentContext,
    api_target: str,
    log_msg: str
)

Logs a call to the api of the agents implementation library / framework.

View Source

    def on_api_log(self, agent_context: AgentContext, api_target: str, log_msg: str):

        """Logs a call to the api of the agents implementation library / framework."""

        pass

on_gym_init_begin

def on_gym_init_begin(
    self,
    agent_context: easyagents.core.AgentContext
)

called when the monitored environment begins the instantiation of a new gym environment.

Args: agent_context: api_context passed to calling agent

View Source

    def on_gym_init_begin(self, agent_context: AgentContext):

        """called when the monitored environment begins the instantiation of a new gym environment.

            Args:

                agent_context: api_context passed to calling agent

        """

on_gym_init_end

def on_gym_init_end(
    self,
    agent_context: easyagents.core.AgentContext
)

called when the monitored environment completed the instantiation of a new gym environment.

Args: agent_context: api_context passed to calling agent

View Source

    def on_gym_init_end(self, agent_context: AgentContext):

        """called when the monitored environment completed the instantiation of a new gym environment.

        Args:

            agent_context: api_context passed to calling agent

        """

        pass

on_gym_reset_begin

def on_gym_reset_begin(
    self,
    agent_context: easyagents.core.AgentContext,
    **kwargs
)

Before a call to gym.reset

Args: agent_context: api_context passed to calling agent kwargs: the args to be passed to the underlying environment

View Source

    def on_gym_reset_begin(self, agent_context: AgentContext, **kwargs):

        """Before a call to gym.reset

            Args:

                agent_context: api_context passed to calling agent

                kwargs: the args to be passed to the underlying environment

        """

on_gym_reset_end

def on_gym_reset_end(
    self,
    agent_context: easyagents.core.AgentContext,
    reset_result: Tuple,
    **kwargs
)

After a call to gym.reset was completed

Args: agent_context: api_context passed to calling agent reset_result: object returned by gym.reset kwargs: args passed to gym.reset

View Source

    def on_gym_reset_end(self, agent_context: AgentContext, reset_result: Tuple, **kwargs):

        """After a call to gym.reset was completed

        Args:

            agent_context: api_context passed to calling agent

            reset_result: object returned by gym.reset

            kwargs: args passed to gym.reset

        """

        pass

on_gym_step_begin

def on_gym_step_begin(
    self,
    agent_context: easyagents.core.AgentContext,
    action
)

Before a call to gym.step

Args: agent_context: api_context passed to calling agent action: the action to be passed to the underlying environment

View Source

    def on_gym_step_begin(self, agent_context: AgentContext, action):

        """Before a call to gym.step

        Args:

            agent_context: api_context passed to calling agent

            action: the action to be passed to the underlying environment

        """

        pass

on_gym_step_end

def on_gym_step_end(
    self,
    agent_context: easyagents.core.AgentContext,
    action,
    step_result: Tuple
)

After a call to gym.step was completed

Args: agent_context: api_context passed to calling agent action: the action to be passed to the underlying environment step_result: (observation,reward,done,info) tuple returned by gym.step

View Source

    def on_gym_step_end(self, agent_context: AgentContext, action, step_result: Tuple):

        """After a call to gym.step was completed

        Args:

            agent_context: api_context passed to calling agent

            action: the action to be passed to the underlying environment

            step_result: (observation,reward,done,info) tuple returned by gym.step

        """

        pass

on_log

def on_log(
    self,
    agent_context: easyagents.core.AgentContext,
    log_msg: str
)

Logs a general message

View Source

    def on_log(self, agent_context: AgentContext, log_msg: str):

        """Logs a general message"""

        pass

on_play_begin

def on_play_begin(
    self,
    agent_context: easyagents.core.AgentContext
)

Called once at the entry of an agent.play() call (during play or eval, but not during train).

View Source

    def on_play_begin(self, agent_context: core.AgentContext):

        agent_context.play.max_steps_per_episode = self._max_steps_per_episode

        if isinstance(agent_context, core.EpisodesTrainContext):

            agent_context.num_episodes_per_iteration = self._num_episodes_per_iteration

on_play_end

def on_play_end(
    self,
    agent_context: easyagents.core.AgentContext
)

Called once before exiting an agent.play() call (during play or eval, but not during train)

View Source

    def on_play_end(self, agent_context: AgentContext):

        """Called once before exiting an agent.play() call (during play or eval, but not during train)"""

on_play_episode_begin

def on_play_episode_begin(
    self,
    agent_context: easyagents.core.AgentContext
)

Called once at the start of new episode to be played (during play or eval, but not during train).

View Source

    def on_play_episode_begin(self, agent_context: AgentContext):

        """Called once at the start of new episode to be played (during play or eval, but not during train). """

on_play_episode_end

def on_play_episode_end(
    self,
    agent_context: easyagents.core.AgentContext
)

Called once after an episode is done or stopped (during play or eval, but not during train).

View Source

    def on_play_episode_end(self, agent_context: AgentContext):

        """Called once after an episode is done or stopped (during play or eval, but not during train)."""

on_play_step_begin

def on_play_step_begin(
    self,
    agent_context: easyagents.core.AgentContext,
    action
)

Called once before a new step is taken in the current episode (during play or eval, but not during train).

Args: agent_context: the context describing the agents current configuration action: the action to be passed to the upcoming gym_env.step call

View Source

    def on_play_step_begin(self, agent_context: AgentContext, action):

        """Called once before a new step is taken in the current episode (during play or eval, but not during train).

            Args:

                 agent_context: the context describing the agents current configuration

                 action: the action to be passed to the upcoming gym_env.step call

        """

on_play_step_end

def on_play_step_end(
    self,
    agent_context: easyagents.core.AgentContext,
    action,
    step_result: Tuple
)

Called once after a step is completed in the current episode (during play or eval, but not during train).

View Source

    def on_play_step_end(self, agent_context: AgentContext, action, step_result: Tuple):

        """Called once after a step is completed in the current episode (during play or eval, but not during train)."""

on_train_begin

def on_train_begin(
    self,
    agent_context: easyagents.core.AgentContext
)

Called once at the entry of an agent.train() call.

View Source

    def on_train_begin(self, agent_context: core.AgentContext):

        tc = agent_context.train

        if self._num_iterations is None:

            self._num_iterations = 10

        if isinstance(tc, core.EpisodesTrainContext):

            ec: core.EpisodesTrainContext = tc

            ec.num_episodes_per_iteration = self._num_episodes_per_iteration

            ec.num_epochs_per_iteration = 1

        if isinstance(tc, core.StepsTrainContext):

            dc: core.StepsTrainContext = tc

            if self._num_iterations is None:

                self._num_iterations = 5 * self._num_episodes_per_iteration * self._max_steps_per_episode

            dc.num_steps_buffer_preload = math.ceil(self._num_iterations / 10)

        tc.num_iterations = self._num_iterations

        tc.num_iterations_between_eval = math.ceil(tc.num_iterations/3)

        tc.num_episodes_per_eval = self._num_episodes_per_iteration

        tc.max_steps_per_episode = self._max_steps_per_episode

on_train_end

def on_train_end(
    self,
    agent_context: easyagents.core.AgentContext
)

Called once before exiting an agent.train() call

View Source

    def on_train_end(self, agent_context: AgentContext):

        """Called once before exiting an agent.train() call"""

on_train_iteration_begin

def on_train_iteration_begin(
    self,
    agent_context: easyagents.core.AgentContext
)

Called once at the start of a new iteration.

View Source

    def on_train_iteration_begin(self, agent_context: AgentContext):

        """Called once at the start of a new iteration. """

on_train_iteration_end

def on_train_iteration_end(
    self,
    agent_context: easyagents.core.AgentContext
)

Called once after the current iteration is completed

View Source

    def on_train_iteration_end(self, agent_context: AgentContext):

        """Called once after the current iteration is completed"""