gameboy_worlds.utils.fundamental

  1# This file contains all the fundamental utilities that do not rely on any other file.
  2import os
  3import logging
  4import importlib.util
  5from typing import Dict
  6
  7
  8class RelativePathFormatter(logging.Formatter):
  9    def format(self, record):
 10        # record.pathname is the full system path
 11        path = record.pathname
 12
 13        # Look for 'src' and keep everything after it
 14        if "src" in path:
 15            # Splits at 'src', takes the last part, and removes leading slashes
 16            record.custom_path = "gameboy_worlds" + path.split("gameboy_worlds")[-1]
 17        else:
 18            # Fallback to just the filename if 'src' isn't found
 19            record.custom_path = record.filename
 20
 21        return super().format(record)
 22
 23
 24def get_logger(
 25    level: int = logging.INFO, filename: str = None, add_console: bool = True
 26) -> logging.Logger:
 27    """
 28    Sets up and returns a logger with specified configurations.
 29    Args:
 30        level (int, optional): Logging level. Defaults to logging.INFO.
 31        filename (str, optional): If provided, logs will be written to this file. Defaults to None.
 32        add_console (bool, optional): If True, logs will also be printed to the console. Defaults to True.
 33    Returns:
 34        logging.Logger: Configured logger instance.
 35    """
 36    fmt_str = "%(asctime)s, [%(levelname)s, %(custom_path)s:%(lineno)d] %(message)s"
 37    # Note: deliberately not passing format=fmt_str here — fmt_str requires
 38    # record.custom_path, which only RelativePathFormatter (below) sets. The
 39    # root logger's handler uses a plain Formatter, so giving it fmt_str
 40    # would raise a KeyError whenever another logger (e.g. asyncio) propagates
 41    # a record to root.
 42    logging.basicConfig()
 43    logger = logging.getLogger("GameBoyWorlds-Server")
 44    if add_console:
 45        logger.handlers.clear()
 46        console_handler = logging.StreamHandler()
 47        log_formatter = RelativePathFormatter(fmt_str)
 48        console_handler.setFormatter(log_formatter)
 49        logger.addHandler(console_handler)
 50    if filename is not None:
 51        file_handler = logging.FileHandler(filename, mode="a")
 52        log_formatter = RelativePathFormatter(fmt_str)
 53        file_handler.setFormatter(log_formatter)
 54        logger.addHandler(file_handler)
 55    if level is not None:
 56        logger.setLevel(level)
 57        logger.propagate = False
 58    return logger
 59
 60
 61def meta_dict_to_str(
 62    meta_dict: dict,
 63    *,
 64    print_mode: bool = False,
 65    n_indents: int = 1,
 66    skip_write_timestamp: bool = True,
 67) -> str:
 68    """
 69    Converts a metadata dictionary to a string representation.
 70    Args:
 71        meta_dict (dict): The metadata dictionary to convert.
 72        print_mode (bool, optional): If True, formats the string for printing with indentation. Defaults to False.
 73        n_indents (int, optional): Number of indentation levels for print mode. Defaults to 1.
 74        skip_write_timestamp (bool, optional): If True, skips the 'write_timestamp' key in non-print mode. Defaults to True.
 75    Returns:
 76        str: String representation of the metadata dictionary.
 77    """
 78    keys = list(meta_dict.keys())
 79    # error out if None is a key
 80    if None in keys:
 81        raise ValueError("None cannot be a key in meta_dict")
 82    keys.sort()
 83    meta_str = ""
 84    for key in keys:
 85        if print_mode:
 86            indent = "\t" * n_indents
 87            element = meta_dict[key]
 88            element_str = None
 89            if isinstance(element, dict):
 90                element_str = "\n" + meta_dict_to_str(
 91                    element, print_mode=True, n_indents=n_indents + 1
 92                )
 93            else:
 94                element_str = str(element)
 95            meta_str += f"{indent}{key}: {element_str}\n"
 96        else:
 97            if skip_write_timestamp and key == "write_timestamp":
 98                continue
 99            meta_str += f"{key.lower().strip()}_{str(meta_dict[key]).lower().strip()}"
100    return meta_str
101
102
103def logger_print_dict(logger: logging.Logger, meta_dict: dict, n_indents: int = 1):
104    """
105    Logs the string representation of a metadata dictionary using the provided logger.
106    Args:
107        logger (logging.Logger): The logger to use for logging.
108        meta_dict (dict): The metadata dictionary to log.
109        n_indents (int, optional): Number of indentation levels for formatting. Defaults to 1.
110    """
111    meta_dict_str = meta_dict_to_str(
112        meta_dict, print_mode=True, n_indents=n_indents, skip_write_timestamp=False
113    )
114    logger.info(meta_dict_str)
115
116
117def file_makedir(file_path: str):
118    """
119    Ensures that the directory for the given file path exists. If not, it creates the necessary directories.
120    Args:
121        file_path (str): The file path for which to ensure the directory exists.
122    """
123    dirname = os.path.dirname(file_path)
124    if dirname != "" and not os.path.exists(dirname):
125        os.makedirs(dirname)
126    return
127
128
129def module_installed(name: str) -> bool:
130    """
131    Checks if a module with the given name is installed.
132    Args:
133        name (str): The name of the module to check.
134    Returns:
135        bool: True if the module is installed, False otherwise.
136    """
137    spec = importlib.util.find_spec(name)
138    return spec is not None
139
140
141def check_optional_installs(warn=False) -> Dict[str, bool]:
142    """
143    Check for installs of optional modules
144
145    Args:
146        warn: whether to log a warning if not found.
147
148    Returns:
149        optionals (dict): a dictionary where keys are optional config modes, and values are whether the packages required for basic imports are installed.
150        This does not check internal requirements (e.g. `einops` may be needed for some models, etc.)
151    """
152    config_imports = {
153        "vlm": ["transformers", "torch", "accelerate", "openai"],
154    }
155    if warn:
156        logger = get_logger()
157    configs = {}
158    for config in config_imports:
159        not_importable = []
160        for module in config_imports[config]:
161            if not module_installed(module):
162                not_importable.append(module)
163        if len(not_importable) > 0 and warn:
164            logger.warning(
165                f'Unable to find imports for the following modules of the {config} setting: {not_importable}. Some features will not be enabled.\nTo fix this, run `uv pip install -e ".[{config}]"` in the GameBoyWorlds repo.'
166            )
167        configs[config] = len(not_importable) == 0
168    return configs