Spaces:
Running
Running
| import gradio as gr | |
| from prodiapy import Prodia | |
| from PIL import Image | |
| from io import BytesIO | |
| import requests | |
| import random | |
| import os | |
| import base64 | |
| import json | |
| import time | |
| class Prodia: | |
| def __init__(self, api_key=os.getenv("PRODIA_API_KEY"), base=None): | |
| self.base = base or "https://api.prodia.com/v1" | |
| self.headers = { | |
| "X-Prodia-Key": api_key | |
| } | |
| def faceswap(self, params): | |
| response = self._post(f"{self.base}/faceswap", params) | |
| return response.json() | |
| def get_job(self, job_id): | |
| response = self._get(f"{self.base}/job/{job_id}") | |
| return response.json() | |
| def wait(self, job): | |
| job_result = job | |
| while job_result['status'] not in ['succeeded', 'failed']: | |
| time.sleep(0.25) | |
| job_result = self.get_job(job['job']) | |
| return job_result | |
| def _post(self, url, params): | |
| headers = { | |
| **self.headers, | |
| "Content-Type": "application/json" | |
| } | |
| response = requests.post(url, headers=headers, data=json.dumps(params)) | |
| if response.status_code != 200: | |
| raise Exception(f"Bad Prodia Response: {response.status_code}") | |
| return response | |
| def _get(self, url): | |
| response = requests.get(url, headers=self.headers) | |
| if response.status_code != 200: | |
| raise Exception(f"Bad Prodia Response: {response.status_code}") | |
| return response | |
| client = Prodia() | |
| def infer(source, target): | |
| if source_image is None or target_image is None: | |
| return | |
| source_url = upload_image(source) | |
| target_url = upload_image(target) | |
| job = client.faceswap({ | |
| "sourceUrl": source_url, | |
| "targetUrl": target_url | |
| }) | |
| res = client.wait(job) | |
| if res['status'] == "failed": | |
| return | |
| return res['imageUrl'] | |
| def upload_image(file): | |
| files = {'file': open(file, 'rb')} | |
| img_id = requests.post(os.getenv("IMAGE_API_1"), files=files).json()['id'] | |
| payload = { | |
| "content": "", | |
| "nonce": f"{random.randint(1, 10000000)}H9X42KSEJFNNH", | |
| "replies": [], | |
| "attachments": [img_id] | |
| } | |
| res = requests.post(os.getenv("IMAGE_API_2"), json=payload, headers={"x-session-token": os.getenv("SESSION_TOKEN")}) | |
| return f"{os.getenv('IMAGE_API_1')}/{img_id}/{res.json()['attachments'][0]['filename']}" | |
| def image_to_base64(image: Image): | |
| buffered = BytesIO() | |
| image.save(buffered, format="PNG") | |
| img_str = base64.b64encode(buffered.getvalue()) | |
| return img_str.decode('utf-8') | |
| with gr.Blocks() as demo: | |
| with gr.Column(): | |
| gr.HTML("<h1><center>Face Swap</center></h1>") | |
| with gr.Row(): | |
| with gr.Row(): | |
| source_image = gr.Image(type="filepath", label="Source Image") | |
| target_image = gr.Image(type="filepath", label="Target Image") | |
| with gr.Column(): | |
| result = gr.Image() | |
| run_button = gr.Button("Swap Face", variant="primary") | |
| gr.Examples( | |
| fn=infer, | |
| inputs=[source_image, target_image], | |
| outputs=[result], | |
| theme="NoCrypt/miku" | |
| ) | |
| run_button.click(fn=infer, inputs=[source_image, target_image], outputs=[result]) | |
| demo.queue(max_size=20, api_open=False).launch(show_api=False, max_threads=400) | |