|
|
|
|
|
from __future__ import annotations |
|
|
|
import argparse |
|
import functools |
|
import os |
|
import pathlib |
|
import subprocess |
|
import sys |
|
import tarfile |
|
|
|
|
|
command = 'pip install -U gradio==2.7.0' |
|
subprocess.call(command.split()) |
|
|
|
import gradio as gr |
|
import huggingface_hub |
|
import PIL.Image |
|
import torch |
|
import torchvision |
|
|
|
sys.path.insert(0, 'bizarre-pose-estimator') |
|
|
|
from _util.twodee_v0 import I as ImageWrapper |
|
|
|
TOKEN = os.environ['TOKEN'] |
|
|
|
MODEL_REPO = 'hysts/bizarre-pose-estimator-models' |
|
MODEL_FILENAME = 'tagger.pth' |
|
LABEL_FILENAME = 'tags.txt' |
|
|
|
|
|
def parse_args() -> argparse.Namespace: |
|
parser = argparse.ArgumentParser() |
|
parser.add_argument('--device', type=str, default='cpu') |
|
parser.add_argument('--score-slider-step', type=float, default=0.05) |
|
parser.add_argument('--score-threshold', type=float, default=0.5) |
|
parser.add_argument('--theme', type=str, default='dark-grass') |
|
parser.add_argument('--live', action='store_true') |
|
parser.add_argument('--share', action='store_true') |
|
parser.add_argument('--port', type=int) |
|
parser.add_argument('--disable-queue', |
|
dest='enable_queue', |
|
action='store_false') |
|
parser.add_argument('--allow-flagging', type=str, default='never') |
|
parser.add_argument('--allow-screenshot', action='store_true') |
|
return parser.parse_args() |
|
|
|
|
|
def load_sample_image_paths() -> list[pathlib.Path]: |
|
image_dir = pathlib.Path('images') |
|
if not image_dir.exists(): |
|
dataset_repo = 'hysts/sample-images-TADNE' |
|
path = huggingface_hub.hf_hub_download(dataset_repo, |
|
'images.tar.gz', |
|
repo_type='dataset', |
|
use_auth_token=TOKEN) |
|
with tarfile.open(path) as f: |
|
f.extractall() |
|
return sorted(image_dir.glob('*')) |
|
|
|
|
|
def load_model(device: torch.device) -> torch.nn.Module: |
|
path = huggingface_hub.hf_hub_download(MODEL_REPO, |
|
MODEL_FILENAME, |
|
use_auth_token=TOKEN) |
|
state_dict = torch.load(path) |
|
model = torchvision.models.resnet50(num_classes=1062) |
|
model.load_state_dict(state_dict) |
|
model.to(device) |
|
model.eval() |
|
return model |
|
|
|
|
|
def load_labels() -> list[str]: |
|
label_path = huggingface_hub.hf_hub_download(MODEL_REPO, |
|
LABEL_FILENAME, |
|
use_auth_token=TOKEN) |
|
with open(label_path) as f: |
|
labels = [line.strip() for line in f.readlines()] |
|
return labels |
|
|
|
|
|
@torch.inference_mode() |
|
def predict(image: PIL.Image.Image, score_threshold: float, |
|
device: torch.device, model: torch.nn.Module, |
|
labels: list[str]) -> dict[str, float]: |
|
data = ImageWrapper(image).resize_square(256).alpha_bg( |
|
c='w').convert('RGB').tensor() |
|
data = data.to(device).unsqueeze(0) |
|
|
|
preds = model(data)[0] |
|
preds = torch.sigmoid(preds) |
|
preds = preds.cpu().numpy().astype(float) |
|
|
|
res = dict() |
|
for prob, label in zip(preds, labels): |
|
if prob < score_threshold: |
|
continue |
|
res[label] = prob |
|
return res |
|
|
|
|
|
def main(): |
|
gr.close_all() |
|
|
|
args = parse_args() |
|
device = torch.device(args.device) |
|
|
|
image_paths = load_sample_image_paths() |
|
examples = [[path.as_posix(), args.score_threshold] |
|
for path in image_paths] |
|
|
|
model = load_model(device) |
|
labels = load_labels() |
|
|
|
func = functools.partial(predict, |
|
device=device, |
|
model=model, |
|
labels=labels) |
|
func = functools.update_wrapper(func, predict) |
|
|
|
repo_url = 'https://github.com/ShuhongChen/bizarre-pose-estimator' |
|
title = 'ShuhongChen/bizarre-pose-estimator (tagger)' |
|
description = f'A demo for {repo_url}' |
|
article = None |
|
|
|
gr.Interface( |
|
func, |
|
[ |
|
gr.inputs.Image(type='pil', label='Input'), |
|
gr.inputs.Slider(0, |
|
1, |
|
step=args.score_slider_step, |
|
default=args.score_threshold, |
|
label='Score Threshold'), |
|
], |
|
gr.outputs.Label(label='Output'), |
|
theme=args.theme, |
|
title=title, |
|
description=description, |
|
article=article, |
|
examples=examples, |
|
allow_screenshot=args.allow_screenshot, |
|
allow_flagging=args.allow_flagging, |
|
live=args.live, |
|
).launch( |
|
enable_queue=args.enable_queue, |
|
server_port=args.port, |
|
share=args.share, |
|
) |
|
|
|
|
|
if __name__ == '__main__': |
|
main() |
|
|
