Newer
Older
from collections import defaultdict
from functools import partial
from typing import Callable
# TODO add player_id as kwarg to all hooks -> pass player id to all methods
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."""
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"
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
"""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_name` and `pos`."""
GAME_ENDED_STEP = "game_ended_step"
PRE_STATE = "pre_state"
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 = "dispenser_pick_up"
POST_DISPENSER_PICK_UP = "dispenser_pick_up"
CUTTING_BOARD_PROGRESS = "cutting_board_progress"
CUTTING_BOARD_100 = "cutting_board_100"
CUTTING_BOARD_START_INTERACT = "cutting_board_start_interaction"
CUTTING_BOARD_END_INTERACT = "cutting_board_end_interact"
PRE_SERVING = "pre_serving"
POST_SERVING = "post_serving"
NO_SERVING = "no_serving"
# TODO drop off
DIRTY_PLATE_ARRIVES = "dirty_plate_arrives"
TRASHCAN_USAGE = "trashcan_usage"
PLATE_CLEANED = "plate_cleaned"
SINK_START_INTERACT = "sink_start_interact"
SINK_END_INTERACT = "sink_end_interact"
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"
SERVE_NOT_ORDERED_MEAL = "serve_not_ordered_meal"
SERVE_WITHOUT_PLATE = "serve_without_plate"
COMPLETED_ORDER = "completed_order"
INIT_ORDERS = "init_orders"
NEW_ORDERS = "new_orders"
ACTION_ON_NOT_REACHABLE_COUNTER = "action_on_not_reachable_counter"
ACTION_PUT = "action_put"
ACTION_INTERACT_START = "action_interact_start"
class Hooks:
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):
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)
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def add_dummy_callbacks(env):
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(
SINK_START_INTERACT,
partial(
print_hook_callback,
text="You started to use the Sink!",
),
)
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!",
),
)