app.py

import os
import json
import time
import base64
import requests
import gradio as gr

username = os.environ.get("CLIENT_ID")
passwd = os.environ.get("CLIENT_SECRETS")

def authorization():
    url = "https://stage.id.trimblecloud.com/oauth/token"
    credential_pair = f"{username}:{passwd}"

    payload = "grant_type=client_credentials&scope=DLPointCloudSegmentation"
    headers = {
        "Content-Type": "application/x-www-form-urlencoded",
        "Authorization": f"Basic {base64.b64encode(credential_pair.encode('utf-8')).decode('utf-8')}",
    }

    response = requests.request("POST", url, headers=headers, data=payload)

    response.raise_for_status()
    print("Succesfully authenticated")
    auth_token = json.loads(response.text)["access_token"]

    return auth_token


def create_file(auth_token, input_filename):
    url = "https://cloud.stage.api.trimblecloud.com/dataocean/api/3.0/api/files"

    payload = json.dumps({"file": {"path": input_filename, "regions": ["us1"]}})
    headers = {
        "Authorization": f"Bearer {auth_token}",
        "Content-Type": "application/json",
    }

    response = requests.request("POST", url, headers=headers, data=payload)

    response.raise_for_status()
    print("File created successfully")
    file_upload_url = json.loads(response.text)["file"]["upload"]["url"]
    return file_upload_url


def upload_file(url, file):
    with open(file.name, "rb") as lasFile:
        payload = lasFile.read()

    headers = {"Content-Type": "application/octet-stream"}

    response = requests.request("PUT", url, headers=headers, data=payload)

    response.raise_for_status()
    print("Upload was successful")


def start_execution(auth_token, input_filename, output_filename="output.las"):
    url = "https://cloud.stage.api.trimblecloud.com/Processing/api/1/api/executions"

    payload = json.dumps(
        {
            "execution": {
                "procedure_id": "a7c4f9c3-b21a-4c9c-b4df-3dc6ba8934d9",
                "region": "aws-us1",
                "parameters": {
                    "source_path": input_filename,
                    "regions": ["us1"],
                    "output_path": output_filename,
                },
            }
        }
    )
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {auth_token}",
    }

    response = requests.request("POST", url, headers=headers, data=payload)

    response.raise_for_status()
    print("Execution has started")

    execution_id = json.loads(response.text)["execution"]["id"]

    return execution_id


def track_execution(auth_token, execution_id, output_filename):
    url = "https://cloud.stage.api.trimblecloud.com/Processing/api/1/api/executions"

    payload = {}
    headers = {
        "Authorization": f"Bearer {auth_token}",
    }

    response = requests.request("GET", url, headers=headers, data=payload)

    status = json.loads(response.text)["items"][0]["execution_status"]

    while status != "FINISHED":
        response = requests.request("GET", url, headers=headers, data=payload)
        status = json.loads(response.text)["items"][0]["execution_status"]
        time.sleep(1)

    return download_output(auth_token, output_filename)


def download_output(auth_token, output_filename):
    url = f"https://cloud.stage.api.trimblecloud.com/dataocean/api/3.0/api/files?path={output_filename}"

    payload = ""
    headers = {
        "Authorization": f"Bearer {auth_token}",
        "Content-Type": "application/json",
    }

    response = requests.request("GET", url, headers=headers, data=payload)

    response.raise_for_status()
    print("File downloading")
    response = json.loads(response.text)
    download_url = response["file"]["download"]["url"]

    return download_url

def predict(input_file):
    input_filename = "input.las"
    output_filename = "output.las"
    auth_token = authorization()
    file_upload_url = create_file(auth_token, input_filename)
    upload_file(file_upload_url, input_file)
    execution_id = start_execution(
        auth_token, input_filename, output_filename
    )
    download_url = track_execution(auth_token, execution_id, output_filename)

    html_content = f'<a href="{download_url}">Download output file</a>'


    return "Inference has finished. Click the download button to access the output file", html_content

demo = gr.Interface(
    title="Point Cloud inference on the Trimble Cloud",
    fn=predict,
    inputs=gr.File(file_types=[".las"], file_count="single"),
    outputs=[gr.Textbox(), "html"],
)

demo.queue(default_enabled=False).launch()