mpsd_software.py

def environment_status(mpsd_release: str, root_dir: Union[str, Path]) -> dict:
    """Show status of release in installation.

    Parameters
    ----------
    mpsd_release : str
        A string representing the MPSD release version.
    root_dir : pathlib.Path
        A Path object pointing to the root directory of the installation.
        Expect a subfolder root/mpsd_release in which we search for the
        toolchains.

    Returns
    -------
    toolchain_map : dict
        A dictionary containing available microarchitectures as keys and
        a list of available package_sets as values for each microarchitecture.

        Note: only toolchains can be reported at the moment (i.e. package_sets
        such as global and global_generic are missing, even if installed).

    """
    msg = f"Showing status of release {mpsd_release} in {root_dir}"
    logging.info(msg)
    plog = logging.getLogger("print")
    release_base_dir = root_dir / mpsd_release
    microarch = get_native_microarchitecture()
    toolchain_dir = release_base_dir / microarch
    spack_dir = toolchain_dir / "spack"

    # if the mpsd_release does not exist:
    if not release_base_dir.exists():
        logging.debug(f"Directory {str(release_base_dir)} does not exist.")
        logging.error(f"MPSD release '{mpsd_release}' is not installed.")
        return None

    # if the mpds_release directory exists but the spack repository is not fully
    # cloned - indicates some kind of incomplete installation:
    if not spack_dir.exists():
        logging.debug(f"Looking for files in {spack_dir}")
        logging.error(
            f"MPSD release '{mpsd_release}' has not been completely installed."
        )

        return None

    # find all folders for all microarch in the release directory
    # except for the blacklisted files
    black_listed_files = [
        config_vars["cmd_log_file"],
        "spack-environments",
        "logs",
        "mpsd-spack-cache",
    ]

    list_of_microarchs_candidates = os.listdir(release_base_dir)
    list_of_microarchs = [
        x for x in list_of_microarchs_candidates if x not in black_listed_files
    ]
    logging.debug(f"{list_of_microarchs=}")

    toolchain_map = {}
    for microarch in list_of_microarchs:
        # get a list of all the toolchains in the microarch
        possible_toolchains = (release_base_dir / microarch).glob(
            "lmod/Core/toolchains/*.lua"
        )
        # append toolchain which is the name of the file without the .lua extension
        toolchain_map[microarch] = [toolchain.stem for toolchain in possible_toolchains]

    logging.debug(f"{toolchain_map=}")

    # pretty print the toolchain map key as the heading
    # and the value as the list of toolchains
    plog.info(f"Installed toolchains ({mpsd_release}):\n")
    for microarch, toolchains in toolchain_map.items():
        plog.info(f"- {microarch}")
        for toolchain in toolchains:
            plog.info(f"    {toolchain}")
        plog.info(f"    [module use {str(release_base_dir / microarch / 'lmod/Core')}]")
        plog.info("")
    return toolchain_map


def initialize_environment(root_dir: Path) -> None:
    """Initialize the software environment.

    This creates a hidden file ``.mpsd-software-root`` to tag the location for
    as the root of the installation. All compiled files, logs etc are written in
    or below this subdirectory.

    Parameters
    ----------
    root_dir : pathlib.Path
        A Path object pointing to the current directory where the script was called.

    """
    # check if the root_dir is not already initialized
    init_file = root_dir / config_vars["init_file"]
    if init_file.exists():
        logging.getLogger("print").info(
            f"Error: Directory {str(root_dir)} is already initialized."
        )
        sys.exit(1)
    else:
        # create the init file
        init_file.touch()
        # note the execution in the execution summary log
        # create the log file and fill it with the headers
        record_script_execution_summary(root_dir=root_dir)
        # record the msg in the log file
        record_script_execution_summary(
            root_dir=root_dir,
            msg=f"Initialising MPSD software instance at {root_dir}.",
        )


def get_root_dir() -> Path:
    """Get the root directory of the installation.

    Look for the hidden file ``.mpsd-software-root``
    (defined in config_vars["init_file"])
    in the current directory, or any parent directory.
    If found, return the path to the root directory
    of the MPSD software instance.
    If not found, exit with an error message.

    Returns
    -------
    root_dir : pathlib.Path
        A Path object pointing to the root directory of the installation.
        This folder contains the hidden file ``.mpsd-software-root``,
        ``mpsd_releases`` ( for eg ``dev-23a``) and ``mpsd-spack-cache``.


    """
    # check if the root_dir is not already initialized
    script_call_dir = Path.cwd()
    init_file = script_call_dir / config_vars["init_file"]
    if not init_file.exists():
        for parent_folder in script_call_dir.parents:
            init_file = parent_folder / config_vars["init_file"]
            if init_file.exists():
                script_call_dir = parent_folder
                return script_call_dir
        logging.getLogger("print").info(
            "Error: Couldnt find MPSD software instance"
            "in the current directory or any parent directory. \n"
            f"Directory {str(script_call_dir)} is not a MPSD software instance. \n"
            "Please run 'mpsd-software init' to "
            "initialise the software instance here, \n"
            "or switch to a directory which is already initialised.\n \n"
            f"Hint: Look for the hidden file {config_vars['init_file']}"
            " to check if a directory is initialised"
        )
        sys.exit(1)
    else:
        return script_call_dir


