cv-analysis-service/cv_analysis/utils/utils.py

from __future__ import annotations

import itertools

import cv2
import numpy as np
from PIL import Image
from funcy import first, iterate, keep, lmap, repeatedly
from numpy import generic


def copy_and_normalize_channels(image):

    if isinstance(image, Image.Image):
        image = np.array(image)

    image = image.copy()
    try:
        image = cv2.cvtColor(image, cv2.COLOR_GRAY2BGR)
    except cv2.error:
        pass

    return image


def npconvert(ob):
    if isinstance(ob, generic):
        return ob.item()
    raise TypeError


def lift(fn):
    def lifted(coll):
        yield from map(fn, coll)

    return lifted


def star(fn):
    def starred(args):
        return fn(*args)

    return starred


def lstarkeep(fn, coll):
    return list(starkeep(fn, coll))


def starkeep(fn, coll):
    yield from keep(star(fn), coll)


def until(cond, func, *args, **kwargs):
    return first(filter(cond, iterate(func, *args, **kwargs)))


def conj(x, xs):
    return [x, *xs]


def rconj(xs, x):
    return [*xs, x]


def make_merger_sentinel():
    def no_new_mergers(records):
        nonlocal number_of_records_so_far

        number_of_records_now = len(records)

        if number_of_records_now == number_of_records_so_far:
            return True

        else:
            number_of_records_so_far = number_of_records_now
            return False

    number_of_records_so_far = -1

    return no_new_mergers


def zipmap(fn, boxes, n=2):
    rets = lmap(list, zip(*map(fn, boxes)))
    yield from repeatedly(lambda: [], n) if len(rets) < n else rets


def every_nth(n, iterable):
    return itertools.islice(iterable, 0, None, n)