scripts/tools/utils/git_utils/common.py

from tqdm import tqdm
import json
from PIL import Image
import base64
import random
import os
import argparse
import sys
import os.path as op
import yaml
import logging
from pprint import pformat


class Config(object):
    def __init__(self, default, overwrite):
        self.default = default
        self.overwrite = overwrite

    def get(self, k):
        if dict_has_path(self.default, k):
            base = dict_get_path_value(self.default, k)
        else:
            base = None
        if dict_has_path(self.overwrite, k):
            over = dict_get_path_value(self.overwrite, k)
            if isinstance(base, dict):
                assert isinstance(over, dict)
                base.update(over)
            else:
                base = over
        return base

    def __getattr__(self, k):
        return self.get(k)

    def __copy__(self):
        return Config(self.default, self.overwrite)

    def __deepcopy__(self, memo):
        from copy import deepcopy

        return Config(deepcopy(self.default), deepcopy(self.overwrite))

    def get_dict(self):
        import copy

        default = copy.deepcopy(self.default)
        for p in get_all_path(self.overwrite, with_list=False):
            v = dict_get_path_value(self.overwrite, p)
            dict_update_path_value(default, p, v)
        return default


def dict_remove_path(d, p):
    ps = p.split("$")
    assert len(ps) > 0
    cur_dict = d
    need_delete = ()
    while True:
        if len(ps) == 1:
            if len(need_delete) > 0 and len(cur_dict) == 1:
                del need_delete[0][need_delete[1]]
            else:
                del cur_dict[ps[0]]
            return
        else:
            if len(cur_dict) == 1:
                if len(need_delete) == 0:
                    need_delete = (cur_dict, ps[0])
            else:
                need_delete = (cur_dict, ps[0])
            cur_dict = cur_dict[ps[0]]
            ps = ps[1:]


def dict_has_path(d, p, with_type=False):
    ps = p.split("$")
    cur_dict = d
    while True:
        if len(ps) > 0:
            k = dict_parse_key(ps[0], with_type)
            if isinstance(cur_dict, dict) and k in cur_dict:
                cur_dict = cur_dict[k]
                ps = ps[1:]
            elif isinstance(cur_dict, list):
                try:
                    k = int(k)
                except:
                    return False
                cur_dict = cur_dict[k]
                ps = ps[1:]
            else:
                return False
        else:
            return True


def dict_update_nested_dict(a, b, overwrite=True):
    for k, v in b.items():
        if k not in a:
            dict_update_path_value(a, k, v)
        else:
            if isinstance(dict_get_path_value(a, k), dict) and isinstance(v, dict):
                dict_update_nested_dict(dict_get_path_value(a, k), v, overwrite)
            else:
                if overwrite:
                    dict_update_path_value(a, k, v)


def get_mpi_rank():
    if "RANK" in os.environ:
        return int(os.environ["RANK"])
    return int(os.environ.get("OMPI_COMM_WORLD_RANK", "0"))


def get_mpi_local_rank():
    if "LOCAL_RANK" in os.environ:
        return int(os.environ["LOCAL_RANK"])
    return int(os.environ.get("OMPI_COMM_WORLD_LOCAL_RANK", "0"))


def get_mpi_size():
    if "WORLD_SIZE" in os.environ:
        return int(os.environ["WORLD_SIZE"])
    return int(os.environ.get("OMPI_COMM_WORLD_SIZE", "1"))


def dict_ensure_path_key_converted(a):
    for k in list(a.keys()):
        v = a[k]
        if "$" in k:
            parts = k.split("$")
            x = {}
            x_curr = x
            for p in parts[:-1]:
                x_curr[p] = {}
                x_curr = x_curr[p]
            if isinstance(v, dict):
                dict_ensure_path_key_converted(v)
            x_curr[parts[-1]] = v
            dict_update_nested_dict(a, x)
            del a[k]
        else:
            if isinstance(v, dict):
                dict_ensure_path_key_converted(v)


