Personal utils in Python

# -*- coding: utf-8 -*-

import os
import sys
import shutil
from time import perf_counter
from collections import defaultdict
from datetime import timedelta
from typing import Sequence, List, Dict, Hashable, Callable

from config.common import DEBUG  # 项目配置
from logger import base_logger  # 日志模块

utils_logger = base_logger.getChild('utils')

def progressbar(it: Sequence, name: str = 'In Progress', bar_length: int = 20, file=sys.stdout, verbose=False):
    def show(i):
        total = len(it)
        bar_progress = bar_length * i // total
        consumedsec = stepsec - startsec
        file.write(
            f'{name} {it[i-1] if verbose else ""}: '
            f'[{"#" * bar_progress}{"." * (bar_length - bar_progress)}] '
            f'{i}/{total} ({i / total:.2%}) '
            f'Time consumed: {timedelta(seconds=round(consumedsec))}; '
            f'Time left (approx.): {timedelta(seconds=round((total-i) / (i/consumedsec))) if i != 0 else "--"}\r'
        )
        file.flush()

    startsec = stepsec = perf_counter()
    show(0)
    for i, item in enumerate(it, start=1):
        yield item
        stepsec = perf_counter()
        show(i)
    file.write("\n")
    file.flush()


def overwrite_dir(dir_name: str):
    if os.path.isdir(dir_name):
        shutil.rmtree(dir_name)
    os.makedirs(dir_name)


def makedirs_if_not_exists(dir_name: str):
    if not os.path.isdir(dir_name):
        os.makedirs(dir_name)


def group_listofdict_bykey(entries: List[Dict], k: Hashable):
    result = defaultdict(list)
    for entry in entries:
        result[entry[k]].append(entry)
    return result


def debug_run(func: Callable, *args, **kwargs):
    if DEBUG:
        return func(*args, **kwargs)


def timer(func: Callable):
    def wrapper(*args, **kwargs):
        start = perf_counter()
        res = func(*args, **kwargs)
        utils_logger.debug('%.3f sec taken by function %s',
                    perf_counter() - start, func.__name__)
        return res
    return wrapper