python installer

[dieBakterie]

Idea how to migrate from shell and bash to python

First we should make an helper script with the fmt_logging and other stuff that's used in all scripts, like with globalcontrol.sh right now for the shell/bash scripts/installer.
Example:
globalcontrol.py

import tomllib
import argparse
import logging
import os
import time
import threading
import subprocess
import json
import argparse
import sys
import re


export def fmt_logging():
    class ColoredFormatter(logging.Formatter):
        COLORS = {
            "DEBUG": "\033[94m",  # Blue
            "INFO": "\033[92m",  # Green
            "WARNING": "\033[93m",  # Yellow
            "ERROR": "\033[91m",  # Red
            "CRITICAL": "\033[95m",  # Magenta
        }
        RESET = "\033[0m"
        DATE_COLOR = "\033[96m"  # Cyan

        def format(self, record):
            log_color = self.COLORS.get(record.levelname, self.RESET)
            record.levelname = f"{log_color}{record.levelname}{self.RESET}"
            record.asctime = (
                f"{self.DATE_COLOR}{self.formatTime(record, self.datefmt)}{self.RESET}"
            )
            return super().format(record)

    log_level = os.getenv("LOG_LEVEL", "INFO").upper()
    log_format = os.getenv(
        "LOG_FORMAT", "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
    )
    formatter = ColoredFormatter(log_format)
    handler = logging.StreamHandler()
    handler.setFormatter(formatter)
    logging.basicConfig(level=log_level, handlers=[handler])

Then in parse.json.py we do:

from globalcontrol.py import (json argparse sys re fmt_logging os)


def remove_comments(json_data):
    """Remove single-line and multi-line comments from JSON data."""
    json_data = re.sub(r"//.*", "", json_data)  # Remove single-line comments
    json_data = re.sub(
        r"/\*.*?\*/", "", json_data, flags=re.DOTALL
    )  # Remove multi-line comments
    return json_data


def parse_json(json_data, query, skip_comments, raw_output):
    """Parse JSON data and return the queried part."""
    if skip_comments:
        json_data = remove_comments(json_data)
    try:
        data = json.loads(json_data)
        if query == ".":
            result = data
        else:
            keys = re.findall(r'\["(.*?)"\]|(\w+)', query)
            for key in keys:
                key = key[0] or key[1]
                if isinstance(data, list):
                    key = int(key)  # Convert to int if accessing a list index
                data = data[key]
            result = data
        if raw_output and isinstance(result, str):
            return result
        return json.dumps(result, indent=4)
    except (json.JSONDecodeError, KeyError, IndexError, ValueError) as e:
        return f"Error: {e}"


def update_json(json_data, key, value, skip_comments):
    """Update the JSON data with the specified key and value."""
    if skip_comments:
        json_data = remove_comments(json_data)
    try:
        data = json.loads(json_data)
        keys = re.findall(r'\["(.*?)"\]|(\w+)', key)
        d = data
        for k in keys[:-1]:
            k = k[0] or k[1]
            if k not in d:
                d[k] = {}
            d = d[k]
        final_key = keys[-1][0] or keys[-1][1]
        d[final_key] = value
        result = json.dumps(data, indent=4)
        return result, None
    except (json.JSONDecodeError, KeyError, IndexError, ValueError) as e:
        return None, f"Error: {e}"


def arg_parser():
    parser = argparse.ArgumentParser(description="A simple JSON parser similar to jq.")
    parser.add_argument(
        "file",
        type=str,
        help="The JSON file to parse, or '-' to read from stdin.",
    )
    parser.add_argument(
        "--query",
        "-Q",
        help="The query to apply to the JSON data.",
    )
    parser.add_argument(
        "--skip-comments",
        "-C",
        action="store_true",
        help="Skip comments in the JSON data.",
    )
    parser.add_argument(
        "--raw-output",
        "-R",
        action="store_true",
        help="Remove quotes from output if the result is a string.",
    )
    parser.add_argument(
        "--update",
        "-U",
        nargs=2,
        metavar=("KEY", "VALUE"),
        help="Update the JSON data with the specified key and value.",
    )
    return parser.parse_args()


