Source code for bob.devtools.bootstrap

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""Bootstraps a new miniconda installation and prepares it for development."""

import glob
import hashlib
import logging
import os
import platform
import shutil
import subprocess
import sys
import time

add_pip_as_python_dependency: false #!final
always_yes: true #!final
anaconda_upload: false #!final
channel_priority: strict #!final
conda_build: #!final
  pkg_format: '2'
default_channels: #!final
  - conda-forge
channel_alias: #!final
quiet: true #!final
remote_connect_timeout_secs: 120.0 #!final
remote_max_retries: 50 #!final
remote_read_timeout_secs: 180.0 #!final
show_channel_urls: true #!final
ssl_verify: false #!final

_SERVER = ""

    ("weeks", 604800),  # 60 * 60 * 24 * 7
    ("days", 86400),  # 60 * 60 * 24
    ("hours", 3600),  # 60 * 60
    ("minutes", 60),
    ("seconds", 1),
"""Time intervals that make up human readable time slots"""

logger = logging.getLogger(__name__)

[docs]def set_environment(name, value, env=os.environ): """Function to setup the environment variable and print debug message. Args: name: The name of the environment variable to set value: The value to set the environment variable to env: Optional environment (dictionary) where to set the variable at """ env[name] = value'environ["%s"] = %s', name, value) return value
[docs]def human_time(seconds, granularity=2): """Returns a human readable time string like "1 day, 2 hours".""" result = [] for name, count in _INTERVALS: value = seconds // count if value: seconds -= value * count if value == 1: name = name.rstrip("s") result.append("{} {}".format(int(value), name)) else: # Add a blank if we're in the middle of other values if len(result) > 0: result.append(None) if not result: if seconds < 1.0: return "%.2f seconds" % seconds else: if seconds == 1: return "1 second" else: return "%d seconds" % seconds return ", ".join([x for x in result[:granularity] if x is not None])
[docs]def run_cmdline(cmd, env=None, **kwargs): """Runs a command on a environment, logs output and reports status. Parameters: cmd (list): The command to run, with parameters separated on a list env (dict, Optional): Environment to use for running the program on. If not set, use :py:obj:`os.environ`. """ if env is None: env = os.environ"(system) %s" % " ".join(cmd)) start = time.time() p = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=env, bufsize=1, universal_newlines=True, **kwargs, ) for line in iter(p.stdout.readline, ""): sys.stdout.write(line) sys.stdout.flush() if p.wait() != 0: raise RuntimeError( "command `%s' exited with error state (%d)" % (" ".join(cmd), p.returncode) ) total = time.time() - start"command took %s" % human_time(total))
[docs]def touch(path): """Python-implementation of the "touch" command-line application.""" with open(path, "a"): os.utime(path, None)
[docs]def merge_conda_cache(cache, prefix, name): """Merges conda pkg caches and conda-bld folders. Args: cache: The cached directory (from previous builds) prefix: The current prefix (root of conda installation) name: The name of the current package """ pkgs_dir = os.path.join(prefix, "pkgs") pkgs_urls_txt = os.path.join(pkgs_dir, "urls.txt") if not os.path.exists(pkgs_dir):"mkdir -p %s", pkgs_dir) os.makedirs(pkgs_dir)"touch %s", pkgs_urls_txt) touch(pkgs_urls_txt) # move packages on cache/pkgs to pkgs_dir cached_pkgs_dir = os.path.join(cache, "pkgs") cached_packages = glob.glob(os.path.join(cached_pkgs_dir, "*.tar.bz2")) cached_packages.extend(glob.glob(os.path.join(cached_pkgs_dir, "*.conda"))) cached_packages = [ k for k in cached_packages if not k.startswith(name + "-") ]"Merging %d cached conda packages", len(cached_packages)) for k in cached_packages: dst = os.path.join(pkgs_dir, os.path.basename(k)) logger.debug("(move) %s -> %s", k, dst) os.rename(k, dst) # merge urls.txt files"Merging urls.txt files from cache...") cached_pkgs_urls_txt = os.path.join(cached_pkgs_dir, "urls.txt") if not os.path.exists(cached_pkgs_urls_txt): with open(pkgs_urls_txt, "rb") as f1: data = set(f1.readlines()) data = sorted(list(data)) else: # use both cached and actual conda package caches with open(pkgs_urls_txt, "rb") as f1, open( cached_pkgs_urls_txt, "rb" ) as f2: data = set(f1.readlines() + f2.readlines()) data = sorted(list(data)) with open(pkgs_urls_txt, "wb") as f: f.writelines(data) pkgs_urls = os.path.join(pkgs_dir, "urls") touch(pkgs_urls) # move conda-bld build results cached_conda_bld = os.path.join(cache, "conda-bld") if os.path.exists(cached_conda_bld): dst = os.path.join(prefix, "conda-bld")"(move) %s -> %s", cached_conda_bld, dst) os.rename(cached_conda_bld, dst)
[docs]def ensure_miniconda_sh(): """Retrieves the miniconda3 installer for the current system. Checks the hash of the miniconda3 installer against the expected version, if that does not match, erase existing installer and re-downloads new installer. """ # WARNING: if you update this version, remember to update hashes below # AND our "mirror" in the internal webserver path = "" if platform.system().lower() == "darwin": # apple silicon system = "MacOSX" if platform.machine().lower() == "arm64": sha256 = "de7c7f229d05104de802f1f729a595736b08139c4ae59ba8ba0049050d63c98f" machine = "arm64" else: # intel sha256 = "5455900cf1298f21333b7c0d1ec159952e1ef5426563cc97eb7e42053d608afc" machine = "x86_64" else: system = "Linux" if platform.machine().lower() == "aarch64": # raspberry pi sha256 = "7f2a0282819f1a12bb44628a836b96fe9dfb21a5920fb8331325c47f91bf9a81" machine = "aarch64" else: # intel sha256 = "0be3654cc3b9c43d3aeeeca5efe6d2f31e9f7711702f3818529b367b3db677fb" machine = "x86_64" path = path % (system, machine) if os.path.exists(""):"(check) sha256 (== %s?)", sha256) actual_sha256 = hashlib.sha256( open("", "rb").read() ).hexdigest() if actual_sha256 == sha256:"Re-using cached miniconda3 installer (hash matches)") return else: "Erasing cached miniconda3 installer (%s does NOT match)", actual_sha256, ) os.unlink("") # re-downloads installer import urllib.request dst = """(download) %s -> %s...", path, dst) response = urllib.request.urlopen(path) with open(dst, "wb") as f: f.write( # checks that the checksum is correct on this file actual_sha256 = hashlib.sha256( open("", "rb").read() ).hexdigest() if actual_sha256 != sha256: os.unlink("") raise RuntimeError( "Just downloaded miniconda3 installer sha256 checksum (%s) does " "NOT match expected value (%s). Removing downloaded installer. " "A wrong checksum may end up making the CI download too many copies " "and be banned! You must fix this ASAP." % ( actual_sha256, sha256, ) )
[docs]def install_miniconda(prefix, name): """Creates a new miniconda installation. Args: prefix: The path leading to the (new) root of the miniconda installation name: The name of this package """"Installing miniconda in %s...", prefix) ensure_miniconda_sh() cached = None if os.path.exists(prefix): # this is the previous cache, move it cached = prefix + ".cached" if os.path.exists(cached):"(rmtree) %s", cached) shutil.rmtree(cached)"(move) %s -> %s", prefix, cached) os.rename(prefix, cached) # remove /opt/conda entries from path to avoid conflicts env = os.environ.copy() env["PATH"] = ":".join( [p for p in env["PATH"].split(":") if not p.startswith("/opt/conda/")] ) run_cmdline(["bash", "", "-b", "-p", prefix], env=env) if cached is not None: merge_conda_cache(cached, prefix, name) shutil.rmtree(cached)
[docs]def get_channels( public, stable, server, intranet, group, add_dependent_channels=False ): """Returns the relevant conda channels to consider if building project. The subset of channels to be returned depends on the visibility and stability of the package being built. Here are the rules: * public and stable: returns the public stable channel * public and not stable: returns the public beta channel * not public and stable: returns both public and private stable channels * not public and not stable: returns both public and private beta channels Public channels have priority over private channles, if turned. Args: public: Boolean indicating if we're supposed to include only public channels stable: Boolean indicating if we're supposed to include stable channels server: The base address of the server containing our conda channels intranet: Boolean indicating if we should add "private"/"public" prefixes on the conda paths group: The group of packages (gitlab namespace) the package we're compiling is part of. Values should match URL namespaces currently available on our internal webserver. Currently, only "bob" or "beat" will work. add_dependent_channels: If True, will add the conda-forge channel to the list Returns: a list of channels that need to be considered. """ if (not public) and (not intranet): raise RuntimeError( "You cannot request for private channels and set" " intranet=False (server=%s) - these are conflicting options" % server ) channels = [] channels_dict = {} # do not use '/public' urls for public channels prefix = "/software/" + group if stable: channels += [server + prefix + "/conda"] channels_dict["public/stable"] = channels[-1] else: channels += [server + prefix + "/conda/label/beta"] # allowed betas channels_dict["public/beta"] = channels[-1] if not public: prefix = "/private" if stable: # allowed private channels channels += [server + prefix + "/conda"] channels_dict["private/stable"] = channels[-1] else: channels += [server + prefix + "/conda/label/beta"] # allowed betas channels_dict["private/beta"] = channels[-1] upload_channel = channels_dict[ "{}/{}".format( "public" if public else "private", "stable" if stable else "beta" ) ] if add_dependent_channels: channels += ["conda-forge"] return channels, upload_channel
[docs]def setup_logger(logger, level): """Sets-up the logging for this command at level ``INFO``""" warn_err = logging.StreamHandler(sys.stderr) warn_err.setLevel(logging.WARNING) logger.addHandler(warn_err) # debug and info messages are written to sys.stdout class _InfoFilter: def filter(self, record): return record.levelno <= logging.INFO debug_info = logging.StreamHandler(sys.stdout) debug_info.setLevel(logging.DEBUG) debug_info.addFilter(_InfoFilter()) logger.addHandler(debug_info) formatter = logging.Formatter("%(levelname)s@%(asctime)s: %(message)s") for handler in logger.handlers: handler.setFormatter(formatter) if level not in range(0, 4): raise ValueError( "The verbosity level %d does not exist. Please reduce the number of " "'--verbose' parameters in your command line" % level ) # set up the verbosity level of the logging system log_level = { 0: logging.ERROR, 1: logging.WARNING, 2: logging.INFO, 3: logging.DEBUG, }[level] logger.setLevel(log_level)
if __name__ == "__main__": import argparse parser = argparse.ArgumentParser( description="Bootstraps a new miniconda " "installation and prepares it for development" ) parser.add_argument( "command", choices=["build", "local", "channel"], help="How to prepare the current environment. Use: ``build``, to " "build bob.devtools, ``local``, to bootstrap deploy or pypi " "stages for bob.devtools builds, ``channel`` channel to bootstrap " "CI environment for beta/stable builds", ) parser.add_argument( "envname", nargs="?", default="bdt", help="The name of the conda environment that will host bdt " "[default: %(default)s]", ) parser.add_argument( "-n", "--name", default=os.environ.get("CI_PROJECT_NAME", "bob.devtools"), help="The name of the project being built [default: %(default)s]", ) parser.add_argument( "-c", "--conda-root", default=os.environ.get( "CONDA_ROOT", os.path.realpath(os.path.join(os.curdir, "miniconda")) ), help="The location where we should install miniconda " "[default: %(default)s]", ) parser.add_argument( "-t", "--tag", default=os.environ.get("CI_COMMIT_TAG", None), help="If building a tag, pass it with this flag [default: %(default)s]", ) parser.add_argument( "-p", "--python", default=os.environ.get("PYTHON_VERSION", None), help="If a specific python version must be used, " "bootstraps a new environment with it [default: %(default)s]", ) parser.add_argument( "--no-proxy", default=("BDT_NO_PROXY" in os.environ), action="store_true", help="If specified, bootstrap will not use the reverse caching " "proxy for downloading conda packages. You may also set the " "environment variable BDT_NO_PROXY to any value, to disable " "this feature.", ) parser.add_argument( "--verbose", "-v", action="count", default=0, help="Increases the verbosity level. We always prints error and " "critical messages. Use a single ``-v`` to enable warnings, " "two ``-vv`` to enable information messages and three ``-vvv`` " "to enable debug messages [default: %(default)s]", ) args = parser.parse_args() setup_logger(logger, args.verbose) condarc = os.path.join(args.conda_root, "condarc") install_miniconda(args.conda_root, conda_bin = os.path.join(args.conda_root, "bin", "mamba") # creates the condarc file condarc = os.path.join(args.conda_root, "condarc")"(create) %s", condarc) with open(condarc, "wt") as f: if args.no_proxy:"Disabling reverse caching proxy for conda channels...") f.write(_BASE_CONDARC) else:"Enabling reverse caching proxy for conda channels...") # Replaces and # by our proxies, so it is optimized for a CI build. Notice we # consider this script is only executed in this context. The URL # should NOT work outside of Idiap's network. f.write( _BASE_CONDARC.replace( "", "", ).replace( "", "", ) ) # These are the same versions as in bob.devtools/conda/meta.yaml conda_version = "22" conda_build_version = "3" mamba_version = "1" boa_version = "0.14" conda_verbosity = [] # if args.verbose >= 2: # conda_verbosity = ['-v'] if args.verbose >= 3: conda_verbosity = ["-vv"] # print conda information for debugging purposes run_cmdline([conda_bin, "info"] + conda_verbosity) should_install_git = ["git"] # check if platform is mac, then don't install git # see: if platform.system() == "Darwin": should_install_git = [] if args.command == "build": # clean conda cache and packages before building run_cmdline([conda_bin, "clean", "--all"]) # Just use the conda-forge channels when self building run_cmdline( [conda_bin, "install", "--yes"] + conda_verbosity + [ "-c", "conda-forge", "-n", "base", "python", "conda=%s" % conda_version, "conda-build=%s" % conda_build_version, "mamba=%s" % mamba_version, "boa=%s" % boa_version, "click", "twine", # required for checking readme of python (zip) distro ] + should_install_git ) elif args.command == "local": # index the locally built packages run_cmdline( [conda_bin, "install", "--yes"] + conda_verbosity + [ "-c", "conda-forge", "-n", "base", "python", "conda=%s" % conda_version, "conda-build=%s" % conda_build_version, "mamba=%s" % mamba_version, "boa=%s" % boa_version, "twine", # required for checking readme of python (zip) distro ] + should_install_git ) conda_bld_path = os.path.join(args.conda_root, "conda-bld") run_cmdline([conda_bin, "index", conda_bld_path]) channels, _ = get_channels( public=True, stable=True, server=_SERVER, intranet=True, group="bob", add_dependent_channels=True, ) channels = ( ["--override-channels"] + ["--channel=" + conda_bld_path] + ["--channel=%s" % k for k in channels] ) conda_cmd = "install" if args.envname in ("base", "root") else "create" run_cmdline( [conda_bin, conda_cmd, "--yes"] + conda_verbosity + channels + ["-n", args.envname, "bob.devtools"] ) elif args.command == "channel": # installs from channel channels, _ = get_channels( public=True, stable=(args.tag is not None), server=_SERVER, intranet=True, group="bob", add_dependent_channels=True, ) channels = ["--override-channels"] + [ "--channel=%s" % k for k in channels ] conda_cmd = "install" if args.envname in ("base", "root") else "create" cmd = ( [conda_bin, conda_cmd, "--yes"] + conda_verbosity + channels + ["-n", args.envname] ) # can only enforce python version on newly created environments if conda_cmd == "create" and args.python is not None: cmd += [f"python={args.python}"] cmd += ["bob.devtools"] if conda_cmd == "install": cmd += ["--update-specs"] run_cmdline(cmd)