def dict_update_path_value(d, p, v):
    ps = p.split("$")
    while True:
        if len(ps) == 1:
            d[ps[0]] = v
            break
        else:
            if ps[0] not in d:
                d[ps[0]] = {}
            d = d[ps[0]]
            ps = ps[1:]


def dict_parse_key(k, with_type):
    if with_type:
        if k[0] == "i":
            return int(k[1:])
        else:
            return k[1:]
    return k


def dict_get_path_value(d, p, with_type=False):
    ps = p.split("$")
    cur_dict = d
    while True:
        if len(ps) > 0:
            k = dict_parse_key(ps[0], with_type)
            if isinstance(cur_dict, (tuple, list)):
                cur_dict = cur_dict[int(k)]
            else:
                cur_dict = cur_dict[k]
            ps = ps[1:]
        else:
            return cur_dict


def releaseLock(locked_file_descriptor):
    locked_file_descriptor.close()


def print_trace():
    import traceback

    traceback.print_exc()


def hash_sha1(s):
    import hashlib

    if type(s) is not str:
        s = pformat(s)
    return hashlib.sha1(s.encode("utf-8")).hexdigest()


def write_to_file(contxt, file_name, append=False):
    p = os.path.dirname(file_name)
    ensure_directory(p)
    if type(contxt) is str:
        contxt = contxt.encode()
    flag = "wb"
    if append:
        flag = "ab"
    with open(file_name, flag) as fp:
        fp.write(contxt)


def ensure_directory(path):
    if path == "" or path == ".":
        return
    if path != None and len(path) > 0:
        assert not op.isfile(path), "{} is a file".format(path)
        if not os.path.exists(path) and not op.islink(path):
            try:
                os.makedirs(path)
            except:
                if os.path.isdir(path):
                    # another process has done makedir
                    pass
                else:
                    raise


def pilimg_from_base64(imagestring):
    try:
        import io

        jpgbytestring = base64.b64decode(imagestring)
        image = Image.open(io.BytesIO(jpgbytestring))
        image = image.convert("RGB")
        return image
    except:
        return None


def json_dump(obj):
    # order the keys so that each operation is deterministic though it might be
    # slower
    return json.dumps(obj, sort_keys=True, separators=(",", ":"))


def acquireLock(lock_f="/tmp/lockfile.LOCK"):
    import fcntl

    ensure_directory(op.dirname(lock_f))
    locked_file_descriptor = open(lock_f, "w+")
    fcntl.lockf(locked_file_descriptor, fcntl.LOCK_EX)
    return locked_file_descriptor


def get_user_name():
    import getpass

    return getpass.getuser()


def limited_retry_agent(num, func, *args, **kwargs):
    for i in range(num):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            logging.warning(
                "fails with \n{}: tried {}/{}-th time".format(
                    e,
                    i + 1,
                    num,
                )
            )
            import time

            print_trace()
            if i == num - 1:
                raise
            t = random.random() * 5
            time.sleep(t)


def exclusive_open_to_read(fname, mode="r"):
    disable_lock = os.environ.get("QD_DISABLE_EXCLUSIVE_READ_BY_LOCK")
    if disable_lock is not None:
        disable_lock = int(disable_lock)
    if not disable_lock:
        user_name = get_user_name()
        lock_fd = acquireLock(
            op.join("/tmp", "{}_lock_{}".format(user_name, hash_sha1(fname)))
        )
    # try:
    # in AML, it could fail with Input/Output error. If it fails, we will
    # use azcopy as a fall back solution for reading
    fp = limited_retry_agent(10, open, fname, mode)
    if not disable_lock:
        releaseLock(lock_fd)
    return fp


def read_to_buffer(file_name):
    with open(file_name, "rb") as fp:
        all_line = fp.read()
    return all_line


def load_list_file(fname):
    with open(fname, "r") as fp:
        lines = fp.readlines()
    result = [line.strip() for line in lines]
    if len(result) > 0 and result[-1] == "":
        result = result[:-1]
    return result


