""" You can add callbacks at specific points in the environment. This "hook" mechanism is defined here. You can add hooks via the `environment_config` under `hook_callbacks` and the here defined `hooks_via_callback_class` function. Each hook get different kwargs. But `env` with the environment object and `hook_ref` with the name of the hook are always present. # Code Documentation """ from __future__ import annotations from abc import abstractmethod from collections import defaultdict from functools import partial from typing import Callable, Any, TYPE_CHECKING, Type if TYPE_CHECKING: from cooperative_cuisine.environment import Environment # --- environment.py --- ITEM_INFO_LOADED = "item_info_load" """Called after the item info is loaded and stored in the env attribute `item_info`. Args: item_info (str): the string content of the item info config file. parsed_item_info (dict[str, ItemInfo]): the parsed item info. """ LAYOUT_FILE_PARSED = "layout_file_parsed" """After the layout file was parsed. No additional kwargs. Everything is stored in the env.""" ITEM_INFO_CONFIG = "item_info_config" """Called before the item info is parsed. Args: item_info_config (str): the string content of the item info config file. """ ENV_INITIALIZED = "env_initialized" """At the end of the __init__ method. Args: environment_config (str): the str content of the environment config file. layout_config (str): the str content of the layout config file. seed (int): the random seed. env_start_time_worldtime (datetime): the "real" world time when the hook was called. """ PRE_PERFORM_ACTION = "pre_perform_action" """Before an action is performed / entered into the environment. `action` kwarg with the entered action. Args: action (Action): the action that will be performed. """ POST_PERFORM_ACTION = "post_perform_action" """After an action is performed / entered into the environment. `action` kwarg with the entered action. Args: action (Action): the action that was performed. """ # TODO Pre and Post Perform Movement PLAYER_ADDED = "player_added" """Called after a player has been added. Kwargs: `player` and `pos`. Args: player (Player): the new player object that was created. pos (npt.NDArray[float]): the 2d position of the player. """ GAME_ENDED_STEP = "game_ended_step" """When the step function is called but the game ended already. No kwargs.""" PRE_STATE = "pre_state" """Called before the state is calculated for a player. Args: player_id (str): The id of the player for which the state is calculated. """ PRE_STEP = "pre_step" """When the step function is called before the execution happens. Currently not active (Optimization). Args: passed_time (timedelta): the time elapsed since the last step call. """ POST_STEP = "post_step" """When the step function is called after the execution. Args: passed_time (timedelta): the time elapsed since the last step call. """ STATE_DICT = "state_dict" """When the dictionary (containing the state) is generated. Args: state (dict): the current state for a player. Should follow the `StateRepresentation`. player_id (str): the id of the player. """ JSON_STATE = "json_state" """When the json string was generated for a state. Args: json_data (str): the json string containing the state. player_id (str): the id of the player. """ PRE_RESET_ENV_TIME = "pre_reset_env_time" """Before the time of the environment is reset. Args: env_time (datetime): the env time before the reset. """ POST_RESET_ENV_TIME = "post_reset_env_time" """After the time of the environment is reset. Args: env_time (datetime): the env time after the reset. """ # --- counters.py --- PRE_COUNTER_PICK_UP = "pre_counter_pick_up" """Before the pick up on a counter is executed. Args: counter (Counter): the counter from which to pick up. on_hands (bool): if the picked up item is put on the free player hands (True) or on a cooking equipment (False). player (str): player id. """ POST_COUNTER_PICK_UP = "post_counter_pick_up" """After the pick up on a counter is executed. Args: counter (Counter): the counter from which to pick up. on_hands (bool): if the picked up item is put on the free player hands (True) or on a cooking equipment (False). return_this (Item | None): what will be put on the player (holding) player (str): player id. player_holding: (Item | None): the item that the player is holding. """ PRE_DISPENSER_PICK_UP = "pre_dispenser_pick_up" """Before the pick up on a dispenser happens. `PRE_COUNTER_PICK_UP` is not called. Args: counter (Counter): the counter(dispenser) from which to pick up. on_hands (bool): if the picked up item is put on the free player hands (True) or on a cooking equipment (False). player (str): player id. """ POST_DISPENSER_PICK_UP = "post_dispenser_pick_up" """After the new item on a dispenser is generated when a pick up happens. Args: counter (Counter): the counter/dispenser from which to pick up. on_hands (bool): if the picked up item is put on the free player hands (True) or on a cooking equipment (False). return_this (Item | None): what will be put on the player (holding) - the new item / the item that was on the counter. player (str): player id. player_holding (Item): What the player was holding. """ PRE_PLATE_DISPENSER_PICK_UP = "pre_plate_dispenser_pick_up" """Before the pick up on a plate dispenser happens. `PRE_COUNTER_PICK_UP` and `PRE_DISPENSER_PICK_UP` is not called. Args: counter (Counter): the plate dispenser from which to pick up. on_hands (bool): if the picked up item is put on the free player hands (True) or on a cooking equipment (False). player (str): player id. """ POST_PLATE_DISPENSER_PICK_UP = "post_plate_dispenser_pick_up" """After the pick up from a plate dispenser is executed. Args: counter (Counter): the counter from which to pick up. player (str): player id. on_hands (bool): if the picked up item is put on the free player hands (True) or on a cooking equipment (False). returned_item (Item | None): what will be put on the player (holding) """ PRE_COUNTER_DROP_OFF = "pre_counter_drop_off" """Before the drop off on a counter is executed. Args: item (Item): the item to drop on the counter. equipment (Item | None | deque[Item]): what is currently on the counter. counter (Counter): the counter from which to drop off. player (str): player id. """ POST_COUNTER_DROP_OFF = "post_counter_drop_off" """Drop off on a free counter (nothing on it). Args: counter (Counter): the counter from which to drop off. player (str): player id. item (Item): the item to drop on the counter. """ DROP_OFF_ON_COOKING_EQUIPMENT = "drop_off_on_cooking_equipment" """When drop off on a counter with a cooking equipment is on it. Before the combining happens. Args: item (Item): the item to drop on the counter. equipment (Item | None | deque[Item]): what is currently on the counter. counter (Counter): the counter from which to drop off. occupied_before (Item): What was on the counter before the drop off. player (str): player id. return_this (Item | None): what will be put on the player (holding) """ PICK_UP_ON_COOKING_EQUIPMENT = "pick_up_on_cooking_equipment" """When pick up from a counter when the player is holding a cooking equipment. Before the combining happens. Args: return_this (Item): the item returned from the counter. occupied_by (Item): What is now in the counter. counter (Counter): the counter from which to pick up. player (str): player id. player_holding (Item): What the player was holding. """ PRE_PLATE_DISPENSER_DROP_OFF = "pre_plate_dispenser_drop_off" """Before something is dropped on a plate dispenser. Args: counter (Counter): the plate dispenser from which to drop off. player (str): player id. item (Item): the item to drop on the counter. """ POST_PLATE_DISPENSER_DROP_OFF = "post_plate_dispenser_drop_off" """Something is dropped on a free plate dispenser. Args: counter (Counter): the counter from which to drop off. player (str): player id. item (Item): the item to drop on the counter. """ DROP_OFF_ON_COOKING_EQUIPMENT_PLATE_DISPENSER = ( "drop_off_on_cooking_equipment_plate_dispenser" ) """After something is dropped on a plate dispenser and is combined with the top plate. Args: counter (Counter): the counter from which to drop off. player (str): player id. item (Item): the item to drop on the counter. equipment (Item | None | deque[Item]): the cooking equipment that combined the items. return_this (Item | None): what will be put on the player (holding) """ DISPENSER_ITEM_RETURNED = "dispenser_item_returned" """Undo the pickup on a dispenser. (Drop off action.) Args: counter (Counter): the dispenser. player (str): player id. item (Item): the item that is returned. """ CUTTING_BOARD_PROGRESS = "cutting_board_progress" """Valid cutting board interaction step. Args: counter (Counter): the cutting board which is used to chop the ingredient. player (str): player id. percent (float): how much percent is added in the progress call. passed_time (timedelta): passed time since the last step call in the environment- """ CUTTING_BOARD_100 = "cutting_board_100" """Chopping finished on a cutting board. Args: counter (Counter): the cutting board. item (Item): the item that was chopped. player (str): player id. before: (Item): What the item was before. """ # --- items.py --- PROGRESS_STARTED = "progress_started" """Progress on a cooking equipment is started. Args: item (CookingEquipment): the cooking equipment that does the progress. """ PROGRESS_FINISHED = "progress_finished" """Progress on a cooking equipment is finished. (Does not include fire.) Args: before (CookingEquipment): the cooking equipment before the progress. item (CookingEquipment): the cooking equipment after the progress. """ PRE_SERVING = "pre_serving" """Called if player wants to drop off on `ServingWindow`. Args: counter (ServingWindow): the serving window the player wants to serve on. item (Item): the item to serve (on the player's hand). env_time (datetime): current env time. player (str): the player id. """ POST_SERVING = "post_serving" """Serving was successful. Args: counter (ServingWindow): the serving window on which the item was served on. item (Item): the item that was served. (Plate) env_time (datetime): current env time. player (str): the player id. """ NO_SERVING = "no_serving" """Serving was not successful. (Rejected from the order manager) Args: counter (ServingWindow): the serving window the player wanted to serve on. item (Item): the item that cannot be serves (now still on the player's hand). env_time (datetime): current env time. player (str): the player id. """ SCORE_CHANGED = "score_changed" """The score was changed. Args: increase (float): the increase of the score. score (float): the new (increased) score. info (str): additional info. """ PLATE_OUT_OF_KITCHEN_TIME = "plate_out_of_kitchen_time" """A plate is out of the kitchen (after successful serving). Args: time_plate_to_add (datetime): the time the plate will arrive back in the kitchen """ DIRTY_PLATE_ARRIVES = "dirty_plate_arrives" """A plate arrived at the kitchen (some time after serving) Args: counter (Counter): """ TRASHCAN_USAGE = "trashcan_usage" """Successful usage of the trashcan. Args: counter (Trashcan): the trashcan used. item (Item): the item the player holding. (If it is a `CookingEquipmeent` only content_list is removed). player (str): the player id. """ ADDED_PLATE_TO_SINK = "added_plate_to_sink" """Dirty Plate dropped on the sink. Args: counter (Sink): the sink, target of the drop off. item (Plate): the dirty plate. player (str): the player id. """ DROP_ON_SINK_ADDON = "drop_on_sink_addon" """Something is dropped on the sink addon and combined with the top plate. Args: counter (SinkAddon): the target counter of the drop off. item (Item): the item which is combined with the top plate. occupied_by: (Item | None): What the sink addon is occupied by. player (str): the player id. """ PICK_UP_FROM_SINK_ADDON = "pick_up_from_sink_addon" """Player picks up the top plate. Args: player (str): the player id. occupied_by (Plate): the top plate that is picked up. counter (SinkAddon): the counter class from it picked up. """ PLATE_CLEANED = "plate_cleaned" """The player finished cleaning a plate. Plate is transferred to the SinkAddon afterwards. Args: counter (Sink): The sink on which the plate was cleaned. player (str): the player id. plate (Item): the plate that was cleaned. plate_before (Item): the plate before cleaning. """ # --- items.py --- ON_ITEM_TRANSITION = "on_item_transition" """A new Effect appears (Fire starts). Args: item (CookingEquipment): the pot, pan, ... on which the fire starts. result (Effect): the fire effect that is now active on the equipment. seconds (float/int): how long it took to create the fire. """ CONTENT_READY = "content_ready" """A meal is ready on a cooking equipment. Args: before (CookingEquipment): the cooking equipment. result (str): Name of the meal. """ PLATED_MEAL = "plated_meal" """A ready meal was plated on a Plate""" COOKING_FINISHED = "cooking_finished" # --- orders.py --- SERVE_NOT_ORDERED_MEAL = "serve_not_ordered_meal" """If a meal does not match to an order but `serve_not_ordered_meals`is active. Args: meal (Item): the meal that is served. meal_name (str): equivalent to meal.name. The name of the meal. player (str): the player id. """ SERVE_WITHOUT_PLATE = "serve_without_plate" """Somehow tried to serve without a plate. Will not be served. Args: item (Item): The item that the player has in his hands. player (str): the player id. """ COMPLETED_ORDER = "completed_order" """A meal was served that completed an order. Args: order (Order): the order that was fulfilled. meal (Item): The meal that was served. item (Item): The plate that was used to serve the meal (with content). relative_order_time (timedelta): the time that the player needed to fulfill the order. remaining_time_ratio (float): the ratio of the remaining time of the order relative to order duration. meal_name (str): name of the meal. player (str): the player id. """ INIT_ORDERS = "init_orders" """The initial orders were generated. Args: init_orders (list[Order]): the init orders. """ NEW_ORDERS = "new_orders" """New orders (not init orders) were generated. Args: new_orders (list[Order]): the new orders. """ ORDER_EXPIRED = "order_expired" """An order expired (took too long). Args: order (Order): the order that expired. """ # --- environment.py --- but player related. ACTION_ON_NOT_REACHABLE_COUNTER = "action_on_not_reachable_counter" """Player wants to interact or pick/drop but no counter is in reach. Args: action (Action): the action with the player id. counter (Counter): closest but not reachable counter. player (str): the player id. """ ACTION_PUT = "action_put" """Player does the put (pick or drop) action Args: action (Action): the action with the player id. counter (Counter): on which the put action is performed. player (str): the player id. """ ACTION_INTERACT_START = "action_interact_start" """Player starts interacting on a counter. Args: action (Action): the action with the player id counter (Counter): on which the player starts the interaction. player (str): the player id. """ # --- effects.py --- ITEM_BURNED = "item_burned" # MISSING """NOT IMPLEMENTED""" NEW_FIRE = "new_fire" """New fire from cooking equipment (Does not include spreading). Args: target (Counter|Item): on which the fire was created. """ FIRE_EXTINGUISHED = "fire_extinguished" """Fire was extinguished. Args: target (Counter|Item): on which the fire was extinguished. player (str): the player id. """ FIRE_SPREADING = "fire_spreading" """Fire spreads on other counter/items after some time. Args: target (Counter|Item): on which the fire is spreading. """ # --- movement.py --- PLAYERS_COLLIDE = "players_collide" """Every frame player collides. Args: collisions (npt.NDArray[bool]): the players which collide. """ # --- players.py --- PLAYER_START_INTERACT = "player_start_interaction" """Same as `ACTION_INTERACT_START` but in the player.py Args: player (str): the player id. counter (Counter): the last interacted counter. item (Item): the item that the player is holding. """ PLAYER_END_INTERACT = "player_end_interact" """The interaction with a counter stopped. Either stopped by the player through button press or move away. Args: player (str): the player id. counter (Counter): the last interacted counter. item (Item): the item that the player is holding. """ # -- extra -- ADDITIONAL_STATE_UPDATE = "additional_state_update" """Update of the additional content of the state. Args: update (dict[str, Any]): update of the additional state content. """ class Hooks: """Represents a collection of hooks and provides methods to register callbacks for hooks and invoke the callbacks when hooks are triggered. Attributes: hooks (defaultdict[list]): A defaultdict containing lists of callbacks for each hook reference. env (any): The environment variable passed to the Hooks instance. Methods: __init__(self, env: Environment) Initializes a new instance of Hooks. __call__(self, hook_ref: str, **kwargs) Invokes the callbacks associated with the specified hook reference. """ def __init__(self, env: Environment): """Constructor for the Hooks object. Args: env (Environment): The environment object to be referenced. """ self.hooks: dict[str, list[Callable]] = defaultdict(list) """The hook callbacks per hook_ref.""" self.env: Environment = env """Reference to the environment object.""" def __call__(self, hook_ref, **kwargs): for callback in self.hooks[hook_ref]: callback(hook_ref=hook_ref, env=self.env, **kwargs) def register_callback(self, hook_ref: str | list[str], callback: Callable): """Register a callback for a hook which is called when the hook is touched during execution. Args: hook_ref: A string or a list of strings representing the reference(s) of the hook(s) to register the callback for. callback: A callable object (function or method) to be registered as a callback. """ if isinstance(hook_ref, (tuple, list, set)): # TODO check for iterable for ref in hook_ref: self.hooks[ref].append(callback) else: self.hooks[hook_ref].append(callback) def print_hook_callback(text, env, **kwargs): """Dummy hook callback. Print env time and hook ref.""" #print(env.env_time, text) class HookCallbackClass: """ Class: HookCallbackClass Represents a callback class for hook events. Attributes: - name: A string representing the name of the callback. - env: An Environment object representing the environment in which the callback is being executed. Methods: - __init__(self, name: str, env: Environment, **kwargs): Initializes a new instance of HookCallbackClass. - __call__(self, hook_ref: str, env: Environment, **kwargs): Abstract method that executes the callback logic when called. Note: - This class is meant to be subclassed and the __call__ method implemented according to specific needs. - The **kwargs parameter allows for additional arguments to be passed to the callback function. Usage Example: ```python # Create an instance of HookCallbackClass callback = HookCallbackClass("my_callback", my_env) # Subclass HookCallbackClass and implement the __call__ method class MyCallback(HookCallbackClass): def __call__(self, hook_ref: str, env: Environment, **kwargs): # Add custom callback logic here # Create an instance of the subclass my_callback = MyCallback("my_callback", my_env) # Call the callback my_callback("hook_reference", my_env) ``` """ def __init__(self, name: str, env: Environment, **kwargs): """Constructor of HookCallbackClass. Args: name: A string representing the name of the callback. env: An instance of the Environment class representing the reference to the environment. **kwargs: Additional keyword arguments. """ self.name: str = name """The name of the callback.""" self.env: Environment = env """Reference to the environment.""" @abstractmethod def __call__(self, hook_ref: str, env: Environment, **kwargs): ... def hooks_via_callback_class( name: str, env: Environment, hooks: list[str], callback_class: Type[HookCallbackClass], callback_class_kwargs: dict[str, Any], ): """Setup hook callback class. Args: name: A string representing the name of the callback class instance. env: An instance of the Environment class. hooks: A list of strings representing the hooks for which the callback class instance needs to be registered. callback_class: A type representing the class of the callback instance to be created. callback_class_kwargs: A dictionary containing additional keyword arguments to be passed to the callback class constructor. """ recorder = callback_class(name=name, env=env, **callback_class_kwargs) for hook in hooks: env.register_callback_for_hook(hook, recorder) def add_dummy_callbacks(env): """Checking the hooks-callback functionality. Args: env: The environment object that represents the system environment. This method adds dummy callbacks to the given environment object. Each callback is registered for a specific hook using the `register_callback_for_hook` method of the environment. The callbacks are defined using the `partial` function from the `functools` module. This allows us to pass additional arguments to the callback while registering it. The `print_hook *_callback` function is used as the callback function, and it prints a message to the console. Here are the hooks and corresponding messages that are registered: 1. SERVE_NOT_ORDERED_MEAL: Prints the message "You tried to serve a meal that was not ordered!" 2. SINK_START_INTERACT: Prints the message "You started to use the Sink!" 3. COMPLETED_ORDER: Prints the message "You completed an order!" 4. TRASHCAN_USAGE: Prints the message "You used the trashcan!" These dummy callbacks can be used for testing or demonstration purposes. """ env.register_callback_for_hook( SERVE_NOT_ORDERED_MEAL, partial( print_hook_callback, text="You tried to served a meal that was not ordered!", ), ) env.register_callback_for_hook( COMPLETED_ORDER, partial( print_hook_callback, text="You completed an order!", ), ) env.register_callback_for_hook( TRASHCAN_USAGE, partial( print_hook_callback, text="You used the trashcan!", ), )