diff --git a/overcooked_simulator/counters.py b/overcooked_simulator/counters.py
index f63d77aac4d9a2ae6a146b164051c23a4ca3e90f..75152544acb3267dca4dd78ae531826168d442dc 100644
--- a/overcooked_simulator/counters.py
+++ b/overcooked_simulator/counters.py
@@ -39,7 +39,7 @@ import uuid
from collections import deque
from collections.abc import Iterable
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 (
@@ -62,26 +62,6 @@ log = logging.getLogger(__name__)
COUNTER_CATEGORY = "Counter"
-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.
@@ -198,11 +178,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):
@@ -219,20 +200,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."""
@@ -390,7 +371,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)
@@ -399,7 +380,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:
@@ -565,17 +546,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):
@@ -586,23 +575,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()
- 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 8f1b39bd52c35fdc923ce714a88b4c7f815fe471..a77f8d78159ee6e6542209230e7bc50a6c0e76df 100644
--- a/overcooked_simulator/game_items.py
+++ b/overcooked_simulator/game_items.py
@@ -26,7 +26,7 @@ import datetime
import logging
import uuid
from enum import Enum
-from typing import Optional
+from typing import Optional, TypedDict
log = logging.getLogger(__name__)
@@ -85,10 +85,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."""
@@ -151,13 +163,12 @@ class Item:
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."""
-
item_category = "Cooking Equipment"
- 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."""
# TODO change content ready just to str (name of the item)?
@@ -171,7 +182,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
@@ -187,9 +198,7 @@ class CookingEquipment(Item):
ingredients = collections.Counter(
item.name for item in self.content_list + other
)
- 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
@@ -204,14 +213,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),
}
break
else:
@@ -279,9 +287,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/gui_2d_vis/drawing.py b/overcooked_simulator/gui_2d_vis/drawing.py
index 415e0ce302c93ce2857f58569e5ed89e41c0aacd..5be3a8aa1892bea0450ed3f6752a61d48ef864a4 100644
--- a/overcooked_simulator/gui_2d_vis/drawing.py
+++ b/overcooked_simulator/gui_2d_vis/drawing.py
@@ -239,6 +239,7 @@ class Visualizer:
case "image":
if "center_offset" in part:
d = np.array(part["center_offset"]) * grid_size
+ pos = np.array(pos)
pos += d
self.draw_image(
diff --git a/overcooked_simulator/overcooked_environment.py b/overcooked_simulator/overcooked_environment.py
index ce79e280b4760a59ab92129f86bd26c905639a1f..90694f30f479e5bf8ec437d97b49e67b46426af3 100644
--- a/overcooked_simulator/overcooked_environment.py
+++ b/overcooked_simulator/overcooked_environment.py
@@ -138,15 +138,6 @@ 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
- }
cooking_counter_equipments = {
cooking_counter: [
@@ -161,13 +152,10 @@ class Environment:
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(
@@ -184,7 +172,9 @@ class Environment:
"E": lambda pos: Dispenser(pos, self.item_info["Cheese"]), # chEEEEse
"G": lambda pos: Dispenser(pos, self.item_info["Sausage"]), # sausaGe
"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(
@@ -205,15 +195,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: CookingCounter(
@@ -223,15 +207,9 @@ 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
"O": lambda pos: CookingCounter(
@@ -241,15 +219,9 @@ class Environment:
occupied_by=CookingEquipment(
name="Peel",
item_info=self.item_info["Peel"],
- 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 == "Peel"
- },
+ transitions=self.filter_item_info(
+ self.item_info, by_equipment_name="Peel"
+ ),
),
),
"F": lambda pos: CookingCounter(
@@ -259,27 +231,18 @@ class Environment:
occupied_by=CookingEquipment(
name="Basket",
item_info=self.item_info["Basket"],
- 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 == "Basket"
- },
+ transitions=self.filter_item_info(
+ self.item_info, by_equipment_name="Basket"
+ ),
),
), # 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,
}
@@ -311,19 +274,19 @@ 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, data) -> dict[str, ItemInfo]:
- """Load `item_info.yml` if only the path is given, create ItemInfo classes and replace equipment strings with item infos."""
+ """Load `item_info.yml`, create ItemInfo classes and replace equipment strings with item infos."""
if self.as_files:
with open(data, "r") as file:
item_lookup = yaml.safe_load(file)
@@ -797,3 +760,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 1f7c27bbd83190ada56e869f60ce863b614cb1b7..45cbf4dd6cb30fb1ef7693c6acc02f6032e2b2de 100644
--- a/tests/test_start.py
+++ b/tests/test_start.py
@@ -5,7 +5,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,
@@ -227,7 +227,15 @@ def test_processing(env_config, layout_config, item_info):
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"],
+ )
+ },
)
env.counters.append(counter)