import gradio as gr
import json
import requests
import pandas as pd
def update_task_options(framework):
config = {
"Custom":["Custom"],
"Diffusers":["text-to-image"],
"Transformers":[
"text-classification",
"zero-shot-classification",
"token-classification",
"question-answering",
"fill-mask",
"summarization",
"translation",
"text2text-generation",
"text-generation",
"feature-extraction",
"image-classification",
"automatic-speech-recognition",
"audio-classification",
"object-detection",
"image-segmentation",
"table-question-answering"
]
}
return gr.Dropdown.update(
choices=config[framework],
value=config[framework][0] if len(config[framework]) > 0 else None
)
def update_regions(provider):
available_regions = []
headers = {
"Content-Type": "application/json",
}
endpoint_url = f"https://api.endpoints.huggingface.cloud/provider/{provider}/region"
response = requests.get(endpoint_url, headers=headers)
for region in response.json()['items']:
if region['status'] == 'available':
available_regions.append(f"{region['region']}/{region['label']}")
return gr.Dropdown.update(
choices=available_regions,
value=available_regions[0] if len(available_regions) > 0 else None
)
def update_compute_options(provider, region):
region = region.split("/")[0]
available_compute_choices = []
headers = {
"Content-Type": "application/json",
}
endpoint_url = f"https://api.endpoints.huggingface.cloud/provider/{provider}/region/{region}/compute"
response = requests.get(endpoint_url, headers=headers)
for compute in response.json()['items']:
if compute['status'] == 'available':
accelerator = compute['accelerator']
numAccelerators = compute['numAccelerators']
memoryGb = compute['memoryGb'].replace("Gi", "GB")
architecture = compute['architecture']
instanceType = compute['instanceType']
type = f"{numAccelerators}vCPU {memoryGb} · {architecture}" if accelerator == "cpu" else f"{numAccelerators}x {architecture}"
available_compute_choices.append(
f"{compute['accelerator'].upper()} [{compute['instanceSize']}] · {type} · {instanceType}"
)
return gr.Dropdown.update(
choices=available_compute_choices,
value=available_compute_choices[0] if len(available_compute_choices) > 0 else None
)
def submit(
hf_token_input,
endpoint_name_input,
provider_selector,
region_selector,
repository_selector,
revision_selector,
task_selector,
framework_selector,
compute_selector,
min_node_selector,
max_node_selector,
security_selector
):
compute_resources = compute_selector.split("·")
accelerator = compute_resources[0][:3].strip()
size_l_index = compute_resources[0].index("[") - 1
size_r_index = compute_resources[0].index("]")
size = compute_resources[0][size_l_index : size_r_index].strip()
type = compute_resources[-1].strip()
payload = {
"accountId": repository_selector.split("/")[0],
"compute": {
"accelerator": accelerator.lower(),
"instanceSize": size[1:],
"instanceType": type,
"scaling": {
"maxReplica": int(max_node_selector),
"minReplica": int(min_node_selector)
}
},
"model": {
"framework": "custom",
"image": {
"huggingface": {}
},
"repository": repository_selector.lower(),
"revision": revision_selector,
"task": task_selector.lower()
},
"name": endpoint_name_input.strip(),
"provider": {
"region": region_selector.split("/")[0].lower(),
"vendor": provider_selector.lower()
},
"type": security_selector.lower()
}
payload = json.dumps(payload)
headers = {
"Authorization": f"Bearer {hf_token_input.strip()}",
"Content-Type": "application/json",
}
endpoint_url = f"https://api.endpoints.huggingface.cloud/endpoint"
print(f"Endpoint: {endpoint_url}")
response = requests.post(endpoint_url, headers=headers, data=payload)
if response.status_code == 400:
return f"{response.text}. Malformed data in {payload}"
elif response.status_code == 401:
return "Invalid token"
elif response.status_code == 409:
return f"Error: {response.text}"
elif response.status_code == 202:
return f"Endpoint {endpoint_name_input} created successfully on {provider_selector.lower()} using {repository_selector.lower()}@{revision_selector}. \n Please check out the progress at https://ui.endpoints.huggingface.co/endpoints."
else:
return f"Something went wrong! - StatusCode:{response.status_code} - Error: {response.text}"
def delete_endpoint(
hf_token_input,
endpoint_name_input
):
response = requests.delete(
f"https://api.endpoints.huggingface.cloud/endpoint/{endpoint_name_input}",
headers = {
"Authorization": f"Bearer {hf_token_input.strip()}",
"Content-Type": "application/json",
}
)
if response.status_code == 401:
return "Invalid token"
elif response.status_code == 404:
return f"Error: {response.text}"
elif response.status_code == 202 or response.status_code == 200:
return f"Endpoint {endpoint_name_input} deleted successfully."
else:
return f"Something went wrong! - StatusCode:{response.status_code} - Error: {response.text}"
def get_all_endpoints(
hf_token_input,
method,
):
response = requests.get(
f"https://api.endpoints.huggingface.cloud/endpoint",
headers = {
"Authorization": f"Bearer {hf_token_input.strip()}",
"Content-Type": "application/json",
})
if response.status_code == 401:
if method == "info":
return gr.DataFrame.update(
value=pd.DataFrame({"Error":["Invalid Token -OR- No Endpoint Found !"]}),
)
else:
return gr.Dropdown.update(
value="Invalid token or No endpoints found!",
)
elif response.status_code == 200:
endpoints_json = response.json()
if method == "info":
endpoints_df = pd.DataFrame(columns=["name", "model", "provider", "compute", "status", "minReplica", "maxReplica", "createdAt", "updatedAt"])
for endpoint in endpoints_json["items"]:
endpoints_df = endpoints_df.append({
"name": endpoint["name"],
"model": endpoint["model"]["repository"] + "@" + endpoint["model"]["revision"],
"provider": endpoint["provider"]["vendor"] + "/" + endpoint["provider"]["region"],
"compute": endpoint["compute"]["instanceType"] + "·" + endpoint["compute"]["instanceSize"] + " [" + endpoint["compute"]["accelerator"] + "]",
"status": endpoint["status"]["state"],
"minReplica": endpoint["compute"]["scaling"]["minReplica"],
"maxReplica": endpoint["compute"]["scaling"]["maxReplica"],
"createdAt": endpoint["status"]["createdAt"],
"updatedAt": endpoint["status"]["updatedAt"],
}, ignore_index=True)
endpoints_df.columns = ["Endpoint Name", "Model Name @ Revision", "Provider", "Instance Type", "Status", "Min Replica", "Max Replica", "Created At", "Updated At"]
return gr.DataFrame.update(
value=endpoints_df,
)
else:
return gr.Dropdown.update(
choices=[endpoint["name"] for endpoint in endpoints_json["items"]],
value=endpoints_json["items"][0]["name"],
)
def update_endpoint(
hf_token_input,
endpoint_name_input,
min_node_selector,
max_node_selector,
instance_type,
):
payload ={
"compute": {
"instanceSize": instance_type.split("·")[0].split("[")[1].split("]")[0],
"instanceType": instance_type.split("·")[-1].strip(),
"scaling": {
"maxReplica": int(max_node_selector),
"minReplica": int(min_node_selector)
}
}}
payload = json.dumps(payload)
response = requests.put(
f"https://api.endpoints.huggingface.cloud/endpoint/{endpoint_name_input}",
headers = {
"Authorization": f"Bearer {hf_token_input.strip()}",
"Content-Type": "application/json",
},
data=payload,
)
if response.status_code == 401:
return "Invalid token"
elif response.status_code == 404:
return f"Error: {response.text}"
elif response.status_code == 202 or response.status_code == 200:
return f"Endpoint {endpoint_name_input} updated successfully."
else:
return f"Something went wrong! - StatusCode:{response.status_code} - Error: {response.text}"
def get_endpoint_logs(
hf_token_input,
endpoint_name_input,
):
response = requests.get(
f"https://api.endpoints.huggingface.cloud/endpoint/{endpoint_name_input}/logs",
headers = {
"Authorization": f"Bearer {hf_token_input.strip()}",
"Content-Type": "application/json",
})
if response.status_code == 401:
return "Invalid token or No logs found!"
elif response.status_code == 200:
return response.text
elif response.status_code == 404:
return f"Error: {response.text}"
with gr.Blocks() as interface:
gr.Markdown("""
#### Your 🤗 Access Token (Required)
""")
hf_token_input = gr.Textbox(
show_label=False,
placeholder="Your personal/organization Huggingface Access Token. Get it from (https://huggingface.co/settings/tokens)",
type="password"
)
# Get All Endpoints Info
with gr.Tab("Info"):
gr.Markdown("""
### All Deployed Endpoints
""")
endpoints_table = gr.Dataframe(
headers=["Endpoint Name", "Model Name", "Provider", "Instance Type", "Status", "Min Replica", "Max Replica", "Created At", "Updated At"],
col_count=(9, "fixed"),
)
endpoint_info_button = gr.Button(value="Get Info")
# Deploy Endpoint
with gr.Tab("Deploy Endpoint"):
gr.Markdown(
"""
###