setup/setup_common.py

import subprocess
import os
import re
import sys
import logging
import shutil
import datetime
import pkg_resources

errors = 0  # Define the 'errors' variable before using it
log = logging.getLogger('sd')

def check_python_version():
    """

    Check if the current Python version is within the acceptable range.

    

    Returns:

    bool: True if the current Python version is valid, False otherwise.

    """
    min_version = (3, 10, 9)
    max_version = (3, 11, 0)
    
    from packaging import version
    
    try:
        current_version = sys.version_info
        log.info(f"Python version is {sys.version}")
        
        if not (min_version <= current_version < max_version):
            log.error(f"The current version of python ({current_version}) is not appropriate to run Kohya_ss GUI")
            log.error("The python version needs to be greater or equal to 3.10.9 and less than 3.11.0")
            return False
        return True
    except Exception as e:
        log.error(f"Failed to verify Python version. Error: {e}")
        return False

def update_submodule(quiet=True):
    """

    Ensure the submodule is initialized and updated.

    

    This function uses the Git command line interface to initialize and update 

    the specified submodule recursively. Errors during the Git operation

    or if Git is not found are caught and logged.

    

    Parameters:

    - quiet: If True, suppresses the output of the Git command.

    """
    git_command = ["git", "submodule", "update", "--init", "--recursive"]
    
    if quiet:
        git_command.append("--quiet")
        
    try:
        # Initialize and update the submodule
        subprocess.run(git_command, check=True)
        log.info("Submodule initialized and updated.")
        
    except subprocess.CalledProcessError as e:
        # Log the error if the Git operation fails
        log.error(f"Error during Git operation: {e}")
    except FileNotFoundError as e:
        # Log the error if the file is not found
        log.error(e)

# def read_tag_version_from_file(file_path):
#     """
#     Read the tag version from a given file.

#     Parameters:
#     - file_path: The path to the file containing the tag version.

#     Returns:
#     The tag version as a string.
#     """
#     with open(file_path, 'r') as file:
#         # Read the first line and strip whitespace
#         tag_version = file.readline().strip()
#     return tag_version

