app.py

import pathlib
import shlex
import subprocess
import gradio as gr
import PIL.Image
import time
import os
import requests
import json
import gradio as gr

# Define the base URL, the API key and Paths
base_url = "https://hyperhuman.deemos.com/api/v2"
api_key = os.getenv('API')
image_path = "/content/sample_data/wooden_chair.jpg"

result_path = "./"


# Define the headers for the requests
headers = {
    "Authorization": f"Bearer {api_key}",
    "Content-Type": "application/json"
}

def submit_task(image_path: str):
    url = f"{base_url}/rodin"
    #image_path = "/content/sample_data/wooden_chair.jpg"
    # Get the uploaded file
     #if uploader.value:
       #uploaded_file = next(iter(uploader.value.values()))
       #if 'name' not in uploaded_file:
         #print("Error: 'name' not found in task response")
         #return


       #image_path = uploaded_file['metadata']['name']

          # Save the uploaded file to disk
   # with open(image_path, 'wb') as f:
   #      f.write(uploaded_file['content'])
    #else:
     # print("No file uploaded. Please upload an image.")

    # Read the image file
    with open(image_path, 'rb') as image_file:
        image_data = image_file.read()

    # Prepare the multipart form data
    files = {
        'images': (os.path.basename(image_path), image_data, 'image/jpeg')
    }

 
    data = {
        'tier': 'Regular'
    }

    # Prepare the headers.
    headers = {
        'Authorization': f'Bearer {api_key}',
    }

    # Note that we are not sending the data as JSON, but as form data.
    # This is because we are sending a file as well.
    response = requests.post(url, files=files, data=data, headers=headers)
    print(response)
    return response.json()

# Function to check the status of a task
def check_status(subscription_key):
    url = f"{base_url}/status"
    data = {
        "subscription_key": subscription_key
    }
    response = requests.post(url, headers=headers, json=data)
    return response.json()

# Function to download the results of a task
def download_results(task_uuid):
    url = f"{base_url}/download"
    data = {
        "task_uuid": task_uuid
    }
    response = requests.post(url, headers=headers, json=data)
    return response.json()



def submit_task2(image_path: str):
  # Submit the task and get the task UUID
  task_response = submit_task(image_path)
  if 'uuid' not in task_response:
      print("Error: 'uuid' not found in task response")
      return

  task_uuid = task_response['uuid']
  subscription_key = task_response['jobs']['subscription_key']

  # Poll the status endpoint every 5 seconds until the task is done
  status = []
  while len(status) == 0 or not all(s['status'] in ['Done', 'Failed'] for s in status):
      time.sleep(5)
      status_response = check_status(subscription_key)
      status = status_response['jobs']
      for s in status:
          print(f"job {s['uuid']}: {s['status']}")

  # Download the results once the task is done
  download_response = download_results(task_uuid)
  download_items = download_response['list']

  # Print the download URLs and download them locally.
  for item in download_items:
      print(f"File Name: {item['name']}, URL: {item['url']}")
      dest_fname = os.path.join(result_path, item['name'])
      os.makedirs(os.path.dirname(dest_fname), exist_ok=True)
      with open(dest_fname, 'wb') as f:
          response = requests.get(item['url'])
          f.write(response.content)
          print(f"Downloaded {dest_fname}")
          return dest_fname






def run(image_path: str) -> str:

   return submit_task2(image_path)

with gr.Blocks() as demo:
   with gr.Group():
        image = gr.Image(label="Input image", show_label=False, type="filepath")
        run_button = gr.Button("Run")
        result = gr.Model3D(label="Result", show_label=False)

        run_button.click(
            fn=run,
            inputs=[
                image
            ],
            outputs=result,
            api_name="image-to-3d",
            concurrency_id="gpu",
            concurrency_limit=1,
        )

demo.launch(auth=("user", os.getenv('PASS')),share=True,debug=True)