def main():
    fmt_logging()
    args = arg_parser()
    if args.file == "-":
        json_data = sys.stdin.read()
    else:
        with open(args.file, "r+", encoding='UTF-8') as f:
            json_data = f.read()

    if args.update:
        key, value = args.update
        result, error = update_json(json_data, key, value, args.skip_comments)
        if error:
            # print(error, file=sys.stderr)
            logging.error(error, file=sys.stderr)

            sys.exit(1)
        if args.file == "-":
            print(result)
        else:
            with open(args.file, "w", encoding='UTF-8') as f:
                f.write(result)
    elif args.query:
        result = parse_json(json_data, args.query, args.skip_comments, args.raw_output)
        print(result)
    else:
        logging.error(
            "Error: Either --query or --update must be specified.", file=sys.stderr
        )
        sys.exit(1)


if __name__ == "__main__":
    main()

could be that we've to run the globalcontrol.py script in the python scripts like for the shell/bash scripts via subprocess or we run the scripts(parse.json.py, ...) via subprocess from the globalcontrol.py via args like --json, --config, ...

[kRHYME7]

@dieBakterie I'll push a local branch soon.
I'll be adding a separate prefix/lib/hydepy. One of its features is a venv wrapper where we can add pythonic libraries, and also dynamic module imports, hehe (just obsessed with speed without dependencies).

The main idea here is upon HyDE installation/updates we will prepare a Venv. And all hydepy scripts will use that as its environment.

It's all good except for the fact that it's slow on loading the click module (<250 ms to load). I already discussed this stuff with @rubiin. Hehe Thanks @rubiin .

I'm planning this ao we can easily integrate the hydepanel.😉

Here's a snippit

hydepy (entrypoint)

import os
import sys
import click
import importlib.util

# Add the current directory to the path
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
try:
    from utils import fmt_logger, get_venv_path  # noqa: E402
except ImportError:
    print("Error: Cannot import lib/utils.py")
    print("Is hyde installed correctly?")
    sys.exit(1)

log = fmt_logger()
sys.path.insert(0, get_venv_path())

module_cache = {}

module_dirs = [
    os.path.expanduser("~/.local/lib/hydepy/modules"),
    "/usr/local/lib/hydepy/modules",
    "/usr/lib/hydepy/modules",
]


def list_modules():
    modules = []
    for module_dir in module_dirs:
        if os.path.exists(module_dir):
            sys.path.insert(0, module_dir)
            for filename in os.listdir(module_dir):
                if filename.endswith(".py") and filename != "__init__.py":
                    module_name = filename[:-3]
                    try:
                        module = importlib.import_module(module_name)
                        if hasattr(module, "hyde") and callable(
                            getattr(module, "hyde")
                        ):
                            modules.append(module_name)
                    except ImportError as e:
                        log.error(f"Error importing module {module_name}: {e}")
                        click.echo(f"Error importing module {module_name}: {e}")
            sys.path.pop(0)
    return modules


class LazyGroup(click.Group):
    """Lazy Load the cached modules, this is to avoid loading all modules at once = faster"""

    def list_commands(self, ctx):
        return list_modules()

    def get_command(self, ctx, cmd_name):
        if cmd_name in module_cache:
            return module_cache[cmd_name]

        for module_dir in module_dirs:
            module_path = os.path.join(module_dir, f"{cmd_name}.py")
            if os.path.exists(module_path):
                try:
                    spec = importlib.util.spec_from_file_location(cmd_name, module_path)
                    module = importlib.util.module_from_spec(spec)
                    spec.loader.exec_module(module)
                    module_cache[cmd_name] = module

                    @click.command(
                        name=cmd_name,
                        context_settings=dict(
                            ignore_unknown_options=True,
                            allow_extra_args=True,
                            help_option_names=[],  # Disable the default --help option
                        ),
                        help=module.hyde.__doc__
                        if hasattr(module, "hyde")
                        else f"Help for {cmd_name}",
                    )
                    @click.pass_context
                    def _command(ctx):
                        try:
                            module.hyde(ctx.args)
                        except Exception as e:
                            log.error(f"Error executing command {cmd_name}: {e}")
                            click.echo(f"Error executing command {cmd_name}: {e}")

                    return _command
                except Exception as e:
                    log.error(f"Error loading module {cmd_name}: {e}")
                    click.echo(f"Error loading module {cmd_name}: {e}")
        return None


