Newer
Older
"""
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
ITEM_INFO_LOADED = "item_info_load"
"""Called after the item info is loaded and stored in the env attribute `item_info`. The kwargs are the passed config
(`item_info`) to the environment from which it was loaded and if it is a file path or the config string (`as_files`)"""
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"
ENV_INITIALIZED = "env_initialized"
"""At the end of the __init__ method. No additional kwargs. Everything is stored in the env."""
PRE_PERFORM_ACTION = "pre_perform_action"
"""Before an action is performed / entered into the environment. `action` kwarg with the entered action."""
POST_PERFORM_ACTION = "post_perform_action"
"""After an action is performed / entered into the environment. `action` kwarg with the entered action."""
# TODO Pre and Post Perform Movement
PLAYER_ADDED = "player_added"
"""Called after a player has been added. Kwargs: `player` and `pos`."""
GAME_ENDED_STEP = "game_ended_step"
PRE_STATE = "pre_state"
PRE_STEP = "pre_step"
POST_STEP = "post_step"
STATE_DICT = "state_dict"
JSON_STATE = "json_state"
PRE_RESET_ENV_TIME = "pre_reset_env_time"
POST_RESET_ENV_TIME = "post_reset_env_time"
# -----
PRE_COUNTER_PICK_UP = "pre_counter_pick_up"
POST_COUNTER_PICK_UP = "post_counter_pick_up"
PRE_COUNTER_DROP_OFF = "pre_counter_drop_off"
POST_COUNTER_DROP_OFF = "post_counter_drop_off"
PRE_DISPENSER_PICK_UP = "pre_dispenser_pick_up"
POST_DISPENSER_PICK_UP = "post_dispenser_pick_up"
DROP_OFF_ON_COOKING_EQUIPMENT = "drop_off_on_cooking_equipment"
PRE_PLATE_DISPENSER_PICK_UP = "pre_plate_dispenser_pick_up"
POST_PLATE_DISPENSER_PICK_UP = "post_plate_dispenser_pick_up"
PRE_PLATE_DISPENSER_DROP_OFF = "pre_plate_dispenser_drop_off"
POST_PLATE_DISPENSER_DROP_OFF = "post_plate_dispenser_drop_off"
CUTTING_BOARD_PROGRESS = "cutting_board_progress"
CUTTING_BOARD_100 = "cutting_board_100"
PROGRESS_STARTED = "progress_started"
PROGRESS_FINISHED = "progress_finished"
PRE_SERVING = "pre_serving"
POST_SERVING = "post_serving"
NO_SERVING = "no_serving"
PLATE_OUT_OF_KITCHEN_TIME = "plate_out_of_kitchen_time"
DIRTY_PLATE_ARRIVES = "dirty_plate_arrives"
TRASHCAN_USAGE = "trashcan_usage"
ADDED_PLATE_TO_SINK = "added_plate_to_sink"
DROP_ON_SINK_ADDON = "drop_on_sink_addon"
PICK_UP_FROM_SINK_ADDON = "pick_up_from_sink_addon"
PLATE_CLEANED = "plate_cleaned"
SERVE_NOT_ORDERED_MEAL = "serve_not_ordered_meal"
SERVE_WITHOUT_PLATE = "serve_without_plate"
ORDER_DURATION_SAMPLE = "order_duration_sample"
COMPLETED_ORDER = "completed_order"
INIT_ORDERS = "init_orders"
NEW_ORDERS = "new_orders"
ORDER_EXPIRED = "order_expired"
ACTION_ON_NOT_REACHABLE_COUNTER = "action_on_not_reachable_counter"
ACTION_PUT = "action_put"
ACTION_INTERACT_START = "action_interact_start"
ITEM_BURNED = "item_burned" # MISSING
NEW_FIRE = "new_fire"
FIRE_SPREADING = "fire_spreading"
PLAYERS_COLLIDE = "players_collide"
PLAYER_START_INTERACT = "player_start_interaction"
PLAYER_END_INTERACT = "player_end_interact"
ON_ITEM_TRANSITION = "on_item_transition"
CONTENT_READY = "content_ready"
"""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):
self.hooks = defaultdict(list)
self.env = env
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!",
),
)