-
Florian Schröder authored
The version in setup.py has been updated from 0.1.0 to 1.0.0. Additionally, the callback for sink interaction has been removed from hooks.py. The release date of version 1.0.0 has also been confirmed in the CHANGELOG.
Florian Schröder authoredThe version in setup.py has been updated from 0.1.0 to 1.0.0. Additionally, the callback for sink interaction has been removed from hooks.py. The release date of version 1.0.0 has also been confirmed in the CHANGELOG.
hooks.py 9.74 KiB
"""
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 `extra_setup_functions` 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"
class Hooks:
"""
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)
Initializes a new instance of Hooks.
Args:
env (any): The environment variable to be stored in the env attribute.
__call__(self, hook_ref, **kwargs)
Invokes the callbacks associated with the specified hook reference.
Args:
hook_ref (str): The hook reference to trigger the callbacks for.
**kwargs: Additional keyword arguments to be passed to the callbacks.
"""
def __init__(self, env):
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):
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):
self.name = name
"""The name of the callback."""
self.env = 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],
):
"""Function to reference in the `environment_config.yml` to add functionality via hooks and a configured 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!",
),
)