@click.command(cls=LazyGroup, invoke_without_command=True)
@click.pass_context
def cli(ctx):
    log.debug(f"Starting hydepy... Using logger: {log.get_logger_type()}")
    ctx.ensure_object(dict)
    if ctx.invoked_subcommand is None:
        click.echo(ctx.get_help())
        ctx.exit()


if __name__ == "__main__":
    cli()
    os._exit

utils.py

import os
import sys
import importlib


def get_venv_path():
    """Set up the virtual environment path and modify sys.path."""
    # Determine the virtual environment path using XDG Base Directory Specification
    xdg_state_home = os.getenv("XDG_STATE_HOME", os.path.expanduser("~/.local/state"))
    venv_path = os.path.join(xdg_state_home, "hydepy", "venv")
    # Fall back to $HOME/.local/lib/hyde/venv if XDG_state_HOME is not set
    if not os.path.exists(venv_path):
        venv_path = os.path.expanduser("~/.local/state/hydepy/venv")
    site_packages_path = os.path.join(
        venv_path,
        "lib",
        f"python{sys.version_info.major}.{sys.version_info.minor}",
        "site-packages",
    )
    sys.path.insert(0, site_packages_path)
    return venv_path


def fmt_logger():
    """Format logger to provide a unified interface."""
    log_level = os.getenv("LOG_LEVEL")
    if not log_level:

        class NoOpLogger:
            def __getattr__(self, name):
                def no_op(*args, **kwargs):
                    pass

                return no_op

            def get_logger_type(self):
                return "NoOpLogger"

        return NoOpLogger()

    log_level = log_level.upper()

    # Dynamically import logging or loguru
    try:
        log = importlib.import_module("loguru")
        log.logger.remove()
        log.logger.add(sys.stderr, level=log_level)
        logger_type = "loguru"
    except ModuleNotFoundError:
        import logging as log

        log.basicConfig(level=getattr(log, log_level))
        logger_type = "logging"

    class UnifiedLogger:
        def __init__(self, logger, logger_type):
            self.logger = logger
            self.logger_type = logger_type

        def debug(self, msg, *args, **kwargs):
            if hasattr(self.logger, "logger"):
                self.logger.logger.debug(msg, *args, **kwargs)
            else:
                self.logger.debug(msg, *args, **kwargs)

        def info(self, msg, *args, **kwargs):
            if hasattr(self.logger, "logger"):
                self.logger.logger.info(msg, *args, **kwargs)
            else:
                self.logger.info(msg, *args, **kwargs)

        def warning(self, msg, *args, **kwargs):
            if hasattr(self.logger, "logger"):
                self.logger.logger.warning(msg, *args, **kwargs)
            else:
                self.logger.warning(msg, *args, **kwargs)

        def error(self, msg, *args, **kwargs):
            if hasattr(self.logger, "logger"):
                self.logger.logger.error(msg, *args, **kwargs)
            else:
                self.logger.error(msg, *args, **kwargs)

        def critical(self, msg, *args, **kwargs):
            if hasattr(self.logger, "logger"):
                self.logger.logger.critical(msg, *args, **kwargs)
            else:
                self.logger.critical(msg, *args, **kwargs)

        def get_logger_type(self):
            return self.logger_type

    return UnifiedLogger(log, logger_type)


def load_modules(module_dir):
    modules = []
    sys.path.insert(0, module_dir)
    for filename in os.listdir(module_dir):
        if filename.endswith(".py") and filename != "__init__.py":
            module_name = filename[:-3]
            try:
                module = importlib.import_module(module_name)
                if hasattr(module, "hyde") and callable(getattr(module, "hyde")):
                    modules.append(module_name)
            except ImportError:
                continue
    sys.path.pop(0)
    return modules


def load_module(module_dir, command):
    sys.path.insert(0, module_dir)
    try:
        module = importlib.import_module(command)
    except ModuleNotFoundError:
        module = None
    sys.path.pop(0)
    return module


# Call get_venv_path() to set up the virtual environment path
sys.path.insert(0, get_venv_path())

I'll be very grateful as I can see that average HyDE users know a bunch of python.

[dieBakterie]

So we'll write our own python libraries/modules for HyDE?

[kRHYME7]

#236 (comment)
And yes

[dieBakterie]

Wanna keep and use this discussion for planning the migration and pre-work needed for python transfer? Like which/what libs we use, when we write our own code and when we use built-in and when external libs... pyamdgpu is an external lib since it has to be installed as an package.