utils.py

import logging
import os
import sys
import uuid
from datetime import datetime
from enum import Enum

from overcooked_simulator import ROOT_DIR


def create_init_env_time():
    """Init time of the environment time, because all environments should have the same internal time."""
    return datetime(
        year=2000, month=1, day=1, hour=0, minute=0, second=0, microsecond=0
    )


def custom_asdict_factory(data):
    def convert_value(obj):
        if isinstance(obj, Enum):
            return obj.value
        return obj

    return dict((k, convert_value(v)) for k, v in data)


def setup_logging(enable_websocket_logging=False):
    path_logs = ROOT_DIR.parent / "logs"
    os.makedirs(path_logs, exist_ok=True)
    logging.basicConfig(
        level=logging.DEBUG,
        format="%(asctime)s %(levelname)-8s %(name)-50s %(message)s",
        handlers=[
            logging.FileHandler(
                path_logs / f"{datetime.now().strftime('%Y-%m-%d_%H-%M-%S')}_debug.log",
                encoding="utf-8",
            ),
            logging.StreamHandler(sys.stdout),
        ],
    )
    logging.getLogger("matplotlib").setLevel(logging.WARNING)
    if not enable_websocket_logging:
        logging.getLogger("asyncio").setLevel(logging.ERROR)
        logging.getLogger("asyncio.coroutines").setLevel(logging.ERROR)
        logging.getLogger("websockets.server").setLevel(logging.ERROR)
        logging.getLogger("websockets.protocol").setLevel(logging.ERROR)
        logging.getLogger("websockets.client").setLevel(logging.ERROR)


def url_and_port_arguments(parser):
    parser.add_argument(
        "-url",
        "--url",
        "--host",
        type=str,
        default="localhost",
        help="Overcooked game server host url.",
    )
    parser.add_argument(
        "-p",
        "--port",
        type=int,
        default=8000,
        help="Port number for the game engine server",
    )


def disable_websocket_logging_arguments(parser):
    parser.add_argument(
        "--enable-websocket-logging" "", action="store_true", default=True
    )


def add_list_of_manager_ids_arguments(parser):
    parser.add_argument(
        "-m",
        "--manager_ids",
        nargs="+",
        type=str,
        default=[uuid.uuid4().hex],
        help="List of manager IDs that can create environments.",
    )