check.py

import math
import os
import random
from functools import lru_cache

import numpy as np
from PIL import Image
from huggingface_hub import HfFileSystem, HfApi
from imgutils.utils import open_onnx_model
from natsort import natsorted

hf_token = os.environ.get('HF_TOKEN')
hf_fs = HfFileSystem(token=hf_token)
hf_client = HfApi(token=hf_token)

REPOSITORY = 'mf666/shit-checker'
MODELS = natsorted([
    os.path.splitext(os.path.relpath(file, REPOSITORY))[0]
    for file in hf_fs.glob(f'{REPOSITORY}/*.onnx')
])
DEFAULT_MODEL = 'mobilenet.xs.v2'


@lru_cache()
def _open_model(model_name):
    return open_onnx_model(hf_client.hf_hub_download(REPOSITORY, f'{model_name}.onnx'))


_DEFAULT_ORDER = 'HWC'


def _get_hwc_map(order_):
    return tuple(_DEFAULT_ORDER.index(c) for c in order_.upper())


def _encode_channels(image, channels_order='CHW', is_float=True):
    array = np.asarray(image.convert('RGB'))
    array = np.transpose(array, _get_hwc_map(channels_order))
    if not is_float:
        assert array.dtype == np.uint8
    else:
        array = (array / 255.0).astype(np.float32)
        assert array.dtype == np.float32
    return array


def _img_encode(image, size=(384, 384), normalize=(0.5, 0.5)):
    image = image.resize(size, Image.BILINEAR)
    data = _encode_channels(image, channels_order='CHW')

    if normalize is not None:
        mean_, std_ = normalize
        mean = np.asarray([mean_]).reshape((-1, 1, 1))
        std = np.asarray([std_]).reshape((-1, 1, 1))
        data = (data - mean) / std

    return data.astype(np.float32)


def _raw_predict(images, model_name=DEFAULT_MODEL):
    items = []
    for image in images:
        items.append(_img_encode(image.convert('RGB')))
    input_ = np.stack(items)
    output, = _open_model(model_name).run(['output'], {'input': input_})
    return output.mean(axis=0)


def predict(image, model_name=DEFAULT_MODEL, max_batch_size=8):
    area = image.width * image.height
    batch_size = int(max(min(math.ceil(area / (384 * 384)) + 1, max_batch_size), 1))
    blocks = []
    for _ in range(batch_size):
        x0 = random.randint(0, max(0, image.width - 384))
        y0 = random.randint(0, max(0, image.height - 384))
        x1 = min(x0 + 384, image.width)
        y1 = min(y0 + 384, image.height)
        blocks.append(image.crop((x0, y0, x1, y1)))

    scores = _raw_predict(blocks, model_name)
    return dict(zip(['shat', 'normal'], map(lambda x: x.item(), scores)))