-
Fabian Heinrich authoredFabian Heinrich authored
recording.py 8.54 KiB
"""
Record events in jsonl-files.
- `USER_LOG_DIR` corresponds to the log directory of your system for programs. See [platformdirs](
https://pypi.org/project/platformdirs/) -> `user_log_dir`.
- `ROOT_DIR` to the location of the cooperative_cuisine.
- `ENV_NAME` to the name of the environment.
```yaml
hook_callbacks:
json_states:
hooks: [ json_state ]
callback_class: !!python/name:cooperative_cuisine.recording.FileRecorder ''
callback_class_kwargs:
record_path: USER_LOG_DIR/ENV_NAME/json_states.jsonl
actions:
hooks: [ pre_perform_action ]
callback_class: !!python/name:cooperative_cuisine.recording.FileRecorder ''
callback_class_kwargs:
record_path: USER_LOG_DIR/ENV_NAME/LOG_RECORD_NAME.jsonl
random_env_events:
hooks: [ order_duration_sample, plate_out_of_kitchen_time ]
callback_class: !!python/name:cooperative_cuisine.recording.FileRecorder ''
callback_class_kwargs:
record_path: USER_LOG_DIR/ENV_NAME/LOG_RECORD_NAME.jsonl
add_hook_ref: true
env_configs:
hooks: [ env_initialized, item_info_config ]
callback_class: !!python/name:cooperative_cuisine.recording.FileRecorder ''
callback_class_kwargs:
record_path: USER_LOG_DIR/ENV_NAME/LOG_RECORD_NAME.jsonl
add_hook_ref: true
```
"""
import json
import logging
import os
import sys
import traceback
from pathlib import Path
from string import Template
import yaml
from cooperative_cuisine import ROOT_DIR
from cooperative_cuisine.counters import Counter
from cooperative_cuisine.environment import Environment
from cooperative_cuisine.hooks import HookCallbackClass
from cooperative_cuisine.items import Item, Effect
from cooperative_cuisine.orders import Order
from cooperative_cuisine.utils import NumpyAndDataclassEncoder, expand_path
log = logging.getLogger(__name__)
"""The logger for this module."""
class FileRecorder(HookCallbackClass):
"""This class is responsible for recording data to a file."""
def __init__(
self,
name: str,
env: Environment,
record_path: str = "USER_LOG_DIR/ENV_NAME/LOG_RECORD_NAME.jsonl",
add_hook_ref: bool = False,
**kwargs,
):
"""Constructor of FileRecorder.
Args:
name (str): The name of the recorder. This name is used to replace the placeholder "LOG_RECORD_NAME" in the default log file path.
env (Environment): The environment in which the recorder is being used.
record_path (str, optional): The path to the log file. Defaults to "USER_LOG_DIR/ENV_NAME/LOG_RECORD_NAME.jsonl".
add_hook_ref (bool, optional): Indicates whether to add a hook reference to the recorded data. Defaults to False.
**kwargs: Additional keyword arguments.
"""
super().__init__(name, env, **kwargs)
self.add_hook_ref: bool = add_hook_ref
"""Indicates whether to add a hook reference to the recorded data. Default value is False."""
record_path = record_path.replace("LOG_RECORD_NAME", name)
record_path = Path(expand_path(record_path, env_name=env.env_name))
self.record_path: Path = record_path
"""The path to the log file. Default value is "USER_LOG_DIR/ENV_NAME/LOG_RECORD_NAME.jsonl"."""
log.info(f"Recorder record for {name} in file://{record_path}")
os.makedirs(record_path.parent, exist_ok=True)
def __call__(self, hook_ref: str, env: Environment, **kwargs):
for key, item in kwargs.items():
if isinstance(item, (Counter, Item, Effect)):
kwargs[key] = item.to_dict()
elif isinstance(item, Order):
order_state = {
"id": item.uuid,
"category": "Order",
"meal": item.meal.name,
"start_time": item.start_time.isoformat(),
"max_duration": item.max_duration.total_seconds(),
}
kwargs[key] = order_state
elif isinstance(item, list):
new_list = []
for i in item:
if isinstance(i, Order):
new_list.append(
{
"id": i.uuid,
"category": "Order",
"meal": i.meal.name,
"start_time": i.start_time.isoformat(),
"max_duration": i.max_duration.total_seconds(),
}
)
else:
new_list.append(i.to_dict())
kwargs[key] = new_list
try:
record = (
json.dumps(
{
"env_time": env.env_time.isoformat(),
**({"hook_ref": hook_ref} if self.add_hook_ref else {}),
**kwargs,
},
cls=NumpyAndDataclassEncoder,
)
+ "\n"
)
with open(self.record_path, "a") as record_file:
record_file.write(record)
except TypeError as e:
traceback.print_exception(e)
log.info(
f"Not JSON serializable Record, hook: {hook_ref}, kwargs: {kwargs}"
)
def print_recorded_events_human_readable(jsonl_path: Path):
"""This function prints a game_event recording in human-readable form.
Args:
jsonl_path: Path to the file with recorded game events.
"""
def stringify_item(item_) -> str | None:
if isinstance(item_, float):
return str(item_)
if isinstance(item_, str):
return item_
if isinstance(item_, list) and len(item_) == 1:
item_ = item_[0]
if item_:
content = None
if item_ and item_["type"] in [
"Plate",
"Pot",
"Basket",
"Peel",
"Pan",
]:
if item_ != "Plate" and item_["content_ready"]:
content_ready = stringify_item(item_["content_ready"])
return f"{item_['type']}{f'({content_ready})'}"
content = [stringify_item(i) for i in item_["content_list"]]
if not content:
content = "None"
return f"{item_['type']}{f'({content})' if content else ''}"
else:
return None
with open(ROOT_DIR / "configs" / "human_readable_print_templates.yaml", "r") as f:
string_templates = yaml.safe_load(f)
column_size = 20
with open(jsonl_path, "r") as jsonl_file:
for line in jsonl_file:
record = json.loads(line)
hook = record["hook_ref"]
for dict_key in record.keys():
match dict_key:
case "hook_ref" | "env_time" | "on_hands":
pass
case "counter":
if not record[dict_key]:
record[dict_key] = "None"
else:
occupied = record[dict_key]["occupied_by"]
if occupied:
if isinstance(occupied, list):
occupied = [stringify_item(i) for i in occupied]
else:
occupied = stringify_item(occupied)
record[dict_key] = f"{record[dict_key]['type']}({occupied})"
case "order":
order = f"Order({record[dict_key]['meal']})"
record[dict_key] = order
case "new_orders":
orders = [f"Order({o['meal']})" for o in record[dict_key]]
record[dict_key] = orders
case _:
try:
record[dict_key] = stringify_item(record[dict_key])
except (KeyError, TypeError) as e:
print(hook, dict_key, record[dict_key], type(e), e)
if hook in string_templates.keys():
string_template = Template(string_templates[hook])
print(string_template.substitute(**dict(record.items())))
else:
print(hook)
for key, item in record.items():
print(f" - {(key+':').ljust(column_size)}{item}")
if __name__ == "__main__":
json_lines_path: Path = Path(sys.argv[1])
print_recorded_events_human_readable(json_lines_path)