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'Download 