Add random instance to make randomness controllable

The random module is replaced with random instance across multiple classes within 'overcooked_simulator' to make the randomness in the Overcooked simulator controllable. The 'random' instance is initialised with a seed in 'overcooked_environment' and then passed to the classes that require randomness. This improves reproducibility of experiments, as the randomness can be controlled by the seed value.

overcooked_simulator/counter_factory.py

@@ -33,6 +33,7 @@ layout_chars:
 """
 import inspect
 import sys
+from random import Random
 from typing import Any, Type, TypeVar
 
 import numpy as np
@@ -113,6 +114,7 @@ class CounterFactory:
         plate_config: PlateConfig,
         order_and_score: OrderAndScoreManager,
         hook: Hooks,
+        random: Random,
     ) -> None:
         """Constructor for the `CounterFactory` class. Set up the attributes necessary to instantiate the counters.
 
@@ -171,6 +173,9 @@ class CounterFactory:
         self.hook = hook
         """Reference to the hook manager."""
 
+        self.random = random
+        """Random instance."""
+
     def get_counter_object(self, c: str, pos: npt.NDArray[float]) -> Counter:
         """Create and returns a counter object based on the provided character and position."""
 
@@ -216,6 +221,7 @@ class CounterFactory:
                     ),
                     "plate_config": self.plate_config,
                     "dispensing": self.item_info[Plate.__name__],
+                    "random": self.random,
                 }
             )
         elif issubclass(counter_class, ServingWindow):

overcooked_simulator/counters.py

@@ -40,6 +40,7 @@ import uuid
 from collections import deque
 from collections.abc import Iterable
 from datetime import datetime, timedelta
+from random import Random
 from typing import TYPE_CHECKING, Optional, Callable, Set
 
 from overcooked_simulator.hooks import (
@@ -70,7 +71,6 @@ if TYPE_CHECKING:
         OrderAndScoreManager,
     )
 
-import numpy as np
 import numpy.typing as npt
 
 from overcooked_simulator.game_items import (
@@ -456,6 +456,7 @@ class PlateDispenser(Counter):
         dispensing: ItemInfo,
         plate_config: PlateConfig,
         plate_transitions: dict[str, ItemInfo],
+        random: Random,
         **kwargs,
     ) -> None:
         super().__init__(**kwargs)
@@ -473,6 +474,8 @@ class PlateDispenser(Counter):
         `out_of_kitchen_timer` list every frame."""
         self.plate_transitions: dict[str, ItemInfo] = plate_transitions
         """Transitions for the plates. Relevant for the sink, because a plate can become a clean one there."""
+        self.random = random
+        """Random instance."""
         self.setup_plates()
 
     def pick_up(self, on_hands: bool = True) -> Item | None:
@@ -497,9 +500,9 @@ class PlateDispenser(Counter):
         """Is called from the serving window to add a plate out of kitchen."""
         # not perfect identical to datetime.now but based on framerate enough.
         time_plate_to_add = env_time + timedelta(
-            seconds=np.random.uniform(
-                low=self.plate_config.plate_delay[0],
-                high=self.plate_config.plate_delay[1],
+            seconds=self.random.uniform(
+                a=self.plate_config.plate_delay[0],
+                b=self.plate_config.plate_delay[1],
             )
         )
         log.debug(f"New plate out of kitchen until {time_plate_to_add}")

overcooked_simulator/order.py

@@ -46,11 +46,11 @@ from __future__ import annotations
 
 import dataclasses
 import logging
-import random
 import uuid
 from abc import abstractmethod
 from collections import deque
 from datetime import datetime, timedelta
+from random import Random
 from typing import Callable, Tuple, Any, Deque, Protocol, TypedDict, Type
 
 from overcooked_simulator.game_items import Item, Plate, ItemInfo
@@ -137,9 +137,11 @@ class OrderGeneration:
         ```
     """
 
-    def __init__(self, available_meals: dict[str, ItemInfo], **kwargs):
+    def __init__(self, available_meals: dict[str, ItemInfo], random: Random, **kwargs):
         self.available_meals: list[ItemInfo] = list(available_meals.values())
         """Available meals restricted through the `environment_config.yml`."""
+        self.random = random
+        """Random instance."""
 
     @abstractmethod
     def init_orders(self, now) -> list[Order]:
@@ -161,11 +163,21 @@ class OrderGeneration:
 class OrderAndScoreManager:
     """The Order and Score Manager that is called from the serving window."""
 