def main():
    """Execute main entry point."""
    parser = argparse.ArgumentParser(
        description=about_intro,
        epilog=about_epilog,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument(
        "-l",
        dest="loglevel",
        choices=["warning", "info", "debug"],
        required=False,
        default="warning",
        help="Set the log level",
    )

    parser.add_argument("--version", action="version", version=__version__)

    subparsers = parser.add_subparsers(
        dest="action", title="actions", description="valid actions", required=True
    )
    subparsers.required = True
    list_of_cmds = [
        ("init", "Initialise the MPSD software instance in the current directory"),
        ("available", "What is available for installation?"),
        ("install", "Install a software environment"),
        # ("reinstall", "Reinstall a package_set"),
        # ("remove", "Remove a package set"),
        # ("start-new", "Start a new MPSD software release version"),
        ("status", "Show status: what is installed?"),
        ("prepare", "Prepare installation of MPSD-release (dev only)"),
    ]
    for cmd, help_text in list_of_cmds:
        subp = subparsers.add_parser(cmd, help=help_text)

        if cmd == "start-new":
            subp.add_argument(
                "--from-release",
                dest="from_release",
                type=str,
                required=True,
                help="Release version to start from",
            )
            subp.add_argument(
                "--to-release",
                dest="to_release",
                type=str,
                required=True,
                help="Release version to create",
            )

        else:
            # all commands except init need a release version
            if cmd != "init":
                subp.add_argument(
                    "release",
                    type=str,
                    help="Release version to prepare, install, reinstall or remove",
                )
            if cmd in ["install", "reinstall", "remove"]:
                # "install" command needs additional documentation
                package_set_help = (
                    f"One or more package sets (like toolchains) to be {cmd}ed. "
                    "Use 'ALL' to refer to all available package sets."
                )

                subp.add_argument(
                    "package_set",  # first option defines attribute
                    # name `args.package_set` in `args = parser_args()`
                    type=str,
                    nargs="+",
                    default="NONE",
                    help=package_set_help,
                )
                subp.add_argument(
                    "--enable-build-cache",
                    action="store_true",
                    help=(
                        "Enable Spack build cache. Useful for reinstallation but "
                        "consumes time and disk space."
                    ),
                )
    # Carry out the action
    args = parser.parse_args()

    # Set up logging without file handle:
    # this is used in the init action and for logging the
    # get_root_dir() function
    set_up_logging(args.loglevel)

    # Check if the action is init
    # if so, call the init function and exit
    if args.action == "init":
        initialize_environment(Path(os.getcwd()))
        sys.exit(0)

    # root dir is the place where this script is called from
    root_dir = get_root_dir()

    # set up logging ( with file handler) for all actions (except init)
    log_file = get_installer_log_file_path(
        args.release,
        args.action,
        root_dir,
    )
    set_up_logging(
        args.loglevel,
        log_file,
    )

    # sanity check for common mistakes in command line arguments
    if args.release.endswith("/"):  # happens easily with autocompletion
        logging.error(
            f"You provided mpsd-release='{args.release}'. "
            f"Did you mean '{args.release.removesuffix('/')}'?"
        )
        sys.exit(1)

    # Check the command and run related function
    if args.action == "remove":
        remove_environment(args.release, root_dir, args.package_set)
    elif args.action == "start-new":
        start_new_environment(args.from_release, args.to_release, root_dir)
    elif args.action == "install":
        install_environment(
            args.release, args.package_set, root_dir, args.enable_build_cache
        )
    elif args.action == "status":
        _ = environment_status(args.release, root_dir)
    elif args.action == "prepare":
        prepare_environment(args.release, root_dir)
    elif args.action == "available":
        get_available_package_sets(args.release)
    else:
        message = (
            f"No known action found ({args.action=}). Should probably never happen."
        )
        logging.error(message)
        raise NotImplementedError(message)


if __name__ == "__main__":
    main()