Utilities¶

kudio.util.tools ¶

Small generic helpers: timing decorators and list utilities.

Classes:

Name	Description
`Timer`	Decorator: run func in a daemon thread, optionally joining for at most

Functions:

Name	Description
`timer`	Decorator: log the wall-clock run time of func.
`find_duplicate`	Return the `Counter` dict of file_list, logging duplicated entries.

Timer ¶

Timer(time_delay: float)

Decorator: run func in a daemon thread, optionally joining for at most time_delay seconds.

>>> @Timer(time_delay=3)
... def job(): ...
>>> thread = job()   # returns the started thread

Source code in kudio/util/tools.py

def __init__(self, time_delay: float):
    self.time_delay = time_delay

timer ¶

timer(func: Callable) -> Callable

Decorator: log the wall-clock run time of func.

Source code in kudio/util/tools.py

def timer(func: Callable) -> Callable:
    """Decorator: log the wall-clock run time of *func*."""

    @wraps(func)
    def wrapper(*args, **kwargs):
        before = time.time()
        result = func(*args, **kwargs)
        log.info("%s(): %.7f s", func.__name__, time.time() - before)
        return result

    return wrapper

find_duplicate ¶

find_duplicate(
    file_list: Iterable[Hashable],
) -> Dict[Hashable, int]

Return the Counter dict of file_list, logging duplicated entries.

Source code in kudio/util/tools.py

def find_duplicate(file_list: Iterable[Hashable]) -> Dict[Hashable, int]:
    """Return the ``Counter`` dict of *file_list*, logging duplicated entries."""
    counts = dict(Counter(file_list))
    duplicates = {key: value for key, value in counts.items() if value > 1}
    if duplicates:
        log.info("Duplicates: %s", duplicates)
    return counts

kudio.util.parallel ¶

Parallel batch processing helpers.

Functions:

Name	Description
`map_waves`	Apply fn to every item, optionally across `workers` processes.

map_waves ¶

map_waves(
    fn: Callable,
    items: Iterable,
    workers: int = 1,
    desc: Optional[str] = None,
    chunksize: int = 1,
) -> List

Apply fn to every item, optionally across workers processes.

With workers <= 1 runs sequentially (easier to debug); otherwise uses a multiprocessing pool. fn must be top-level (picklable) for workers > 1.

Source code in kudio/util/parallel.py

def map_waves(fn: Callable, items: Iterable, workers: int = 1,
              desc: Optional[str] = None, chunksize: int = 1) -> List:
    """Apply *fn* to every item, optionally across ``workers`` processes.

    With ``workers <= 1`` runs sequentially (easier to debug); otherwise uses a
    multiprocessing pool. *fn* must be top-level (picklable) for ``workers > 1``.
    """
    items = list(items)
    if workers and workers > 1 and len(items) > 1:
        with Pool(workers) as pool:
            return list(tqdm(pool.imap(fn, items, chunksize=chunksize),
                             total=len(items), desc=desc or "map_waves"))
    return [fn(x) for x in tqdm(items, desc=desc or "map_waves")]

kudio.util.colors ¶

ANSI-colored console helpers used by the CLI tools and the GUI.

Colors ¶

ANSI escape sequences (SGR codes).

kudio.util.conv ¶

Data conversion/export helpers.

Functions:

Name	Description
`data2xlsx`	Write a dict or DataFrame to a sheet of an `.xlsx` file.
`lowercase`	The strings in lst, lower-cased. Anything that is not a string is
`uppercase`	The strings in lst, upper-cased. Anything that is not a string is

data2xlsx ¶

data2xlsx(
    data: Union[dict, DataFrame],
    xlsx_dir: Union[str, Path],
    sheet_name: str,
    index: bool = False,
    fit_column: bool = True,
) -> Optional[str]

Write a dict or DataFrame to a sheet of an .xlsx file.

If the file already exists, the sheet is replaced; other sheets are kept. With fit_column=True, column widths are auto-sized to the content.

Source code in kudio/util/conv.py

def data2xlsx(data: Union[dict, pd.DataFrame], xlsx_dir: Union[str, Path],
              sheet_name: str, index: bool = False,
              fit_column: bool = True) -> Optional[str]:
    """Write a dict or DataFrame to a sheet of an ``.xlsx`` file.

    If the file already exists, the sheet is replaced; other sheets are kept.
    With ``fit_column=True``, column widths are auto-sized to the content.
    """
    pd = _pandas()
    if isinstance(data, dict):
        try:
            data_frame = pd.DataFrame.from_dict(data, orient='columns')
        except Exception as e:
            log.error("data2xlsx: bad data: %s", e)
            return f'[data2xlsx] data error: {e}'
    elif isinstance(data, pd.DataFrame):
        data_frame = data
    else:
        return '[data2xlsx] check the data type'

    xlsx_dir = Path(xlsx_dir)
    xlsx_dir.parent.mkdir(parents=True, exist_ok=True)

    # one writer for both cases: a new file used to skip fit_column entirely
    append = xlsx_dir.is_file()
    with pd.ExcelWriter(xlsx_dir, engine='openpyxl', mode='a' if append else 'w',
                        **({'if_sheet_exists': 'replace'} if append else {})) as writer:
        data_frame.to_excel(writer, sheet_name=sheet_name, index=index)
        if fit_column:
            from openpyxl.utils import get_column_letter
            worksheet = writer.book[sheet_name]
            for column_cells in worksheet.columns:
                length = max(len(str(cell.value)) for cell in column_cells)
                worksheet.column_dimensions[
                    get_column_letter(column_cells[0].column)].width = length
    return None

lowercase ¶

lowercase(lst: Iterable) -> List[str]

The strings in lst, lower-cased. Anything that is not a string is left out.

Source code in kudio/util/conv.py

def lowercase(lst: Iterable) -> List[str]:
    """The strings in *lst*, lower-cased. Anything that is not a string is
    left out."""
    return [_.lower() for _ in lst if isinstance(_, str)]

uppercase ¶

uppercase(lst: Iterable) -> List[str]

The strings in lst, upper-cased. Anything that is not a string is left out.

Source code in kudio/util/conv.py

def uppercase(lst: Iterable) -> List[str]:
    """The strings in *lst*, upper-cased. Anything that is not a string is
    left out."""
    return [_.upper() for _ in lst if isinstance(_, str)]

kudio.util.visual ¶

Plotting helpers: waveforms, spectrograms, training histories.

Classes:

Name	Description
`Common`	Base figure wrapper with save/show/close helpers.
`WaveVisualizer`	Plot waveform + spectrogram side by side for one or more files.

Functions:

Name	Description
`history_plot`	Plot every history curve whose key contains metrics (Keras History).

Common ¶

Common(fig_dir: str = 'kd_figs', is_save: bool = False)

Base figure wrapper with save/show/close helpers.

Methods:

Name	Description
`plot_stft_w_2mfe`	Waveform + two mel spectrograms (full band vs. `sub_mfe_fq`).