#!/usr/bin/env python3
# =========================================================================
# Program: iota2
#
# Copyright (c) CESBIO. All rights reserved.
#
# See LICENSE for details.
#
# This software is distributed WITHOUT ANY WARRANTY; without even
# the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
# PURPOSE. See the above copyright notices for more information.
#
# =========================================================================
"""This module contains decorators and utils functions."""
import argparse
import inspect
import logging
import os
import subprocess # pylint: disable=no-name-in-module
from collections.abc import Callable, Generator, Mapping
from dataclasses import dataclass, fields, is_dataclass
from functools import wraps
from timeit import default_timer as timer
from typing import Any
LOGGER = logging.getLogger("distributed.worker")
[docs]@dataclass
class TaskConfig:
"""
Dataclass containing ressources for tasks
"""
logger: logging.Logger = LOGGER
ram: int = 128
[docs]def is_empty_dataclass(dataclass_inst: object) -> bool:
"""Check if at least one field of the dataclass is set."""
if not is_dataclass(dataclass_inst):
raise TypeError(f"{dataclass_inst} is not a dataclass.")
return not any(
getattr(dataclass_inst, field.name) is not None
for field in fields(dataclass_inst)
)
[docs]def chunker(list_of_things: list, chunk_size: int) -> Generator:
"""
Yield an iterator, splitting the input list by chunks of given size
Parameters
----------
list_of_things:
Input list
chunk_size:
Number of elements by list
"""
for pos in range(0, len(list_of_things), chunk_size):
yield list_of_things[pos : pos + chunk_size]
[docs]def print_types(func: Callable) -> Callable:
"""
Decorator to print types of arguments when calling a function.
"""
def wrapper(*args: list, **kwargs: dict) -> Any:
print(f"-----> calling function '{func.__name__}'")
for arg, arg_signature in zip(args, inspect.signature(func).parameters):
print(f"argument '{arg_signature}' with type {type(arg)}")
for arg_name, arg_val in kwargs.items():
print(f"argument '{arg_name}' with type {type(arg_val)}")
return func(*args, **kwargs)
return wrapper
# pylint: disable=too-few-public-methods
[docs]class RemoveInStringList:
"""Remove element decorator to remove strings in a list."""
[docs] def __init__(self, *args: str):
self.pattern_list = args
def __str__(self) -> str:
return self.__class__.__name__
def __call__(self, fun: Callable) -> Callable:
@wraps(fun)
def wrapped_f(*args: list) -> list:
results = fun(*args)
results_filtered = []
for elem in results:
pattern_found = False
for pattern in self.pattern_list:
if pattern in elem:
pattern_found = True
break
if pattern_found is False:
results_filtered.append(elem)
return results_filtered
return wrapped_f
[docs]def run(
cmd: str,
desc: str | None = None,
env: Mapping[str, str] | None = None,
logger: logging.Logger = LOGGER,
) -> int:
"""Launch a system command and raise an execption if fail.
Parameters
----------
cmd:
the system command to be launched
desc:
an optional description of the command for log_dir
env:
the environ variable if None, os.environ is used
logger:
by default module LOGGER value is used
"""
if env is None:
env = os.environ
# Create subprocess
start = timer()
logger.debug(f"run command : {cmd}")
with subprocess.Popen(
cmd, env=env, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT
) as proc:
# Get output as strings
out, err = proc.communicate()
output_log = str(out.rstrip()) if out else None
err_log = str(err) if err else None
# Get return code
rtc = proc.returncode
stop = timer()
# Log outputs
if desc is not None:
logger.debug(desc)
if output_log is not None:
logger.debug(f"out/err: {output_log}")
logger.debug(f"Done in {stop - start} seconds")
# Log error code
if rtc != 0:
logger.error(f"Command {cmd} exited with non-zero return code {rtc}")
exception_msg = f"Launch command fail : {cmd} \n {output_log} \n {err_log}"
raise Exception(exception_msg)
return rtc
[docs]def str2bool(val: str) -> bool:
"""Convert a string vue to boolean.
usage: use in argParse as function to parse options
Parameters
----------
val:
the value to convert
"""
if val.lower() in ("yes", "true", "t", "y", "1"):
return True
if val.lower() in ("no", "false", "f", "n", "0"):
return False
raise argparse.ArgumentTypeError("Boolean value expected.")
[docs]def is_nomenclature_castable_to_int(labels_table: dict[int, int | str]) -> bool:
"""Try to convert labels to int."""
all_castable = []
for _, user_label in labels_table.items():
try:
_ = int(user_label)
all_castable.append(True)
except ValueError:
all_castable.append(False)
return all(all_castable)