Utilities

---

---

In this file implemented the helper function

timing(f)

Decorator to measure and print execution time of a function. If it's a class method, logs as 'ClassName.method'. Else just 'function'.

Parameters:

Name	Type	Description	Default
f	callable	The function to be measured.	required

Returns:

Type	Description
callable	The wrapped function.

Source code in src/utils.py

def timing(f: callable) -> object:
    """
    Decorator to measure and print execution time of a function.
    If it's a class method, logs as 'ClassName.method'. Else just 'function'.

    Parameters
    ----------
    f : callable
        The function to be measured.

    Returns
    -------
    callable
        The wrapped function.
    """

    if asyncio.iscoroutinefunction(f):
        @wraps(f)
        async def wrap(*args, **kwargs) -> Tuple[Any, float]:
            t_0 = time()
            result = await f(*args, **kwargs)
            t_1 = time()
            return _log_duration(f, args, t_0, t_1, result)
    else:
        @wraps(f)
        def wrap(*args, **kwargs) -> Tuple[Any, float]:
            t_0 = time()
            result = f(*args, **kwargs)
            t_1 = time()
            return _log_duration(f, args, t_0, t_1, result)

    return wrap

generate_unit_vector(length)

Generate and normalize a vector of given length

Source code in src/utils.py

def generate_unit_vector(length: int):
    """Generate and normalize a vector of given length"""
    vector = np.random.uniform(size=length)
    normalized = vector / np.linalg.norm(vector)

    return normalized

json_save(path, obj)

Save supported files to json format

Parameters:

Name	Type	Description	Default
path	Union[str, Path]	path to save object	required

---

obj: Dict object

Source code in src/utils.py

def json_save(path: Union[str, Path], obj: Dict):
    """
    Save supported files to json format
    Parameters
    ----------
    path: Union[str, Path]
        path to save object

    -------
    obj: Dict
        object

    Returns
    ---------
    """
    with open(path, 'w', encoding='utf-8') as file:
        json.dump(obj, file)

json_load(path, encoding='utf-8')

Load json files from path

Parameters:

Name	Type	Description	Default
path	Union[str, Path]	path to save object	required
encoding	str	encoding type	'utf-8'

Returns:

Name	Type	Description
obj	object	Contents of the file

Source code in src/utils.py

def json_load(path: Union[str, Path], encoding='utf-8'):
    """
    Load json files from path

    Parameters
    ----------
    path : Union[str,Path]
        path to save object
    encoding : str
        encoding type

    Returns
    -------
    obj: object
        Contents of the file

    """
    with open(path, 'r', encoding=encoding) as file:
        obj = json.load(file)
        return obj

generate_guid(k=6)

Generate random id

Parameters:

Name	Type	Description	Default
k	int	length of guid	6

Returns:

Name	Type	Description
guid	str	random id

Source code in src/utils.py

def generate_guid(k: int = 6) -> str:
    """
    Generate random id
    Parameters
    ----------
    k : int
        length of guid

    Returns
    -------
    guid : str
        random id
    """

    guid = ''.join(random.choices(string.ascii_uppercase + string.digits, k=k))
    return guid

add_ids_to_dicts(ids, dicts)

Add ids to corresponding dictionaries

Parameters:

Name	Type	Description	Default
ids	List[int]	ids to add to the dictionaries in the same order as the dicts themselves	required
dicts	List[dict]		required

Returns:

Name	Type	Description
out	List[dict]	updated dictionaries

Source code in src/utils.py

def add_ids_to_dicts(ids: List[int],
                     dicts: List[dict]) -> List[dict]:
    """
    Add ids to corresponding dictionaries
    Parameters
    ----------
    ids : List[int]
        ids to add to the dictionaries in the same order as the dicts themselves
    dicts : List[dict]

    Returns
    -------
    out : List[dict]
        updated dictionaries

    """
    for id_, pred in zip(ids, dicts):
        pred["id"] = id_
    return dicts

argmax_dict(dict_)

Return dict item with maximal value

Parameters:

Name	Type	Description	Default
dict_	dict	any dictionary with numeric values	required

Returns:

Name	Type	Description
out	dict	element(s) with maximal value

Source code in src/utils.py

def argmax_dict(dict_: dict) -> dict:
    """
    Return dict item with maximal value

    Parameters
    ----------
    dict_ : dict
        any dictionary with numeric values

    Returns
    -------
    out : dict
        element(s) with maximal value
    """
    return {k: v for k, v in dict_.items() if v == max(dict_.values())}