-    def __init__(self, order_config, available_meals: dict[str, ItemInfo], hook: Hooks):
+    def __init__(
+        self,
+        order_config,
+        available_meals: dict[str, ItemInfo],
+        hook: Hooks,
+        random: Random,
+    ):
+        self.random = random
+        """Random instance."""
         self.score: float = 0.0
         """The current score of the environment."""
         self.order_gen: OrderGeneration = order_config["order_gen_class"](
-            available_meals=available_meals, kwargs=order_config["order_gen_kwargs"]
+            available_meals=available_meals,
+            random=random,
+            kwargs=order_config["order_gen_kwargs"],
         )
         """The order generation."""
         self.serving_not_ordered_meals: Callable[
@@ -511,8 +523,8 @@ class RandomOrderGeneration(OrderGeneration):
     ```
     """
 
-    def __init__(self, available_meals: dict[str, ItemInfo], **kwargs):
-        super().__init__(available_meals, **kwargs)
+    def __init__(self, available_meals: dict[str, ItemInfo], random: Random, **kwargs):
+        super().__init__(available_meals, random, **kwargs)
         self.kwargs: RandomOrderKwarg = RandomOrderKwarg(**kwargs["kwargs"])
         self.next_order_time: datetime | None = datetime.max
         self.number_cur_orders: int = 0
@@ -524,7 +536,7 @@ class RandomOrderGeneration(OrderGeneration):
         if not self.kwargs.sample_on_serving:
             self.create_random_next_time_delta(now)
         return self.create_orders_for_meals(
-            random.choices(self.available_meals, k=self.kwargs.num_start_meals),
+            self.random.choices(self.available_meals, k=self.kwargs.num_start_meals),
             now,
             self.kwargs.sample_on_serving,
         )
@@ -547,7 +559,7 @@ class RandomOrderGeneration(OrderGeneration):
             self.needed_orders = max(self.needed_orders, 0)
             self.number_cur_orders += len(new_finished_orders)
             return self.create_orders_for_meals(
-                random.choices(self.available_meals, k=len(new_finished_orders)),
+                self.random.choices(self.available_meals, k=len(new_finished_orders)),
                 now,
             )
         if self.next_order_time <= now:
@@ -560,7 +572,7 @@ class RandomOrderGeneration(OrderGeneration):
                     self.next_order_time = datetime.max
                 self.number_cur_orders += 1
                 return self.create_orders_for_meals(
-                    [random.choice(self.available_meals)],
+                    [self.random.choice(self.available_meals)],
                     now,
                 )
         return []
@@ -575,7 +587,7 @@ class RandomOrderGeneration(OrderGeneration):
             else:
                 duration = timedelta(
                     seconds=getattr(
-                        random, self.kwargs.order_duration_random_func["func"]
+                        self.random, self.kwargs.order_duration_random_func["func"]
                     )(**self.kwargs.order_duration_random_func["kwargs"])
                 )
             log.info(f"Create order for meal {meal} with duration {duration}")
@@ -601,7 +613,7 @@ class RandomOrderGeneration(OrderGeneration):
 
     def create_random_next_time_delta(self, now: datetime):
         self.next_order_time = now + timedelta(
-            seconds=getattr(random, self.kwargs.sample_on_dur_random_func["func"])(
+            seconds=getattr(self.random, self.kwargs.sample_on_dur_random_func["func"])(
                 **self.kwargs.sample_on_dur_random_func["kwargs"]
             )
         )

overcooked_simulator/overcooked_environment.py

@@ -4,11 +4,11 @@ import dataclasses
 import inspect
 import json
 import logging
-import random
 import sys
 from datetime import timedelta, datetime
 from enum import Enum
 from pathlib import Path
+from random import Random
 from typing import Literal, TypedDict, Callable
 
 import numpy as np
@@ -123,7 +123,10 @@ class Environment:
         layout_config: Path | str,
         item_info: Path | str,
         as_files: bool = True,
+        seed: int = 56789223842348,
     ):
+        self.random: Random = Random(seed)
+        """Random instance."""
         self.hook: Hooks = Hooks(self)
         """Hook manager. Register callbacks and create hook points with additional kwargs."""
 
@@ -174,6 +177,7 @@ class Environment:
                 if info.type == ItemType.Meal and item in self.allowed_meal_names
             },
             hook=self.hook,
+            random=self.random,
         )
         """The manager for the orders and score update."""
 
@@ -198,6 +202,7 @@ class Environment:
             ),
             order_and_score=self.order_and_score,
             hook=self.hook,
+            random=self.random,
         )
 
         (
@@ -600,11 +605,13 @@ class Environment:
         self.players[player.name] = player
         if player.pos is None:
             if len(self.designated_player_positions) > 0:
-                free_idx = random.randint(0, len(self.designated_player_positions) - 1)
+                free_idx = self.random.randint(
+                    0, len(self.designated_player_positions) - 1
+                )
                 player.move_abs(self.designated_player_positions[free_idx])
                 del self.designated_player_positions[free_idx]
             elif len(self.free_positions) > 0:
-                free_idx = random.randint(0, len(self.free_positions) - 1)
+                free_idx = self.random.randint(0, len(self.free_positions) - 1)
                 player.move_abs(self.free_positions[free_idx])
                 del self.free_positions[free_idx]
             else: