initial n steps

marl_factory_grid/environment/factory.py

@@ -136,6 +136,7 @@ class Factory(gym.Env):
 
         # All is set up, trigger entity spawn with variable pos
         self.state.rules.do_all_reset(self.state)
+        self.state.rules.do_all_post_spawn_reset(self.state)
 
         # Build initial observations for all agents
         self.obs_builder.reset(self.state)

marl_factory_grid/environment/groups/global_entities.py

@@ -4,15 +4,17 @@ from random import shuffle
 from typing import Dict
 
 from marl_factory_grid.environment.groups.objects import Objects
-from marl_factory_grid.utils.helpers import POS_MASK
+from marl_factory_grid.utils.helpers import POS_MASK_8, POS_MASK_4
 
 
 class Entities(Objects):
     _entity = Objects
 
-    @staticmethod
-    def neighboring_positions(pos):
-        return [tuple(x) for x in (POS_MASK + pos).reshape(-1, 2)]
+    def neighboring_positions(self, pos):
+        return [tuple(x) for x in (POS_MASK_8 + pos).reshape(-1, 2) if tuple(x) in self._floor_positions]
+
+    def neighboring_4_positions(self, pos):
+        return [tuple(x) for x in (POS_MASK_4 + pos) if tuple(x) in self._floor_positions]
 
     def get_entities_near_pos(self, pos):
         return [y for x in itemgetter(*self.neighboring_positions(pos))(self.pos_dict) for y in x]

marl_factory_grid/environment/rules.py

@@ -1,7 +1,10 @@
 import abc
+import random
 from random import shuffle
 from typing import List, Collection
 
+import numpy as np
+
 from marl_factory_grid.environment import rewards as r, constants as c
 from marl_factory_grid.environment.entity.agent import Agent
 from marl_factory_grid.utils import helpers as h
@@ -37,6 +40,15 @@ class Rule(abc.ABC):
         TODO
 
 
+        :return:
+        """
+        return []
+
+    def on_reset_post_spawn(self, state) -> List[TickResult]:
+        """
+        TODO
+
+
         :return:
         """
         return []
@@ -230,3 +242,33 @@ class WatchCollisions(Rule):
             if inter_entity_collision_detected or collision_in_step:
                 return [DoneResult(validity=c.VALID, identifier=c.COLLISION, reward=self.reward_at_done)]
         return []
+
+
+class DoRandomInitialSteps(Rule):
+    def __init__(self, random_steps: 10):
+        """
+        Special rule which spawns destinations, that are bound to a single agent a fixed set of positions.
+        Useful for introducing specialists, etc. ..
+
+        !!! This rule does not introduce any reward or done condition.
+
+        :param random_steps:  Number of random steps agents perform in an environment.
+                                Useful in the `N-Puzzle` configuration.
+        """
+        super().__init__()
+        self.random_steps = random_steps
+
+    def on_reset_post_spawn(self, state):
+        state.print("Random Initial Steps initiated....")
+        for _ in range(self.random_steps):
+            # Find free positions
+            free_pos = state.random_free_position
+            neighbor_positions = state.entities.neighboring_4_positions(free_pos)
+            random.shuffle(neighbor_positions)
+            chosen_agent = h.get_first(state[c.AGENT].by_pos(neighbor_positions.pop()))
+            assert isinstance(chosen_agent, Agent)
+            valid = chosen_agent.move(free_pos, state)
+            valid_str = " not" if not valid else ""
+            state.print(f"Move {chosen_agent.name} from {chosen_agent.last_pos} "
+                        f"to {chosen_agent.pos} was{valid_str} valid.")
+        pass