Skip to content
Snippets Groups Projects
Commit bb1339ae authored by Florian Schröder's avatar Florian Schröder
Browse files

Refactor item transition implementation in simulator 

The transition implementation for items in the simulator has been updated. This refactoring introduces the filter_item_info function, replacing the previous verbose, repeating pattern with filter_item_info method calls in the overcooked_environment.py module. This makes the code more efficient and readable. Test adjustments and additional type hints were also included.
parent cef13156
No related branches found
No related tags found
1 merge request!32Resolve "replace transitions values from dict just to the iteminfo because it already stores all the info"
Pipeline #44363 passed
Stage: test
Hide whitespace changes
Inline Side-by-side
overcooked_simulator/counters.py
+ 36
− 50
View file @ bb1339ae
......@@ -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."""
......
overcooked_simulator/game_items.py
+ 23
− 16
View file @ bb1339ae
......@@ -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,
)
......
overcooked_simulator/overcooked_environment.py
+ 43
− 45
View file @ bb1339ae
......@@ -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
tests/test_start.py
+ 11
− 3
View file @ bb1339ae
......@@ -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."
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment