"""All counters are derived from the `Counter` class. Counters implement the `Counter.pick_up` method, which defines
what should happen when the agent wants to pick something up from the counter. On the other side,
the `Counter.drop_off` method receives the item what should be put on the counter. Before that the
`Counter.can_drop_off` method checked if the item can be put on the counter. The progress on Counters or on objects
on the counters are handled via the Counters. They have the task to delegate the progress call via the `progress`
method, e.g., the `CuttingBoard.progress`. The environment class detects which classes in this module have the
`progress` method defined and on instances of these classes the progress will be delegated.
Inside the item_info.yaml, equipment needs to be defined. It includes counters that are part of the
interaction/requirements for the interaction.
CuttingBoard:
type: Equipment
Sink:
type: Equipment
Stove:
type: Equipment
The defined counter classes are:
- `Counter`
- `CuttingBoard`
- `ServingWindow`
- `Dispenser`
- `PlateDispenser`
- `Trashcan`
- `CookingCounter`
- `Sink`
- `SinkAddon`
## Code Documentation
"""
from __future__ import annotations
import copy
import dataclasses
import logging
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, Callable, Set
from cooperative_cuisine.hooks import (
Hooks,
POST_DISPENSER_PICK_UP,
PRE_DISPENSER_PICK_UP,
CUTTING_BOARD_PROGRESS,
CUTTING_BOARD_100,
PRE_COUNTER_PICK_UP,
POST_COUNTER_PICK_UP,
PRE_SERVING,
POST_SERVING,
NO_SERVING,
DIRTY_PLATE_ARRIVES,
TRASHCAN_USAGE,
PLATE_CLEANED,
ADDED_PLATE_TO_SINK,
DROP_ON_SINK_ADDON,
PICK_UP_FROM_SINK_ADDON,
PLATE_OUT_OF_KITCHEN_TIME,
DROP_OFF_ON_COOKING_EQUIPMENT,
POST_COUNTER_DROP_OFF,
PRE_COUNTER_DROP_OFF,
POST_PLATE_DISPENSER_DROP_OFF,
PRE_PLATE_DISPENSER_DROP_OFF,
PRE_PLATE_DISPENSER_PICK_UP,
POST_PLATE_DISPENSER_PICK_UP,
DISPENSER_ITEM_RETURNED,
DROP_OFF_ON_COOKING_EQUIPMENT_PLATE_DISPENSER, PICK_UP_ON_COOKING_EQUIPMENT,
)
from cooperative_cuisine.state_representation import CounterState
if TYPE_CHECKING:
from cooperative_cuisine.environment import (
OrderManager,
)
import numpy as np
import numpy.typing as npt
from cooperative_cuisine.items import (
Item,
CookingEquipment,
Plate,
ItemInfo,
EffectType,
Effect,
)
log = logging.getLogger(__name__)
"""The logger for this module."""
class Counter:
"""Simple class for a counter at a specified position (center of counter). Can hold things on top.
The character `#` in the `layout` file represents the standard Counter.
"""
def __init__(
self,
pos: npt.NDArray[float],
hook: Hooks,
occupied_by: Item | None = None,
uid: str = None,
**kwargs,
):
"""Constructor setting the arguments as attributes.
Args:
pos: Position of the counter in the environment. 2-element vector.
occupied_by: The item on top of the counter.
"""
self.uuid: str = uuid.uuid4().hex if uid is None else None
"""A unique id for better tracking in GUIs with assets which instance moved or changed."""
self.pos: npt.NDArray[float] = pos
"""The position of the counter."""
self.occupied_by: Item | Iterable[Item] | None = occupied_by
"""What is on top of the counter, e.g., `Item`s."""
self.active_effects: list[Effect] = []
"""The effects that currently affect the usage of the counter."""
self.hook: Hooks = hook
"""Reference to the hook manager."""
self.orientation: npt.NDArray[float] = np.array([0, 1], dtype=float)
"""In what direction the counter is facing."""
@property
def occupied(self) -> bool:
"""Is something on top of the counter."""
return self.occupied_by is not None
def set_orientation(self, orientation: npt.NDArray[float]):
"""This method sets the orientation of an object to the specified numpy array.
The provided numpy array is normalized if its norm is not equal to 1, ensuring that it represents a valid
orientation. The resulting orientation is then assigned to the 'orientation' attribute of the object. If the
norm of the provided numpy array is already equal to 1, it is assigned directly to the 'orientation'
attribute without normalization.
Args:
orientation: A 2D numpy array representing the orientation of an object.
"""
if not np.isclose(np.linalg.norm(orientation), 1):
self.orientation = orientation / np.linalg.norm(orientation)
else:
self.orientation = orientation
def pick_up(
self, on_hands: bool = True, player: str = "0", player_holding: Item | None = None,
) -> Item | None | Iterable[Item]:
"""Gets called upon a player performing the pickup action. If the counter can give something to
the player, it does so. In the standard counter this is when an item is on the counter.
Args:
on_hands: Will the item be put on empty hands or on a cooking equipment.
player: The player name that tries to pick up from the counter.
player_holding: What the player is currently holding.
Returns:
The item which the counter is occupied by. None if nothing is there.
"""
self.hook(PRE_COUNTER_PICK_UP, counter=self, on_hands=on_hands, player=player)
if on_hands:
if self.occupied_by:
occupied_by = self.occupied_by
self.occupied_by = None
self.hook(
POST_COUNTER_PICK_UP,
counter=self,
on_hands=on_hands,
return_this=occupied_by,
player=player,
player_holding=player_holding,
)
return occupied_by
return None
if self.occupied_by and isinstance(self.occupied_by, CookingEquipment):
return_this = self.occupied_by.release()
self.hook(
PICK_UP_ON_COOKING_EQUIPMENT,
return_this=return_this,
occupied_by=self.occupied_by,
counter=self,
player=player,
player_holding=player_holding,
)
return return_this
occupied_by = self.occupied_by
self.occupied_by = None
self.hook(
POST_COUNTER_PICK_UP,
counter=self,
on_hands=on_hands,
return_this=occupied_by,
player=player,
player_holding=player_holding,
)
return occupied_by
def can_drop_off(self, item: Item) -> bool:
"""Checks whether an item by the player can be dropped of. More relevant for example with
ingredient dispensers, which should always be occupied and cannot take an item.
Args:
item: The item for which to check, if it can be placed on the counter.
Returns:
True if the item can be placed on the counter, False if not.
"""
return self.occupied_by is None or self.occupied_by.can_combine(item)
def drop_off(self, item: Item, player: str = "0") -> Item | None:
"""Takes the thing dropped of by the player.
Args:
item: The item to be placed on the counter.
player: The player name that tries to drop something on the counter.
Returns:
Item or None what should be put back on the players hand, e.g., the cooking equipment.
"""
self.hook(
PRE_COUNTER_DROP_OFF,
item=item,
equipment=self.occupied_by,
counter=self,
player=player,
)
if self.occupied_by is None:
self.occupied_by = item
self.hook(
POST_COUNTER_DROP_OFF,
counter=self,
player=player,
item=item,
)
elif self.occupied_by.can_combine(item):
before_combine = copy.copy(item)
occupied_before_combine = copy.deepcopy(self.occupied_by)
return_this = self.occupied_by.combine(item)
self.hook(
DROP_OFF_ON_COOKING_EQUIPMENT,
item=before_combine,
equipment=self.occupied_by,
counter=self,
occupied_before=occupied_before_combine,
player=player,
return_this=return_this,
)
return return_this
return None
def __repr__(self):
return (
f"{self.__class__.__name__}(pos={self.pos},occupied_by={self.occupied_by})"
)
def do_tool_interaction(self, passed_time: timedelta, tool: Item, player: str):
"""Progress call for player that interact with a counter (optionally via a tool).
Args:
passed_time: The amount of time that has passed between the environment step calls.
tool: The tool being used for interaction.
This method performs tool interaction on the specified tool. It can handle cases where the location is
occupied by a single item or a deque of items.
If the location is occupied by a deque of items, the method will iterate through each item and attempt to
perform the tool interaction. If succesful, it will set the 'successful' flag to True.
If the location is occupied by a single item, the method will directly call the 'do_single_tool_interaction'
method to perform the interaction.
If none of the attempts were successful, the method will call the 'do_single_tool_interaction' method on
itself to perform the interaction.
"""
successful = False
if self.occupied_by:
if isinstance(self.occupied_by, deque):
for item in self.occupied_by:
successful |= self._do_single_tool_interaction(
passed_time, tool, item, player
)
else:
successful = self._do_single_tool_interaction(
passed_time, tool, self.occupied_by, player
)
if not successful:
self._do_single_tool_interaction(passed_time, tool, self, player)
@staticmethod
def _do_single_tool_interaction(
passed_time: timedelta, tool: Item, target: Item | Counter, player: str
) -> bool:
"""This method is used to perform a single tool interaction on a target entity.
It calculates the progress of the interaction based on the amount of time passed and the properties of the
tool and target entity. If the tool has suitable effects for the target entity, the progress percentage is
updated and the method returns True. If the progress exceeds the maximum value, the effect is removed from
the target entity and the method returns True. Otherwise, the method returns False indicating that the tool
interaction was unsuccessful.
Args:
passed_time: A timedelta object representing the amount of time passed.
tool: An Item object representing the tool being used.
target: An Item or Counter object representing the target entity.
Returns:
A boolean value indicating whether the tool interaction was successful.
"""
suitable_effects = [
e for e in target.active_effects if e.name in tool.item_info.needs
]
if suitable_effects:
effect = suitable_effects[0]
percent = passed_time.total_seconds() / tool.item_info.seconds
effect.progres_percentage += percent
if effect.progres_percentage > 1.0:
effect.item_info.manager.remove_active_effect(effect, target, player)
target.active_effects.remove(effect)
return True
return False
def do_hand_free_interaction(
self, passed_time: timedelta, now: datetime, player: str
):
"""Called by environment step function for time progression.
Args:
passed_time: the time passed since the last progress call
now: the current env time. **Not the same as `datetime.now`**.
player: Name of the player doing the interaction.
"""
...
def to_dict(self) -> CounterState:
"""For the state representation. Only the relevant attributes are put into the dict."""
return {
"id": self.uuid,
"category": "Counter",
"type": self.__class__.__name__,
"pos": self.pos.tolist(),
"orientation": self.orientation.tolist(),
"occupied_by": None
if self.occupied_by is None
else (
self.occupied_by.to_dict()
if isinstance(self.occupied_by, Item)
else [o.to_dict() for o in self.occupied_by]
),
"active_effects": [e.to_dict() for e in self.active_effects],
}
class EdgeCounter(Counter):
...
class CuttingBoard(Counter):
"""Cutting ingredients on. The requirement in a new object could look like.
ChoppedTomato:
type: Ingredient
needs: [ Tomato ]
seconds: 4.0
equipment: CuttingBoard
The character `C` in the `layout` file represents the CuttingBoard.
"""
def __init__(self, transitions: dict[str, ItemInfo], **kwargs):
self.transitions: dict[str, ItemInfo] = transitions
"""The allowed transitions to a new item. Keys are the resulting items and the `ItemInfo` (value) contains
the needed items in the `need` attribute."""
self.inverted_transition_dict: dict[str, ItemInfo] = {
info.needs[0]: info for name, info in self.transitions.items()
}
"""For faster accessing the needed item. Keys are the ingredients that the player can put and chop on the
board."""
super().__init__(**kwargs)
def do_hand_free_interaction(
self, passed_time: timedelta, now: datetime, player: str
):
"""Called by environment step function for time progression.
Args:
passed_time: the time passed since the last progress call
now: the current env time. **Not the same as `datetime.now`**.
player: Name of the player doing the interaction.
Checks if the item on the board is in the allowed transitions via a Cutting board. Pass the progress call to
the item on the board. If the progress on the item reaches 100% it changes the name of the item based on the
"goal" name in the transition definition.
"""
if (
self.occupied
and self.occupied_by.name in self.inverted_transition_dict
and not any(
e.item_info.effect_type == EffectType.Unusable
for e in self.occupied_by.active_effects
)
and not any(
e.item_info.effect_type == EffectType.Unusable
for e in self.active_effects
)
):
percent = (
passed_time.total_seconds()
/ self.inverted_transition_dict[self.occupied_by.name].seconds
)
self.occupied_by.progress(
equipment=self.__class__.__name__, percent=percent
)
self.hook(
CUTTING_BOARD_PROGRESS,
counter=self,
player=player,
percent=percent,
passed_time=passed_time,
)
if self.occupied_by.progress_percentage == 1.0:
item_before = copy.copy(self.occupied_by)
self.occupied_by.reset()
self.occupied_by.name = self.inverted_transition_dict[
self.occupied_by.name
].name
self.hook(CUTTING_BOARD_100, counter=self, item=self.occupied_by, player=player, before=item_before)
class ServingWindow(Counter):
"""The orders and scores are updated based on completed and dropped off meals. The plate dispenser is pinged for
the info about a plate outside the kitchen.
All items in the `item_info.yml` with the type meal are considered to be servable, if they are ordered. Not
ordered meals can also be served, if a `serving_not_ordered_meals` function is set in the `environment_config.yml`.
The plate dispenser will put after some time a dirty plate on itself after a meal was served.
The character `W` in the `layout` file represents the ServingWindow.
"""
def __init__(
self,
order_manager: OrderManager,
meals: set[str],
env_time_func: Callable[[], datetime],
plate_dispenser: PlateDispenser = None,
**kwargs,
):
self.order_manager: OrderManager = order_manager
"""Reference to the OrderAndScoreManager class. It determines which meals can be served and it manages the
score."""
self.plate_dispenser: PlateDispenser = plate_dispenser
"""Served meals are mentioned on the plate dispenser. So that the plate dispenser can add a dirty plate after
some time."""
self.meals: set[str] = meals
"""All allowed meals by the `environment_config.yml`."""
self.env_time_func: Callable[[], datetime] = env_time_func
"""Reference to get the current env time by calling the `env_time_func`."""
super().__init__(**kwargs)
def drop_off(self, item, player: str = "0") -> Item | None:
env_time = self.env_time_func()
self.hook(
PRE_SERVING,
counter=self,
item=item,
env_time=env_time,
player=player,
)
if self.order_manager.serve_meal(item=item, env_time=env_time, player=player):
if self.plate_dispenser is not None:
self.plate_dispenser.update_plate_out_of_kitchen(env_time=env_time)
self.hook(POST_SERVING, counter=self, item=item, env_time=env_time, player=player)
return None
self.hook(NO_SERVING, counter=self, item=item, env_time=env_time, player=player)
return item
def can_drop_off(self, item: Item) -> bool:
if any(
e.item_info.effect_type == EffectType.Unusable for e in self.active_effects
):
return False
if any(
e.item_info.effect_type == EffectType.Unusable for e in item.active_effects
):
return False
return isinstance(item, CookingEquipment) and (
(item.content_ready is not None and item.content_ready.name in self.meals)
or (len(item.content_list) == 1 and item.content_list[0].name in self.meals)
)
def pick_up(self, on_hands: bool = True, player: str = "0", player_holding: Item | None = None,) -> Item | None:
pass
def add_plate_dispenser(self, plate_dispenser):
self.plate_dispenser = plate_dispenser
class Dispenser(Counter):
"""The class for all dispensers except plate dispenser. Here ingredients can be grabbed from the player/agent.
At the moment all ingredients have an unlimited stock.
The character for each dispenser in the `layout` file is defined in the `environment_config.yml`:
```yaml
T: Tomato
L: Lettuce
N: Onion # N for oNioN
B: Bun
M: Meat
```
In the implementation, an instance of the item to dispense is always on top of the dispenser.
Which also is easier for the visualization of the dispenser.
"""
def __init__(self, dispensing: ItemInfo, undo_dispenser_pickup: bool, **kwargs):
self.dispensing: ItemInfo = dispensing
"""`ItemInfo` what the the Dispenser is dispensing. One ready object always is on top of the counter."""
self.undo_dispenser_pickup: bool = undo_dispenser_pickup
"""Put back ingredients of the same type on a dispenser."""
super().__init__(
occupied_by=self.create_item(),
**kwargs,
)
def pick_up(self, on_hands: bool = True, player: str = "0", player_holding: Item | None = None,) -> Item | None:
self.hook(PRE_DISPENSER_PICK_UP, counter=self, on_hands=on_hands, player=player)
return_this = self.occupied_by
self.occupied_by = self.create_item()
self.hook(
POST_DISPENSER_PICK_UP,
counter=self,
on_hands=on_hands,
return_this=return_this,
player=player,
player_holding=player_holding,
)
return return_this
def drop_off(self, item: Item, player: str = "0") -> Item | None:
self.hook(DISPENSER_ITEM_RETURNED, player=player, counter=self, item=item)
if self.occupied_by.can_combine(item):
return self.occupied_by.combine(item)
def can_drop_off(self, item: Item) -> bool:
return self.occupied_by.can_combine(item) or (
self.undo_dispenser_pickup and item.name == self.dispensing.name
)
def __repr__(self):
return f"{self.dispensing.name}Dispenser"
def create_item(self) -> Item:
"""Create a new item to put on the dispenser after the previous one was picked up."""
kwargs = {
"name": self.dispensing.name,
"item_info": self.dispensing,
}
return Item(**kwargs)
def to_dict(self) -> CounterState:
d = super().to_dict()
d.update((("type", self.__repr__()),))
return d
@dataclasses.dataclass
class PlateConfig:
"""Configure the initial and behavior of the plates in the environment."""
clean_plates: int = 0
"""clean plates at the start."""
dirty_plates: int = 3
"""dirty plates at the start."""
plate_delay: list[int, int] = dataclasses.field(default_factory=lambda: [5, 10])
"""The uniform sampling range for the plate delay between serving and return in seconds."""
return_dirty: bool = True
"""Specifies if plates are returned dirty or clean to the plate dispenser."""
class PlateDispenser(Counter):
"""At the moment, one and only one plate dispenser must exist in an environment, because only at one place the dirty
plates should arrive.
How many plates should exist at the start of the level on the
plate dispenser is defined in the `environment_config.yml`:
```yaml
plates:
clean_plates: 1
dirty_plates: 2
plate_delay: [ 5, 10 ]
# seconds until the dirty plate arrives.
```
The character `P` in the `layout` file represents the PlateDispenser.
"""
def __init__(
self,
dispensing: ItemInfo,
plate_config: PlateConfig,
plate_transitions: dict[str, ItemInfo],
random: Random,
**kwargs,
) -> None:
super().__init__(**kwargs)
self.dispensing: ItemInfo = dispensing
"""Plate ItemInfo."""
self.occupied_by: deque = deque()
"""The queue of plates. New dirty ones are put at the end and therefore under the current plates."""
self.out_of_kitchen_timer: list[datetime] = []
"""Internal timer for how many plates are out of kitchen and how long."""
self.plate_config: PlateConfig = plate_config
"""The config how many plates are present in the kitchen at the beginning (and in total) and the config for
the random "out of kitchen" timer."""
self.next_plate_time: datetime = datetime.max
"""For efficient checking if dirty plates should be created, instead of looping through the
`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()
"""
PRE_PLATE_DISPENSER_PICK_UP = "pre_plate_dispenser_pick_up"
POST_PLATE_DISPENSER_PICK_UP = "post_plate_dispenser_pick_up"
"""
def pick_up(self, on_hands: bool = True, player: str = "0", player_holding: Item | None = None,) -> Item | None:
self.hook(
PRE_PLATE_DISPENSER_PICK_UP,
counter=self,
on_hands=on_hands,
player=player,
)
if self.occupied_by:
returned_item = self.occupied_by.pop()
self.hook(
POST_PLATE_DISPENSER_PICK_UP,
counter=self,
player=player,
on_hands=on_hands,
returned_item=returned_item,
)
return returned_item
def can_drop_off(self, item: Item) -> bool:
return not self.occupied_by or self.occupied_by[-1].can_combine(item)
def drop_off(self, item: Item, player: str = "0") -> Item | None:
self.hook(
PRE_PLATE_DISPENSER_DROP_OFF,
counter=self,
player=player,
item=item,
)
# unnecessary hooks?
if not self.occupied_by:
self.occupied_by.append(item)
self.hook(
POST_PLATE_DISPENSER_DROP_OFF,
counter=self,
player=player,
item=item,
)
elif self.occupied_by[-1].can_combine(item):
returned_item = self.occupied_by[-1].combine(item)
self.hook(
DROP_OFF_ON_COOKING_EQUIPMENT_PLATE_DISPENSER,
counter=self,
player=player,
item=item,
equipment=self.occupied_by[-1],
return_this=returned_item,
)
return returned_item
return None
def add_dirty_plate(self):
"""Add a dirty plate after a timer is completed."""
self.occupied_by.appendleft(
self.create_item(clean=not self.plate_config.return_dirty)
)
def update_plate_out_of_kitchen(self, env_time: datetime):
"""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=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}")
self.out_of_kitchen_timer.append(time_plate_to_add)
if time_plate_to_add < self.next_plate_time:
self.next_plate_time = time_plate_to_add
self.hook(PLATE_OUT_OF_KITCHEN_TIME, time_plate_to_add=time_plate_to_add)
def setup_plates(self):
"""Create plates based on the config. Clean and dirty ones."""
if self.plate_config.dirty_plates > 0:
log.info(f"Setup {self.plate_config.dirty_plates} dirty plates.")
self.occupied_by.extend(
[self.create_item() for _ in range(self.plate_config.dirty_plates)]
)
if self.plate_config.clean_plates > 0:
#log.info(f"Setup {self.plate_config.dirty_plates} clean plates.")
self.occupied_by.extend(
[
self.create_item(clean=True)
for _ in range(self.plate_config.clean_plates)
]
)
def progress(self, passed_time: timedelta, now: datetime):
"""Check if plates arrive from outside the kitchen and add a dirty plate accordingly"""
if self.next_plate_time <= now:
idx_delete = []
for i, times in enumerate(self.out_of_kitchen_timer):
if times <= now:
self.hook(DIRTY_PLATE_ARRIVES, counter=self, times=times, now=now)
idx_delete.append(i)
log.debug("Add dirty plate")
self.add_dirty_plate()
for idx in reversed(idx_delete):
self.out_of_kitchen_timer.pop(idx)
self.next_plate_time = (
min(self.out_of_kitchen_timer)
if self.out_of_kitchen_timer
else datetime.max
)
def create_item(self, clean: bool = False) -> Plate:
"""Create a plate.
Args:
clean: Whether the plate is clean or dirty.
"""
kwargs = {
"clean": clean,
"transitions": self.plate_transitions,
"item_info": self.dispensing,
"hook": self.hook,
}
return Plate(**kwargs)
class Trashcan(Counter):
"""Ingredients and content on a cooking equipment can be removed from the environment via the trash.
The character `X` in the `layout` file represents the Trashcan.
"""
def __init__(self, **kwargs):
super().__init__(**kwargs)
def pick_up(self, on_hands: bool = True, player: str = "0", player_holding: Item | None = None,) -> Item | None:
pass
def drop_off(self, item: Item, player: str = "0") -> Item | None:
if any(
e.item_info.effect_type == EffectType.Unusable for e in item.active_effects
) or any(
e.item_info.effect_type == EffectType.Unusable for e in self.active_effects
):
return item
if isinstance(item, CookingEquipment):
self.hook(TRASHCAN_USAGE, counter=self, item=item, player=player)
item.reset_content()
item.reset()
return item
self.hook(TRASHCAN_USAGE, counter=self, item=item, player=player)
return None
def can_drop_off(self, item: Item) -> bool:
return item.name != "Extinguisher"
class CookingCounter(Counter):
"""Cooking machine. Class for the stove, deep fryer, and oven.
The character depends on the cooking equipment on top of it:
```yaml
U: Stove with a pot
Q: Stove with a pan
O: Oven with a (pizza) peel
F: DeepFryer with a basket
```
"""
def __init__(
self,
name: str,
equipments: set[str],
**kwargs,
):
self.name: str = name
"""The type/name of the cooking counter, e.g., Stove, DeepFryer, Oven."""
self.equipments: set[str] = equipments
"""The valid equipment for the cooking counter, e.g., for a Stove {'Pot','Pan'}."""
super().__init__(**kwargs)
def can_drop_off(self, item: Item) -> bool:
if self.occupied_by is None:
return isinstance(item, CookingEquipment) and item.name in self.equipments
else:
return self.occupied_by.can_combine(item)
def progress(self, passed_time: timedelta, now: datetime):
"""Called by environment step function for time progression"""
if (
self.occupied_by
and isinstance(self.occupied_by, CookingEquipment)
and self.occupied_by.name in self.equipments
and self.occupied_by.can_progress()
and not any(
e.item_info.effect_type == EffectType.Unusable
for e in self.active_effects
)
):
self.occupied_by.progress(passed_time, now)
def __repr__(self):
return f"{self.name}(pos={self.pos},occupied_by={self.occupied_by})"
def to_dict(self) -> CounterState:
d = super().to_dict()
d.update((("type", self.name),))
return d
class Sink(Counter):
"""The counter in which the dirty plates can be washed to clean plates.
Needs a `SinkAddon`. The closest is calculated during initialisation, should not be seperated by each other (needs
to touch the sink).
The logic is similar to the CuttingBoard because there is no additional cooking equipment between the object to
progress and the counter. When the progress on the dirty plate is done, it is set to clean and is passed to the
`SinkAddon`.
The character `S` in the `layout` file represents the Sink.
"""
def __init__(
self,
transitions: dict[str, ItemInfo],
sink_addon: SinkAddon = None,
**kwargs,
):
super().__init__(**kwargs)
self.sink_addon: SinkAddon = sink_addon
"""The connected sink addon which will receive the clean plates"""
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) -> bool:
"""If there is a plate in the sink."""
return len(self.occupied_by) != 0
def do_hand_free_interaction(
self, passed_time: timedelta, now: datetime, player: str
):
if (
self.occupied
and self.occupied_by[-1].name in self.transition_needs
and not any(
e.item_info.effect_type == EffectType.Unusable
for e in self.active_effects
)
and not any(
e.item_info.effect_type == EffectType.Unusable
for e in self.sink_addon.active_effects
)
and not any(
e.item_info.effect_type == EffectType.Unusable
for e in self.occupied_by[-1].active_effects
)
and (
not self.sink_addon.occupied_by
or not any(
e.item_info.effect_type == EffectType.Unusable
for e in self.sink_addon.occupied_by[-1].active_effects
)
)
):
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:
plate_before = copy.deepcopy(self.occupied_by[-1])
self.occupied_by[-1].reset()
self.occupied_by[-1].name = name
plate = self.occupied_by.pop()
plate.clean = True
self.hook(PLATE_CLEANED, counter=self, player=player, plate=plate, player_before=plate_before)
self.sink_addon.add_clean_plate(plate)
break
def can_drop_off(self, item: Item) -> bool:
return isinstance(item, Plate) and not item.clean
def drop_off(self, item: Plate, player: str = "0") -> Item | None:
self.occupied_by.appendleft(item)
self.hook(ADDED_PLATE_TO_SINK, counter=self, item=item, player=player)
return None
def pick_up(self, on_hands: bool = True, player: str = "0", player_holding: Item | None = None,) -> Item | None:
return None
def set_addon(self, sink_addon: SinkAddon):
"""Set the closest addon in post_setup."""
self.sink_addon = sink_addon
class SinkAddon(Counter):
"""The counter on which the clean plates appear after cleaning them in the `Sink`
It needs to be set close to/touching the `Sink`.
The character `+` in the `layout` file represents the SinkAddon.
"""
def __init__(self, occupied_by=None, **kwargs):
super().__init__(**kwargs)
# maybe check if occupied by is already a list or deque?
self.occupied_by: deque = deque([occupied_by]) if occupied_by else deque()
"""The stack of clean plates."""
def can_drop_off(self, item: Item) -> bool:
return self.occupied_by and self.occupied_by[-1].can_combine(item)
def drop_off(self, item: Item, player: str = "0") -> Item | None:
self.hook(DROP_ON_SINK_ADDON, counter=self, item=item, occupied_by=self.occupied_by[-1], player=player)
return self.occupied_by[-1].combine(item)
def add_clean_plate(self, plate: Plate):
"""Called from the `Sink` after a plate is cleaned / the progress is complete."""
self.occupied_by.appendleft(plate)
def pick_up(self, on_hands: bool = True, player: str = "0", player_holding: Item | None = None,) -> Item | None:
if self.occupied_by:
self.hook(
PICK_UP_FROM_SINK_ADDON,
player=player,
occupied_by=self.occupied_by[-1],
counter=self,
)
return self.occupied_by.pop()