def init_logging():
    ch = logging.StreamHandler(stream=sys.stdout)
    ch.setLevel(logging.INFO)
    logger_fmt = logging.Formatter(
        "%(asctime)s.%(msecs)03d %(process)d:%(filename)s:%(lineno)s %(funcName)10s(): %(message)s"
    )
    ch.setFormatter(logger_fmt)

    root = logging.getLogger()
    root.handlers = []
    root.addHandler(ch)
    root.setLevel(logging.INFO)


def load_from_yaml_str(s):
    return yaml.load(s, Loader=yaml.UnsafeLoader)


def get_all_path(d, with_type=False, leaf_only=True, with_list=True):
    assert not with_type, "will not support"
    all_path = []

    if isinstance(d, dict):
        for k, v in d.items():
            all_sub_path = get_all_path(
                v, with_type, leaf_only=leaf_only, with_list=with_list
            )
            all_path.extend([k + "$" + p for p in all_sub_path])
            if not leaf_only or len(all_sub_path) == 0:
                all_path.append(k)
    elif (isinstance(d, tuple) or isinstance(d, list)) and with_list:
        for i, _v in enumerate(d):
            all_sub_path = get_all_path(
                _v,
                with_type,
                leaf_only=leaf_only,
                with_list=with_list,
            )
            all_path.extend(["{}$".format(i) + p for p in all_sub_path])
            if not leaf_only or len(all_sub_path) == 0:
                all_path.append("{}".format(i))
    return all_path


def load_from_yaml_file(file_name):
    # do not use QDFile.open as QDFile.open depends on this function
    with exclusive_open_to_read(file_name, "r") as fp:
        # with open(file_name, 'r') as fp:
        data = load_from_yaml_str(fp)
    while isinstance(data, dict) and "_base_" in data:
        b = op.join(op.dirname(file_name), data["_base_"])
        result = load_from_yaml_file(b)
        assert isinstance(result, dict)
        del data["_base_"]
        all_key = get_all_path(data, with_list=False)
        for k in all_key:
            v = dict_get_path_value(data, k)
            dict_update_path_value(result, k, v)
        data = result
    return data


def parse_general_args():
    parser = argparse.ArgumentParser(description="General Parser")
    parser.add_argument("-c", "--config_file", help="config file", type=str)
    parser.add_argument("-p", "--param", help="parameter string, yaml format", type=str)
    parser.add_argument(
        "-bp", "--base64_param", help="base64 encoded yaml format", type=str
    )
    args = parser.parse_args()
    kwargs = {}
    if args.config_file:
        logging.info("loading parameter from {}".format(args.config_file))
        configs = load_from_yaml_file(args.config_file)
        for k in configs:
            kwargs[k] = configs[k]
    if args.base64_param:
        configs = load_from_yaml_str(base64.b64decode(args.base64_param))
        for k in configs:
            if k not in kwargs:
                kwargs[k] = configs[k]
            elif kwargs[k] == configs[k]:
                continue
            else:
                logging.info(
                    "overwriting {} to {} for {}".format(kwargs[k], configs[k], k)
                )
                kwargs[k] = configs[k]
    if args.param:
        configs = load_from_yaml_str(args.param)
        dict_ensure_path_key_converted(configs)
        for k in configs:
            if k not in kwargs:
                kwargs[k] = configs[k]
            elif kwargs[k] == configs[k]:
                continue
            else:
                logging.info(
                    "overwriting {} to {} for {}".format(kwargs[k], configs[k], k)
                )
                kwargs[k] = configs[k]
    return kwargs


def qd_tqdm(*args, **kwargs):
    desc = kwargs.get("desc", "")
    import inspect

    frame = inspect.currentframe()
    frames = inspect.getouterframes(frame)
    frame = frames[1].frame
    line_number = frame.f_lineno
    fname = op.basename(frame.f_code.co_filename)
    message = "{}:{}".format(fname, line_number)

    if "desc" in kwargs:
        kwargs["desc"] = message + " " + desc
    else:
        kwargs["desc"] = message

    if "mininterval" not in kwargs:
        # every 2 secons; default is 0.1 second which is too frequent
        kwargs["mininterval"] = 2

    return tqdm(*args, **kwargs)