Source code for e3_build_tools.utils

"""Module with utils functions to e3_build_tools."""

import enum
import logging
import pathlib
import re
import subprocess
from typing import Dict, Type, Union

import git



logger = logging.getLogger(__name__)





class WrapperType(enum.Enum):
    """Enum of wrapper types."""



    BASE = enum.auto()



    REQUIRE = enum.auto()



    MODULE = enum.auto()






def get_wrapper_type(wrapper: Union[git.Repo, pathlib.Path]) -> WrapperType:
    """Check wrapper type.

    Raises:
        TypeError: If current directory is not a module.
    """
    if isinstance(wrapper, git.Repo):
        wrapper = pathlib.Path(str(wrapper.working_dir))

    if wrapper.name == "e3-base":
        return WrapperType.BASE
    elif wrapper.name == "e3-require":
        return WrapperType.REQUIRE
    elif wrapper.name.startswith("e3-"):
        return WrapperType.MODULE
    raise TypeError("Not a module, stop.")





def read_makefile_definitions(content: str) -> Dict[str, str]:
    """Return the definitions of makefile variables."""
    config_regex = (
        r"^\s*(?P<key>[a-zA-Z0-9_]*)\s*[:?]?=\s*(?P<val>[a-zA-Z0-9\.\/\-\_\+]*).*$"
    )
    config_data = {}
    for line in content.split("\n"):
        m = re.match(config_regex, line)
        if m:
            config_data[m.group("key")] = m.group("val").strip()
    return config_data





def modify_makefile_definitions(content: str, substitutions: Dict[str, str]) -> str:
    """Overwrite the definitions of makefile variables.

    Declarations that are commented out are ignored.
    """
    for var, val in substitutions.items():
        pattern = rf"^[ \t]*({var}[ \t]*[:?]?=[ \t]*)[a-zA-Z0-9\.\/\-\_\+]*(.*)$"
        replacement = rf"\g<1>{val}\g<2>"
        content, changes_made = re.subn(
            pattern,
            replacement,
            content,
            flags=re.MULTILINE,
        )
        if not changes_made:
            logger.debug(f"Variable {var!r} not found in config file")
    return content





def modify_config_file(filename: pathlib.Path, key: str, val: str) -> None:
    """Modify the data in a config file."""
    content = filename.read_text()
    new_content = modify_makefile_definitions(content, {key: val})
    filename.write_text(new_content)

    logger.debug(f"Modified config file {filename!s}: substituted {key!r} with {val!r}")





def check_module_version(version: str) -> bool:
    """Return if module version is in the expected format."""
    version_pattern = re.compile(r"^(\d+\.){2}\d+((-[\w|.]+)+(\+\d+)?|\+\d+)$")
    return version_pattern.match(version) is not None





def check_require_version(version: str) -> bool:
    """Return if require version is in the expected format."""
    version_pattern = re.compile(r"^(\d+\.){2}\d+$")
    return version_pattern.match(version) is not None





def check_base_version(version: str) -> bool:
    """Return if base version is in the expected format.

    Unfortunately epics base does not follow semantic version and sometimes
    can have one digit more, like on version 7.0.6.1.
    """
    version_pattern = re.compile(r"^(\d+\.){2,3}\d+(-\w+)?$")
    return version_pattern.match(version) is not None





def ensure_build_number(version: str) -> str:
    """Return the same string but with `+0` appended if string is a valid version.

    We have to deal with the implicit addition of +0 by require to modules that have
    no build number defined, but we only do this for versions of the form "x.y.z".
    """
    version_pattern = re.compile(r"^\d+\.\d+\.\d+$")
    if version_pattern.match(version) and "+" not in version:
        logger.debug(f"Appending '+0' to version {version!r}")
        version += "+0"
    return version





def increment_build_number(version: str) -> str:
    """Return a version with incremented build number."""
    version = ensure_build_number(version)
    pattern = re.compile(r"([^+]*)\+(.*)")
    match = pattern.match(version)
    if not match:
        return version
    build_number = int(match.group(2)) + 1
    new_version = f"{match.group(1)}+{build_number}"
    logger.debug(f"Incrementing build number {version!r} to {new_version!r}")
    return new_version





def extract_base_version(base_path: str) -> str:
    """Return the version as taken from a path to EPICS base.

    Raises:
        ValueError: If `base_path` is not a valid EPICS base path.
    """
    match = re.search("(?<=base-)[0-9.-]+[a-zA-Z0-9]*", base_path)
    if not match:
        raise ValueError
    return match.group(0)





def deep_merge(first: Type[Dict], second: Type[Dict]) -> Type[Dict]:
    """Return a nested merge of two dictionaries, with `first` taking precedence.

    Based on https://stackoverflow.com/a/56177639

    Each key that is found only in `first` or only in `second` will be included in the output
    collection with its value intact. For any key in common between `first` and `second`, the
    corresponding values will be merged, and `first` will take precedence over `second`.
    """
    if not isinstance(first, dict) or not isinstance(second, dict):
        return second if first is None else first
    else:
        # Compute set of all keys in both dictionaries.
        keys = set(first.keys()) | set(second.keys())

        # Build output dictionary, merging recursively values with common keys,
        # where `None` is used to mean the absence of a value.
        return {key: deep_merge(first.get(key), second.get(key)) for key in keys}





def run_make(path: pathlib.Path, *makeargs) -> subprocess.CompletedProcess:
    """Run GNU make in the target directory with the given arguments."""
    cmd = ["make", "-C", str(path), *makeargs]

    logger.debug(f"Running command '{' '.join(cmd)}'")
    result = subprocess.run(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        universal_newlines=True,
        encoding="utf-8",
    )
    logger.debug(f"stdout: {result.stdout}")
    logger.debug(f"stderr: {result.stderr}")

    return result