diff --git a/overcooked_simulator/counters.py b/overcooked_simulator/counters.py
index 50d76412e0d235ef047dcc6b5ce5756ee0dcd209..6d1196bccf11a16ff9455bf694bb17cb1963cd2d 100644
--- a/overcooked_simulator/counters.py
+++ b/overcooked_simulator/counters.py
@@ -36,7 +36,7 @@ import dataclasses
import logging
from collections import deque
from datetime import datetime, timedelta
-from typing import TYPE_CHECKING, Optional, Callable, TypedDict
+from typing import TYPE_CHECKING, Optional, Callable, Set
if TYPE_CHECKING:
from overcooked_simulator.overcooked_environment import (
@@ -57,26 +57,6 @@ from overcooked_simulator.game_items import (
log = logging.getLogger(__name__)
-class TransitionsValueDict(TypedDict):
- """The values in the transitions dicts of the `CookingEquipment`."""
-
- seconds: int | float
- """The needed seconds to progress for the transition."""
- needs: list[str]
- """The names of the needed items for the transition."""
- info: ItemInfo | str
- """The ItemInfo of the resulting item."""
-
-
-class TransitionsValueByNameDict(TypedDict):
- """The values in the transitions dicts of the `CuttingBoard` and the `Sink`."""
-
- seconds: int | float
- """The needed seconds to progress for the transition."""
- result: str
- """The new name of the item after the transition."""
-
-
class Counter:
"""Simple class for a counter at a specified position (center of counter). Can hold things on top.
@@ -172,11 +152,12 @@ class CuttingBoard(Counter):
The character `C` in the `layout` file represents the CuttingBoard.
"""
- def __init__(
- self, pos: np.ndarray, transitions: dict[str, TransitionsValueByNameDict]
- ):
+ def __init__(self, pos: np.ndarray, transitions: dict[str, ItemInfo]):
self.progressing = False
self.transitions = transitions
+ self.inverted_transition_dict = {
+ info.needs[0]: info for name, info in self.transitions.items()
+ }
super().__init__(pos=pos)
def progress(self, passed_time: timedelta, now: datetime):
@@ -193,20 +174,20 @@ class CuttingBoard(Counter):
if (
self.occupied
and self.progressing
- and self.occupied_by.name in self.transitions
+ and self.occupied_by.name in self.inverted_transition_dict
):
percent = (
passed_time.total_seconds()
- / self.transitions[self.occupied_by.name]["seconds"]
+ / self.inverted_transition_dict[self.occupied_by.name].seconds
)
self.occupied_by.progress(
equipment=self.__class__.__name__, percent=percent
)
if self.occupied_by.progress_percentage == 1.0:
self.occupied_by.reset()
- self.occupied_by.name = self.transitions[self.occupied_by.name][
- "result"
- ]
+ self.occupied_by.name = self.inverted_transition_dict[
+ self.occupied_by.name
+ ].name
def start_progress(self):
"""Starts the cutting process."""
@@ -353,7 +334,7 @@ class PlateDispenser(Counter):
pos: npt.NDArray[float],
dispensing: ItemInfo,
plate_config: PlateConfig,
- plate_transitions: dict,
+ plate_transitions: dict[str, ItemInfo],
**kwargs,
) -> None:
super().__init__(pos=pos, **kwargs)
@@ -362,7 +343,7 @@ class PlateDispenser(Counter):
self.out_of_kitchen_timer = []
self.plate_config = plate_config
self.next_plate_time = datetime.max
- self.plate_transitions: dict[str, TransitionsValueDict] = plate_transitions
+ self.plate_transitions = plate_transitions
self.setup_plates()
def pick_up(self, on_hands: bool = True) -> Item | None:
@@ -505,17 +486,25 @@ class Sink(Counter):
def __init__(
self,
pos: npt.NDArray[float],
- transitions: dict[str, TransitionsValueByNameDict],
+ transitions: dict[str, ItemInfo],
sink_addon: SinkAddon = None,
):
super().__init__(pos=pos)
self.progressing = False
self.sink_addon: SinkAddon = sink_addon
"""The connected sink addon which will receive the clean plates"""
- self.occupied_by = deque()
+ self.occupied_by: deque[Plate] = deque()
"""The queue of dirty plates. Only the one on the top is progressed."""
self.transitions = transitions
"""The allowed transitions for the items in the sink. Here only clean plates transfer from dirty plates."""
+ self.transition_needs: Set[str] = set()
+ """Set of all first needs of the transition item info."""
+
+ for name, info in transitions.items():
+ assert (
+ len(info.needs) >= 1
+ ), "transitions in a Sink need at least one item need."
+ self.transition_needs.update([info.needs[0]])
@property
def occupied(self):
@@ -526,24 +515,21 @@ class Sink(Counter):
if (
self.occupied
and self.progressing
- and self.occupied_by[-1].name in self.transitions
+ and self.occupied_by[-1].name in self.transition_needs
):
- percent = (
- passed_time.total_seconds()
- / self.transitions[self.occupied_by[-1].name]["seconds"]
- )
- self.occupied_by[-1].progress(
- equipment=self.__class__.__name__, percent=percent
- )
- if self.occupied_by[-1].progress_percentage == 1.0:
- self.occupied_by[-1].reset()
- print(self.transitions[self.occupied_by[-1].name]["result"])
- self.occupied_by[-1].name = self.transitions[self.occupied_by[-1].name][
- "result"
- ]
- plate = self.occupied_by.pop()
- plate.clean = True
- self.sink_addon.add_clean_plate(plate)
+ for name, info in self.transitions.items():
+ if info.needs[0] == self.occupied_by[-1].name:
+ percent = passed_time.total_seconds() / info.seconds
+ self.occupied_by[-1].progress(
+ equipment=self.__class__.__name__, percent=percent
+ )
+ if self.occupied_by[-1].progress_percentage == 1.0:
+ self.occupied_by[-1].reset()
+ self.occupied_by[-1].name = name
+ plate = self.occupied_by.pop()
+ plate.clean = True
+ self.sink_addon.add_clean_plate(plate)
+ break
def start_progress(self):
"""Starts the cutting process."""
diff --git a/overcooked_simulator/game_items.py b/overcooked_simulator/game_items.py
index 08e255e3eb139c98589c7ad681b82054bc2c2b75..babd1ec2713aa21ac84472137f339711e0b0932f 100644
--- a/overcooked_simulator/game_items.py
+++ b/overcooked_simulator/game_items.py
@@ -25,7 +25,7 @@ import dataclasses
import datetime
import logging
from enum import Enum
-from typing import Optional
+from typing import Optional, TypedDict
log = logging.getLogger(__name__)
@@ -81,10 +81,22 @@ class ItemInfo:
equipment: ItemInfo | None = dataclasses.field(compare=False, default=None)
"""On which the item can be created. `null`, `~` (None) converts to Plate."""
+ recipe: collections.Counter | None = None
+ """Internally set in CookingEquipment"""
+
def __post_init__(self):
self.type = ItemType(self.type)
+class ActiveTransitionTypedDict(TypedDict):
+ """The values in the active transitions dicts of `CookingEquipment`."""
+
+ seconds: int | float
+ """The needed seconds to progress for the transition."""
+ result: str
+ """The new name of the item after the transition."""
+
+
class Item:
"""Base class for game items which can be held by a player."""
@@ -135,10 +147,10 @@ class CookingEquipment(Item):
"""Pot, Pan, ... that can hold items. It holds the progress of the content (e.g., the soup) in itself (
progress_percentage) and not in the items in the content list."""
- def __init__(self, transitions: dict, *args, **kwargs):
+ def __init__(self, transitions: dict[str, ItemInfo], *args, **kwargs):
super().__init__(*args, **kwargs)
self.transitions = transitions
- self.active_transition: Optional[dict] = None
+ self.active_transition: Optional[ActiveTransitionTypedDict] = None
"""The info how and when to convert the content_list to a new item."""
self.content_ready: Item | None = None
@@ -151,7 +163,7 @@ class CookingEquipment(Item):
log.debug(f"Initialize {self.name}: {self.transitions}")
for transition in self.transitions.values():
- transition["recipe"] = collections.Counter(transition["needs"])
+ transition.recipe = collections.Counter(transition.needs)
def can_combine(self, other) -> bool:
# already cooking or nothing to combine
@@ -168,9 +180,7 @@ class CookingEquipment(Item):
item.name for item in self.content_list + other
)
print(ingredients)
- return any(
- ingredients <= recipe["recipe"] for recipe in self.transitions.values()
- )
+ return any(ingredients <= recipe.recipe for recipe in self.transitions.values())
def combine(self, other) -> Item | None:
return_value = None
@@ -185,14 +195,13 @@ class CookingEquipment(Item):
ingredients = collections.Counter(item.name for item in self.content_list)
for result, transition in self.transitions.items():
- recipe = transition["recipe"]
- if ingredients == recipe:
- if transition["seconds"] == 0:
- self.content_ready = Item(name=result, item_info=transition["info"])
+ if ingredients == transition.recipe:
+ if transition.seconds == 0:
+ self.content_ready = Item(name=result, item_info=transition)
else:
self.active_transition = {
- "seconds": transition["seconds"],
- "result": Item(name=result, item_info=transition["info"]),
+ "seconds": transition.seconds,
+ "result": Item(name=result, item_info=transition),
}
print(f"{self.name} {self.active_transition}, {self.content_list}")
break
@@ -246,9 +255,7 @@ class Plate(CookingEquipment):
"""All meals can be hold by a clean plate"""
super().__init__(
name=self.create_name(),
- transitions={
- k: v for k, v in transitions.items() if not v["info"].equipment
- },
+ transitions={k: v for k, v in transitions.items() if not v.equipment},
*args,
**kwargs,
)
diff --git a/overcooked_simulator/overcooked_environment.py b/overcooked_simulator/overcooked_environment.py
index b60ed369dbeab16833d0a2dad499ec0bac84d59a..daa3a360b222b55b05d358f45a3bec635e079ccf 100644
--- a/overcooked_simulator/overcooked_environment.py
+++ b/overcooked_simulator/overcooked_environment.py
@@ -125,26 +125,14 @@ class Environment:
},
)
"""The manager for the orders and score update."""
- plate_transitions = {
- item: {
- "seconds": info.seconds,
- "needs": info.needs,
- "info": info,
- }
- for item, info in self.item_info.items()
- if info.type == ItemType.Meal
- }
self.SYMBOL_TO_CHARACTER_MAP = {
"#": Counter,
"C": lambda pos: CuttingBoard(
- pos,
- {
- info.needs[0]: {"seconds": info.seconds, "result": item}
- for item, info in self.item_info.items()
- if info.equipment is not None
- and info.equipment.name == "CuttingBoard"
- },
+ pos=pos,
+ transitions=self.filter_item_info(
+ self.item_info, by_equipment_name="CuttingBoard"
+ ),
),
"X": Trashcan,
"W": lambda pos: ServingWindow(
@@ -156,7 +144,9 @@ class Environment:
"T": lambda pos: Dispenser(pos, self.item_info["Tomato"]),
"L": lambda pos: Dispenser(pos, self.item_info["Lettuce"]),
"P": lambda pos: PlateDispenser(
- plate_transitions=plate_transitions,
+ plate_transitions=self.filter_item_info(
+ item_info=self.item_info, by_item_type=ItemType.Meal
+ ),
pos=pos,
dispensing=self.item_info["Plate"],
plate_config=PlateConfig(
@@ -175,15 +165,9 @@ class Environment:
occupied_by=CookingEquipment(
name="Pot",
item_info=self.item_info["Pot"],
- transitions={
- item: {
- "seconds": info.seconds,
- "needs": info.needs,
- "info": info,
- }
- for item, info in self.item_info.items()
- if info.equipment is not None and info.equipment.name == "Pot"
- },
+ transitions=self.filter_item_info(
+ self.item_info, by_equipment_name="Pot"
+ ),
),
), # Stove with pot: U because it looks like a pot
"Q": lambda pos: Stove(
@@ -191,26 +175,18 @@ class Environment:
occupied_by=CookingEquipment(
name="Pan",
item_info=self.item_info["Pan"],
- transitions={
- item: {
- "seconds": info.seconds,
- "needs": info.needs,
- "info": info,
- }
- for item, info in self.item_info.items()
- if info.equipment is not None and info.equipment.name == "Pan"
- },
+ transitions=self.filter_item_info(
+ self.item_info, by_equipment_name="Pan"
+ ),
),
), # Stove with pan: Q because it looks like a pan
"B": lambda pos: Dispenser(pos, self.item_info["Bun"]),
"M": lambda pos: Dispenser(pos, self.item_info["Meat"]),
"S": lambda pos: Sink(
pos,
- transitions={
- info.needs[0]: {"seconds": info.seconds, "result": item}
- for item, info in self.item_info.items()
- if info.equipment is not None and info.equipment.name == "Sink"
- },
+ transitions=self.filter_item_info(
+ item_info=self.item_info, by_equipment_name="Sink"
+ ),
),
"+": SinkAddon,
}
@@ -242,17 +218,17 @@ class Environment:
"""The relative env time when it will stop/end"""
log.debug(f"End time: {self.env_time_end}")
+ @property
+ def game_ended(self) -> bool:
+ """Whether the game is over or not based on the calculated `Environment.env_time_end`"""
+ return self.env_time >= self.env_time_end
+
def get_env_time(self):
"""the internal time of the environment. An environment starts always with the time from `create_init_env_time`.
Utility method to pass a reference to the serving window."""
return self.env_time
- @property
- def game_ended(self) -> bool:
- """Whether the game is over or not based on the calculated `Environment.env_time_end`"""
- return self.env_time >= self.env_time_end
-
def load_item_info(self) -> dict[str, ItemInfo]:
"""Load `item_info.yml`, create ItemInfo classes and replace equipment strings with item infos."""
with open(self.item_info_path, "r") as file:
@@ -708,3 +684,25 @@ class Environment:
"""Reset the env time to the initial time, defined by `create_init_env_time`."""
self.env_time = create_init_env_time()
log.debug(f"Reset env time to {self.env_time}")
+
+ @staticmethod
+ def filter_item_info(
+ item_info: dict[str, ItemInfo],
+ by_item_type: ItemType = None,
+ by_equipment_name: str = None,
+ ) -> dict[str, ItemInfo]:
+ """Filter the item info dict by item type or equipment name"""
+ if by_item_type is not None:
+ return {
+ name: info
+ for name, info in item_info.items()
+ if info.type == by_item_type
+ }
+ if by_equipment_name is not None:
+ return {
+ name: info
+ for name, info in item_info.items()
+ if info.equipment is not None
+ and info.equipment.name == by_equipment_name
+ }
+ return item_info
diff --git a/tests/test_start.py b/tests/test_start.py
index 13ac792cf56ecf5f25d1c0b80e2c5417a7a597ab..ae27dd25a987c5e04c4e03114cc53b859b90095a 100644
--- a/tests/test_start.py
+++ b/tests/test_start.py
@@ -6,7 +6,7 @@ import pytest
from overcooked_simulator import ROOT_DIR
from overcooked_simulator.counters import Counter, CuttingBoard
-from overcooked_simulator.game_items import Item
+from overcooked_simulator.game_items import Item, ItemInfo, ItemType
from overcooked_simulator.overcooked_environment import (
Action,
Environment,
@@ -228,7 +228,15 @@ def test_processing():
counter_pos = np.array([2, 2])
counter = CuttingBoard(
counter_pos,
- transitions={"Tomato": {"seconds": 1, "result": "ChoppedTomato"}},
+ transitions={
+ "ChoppedTomato": ItemInfo(
+ name="ChoppedTomato",
+ seconds=0.5,
+ equipment=ItemInfo(name="CuttingBoard", type=ItemType.Equipment),
+ type=ItemType.Ingredient,
+ needs=["Tomato"],
+ )
+ },
)
sim.env.counters.append(counter)
@@ -248,7 +256,7 @@ def test_processing():
assert tomato.name != "ChoppedTomato", "Tomato is not finished yet."
- time.sleep(1)
+ time.sleep(0.6)
assert tomato.name == "ChoppedTomato", "Tomato should be finished."