Source code for heros.helper

import re
import sys
import os
import logging
import importlib
import hashlib
import ipaddress




def object_name_from_keyexpr(key_expr, ns_objects, realm, endpoint=".*"):
    return re.search(f"{ns_objects}/{realm}/(.*?)/{endpoint}", key_expr).groups()[0]





def full_classname(o):
    """
    Return the fully qualified class name of an object.

    Args:
        o: object

    Returns:
        fully qualified module and class name
    """
    cl = o.__class__
    mod = cl.__module__
    if mod == "__builtin__":
        return cl.__name__  # avoid outputs like '__builtin__.str'
    return ".".join([mod, cl.__name__])





def get_heros_pkg_versions() -> dict:
    """Returns the versions of the installed heros packages

    Returns:
        dict: A dictionary with the package names as keys and the versions as values. If package version is not
            available, the value is "n.a.", if package is not installed the value is "not installed".
    """
    ver_dict = {}
    for module_name in ["heros", "boss", "herosdevices", "herostools", "atomiq"]:
        try:
            module = importlib.import_module(module_name)
            if hasattr(module, "__version__"):
                ver_dict[module.__name__] = module.__version__
            else:
                ver_dict[module.__name__] = "n.a."
        except ImportError:
            ver_dict[module_name] = "not installed"
    return ver_dict





def generate_multicast_address(realm: str) -> str:
    """Generate a deterministic multicast address based on the realm name.

    Note:
        As the multicast range does not include an infinite number of ip addresses
        there is a slim but finite probability that two realm names generate
        the same multicast address.

    Args:
        realm: The realm name.

    Returns:
        str: A multicast IP address in the range 224.0.1.0 to 238.255.255.255.
    """
    hash = int.from_bytes(hashlib.shake_128(realm.encode()).digest(4))

    # Multicast range: 224.0.1.0 to 239.255.255.255
    multicast_start = 0xE0000100  # 224.0.1.0 in integer
    multicast_end = 0xEEFFFFFF  # 238.255.255.255 in integer
    multicast_range = multicast_end - multicast_start + 1
    ip_int = multicast_start + (hash % multicast_range)
    return str(ipaddress.ip_address(ip_int))



##############################################################
# extend logging mechanism
SPAM = 5
setattr(logging, "SPAM", 5)
logging.addLevelName(levelName="SPAM", level=5)




class Logger(logging.Logger):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        if "HEROS_LOG" in os.environ:
            self.setLevel(os.environ["HEROS_LOG"])



    def setLevel(self, level, globally=False):
        """Set logger level; optionally propagate to all existing loggers."""
        if isinstance(level, str):
            level = level.upper()
        try:
            level = int(level)
        except ValueError:
            pass
        super().setLevel(level)
        if globally:
            os.environ["HEROS_LOG"] = str(level)
            for name, logger in logging.root.manager.loggerDict.items():
                if isinstance(logger, logging.Logger):
                    logger.setLevel(level)




    def spam(self, msg, *args, **kwargs):
        """Log a SPAM-level message."""
        if self.isEnabledFor(SPAM):
            self.log(SPAM, msg, *args, **kwargs)






class ColoredFormatter(logging.Formatter):
    """Logging formatter that adds ANSI colors to logging output."""

    RESET = "\033[0m"
    RED = "\033[31m"
    GREEN = "\033[32m"
    YELLOW = "\033[33m"

    def __init__(self, *args, **kwargs):
        self.supports_color = self._supports_color()
        super().__init__(*args, **kwargs)



    def format(self, record):
        if self.supports_color:
            # Windows does not support ANSI escape codes
            if record.levelno == logging.INFO:
                level_color = self.GREEN
            elif record.levelno == logging.WARNING:
                level_color = self.YELLOW
            elif record.levelno == logging.ERROR:
                level_color = self.RED
            elif record.levelno == logging.CRITICAL:
                level_color = self.RED + "\033[1m"
            else:
                level_color = self.RESET

            record.levelname = f"{level_color}{record.levelname}{self.RESET}"

        return super().format(record)




    def _supports_color(self):
        """Check if the current platform supports ANSI colors."""
        supported_platform = sys.platform != "Pocket PC" and (sys.platform != "win32" or "ANSICON" in os.environ)
        ansi_tty = hasattr(sys.stdout, "isatty") and sys.stdout.isatty()
        return supported_platform and ansi_tty




# logger factory


def get_logger(name: str = "heros") -> logging.Logger:
    logging.setLoggerClass(Logger)
    # Set up console handler only once
    if not logging.getLogger().handlers:
        console_handler = logging.StreamHandler()
        console_handler.setFormatter(
            ColoredFormatter(
                fmt="%(asctime)-15s [%(levelname)s] %(name)s [%(filename)s:%(lineno)d %(funcName)s]: %(message)s"
            )
        )
        logging.getLogger().addHandler(console_handler)
        logging.getLogger().setLevel(logging.INFO)  # default to show SPAM messages
    return logging.getLogger(name)



log = get_logger("heros")