pil_to_b64(image, format_='PNG', encoding='ascii', **save_kwargs)

Encode PIL.Image to base64 string (with given encoding)

Parameters:

Name	Type	Description	Default
image	Image	input image	required
format_	str	format to save image in buffer	'PNG'
encoding	str	bytes to string encoding method	'ascii'
**save_kwargs		format-specific options forwarded to Image.save (e.g. compress_level for PNG)	{}

Returns:

Name	Type	Description
image_encoded	str	output string

Source code in src/utils.py

def pil_to_b64(image: Image.Image, format_: str = 'PNG', encoding: str = 'ascii', **save_kwargs) -> str:
    """
    Encode PIL.Image to base64 string (with given encoding)

    Parameters
    ----------
    image : Image.Image
        input image
    format_ : str
        format to save image in buffer
    encoding : str
        bytes to string encoding method
    **save_kwargs :
        format-specific options forwarded to ``Image.save`` (e.g. ``compress_level`` for PNG)

    Returns
    -------
    image_encoded : str
        output string
    """

    buffered = BytesIO()
    image.save(buffered, format=format_, **save_kwargs)
    image_b64 = base64.urlsafe_b64encode(buffered.getvalue())
    image_encoded = image_b64.decode(encoding)

    return image_encoded

b64_to_bytes(image_b64)

Decode a urlsafe base64 string into raw bytes, padding it to a multiple of 4 first.

Parameters:

Name	Type	Description	Default
image_b64	str	base64 image	required

Returns:

Type	Description
bytes	decoded raw bytes

Source code in src/utils.py

def b64_to_bytes(image_b64: str) -> bytes:
    """
    Decode a urlsafe base64 string into raw bytes, padding it to a multiple of 4 first.

    Parameters
    ----------
    image_b64 : str
        base64 image

    Returns
    -------
    bytes
        decoded raw bytes
    """
    mod = len(image_b64) % 4
    if mod:
        image_b64 += "=" * (4 - mod)
    return base64.urlsafe_b64decode(image_b64)

b64_to_pil(image_b64)

Decode base64 converted string into image

Parameters:

Name	Type	Description	Default
image_b64	str	base64 image	required

Returns:

Name	Type	Description
image	Image	out image

Source code in src/utils.py

def b64_to_pil(image_b64: str):
    """
    Decode base64 converted string into image

    Parameters
    ----------
    image_b64 : str
        base64 image

    Returns
    -------
    image : Image.Image
        out image
    """
    return Image.open(BytesIO(b64_to_bytes(image_b64)))

generate_openapi_json(openapi_version='3.1.0')

Generate and save openapi specification of the application

Parameters:

Name	Type	Description	Default
openapi_version	str	openapi version specifier	'3.1.0'

Returns:

Type	Description
None

Source code in src/utils.py

def generate_openapi_json(openapi_version="3.1.0"):
    """
    Generate and save openapi specification of the application

    Parameters
    ----------
    openapi_version : str
        openapi version specifier

    Returns
    -------
    None
    """
    parser = ArgumentParser()
    parser.add_argument('--app_path')
    parser.add_argument('--server_url', default='0.0.0.0:8000')
    args = parser.parse_args()
    app_path = args.app_path
    server_url = args.server_url
    module = importlib.import_module(app_path)
    app = module.app

    openapi_json = get_openapi(
        title=app.title,
        openapi_version=openapi_version,
        version=app.version,
        description=app.description,
        routes=app.routes,
        servers=[{'url': server_url}]
    )
    json_save('openapi.json', openapi_json)

get_subclasses(cls)

Get subclasses of the given class recursively. Function will work even for nested inheritances.

Parameters:

Name	Type	Description	Default
cls	Object	given class	required

Returns:

Name	Type	Description
subclasses	List	all subclasses

Source code in src/utils.py

def get_subclasses(cls):
    """
    Get subclasses of the given class recursively.
    Function will work even for nested inheritances.

    Parameters
    ----------
    cls : Object
        given class

    Returns
    -------
    subclasses : List
        all subclasses
    """
    subclasses = []
    direct_subclasses = cls.__subclasses__()

    for subclass in direct_subclasses:
        subclasses.append(subclass)
        subclasses.extend(get_subclasses(subclass))

    return subclasses