from pathlib import Path from typing import Optional import logging from ruamel.yaml import YAML as _YAML from ruamel.yaml.error import YAMLError as _YAMLError from ppp_logging import DEBUG_LEVEL, log from ppp_utils import deep_freeze, escape_single_quotes class PPPENMappingVariant: """ A class to represent a variant of an extra network mapping. Attributes: condition (str): The condition for the variant. name (str): The name of the variant. parameters (float|str): The parameters for the variant. triggers (list[str]): The triggers for the variant. weight (float): The weight for the variant when multiple variants apply. """ def __init__(self, condition: str, name: str, parameters: float | str, triggers: list[str], weight: float): self.condition: str = condition self.name: str = name self.parameters: float | str = parameters self.triggers: list[str] = triggers self.weight: float = weight class PPPENMapping: """ A extra network mapping object. Attributes: kind (str): The kind of the extra network. name (str): The name of the extra network mapping. file (Path | None): The path to the file where the extranetwork mapping is defined, or None if from inline input. variants (list[PPPENMappingVariant]): The processed variants of the extranetwork mapping. """ def __init__(self, fullpath: Path | None, kind: str, name: str, variants: list[dict]): self.file: Path | None = fullpath self.kind: str = kind self.name: str = name self.variants: list[PPPENMappingVariant] = [ PPPENMappingVariant( **{**{"condition": None, "name": None, "parameters": None, "triggers": None, "weight": 1.0}, **v} ) for v in variants ] def __hash__(self) -> int: t = (self.kind, self.name, deep_freeze(self.variants)) return hash(t) def __sizeof__(self): return ( self.kind.__sizeof__() + self.name.__sizeof__() + (self.file.__sizeof__() if self.file is not None else 0) + self.variants.__sizeof__() ) class PPPExtraNetworkMappings: """ A class to manage extra network mappings. Attributes: extranetwork_maps (dict[str, PPPENMapping]): The extra network mappings. """ DEFAULT_ENMAPPINGS_FOLDER = "extranetworkmappings" def __init__(self, logger=None): self.__logger: logging.Logger = logger self.__debug_level = DEBUG_LEVEL.none self.__enmappings_folders: list[Path] = [] self.__enmappings_files: dict[Path, float] = {} self.__local_enmappings_input_hash: int | None = None self.extranetwork_mappings: dict[str, PPPENMapping] = {} self.cached_mappings = {} def __hash__(self) -> int: return hash(deep_freeze(self.extranetwork_mappings)) def __sizeof__(self): return ( self.extranetwork_mappings.__sizeof__() + self.__enmappings_folders.__sizeof__() + self.__enmappings_files.__sizeof__() + self.cached_mappings.__sizeof__() ) def get_mapping(self, key: str) -> PPPENMapping | None: """ Get an extra network mapping by key. Args: key (str): The key of the extra network mapping in the format "kind:name". Returns: PPPENMapping | None: The extra network mapping if found, or None if not found """ for k, v in self.extranetwork_mappings.items(): if k.lower() == key.lower(): return v return None def refresh_extranetwork_mappings( self, debug_level: DEBUG_LEVEL, enmappings_folders: Optional[list[Path]], enmappings_input: str = None, ): """ Initialize the extra network mappings. """ self.__debug_level = debug_level self.__enmappings_folders = [Path(f) for f in (enmappings_folders or [])] # log(self.__logger, self.__debug_level, logging.INFO, "Refreshing extra network mappings...") # t1 = time.monotonic_ns() self.cached_mappings = {} for fullpath in list(self.__enmappings_files.keys()): if not fullpath.exists() or not any( fullpath.parent.is_relative_to(folder) for folder in self.__enmappings_folders ): self.__remove_extranetwork_mappings_from_path(fullpath) if enmappings_input is None and self.__local_enmappings_input_hash is not None: self.__remove_extranetwork_mappings_from_input() if enmappings_folders is not None or enmappings_input is not None: if enmappings_folders is not None: for f in self.__enmappings_folders: self.__get_extranetwork_mappings_in_path(f) if enmappings_input is not None: self.__get_extranetwork_mappings_in_input(enmappings_input) else: self.extranetwork_mappings = {} self.__enmappings_files = {} self.__local_enmappings_input_hash = None # t2 = time.monotonic_ns() # log(self.__logger, self.__debug_level, logging.INFO, f"Extra network mappings refresh time: {(t2 - t1) / 1_000_000_000:.3f} seconds") def __remove_extranetwork_mappings_from_path(self, full_path: Path, debug=True): """ Clear all extra network mappings from a file. Args: full_path (Path): The path to the file. debug (bool): Whether to print debug messages or not. """ if debug and full_path in self.__enmappings_files: log( self.__logger, self.__debug_level, logging.DEBUG, f"Removing extra network mappings from file: {full_path}", ) if full_path in self.__enmappings_files: del self.__enmappings_files[full_path] for key in list(self.extranetwork_mappings.keys()): if self.extranetwork_mappings[key].file == full_path: del self.extranetwork_mappings[key] def __remove_extranetwork_mappings_from_input(self, debug=True): """ Clear all extra network mappings loaded from inline input. Args: debug (bool): Whether to print debug messages or not. """ if debug and self.__local_enmappings_input_hash is not None: log(self.__logger, self.__debug_level, logging.DEBUG, "Removing extra network mappings from input") self.__local_enmappings_input_hash = None for key in list(self.extranetwork_mappings.keys()): if self.extranetwork_mappings[key].file is None: del self.extranetwork_mappings[key] def __get_extranetwork_mappings_in_file(self, full_path: Path): """ Get all extra network mappings in a file. Args: full_path (Path): The path to the file. """ last_modified = full_path.stat().st_mtime last_modified_cached = self.__enmappings_files.get(full_path, None) if last_modified_cached is not None and last_modified == self.__enmappings_files[full_path]: return extension = full_path.suffix if extension not in (".yaml", ".yml", ".json"): return self.__remove_extranetwork_mappings_from_path(full_path, False) if last_modified_cached is not None: log( self.__logger, self.__debug_level, logging.DEBUG, f"Updating extra network mappings from file: {full_path}", ) self.__get_extranetwork_mappings_in_structured_file(full_path) self.__enmappings_files[full_path] = last_modified def __get_extranetwork_mappings_in_input(self, enmappings_input: str): """ Get all extra network mappings in the string. Args: enmappings_input (str): The input string containing extra network mappings in yaml format. """ new_h = hash(enmappings_input) if new_h == self.__local_enmappings_input_hash: return was_loaded = self.__local_enmappings_input_hash is not None self.__remove_extranetwork_mappings_from_input(False) if was_loaded: log(self.__logger, self.__debug_level, logging.DEBUG, "Updating extra network mappings from input") enmappings_input = enmappings_input.strip() if enmappings_input != "": try: content = _YAML(typ="safe").load(enmappings_input) except _YAMLError as e: log( self.__logger, self.__debug_level, logging.WARNING, f"Invalid format for input extra network mappings: {e}", ) return if content is not None: self.__add_extranetwork_mapping(content, None) self.__local_enmappings_input_hash = new_h def __add_extranetwork_mapping(self, content: dict[str, dict[str, list[dict]]], full_path: Path | None): """ Add an extra network mapping to the extra network mappings dictionary. Args: content (object): The content of the extra network mapping. full_path (Path | None): The path to the file that contains it, or None if from inline input. """ file_str = str(full_path) if full_path is not None else "input" if not isinstance(content, dict): log( self.__logger, self.__debug_level, logging.WARNING, f"Invalid extra network mapping in file '{escape_single_quotes(file_str)}'!", ) return for kind, maps in content.items(): if not isinstance(maps, dict): log( self.__logger, self.__debug_level, logging.WARNING, f"Invalid extra network mapping definition for '{escape_single_quotes(kind)}:*' in file '{escape_single_quotes(file_str)}'!", ) else: for name, variants in maps.items(): key = f"{kind}:{name}" if not isinstance(variants, list): log( self.__logger, self.__debug_level, logging.WARNING, f"Invalid extra network mapping definition for '{escape_single_quotes(key)}' in file '{escape_single_quotes(file_str)}'!", ) elif self.extranetwork_mappings.get(key, None) is not None: f = ( str(self.extranetwork_mappings[key].file) if self.extranetwork_mappings[key].file is not None else "input" ) log( self.__logger, self.__debug_level, logging.WARNING, f"Duplicate extra network mapping '{escape_single_quotes(key)}' in file '{escape_single_quotes(file_str)}' and '{escape_single_quotes(f)}'!", ) elif not isinstance(variants, list) or not all(isinstance(v, dict) for v in variants): log( self.__logger, self.__debug_level, logging.WARNING, f"Invalid extra network mapping definition for '{escape_single_quotes(key)}' in file '{escape_single_quotes(file_str)}'!", ) else: self.extranetwork_mappings[key] = PPPENMapping(full_path, kind, name, variants) def __get_extranetwork_mappings_in_structured_file(self, full_path: Path): """ Get all extra network mappings in a structured file. Args: full_path (Path): The path to the file. """ try: try: with open(full_path, "r", encoding="utf-8") as file: content = _YAML(typ="safe").load(file) except: # pylint: disable=bare-except log( self.__logger, self.__debug_level, logging.WARNING, f"Could not read file '{escape_single_quotes(str(full_path))}' with utf-8 encoding, trying windows-1252...", ) with open(full_path, "r", encoding="windows-1252") as file: content = _YAML(typ="safe").load(file) self.__add_extranetwork_mapping(content, full_path) except Exception as e: # pylint: disable=broad-except log( self.__logger, self.__debug_level, logging.ERROR, f"Error reading extra network mappings from file '{escape_single_quotes(str(full_path))}': {e}", ) def __get_extranetwork_mappings_in_path(self, path: Path): """ Get all extra network mappings in a path. Args: path (Path): The path (folder or file). """ if not path.exists(): log( self.__logger, self.__debug_level, logging.WARNING, f"Extra network mappings path '{escape_single_quotes(str(path))}' does not exist!", ) return if path.is_file(): self.__get_extranetwork_mappings_in_file(path) return for child in path.iterdir(): if child.name.startswith("."): continue self.__get_extranetwork_mappings_in_path(child)