def clone_or_checkout(repo_url, branch_or_tag, directory_name):
    """

    Clone a repo or checkout a specific branch or tag if the repo already exists.

    For branches, it updates to the latest version before checking out.

    Suppresses detached HEAD advice for tags or specific commits.

    Restores the original working directory after operations.



    Parameters:

    - repo_url: The URL of the Git repository.

    - branch_or_tag: The name of the branch or tag to clone or checkout.

    - directory_name: The name of the directory to clone into or where the repo already exists.

    """
    original_dir = os.getcwd()  # Store the original directory
    try:
        if not os.path.exists(directory_name):
            # Directory does not exist, clone the repo quietly
            
            # Construct the command as a string for logging
            # run_cmd = f"git clone --branch {branch_or_tag} --single-branch --quiet {repo_url} {directory_name}"
            run_cmd = ["git", "clone", "--branch", branch_or_tag, "--single-branch", "--quiet", repo_url, directory_name]


            # Log the command
            log.debug(run_cmd)
            
            # Run the command
            process = subprocess.Popen(
                run_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
            )
            output, error = process.communicate()
            
            if error and not error.startswith("Note: switching to"):
                log.warning(error)
            else:
                log.info(f"Successfully cloned sd-scripts {branch_or_tag}")
            
        else:
            os.chdir(directory_name)
            subprocess.run(["git", "fetch", "--all", "--quiet"], check=True)
            subprocess.run(["git", "config", "advice.detachedHead", "false"], check=True)
            
            # Get the current branch or commit hash
            current_branch_hash = subprocess.check_output(["git", "rev-parse", "HEAD"]).strip().decode()
            tag_branch_hash = subprocess.check_output(["git", "rev-parse", branch_or_tag]).strip().decode()
            
            if current_branch_hash != tag_branch_hash:
                run_cmd = f"git checkout {branch_or_tag} --quiet"
                # Log the command
                log.debug(run_cmd)
                
                # Execute the checkout command
                process = subprocess.Popen(run_cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
                output, error = process.communicate()
                
                if error:
                    log.warning(error.decode())
                else:
                    log.info(f"Checked out sd-scripts {branch_or_tag} successfully.")
            else:
                log.info(f"Current branch of sd-scripts is already at the required release {branch_or_tag}.")
    except subprocess.CalledProcessError as e:
        log.error(f"Error during Git operation: {e}")
    finally:
        os.chdir(original_dir)  # Restore the original directory

# setup console and file logging
def setup_logging(clean=False):
    #
    # This function was adapted from code written by vladimandic: https://github.com/vladmandic/automatic/commits/master
    #

    from rich.theme import Theme
    from rich.logging import RichHandler
    from rich.console import Console
    from rich.pretty import install as pretty_install
    from rich.traceback import install as traceback_install

    console = Console(
        log_time=True,
        log_time_format='%H:%M:%S-%f',
        theme=Theme(
            {
                'traceback.border': 'black',
                'traceback.border.syntax_error': 'black',
                'inspect.value.border': 'black',
            }
        ),
    )
    # logging.getLogger("urllib3").setLevel(logging.ERROR)
    # logging.getLogger("httpx").setLevel(logging.ERROR)

    current_datetime = datetime.datetime.now()
    current_datetime_str = current_datetime.strftime('%Y%m%d-%H%M%S')
    log_file = os.path.join(
        os.path.dirname(__file__),
        f'../logs/setup/kohya_ss_gui_{current_datetime_str}.log',
    )

    # Create directories if they don't exist
    log_directory = os.path.dirname(log_file)
    os.makedirs(log_directory, exist_ok=True)

    level = logging.INFO
    logging.basicConfig(
        level=logging.ERROR,
        format='%(asctime)s | %(name)s | %(levelname)s | %(module)s | %(message)s',
        filename=log_file,
        filemode='a',
        encoding='utf-8',
        force=True,
    )
    log.setLevel(
        logging.DEBUG
    )   # log to file is always at level debug for facility `sd`
    pretty_install(console=console)
    traceback_install(
        console=console,
        extra_lines=1,
        width=console.width,
        word_wrap=False,
        indent_guides=False,
        suppress=[],
    )
    rh = RichHandler(
        show_time=True,
        omit_repeated_times=False,
        show_level=True,
        show_path=False,
        markup=False,
        rich_tracebacks=True,
        log_time_format='%H:%M:%S-%f',
        level=level,
        console=console,
    )
    rh.set_name(level)
    while log.hasHandlers() and len(log.handlers) > 0:
        log.removeHandler(log.handlers[0])
    log.addHandler(rh)


def install_requirements_inbulk(requirements_file, show_stdout=True, optional_parm="", upgrade = False):
    if not os.path.exists(requirements_file):
        log.error(f'Could not find the requirements file in {requirements_file}.')
        return

    log.info(f'Installing requirements from {requirements_file}...')

    if upgrade:
        optional_parm += " -U"

    if show_stdout:
        run_cmd(f'pip install -r {requirements_file} {optional_parm}')
    else:
        run_cmd(f'pip install -r {requirements_file} {optional_parm} --quiet')
    log.info(f'Requirements from {requirements_file} installed.')
    


def configure_accelerate(run_accelerate=False):
    #
    # This function was taken and adapted from code written by jstayco
    #

    from pathlib import Path

    def env_var_exists(var_name):
        return var_name in os.environ and os.environ[var_name] != ''

    log.info('Configuring accelerate...')
    
    source_accelerate_config_file = os.path.join(
        os.path.dirname(os.path.abspath(__file__)),
        '..',
        'config_files',
        'accelerate',
        'default_config.yaml',
    )

    if not os.path.exists(source_accelerate_config_file):
        if run_accelerate:
            run_cmd('accelerate config')
        else:
            log.warning(
                f'Could not find the accelerate configuration file in {source_accelerate_config_file}. Please configure accelerate manually by runningthe option in the menu.'
            )
    
    log.debug(
        f'Source accelerate config location: {source_accelerate_config_file}'
    )

    target_config_location = None

    log.debug(
        f"Environment variables: HF_HOME: {os.environ.get('HF_HOME')}, "
        f"LOCALAPPDATA: {os.environ.get('LOCALAPPDATA')}, "
        f"USERPROFILE: {os.environ.get('USERPROFILE')}"
    )
    if env_var_exists('HF_HOME'):
        target_config_location = Path(
            os.environ['HF_HOME'], 'accelerate', 'default_config.yaml'
        )
    elif env_var_exists('LOCALAPPDATA'):
        target_config_location = Path(
            os.environ['LOCALAPPDATA'],
            'huggingface',
            'accelerate',
            'default_config.yaml',
        )
    elif env_var_exists('USERPROFILE'):
        target_config_location = Path(
            os.environ['USERPROFILE'],
            '.cache',
            'huggingface',
            'accelerate',
            'default_config.yaml',
        )

    log.debug(f'Target config location: {target_config_location}')

    if target_config_location:
        if not target_config_location.is_file():
            target_config_location.parent.mkdir(parents=True, exist_ok=True)
            log.debug(
                f'Target accelerate config location: {target_config_location}'
            )
            shutil.copyfile(
                source_accelerate_config_file, target_config_location
            )
            log.info(
                f'Copied accelerate config file to: {target_config_location}'
            )
        else:
            if run_accelerate:
                run_cmd('accelerate config')
            else:
                log.warning(
                    'Could not automatically configure accelerate. Please manually configure accelerate with the option in the menu or with: accelerate config.'
                )
    else:
        if run_accelerate:
            run_cmd('accelerate config')
        else:
            log.warning(
                'Could not automatically configure accelerate. Please manually configure accelerate with the option in the menu or with: accelerate config.'
            )


def check_torch():
    #
    # This function was adapted from code written by vladimandic: https://github.com/vladmandic/automatic/commits/master
    #

    # Check for toolkit
    if shutil.which('nvidia-smi') is not None or os.path.exists(
        os.path.join(
            os.environ.get('SystemRoot') or r'C:\Windows',
            'System32',
            'nvidia-smi.exe',
        )
    ):
        log.info('nVidia toolkit detected')
    elif shutil.which('rocminfo') is not None or os.path.exists(
        '/opt/rocm/bin/rocminfo'
    ):
        log.info('AMD toolkit detected')
    elif (shutil.which('sycl-ls') is not None
    or os.environ.get('ONEAPI_ROOT') is not None
    or os.path.exists('/opt/intel/oneapi')):
        log.info('Intel OneAPI toolkit detected')
    else:
        log.info('Using CPU-only Torch')

    try:
        import torch
        try:
            # Import IPEX / XPU support
            import intel_extension_for_pytorch as ipex
        except Exception:
            pass
        log.info(f'Torch {torch.__version__}')

        if torch.cuda.is_available():
            if torch.version.cuda:
                # Log nVidia CUDA and cuDNN versions
                log.info(
                    f'Torch backend: nVidia CUDA {torch.version.cuda} cuDNN {torch.backends.cudnn.version() if torch.backends.cudnn.is_available() else "N/A"}'
                )
            elif torch.version.hip:
                # Log AMD ROCm HIP version
                log.info(f'Torch backend: AMD ROCm HIP {torch.version.hip}')
            else:
                log.warning('Unknown Torch backend')

            # Log information about detected GPUs
            for device in [
                torch.cuda.device(i) for i in range(torch.cuda.device_count())
            ]:
                log.info(
                    f'Torch detected GPU: {torch.cuda.get_device_name(device)} VRAM {round(torch.cuda.get_device_properties(device).total_memory / 1024 / 1024)} Arch {torch.cuda.get_device_capability(device)} Cores {torch.cuda.get_device_properties(device).multi_processor_count}'
                )
        # Check if XPU is available
        elif hasattr(torch, "xpu") and torch.xpu.is_available():
            # Log Intel IPEX version
            log.info(f'Torch backend: Intel IPEX {ipex.__version__}')
            for device in [
                torch.xpu.device(i) for i in range(torch.xpu.device_count())
            ]:
                log.info(
                    f'Torch detected GPU: {torch.xpu.get_device_name(device)} VRAM {round(torch.xpu.get_device_properties(device).total_memory / 1024 / 1024)} Compute Units {torch.xpu.get_device_properties(device).max_compute_units}'
                )
        else:
            log.warning('Torch reports GPU not available')
        
        return int(torch.__version__[0])
    except Exception as e:
        # log.warning(f'Could not load torch: {e}')
        return 0


# report current version of code
def check_repo_version():
    """

    This function checks the version of the repository by reading the contents of a file named '.release'

    in the current directory. If the file exists, it reads the release version from the file and logs it.

    If the file does not exist, it logs a debug message indicating that the release could not be read.

    """
    if os.path.exists('.release'):
        try:
            with open(os.path.join('./.release'), 'r', encoding='utf8') as file:
                release= file.read()
            
            log.info(f'Kohya_ss GUI version: {release}')
        except Exception as e:
            log.error(f'Could not read release: {e}')
    else:
        log.debug('Could not read release...')
    
# execute git command
def git(arg: str, folder: str = None, ignore: bool = False):
    """

    Executes a Git command with the specified arguments.



    This function is designed to run Git commands and handle their output.

    It can be used to execute Git commands in a specific folder or the current directory.

    If an error occurs during the Git operation and the 'ignore' flag is not set,

    it logs the error message and the Git output for debugging purposes.



    Parameters:

    - arg: A string containing the Git command arguments.

    - folder: An optional string specifying the folder where the Git command should be executed.

               If not provided, the current directory is used.

    - ignore: A boolean flag indicating whether to ignore errors during the Git operation.

               If set to True, errors will not be logged.



    Note:

    This function was adapted from code written by vladimandic: https://github.com/vladmandic/automatic/commits/master

    """
    
    git_cmd = os.environ.get('GIT', "git")
    result = subprocess.run(f'"{git_cmd}" {arg}', check=False, shell=True, env=os.environ, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=folder or '.')
    txt = result.stdout.decode(encoding="utf8", errors="ignore")
    if len(result.stderr) > 0:
        txt += ('\n' if len(txt) > 0 else '') + result.stderr.decode(encoding="utf8", errors="ignore")
    txt = txt.strip()
    if result.returncode != 0 and not ignore:
        global errors
        errors += 1
        log.error(f'Error running git: {folder} / {arg}')
        if 'or stash them' in txt:
            log.error(f'Local changes detected: check log for details...')
        log.debug(f'Git output: {txt}')


def pip(arg: str, ignore: bool = False, quiet: bool = False, show_stdout: bool = False):
    """

    Executes a pip command with the specified arguments.



    This function is designed to run pip commands and handle their output.

    It can be used to install, upgrade, or uninstall packages using pip.

    If an error occurs during the pip operation and the 'ignore' flag is not set,

    it logs the error message and the pip output for debugging purposes.



    Parameters:

    - arg: A string containing the pip command arguments.

    - ignore: A boolean flag indicating whether to ignore errors during the pip operation.

               If set to True, errors will not be logged.

    - quiet: A boolean flag indicating whether to suppress the output of the pip command.

              If set to True, the function will not log any output.

    - show_stdout: A boolean flag indicating whether to display the pip command's output

                    to the console. If set to True, the function will print the output

                    to the console.



    Returns:

    - The output of the pip command as a string, or None if the 'show_stdout' flag is set.

    """
    # arg = arg.replace('>=', '==')
    if not quiet:
        log.info(f'Installing package: {arg.replace("install", "").replace("--upgrade", "").replace("--no-deps", "").replace("--force", "").replace("  ", " ").strip()}')
    log.debug(f"Running pip: {arg}")
    if show_stdout:
        subprocess.run(f'"{sys.executable}" -m pip {arg}', shell=True, check=False, env=os.environ)
    else:
        result = subprocess.run(f'"{sys.executable}" -m pip {arg}', shell=True, check=False, env=os.environ, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        txt = result.stdout.decode(encoding="utf8", errors="ignore")
        if len(result.stderr) > 0:
            txt += ('\n' if len(txt) > 0 else '') + result.stderr.decode(encoding="utf8", errors="ignore")
        txt = txt.strip()
        if result.returncode != 0 and not ignore:
            global errors # pylint: disable=global-statement
            errors += 1
            log.error(f'Error running pip: {arg}')
            log.debug(f'Pip output: {txt}')
        return txt

def installed(package, friendly: str = None):
    """

    Checks if the specified package(s) are installed with the correct version.

    This function can handle package specifications with or without version constraints,

    and can also filter out command-line options and URLs when a 'friendly' string is provided.

    

    Parameters:

    - package: A string that specifies one or more packages with optional version constraints.

    - friendly: An optional string used to provide a cleaner version of the package string

                that excludes command-line options and URLs.



    Returns:

    - True if all specified packages are installed with the correct versions, False otherwise.

    

    Note:

    This function was adapted from code written by vladimandic.

    """
    
    # Remove any optional features specified in brackets (e.g., "package[option]==version" becomes "package==version")
    package = re.sub(r'\[.*?\]', '', package)

    try:
        if friendly:
            # If a 'friendly' version of the package string is provided, split it into components
            pkgs = friendly.split()
            
            # Filter out command-line options and URLs from the package specification
            pkgs = [
                p
                for p in package.split()
                if not p.startswith('--') and "://" not in p
            ]
        else:
            # Split the package string into components, excluding '-' and '=' prefixed items
            pkgs = [
                p
                for p in package.split()
                if not p.startswith('-') and not p.startswith('=')
            ]
            # For each package component, extract the package name, excluding any URLs
            pkgs = [
                p.split('/')[-1] for p in pkgs
            ]

        for pkg in pkgs:
            # Parse the package name and version based on the version specifier used
            if '>=' in pkg:
                pkg_name, pkg_version = [x.strip() for x in pkg.split('>=')]
            elif '==' in pkg:
                pkg_name, pkg_version = [x.strip() for x in pkg.split('==')]
            else:
                pkg_name, pkg_version = pkg.strip(), None

            # Attempt to find the installed package by its name
            spec = pkg_resources.working_set.by_key.get(pkg_name, None)
            if spec is None:
                # Try again with lowercase name
                spec = pkg_resources.working_set.by_key.get(pkg_name.lower(), None)
            if spec is None:
                # Try replacing underscores with dashes
                spec = pkg_resources.working_set.by_key.get(pkg_name.replace('_', '-'), None)

            if spec is not None:
                # Package is found, check version
                version = pkg_resources.get_distribution(pkg_name).version
                log.debug(f'Package version found: {pkg_name} {version}')

                if pkg_version is not None:
                    # Verify if the installed version meets the specified constraints
                    if '>=' in pkg:
                        ok = version >= pkg_version
                    else:
                        ok = version == pkg_version

                    if not ok:
                        # Version mismatch, log warning and return False
                        log.warning(f'Package wrong version: {pkg_name} {version} required {pkg_version}')
                        return False
            else:
                # Package not found, log debug message and return False
                log.debug(f'Package version not found: {pkg_name}')
                return False

        # All specified packages are installed with the correct versions
        return True
    except ModuleNotFoundError:
        # One or more packages are not installed, log debug message and return False
        log.debug(f'Package not installed: {pkgs}')
        return False



# install package using pip if not already installed
def install(

    package,

    friendly: str = None,

    ignore: bool = False,

    reinstall: bool = False,

    show_stdout: bool = False,

):
    """

    Installs or upgrades a Python package using pip, with options to ignode errors,

    reinstall packages, and display outputs.

    

    Parameters:

    - package (str): The name of the package to be installed or upgraded. Can include

      version specifiers. Anything after a '#' in the package name will be ignored.

    - friendly (str, optional): A more user-friendly name for the package, used for

      logging or user interface purposes. Defaults to None.

    - ignore (bool, optional): If True, any errors encountered during the installation

      will be ignored. Defaults to False.

    - reinstall (bool, optional): If True, forces the reinstallation of the package

      even if it's already installed. This also disables any quick install checks. Defaults to False.

    - show_stdout (bool, optional): If True, displays the standard output from the pip

      command to the console. Useful for debugging. Defaults to False.



    Returns:

    None. The function performs operations that affect the environment but does not return

    any value.

    

    Note:

    If `reinstall` is True, it disables any mechanism that allows for skipping installations

    when the package is already present, forcing a fresh install.

    """
    # Remove anything after '#' in the package variable
    package = package.split('#')[0].strip()

    if reinstall:
        global quick_allowed   # pylint: disable=global-statement
        quick_allowed = False
    if reinstall or not installed(package, friendly):
        pip(f'install --upgrade {package}', ignore=ignore, show_stdout=show_stdout)


def process_requirements_line(line, show_stdout: bool = False):
    # Remove brackets and their contents from the line using regular expressions
    # e.g., diffusers[torch]==0.10.2 becomes diffusers==0.10.2
    package_name = re.sub(r'\[.*?\]', '', line)
    install(line, package_name, show_stdout=show_stdout)


def install_requirements(requirements_file, check_no_verify_flag=False, show_stdout: bool = False):
    if check_no_verify_flag:
        log.info(f'Verifying modules installation status from {requirements_file}...')
    else:
        log.info(f'Installing modules from {requirements_file}...')
    with open(requirements_file, 'r', encoding='utf8') as f:
        # Read lines from the requirements file, strip whitespace, and filter out empty lines, comments, and lines starting with '.'
        if check_no_verify_flag:
            lines = [
                line.strip()
                for line in f.readlines()
                if line.strip() != ''
                and not line.startswith('#')
                and line is not None
                and 'no_verify' not in line
            ]
        else:
            lines = [
                line.strip()
                for line in f.readlines()
                if line.strip() != ''
                and not line.startswith('#')
                and line is not None
            ]

        # Iterate over each line and install the requirements
        for line in lines:
            # Check if the line starts with '-r' to include another requirements file
            if line.startswith('-r'):
                # Get the path to the included requirements file
                included_file = line[2:].strip()
                # Expand the included requirements file recursively
                install_requirements(included_file, check_no_verify_flag=check_no_verify_flag, show_stdout=show_stdout)
            else:
                process_requirements_line(line, show_stdout=show_stdout)


def ensure_base_requirements():
    try:
        import rich   # pylint: disable=unused-import
    except ImportError:
        install('--upgrade rich', 'rich')
        
    try:
        import packaging
    except ImportError:
        install('packaging')


def run_cmd(run_cmd):
    try:
        subprocess.run(run_cmd, shell=True, check=False, env=os.environ)
    except subprocess.CalledProcessError as e:
        log.error(f'Error occurred while running command: {run_cmd}')
        log.error(f'Error: {e}')


def delete_file(file_path):
    if os.path.exists(file_path):
        os.remove(file_path)


def write_to_file(file_path, content):
    try:
        with open(file_path, 'w') as file:
            file.write(content)
    except IOError as e:
        print(f'Error occurred while writing to file: {file_path}')
        print(f'Error: {e}')


def clear_screen():
    # Check the current operating system to execute the correct clear screen command
    if os.name == 'nt':  # If the operating system is Windows
        os.system('cls')
    else:  # If the operating system is Linux or Mac
        os.system('clear')