OiO.lk Community platform!

Oio.lk is an excellent forum for developers, providing a wide range of resources, discussions, and support for those in the developer community. Join oio.lk today to connect with like-minded professionals, share insights, and stay updated on the latest trends and technologies in the development field.
  You need to log in or register to access the solved answers to this problem.
  • You have reached the maximum number of guest views allowed
  • Please register below to remove this limitation

How to prevent _was_dead_step() from clearning all the rewards in pettingzoo custom environment? [closed]

  • Thread starter Thread starter Dechen
  • Start date Start date
D

Dechen

Guest
I have a custom pettingzoo AEC environment. When an agent terminates, _was_dead_step() is called to clear out that agent. However, _was_dead_step() uses self._clear_rewards() at the end which sets all rewards for all the remaining agents to 0 as well.

I want to prevent this from happening as I need those rewards further (I am tracking rewards instead of accumulated rewards as that makes more sense in my problem). Is there any way to do so?

Code:
def _clear_rewards(self) -> None:
        """Clears all items in .rewards."""
        for agent in self.rewards:
            self.rewards[agent] = 0
def _was_dead_step(self, action: ActionType) -> None:
        """Helper function that performs step() for dead agents.

        Does the following:

        1. Removes dead agent from .agents, .terminations, .truncations, .rewards, ._cumulative_rewards, and .infos
        2. Loads next agent into .agent_selection: if another agent is dead, loads that one, otherwise load next live agent
        3. Clear the rewards dict

        Examples:
            Highly recommended to use at the beginning of step as follows:

        def step(self, action):
            if (self.terminations[self.agent_selection] or self.truncations[self.agent_selection]):
                self._was_dead_step()
                return
            # main contents of step
        """
        if action is not None:
            raise ValueError("when an agent is dead, the only valid action is None")

        # removes dead agent
        agent = self.agent_selection
        assert (
            self.terminations[agent] or self.truncations[agent]
        ), "an agent that was not dead as attempted to be removed"
        del self.terminations[agent]
        del self.truncations[agent]
        del self.rewards[agent]
        del self._cumulative_rewards[agent]
        del self.infos[agent]
        self.agents.remove(agent)

        # finds next dead agent or loads next live agent (Stored in _skip_agent_selection)
        _deads_order = [
            agent
            for agent in self.agents
            if (self.terminations[agent] or self.truncations[agent])
        ]
        if _deads_order:
            if getattr(self, "_skip_agent_selection", None) is None:
                self._skip_agent_selection = self.agent_selection
            self.agent_selection = _deads_order[0]
        else:
            if getattr(self, "_skip_agent_selection", None) is not None:
                assert self._skip_agent_selection is not None
                self.agent_selection = self._skip_agent_selection
            self._skip_agent_selection = None
        self._clear_rewards()
<p>I have a custom pettingzoo AEC environment. When an agent terminates, _was_dead_step() is called to clear out that agent. However, _was_dead_step() uses self._clear_rewards() at the end which sets all rewards for all the remaining agents to 0 as well.</p>
<p>I want to prevent this from happening as I need those rewards further (I am tracking rewards instead of accumulated rewards as that makes more sense in my problem). Is there any way to do so?</p>
<pre><code>def _clear_rewards(self) -> None:
"""Clears all items in .rewards."""
for agent in self.rewards:
self.rewards[agent] = 0
def _was_dead_step(self, action: ActionType) -> None:
"""Helper function that performs step() for dead agents.

Does the following:

1. Removes dead agent from .agents, .terminations, .truncations, .rewards, ._cumulative_rewards, and .infos
2. Loads next agent into .agent_selection: if another agent is dead, loads that one, otherwise load next live agent
3. Clear the rewards dict

Examples:
Highly recommended to use at the beginning of step as follows:

def step(self, action):
if (self.terminations[self.agent_selection] or self.truncations[self.agent_selection]):
self._was_dead_step()
return
# main contents of step
"""
if action is not None:
raise ValueError("when an agent is dead, the only valid action is None")

# removes dead agent
agent = self.agent_selection
assert (
self.terminations[agent] or self.truncations[agent]
), "an agent that was not dead as attempted to be removed"
del self.terminations[agent]
del self.truncations[agent]
del self.rewards[agent]
del self._cumulative_rewards[agent]
del self.infos[agent]
self.agents.remove(agent)

# finds next dead agent or loads next live agent (Stored in _skip_agent_selection)
_deads_order = [
agent
for agent in self.agents
if (self.terminations[agent] or self.truncations[agent])
]
if _deads_order:
if getattr(self, "_skip_agent_selection", None) is None:
self._skip_agent_selection = self.agent_selection
self.agent_selection = _deads_order[0]
else:
if getattr(self, "_skip_agent_selection", None) is not None:
assert self._skip_agent_selection is not None
self.agent_selection = self._skip_agent_selection
self._skip_agent_selection = None
self._clear_rewards()
</code></pre>
 

Latest posts

Top