opencog-rocca / rocca / agents / navigate.py

# AUTOGENERATED! DO NOT EDIT! File to edit: 03_minerl_navigate_agent.ipynb (unless otherwise specified).

__all__ = ["NavigateAgent"]

# Cell

import time

import gym
import minerl  # noqa

# OpenCog
from opencog.logger import log
from opencog.pln import *
from opencog.type_constructors import *
from opencog.utilities import set_default_atomspace

# ROCCA
from ..envs.wrappers import MineRLWrapper
from ..envs.wrappers.minerl_wrapper import minerl_single_action
from . import OpencogAgent
from .utils import *

from ..utils import *

# Cell


class NavigateAgent(OpencogAgent):
    def __init__(self, env, action_space, p_goal, n_goal, log_level="info"):
        OpencogAgent.__init__(self, env, action_space, p_goal, n_goal, log_level)

    def control_cycle(self):
        """Run one step of observation, decision and env update"""
        agent_log.debug("atomese_obs = {}".format(self.observation))
        obs_record = [
            self.record(o, self.cycle_count, tv=TRUE_TV) for o in self.observation
        ]
        agent_log.debug("obs_record = {}".format(obs_record))

        # Make the goal for that iteration
        goal = self.make_goal()
        agent_log.debug("goal = {}".format(goal))

        # Plan, i.e. come up with cognitive schematics as plans.  Here the
        # goal expiry is 2, i.e. must be fulfilled set for the next two iterations.
        cogscms = self.plan(goal, self.expiry)
        agent_log.debug("cogscms = {}".format(cogscms))

        # Deduce the action distribution
        mxmdl = self.deduce(cogscms)
        agent_log.debug("mxmdl = {}".format(mxmdl_to_str(mxmdl)))

        # Select the next action
        action, pblty = self.decide(mxmdl)
        agent_log.debug(
            "action with probability of success = {}".format(
                act_pblt_to_str((action, pblty))
            )
        )

        # Timestamp the action that is about to be executed
        action_record = self.record(action, self.cycle_count, tv=TRUE_TV)
        agent_log.debug("action_record = {}".format(action_record))
        agent_log.debug("action = {}".format(action))

        # Increment the counter for that action and log it
        self.action_counter[action] += 1
        agent_log.debug("action_counter = {}".format(self.action_counter))

        # Increase the step count and run the next step of the environment
        self.cycle_count += 1
        # TODO gather environment info.
        reward, self.observation, done = self.env.step(
            minerl_single_action(self.env, action)
        )

        self.accumulated_reward += float(reward.out[1].name)
        agent_log.debug("observation = {}".format(self.observation))
        agent_log.debug("reward = {}".format(reward))
        agent_log.debug("accumulated reward = {}".format(self.accumulated_reward))

        reward_record = self.record(reward, self.cycle_count, tv=TRUE_TV)
        agent_log.debug("reward_record = {}".format(reward_record))

        if done:
            return False

        return True