Monitor Hugging Face Endpoint

2025-04-09 16:37 UTC o3-mini-high Open in ChatGPT ↗

Hugging Face Inference Endpoints don’t currently offer a built‐in webhook, event, or callback to signal when a “cold” endpoint (with autoscale set to 0) has finished warming up. In other words, there isn’t an out‐of‐the‐box notification mechanism to tell you (or your users) “I’m ready” once the endpoint is active. Instead, you’ll need to implement your own monitoring solution. Here are some strategies and workarounds:

1. Polling the Endpoint

Overview:
Since there’s no native event, the typical approach is to periodically poll the endpoint to determine when it’s up. Your monitoring script would repeatedly send a request (often a “ping” or health-check) and look for a successful response (HTTP status 200) rather than a service-unavailable status (like HTTP 503).

Example Approach:

Send a test request: Use a lightweight request (e.g., a simple inference call or ping) to the endpoint.
Check the response: If you receive a valid response (200), that’s a sign the endpoint has warmed up.
Trigger a notification: Once the endpoint is ready, you can trigger an alert (an email, a message, or even a POST to your own webhook).

Example in Python:

import time
import requests

# Your Hugging Face inference endpoint URL
endpoint_url = "https://your-huggingface-endpoint"

# Optional: URL of your own notification service (webhook/email)
notification_url = "https://your-notification-webhook"

def check_endpoint():
    try:
        # Note: Adjust the payload as needed for your endpoint
        response = requests.post(endpoint_url, json={"data": "ping"}, timeout=5)
        # 200 OK implies the endpoint is ready
        return response.status_code == 200
    except Exception:
        return False

while not check_endpoint():
    print("Endpoint not ready, waiting...")
    time.sleep(10)  # Poll every 10 seconds; adjust as necessary

print("Endpoint is now ready!")

# Optional: Trigger a notification to your monitoring system
try:
    requests.post(notification_url, json={"message": "Your Hugging Face endpoint is ready!"})
except Exception as e:
    print("Failed to send notification:", e)

This script demonstrates a basic polling mechanism. You can adapt it to include more sophisticated error handling, logging, or integration with your notification system.

2. Using the Hugging Face Dashboard and Logs

While there isn’t a direct webhook, you do have access to logs and metrics through the Hugging Face dashboard. These logs can help you monitor the start-up performance and usage of your endpoint. However, this method will not proactively notify you; you would need to check the dashboard manually or integrate those logs with your own alerting system via custom scripts.

3. Considerations for Keeping the Endpoint Warm

If immediate responsiveness is critical for your application, consider these options:

Set a Minimum Scale:
Although autoscale=0 helps reduce costs by shutting down idle endpoints, setting a non-zero minimum (for example, autoscale=1) can ensure that there’s always at least one instance available to handle requests quickly. This trade-off may be worthwhile if the warm-up latency would adversely affect your users.
Warm-Up Strategies:
Alternatively, if cost is a concern but you want to mitigate the cold start delay, you might schedule periodic “dummy” requests (if it doesn’t affect your billing significantly) to keep the endpoint active. This is a common pattern to reduce latency at the cost of some extra requests.

4. Conclusion

As of now, Hugging Face does not provide a native webhook or callback mechanism to notify you when the inference endpoint becomes ready from a cold start. Your best approach is to implement a monitoring system (such as periodic polling) that checks the endpoint status and then sends a notification via your own infrastructure (such as a custom webhook, email, or messaging service).

For any new developments or official announcements regarding this functionality, it’s a good idea to keep an eye on Hugging Face’s documentation or community forums, as features and integrations evolve over time.

If you have further questions or need additional details on implementing a specific part of this solution, feel free to ask!

Below is an explanation of your options—and some sample code—to achieve a “pause/resume” behavior for your endpoint. Note that Hugging Face doesn’t (yet) offer a dedicated pause/resume API command for endpoints running with a fixed instance count. Instead, you have two main paths:

1. Revisit the Autoscaling Approach

Even if you don’t want full autoscaling behavior in production, enabling autoscaling with a minimum of 0 workers is the simplest way to let an endpoint “pause” when it’s not in use and then “wake” on demand. In this model:

Pause: The endpoint automatically scales down to 0 (and therefore is effectively paused) after a period of inactivity.
Resume: When a new request comes in, the endpoint spins up an instance (with the associated cold-start latency).

This gives you a built‑in, fully scriptless way of “pausing” the endpoint (at the cost of the cold start).

2. Manually Update the Endpoint Configuration via the API

If you’d rather keep a fixed number of workers when active and want to control when they’re “on” or “off” via your own script, one workaround is to update the endpoint configuration through the Hugging Face management API. In effect, you “simulate” a pause by setting the number of active workers to 0 (or any number that means “paused”) and then later “resume” by raising that count back up.

Important:
The ability to update worker counts programmatically is more naturally available when using autoscaling. When autoscaling is disabled, the system is set up for a fixed worker count and may not accept a direct change to “0” using the same API. In that case, you might have to switch to an autoscaling configuration (with a min‑value of 0) and then update the minimum worker count via a script.

Check the latest Hugging Face Inference Endpoints documentation or reach out to support to verify that your endpoint’s configuration supports dynamic updates.

Here’s a simple Python example that uses a PATCH request (using the Hugging Face API) to update the configuration. Replace the placeholders with your endpoint’s ID and your API token. (This example assumes that your endpoint’s management API accepts a parameter such as “min_workers”.):

import requests

# Replace these with your actual values.
API_TOKEN = "your_api_token"
ENDPOINT_ID = "your_endpoint_id"
base_url = f"https://huggingface.co/api/inference-endpoints/{ENDPOINT_ID}"

headers = {"Authorization": f"Bearer {API_TOKEN}"}

def update_endpoint(min_workers):
    """
    Update the endpoint configuration to set the minimum number of workers.
    Setting min_workers to 0 simulates a pause (if allowed);
    setting to a positive number resumes the endpoint.
    """
    data = {"min_workers": min_workers}
    response = requests.patch(base_url, json=data, headers=headers)
    if response.status_code == 200:
        print(f"Endpoint updated successfully: min_workers set to {min_workers}.")
    else:
        print("Failed to update endpoint:", response.text)

# To 'pause' the endpoint
update_endpoint(0)

# ... later, to 'resume' the endpoint (for example, setting to 1)
update_endpoint(1)

3. Next Steps

Check Documentation:
Since Hugging Face features evolve over time, it’s always a good idea to check the official documentation or your dashboard for the latest on endpoint management.
Contact Support:
If you’re on a commercial plan or if your setup is nonstandard, Hugging Face support can provide guidance on whether your endpoint’s configuration supports such dynamic updates.
Consider Your Workflow:
If the goal is to save on costs or only process requests at specific times, setting up autoscaling with a minimum of 0 might be the simplest and most reliable option—even if it means accepting the occasional cold start.

In summary, while there isn’t a dedicated “pause/resume” API for fixed endpoints, you can simulate this behavior by programmatically updating the endpoint’s worker count—provided your endpoint is configured (or reconfigured) to allow scaling from 0 to the desired active count.

This is what their documentation says

Pause and Resume your Endpoint You can pause & resume endpoints to save cost and configurations. Please note that if your endpoint is in a failed state, you will need to create a new endpoint. To pause/resume your endpoint, navigate to the “overview” tab and click the button at top right corner, which will show “Pause endpoint” to pause, or “Resume endpoint” to reactivate the paused endpoint.

When pausing an endpoint the min & max number of replicas will be set to 0. When resuming an endpoint the min & max number of replicas will be set to 1. This allows you to programmatically pause and resume your endpoint by updating the “min_replicas” and “max_replicas” fields in the API. Paused inference endpoints will have the following status: PAUSED. Paused endpoints will be NOT be billed until resumed. Pausing & Resuming an endpoint is a great way to save costs when you don’t need your endpoint to be running. For example, you can easily pause your endpoint during the night or weekends. You should pause your endpoint when you don’t need it for the time being.

The url of your endpoint will remain the same, even if you pause and resume it. This means that you can pause your endpoint and resume it later without having to update your code.

Based on the documentation you shared, you can programmatically pause and resume your Hugging Face Inference Endpoint by updating its replica settings via the API. Specifically:

To pause the endpoint:
Set both min_replicas and max_replicas to 0. This pauses the endpoint (its status becomes PAUSED) and you are not billed until you resume it.
To resume the endpoint:
Set both min_replicas and max_replicas to 1. This brings the endpoint back to an active state without changing the URL.

Below is a sample Python script that demonstrates how to do this using the Hugging Face management API via a PATCH request. Replace the placeholders with your actual API token and endpoint ID.

import requests

# Replace these with your actual values
API_TOKEN = "your_api_token"
ENDPOINT_ID = "your_endpoint_id"
base_url = f"https://huggingface.co/api/inference-endpoints/{ENDPOINT_ID}"

headers = {
    "Authorization": f"Bearer {API_TOKEN}"
}

def update_endpoint(min_replicas, max_replicas):
    """
    Update the endpoint to set the number of replicas.
    Pausing is done by setting both min_replicas and max_replicas to 0.
    Resuming is done by setting them to 1.
    """
    data = {
        "min_replicas": min_replicas,
        "max_replicas": max_replicas
    }
    response = requests.patch(base_url, json=data, headers=headers)
    if response.status_code == 200:
        print(f"Successfully updated endpoint: min_replicas={min_replicas}, max_replicas={max_replicas}.")
    else:
        print(f"Error updating endpoint: {response.text}")

# To pause the endpoint (and avoid billing until resumed):
update_endpoint(0, 0)

# Later, when you want to resume (bringing the endpoint back up):
update_endpoint(1, 1)

A Few Additional Considerations

Endpoint Status and Billing:
When the endpoint is paused (i.e., both replica counts are 0), it won’t be billed. When you resume it by setting the replica counts to 1, it will quickly come back online under the same URL.
Error Handling:
The script includes a basic printout for error handling. You might want to integrate better logging or exception handling depending on your production needs.
Failed State:
According to the documentation, if your endpoint has already entered a failed state, you’ll need to create a new one rather than trying to resume the failed instance.
Monitoring Changes:
You may wish to monitor the endpoint’s status after making the API calls to ensure that your changes have taken effect. The Hugging Face dashboard or log streams can help verify status changes.

This approach allows you to easily integrate pause and resume functionality into your workflow, saving costs when the endpoint isn’t needed. Feel free to ask if you have additional questions or need further details!

Yes, you can easily test the pause/resume functionality from the terminal using a tool like cURL. This allows you to simulate the API calls without writing a full script.

Below are example commands for both pausing and resuming your endpoint. Be sure to replace your_endpoint_id and your_api_token with your actual endpoint ID and Hugging Face API token.

To Pause the Endpoint

This command sets both min_replicas and max_replicas to 0:

curl -X PATCH "https://huggingface.co/api/inference-endpoints/your_endpoint_id" \
     -H "Authorization: Bearer your_api_token" \
     -H "Content-Type: application/json" \
     -d '{"min_replicas": 0, "max_replicas": 0}'

To Resume the Endpoint

This command sets both min_replicas and max_replicas to 1:

curl -X PATCH "https://huggingface.co/api/inference-endpoints/your_endpoint_id" \
     -H "Authorization: Bearer your_api_token" \
     -H "Content-Type: application/json" \
     -d '{"min_replicas": 1, "max_replicas": 1}'

Additional Tips

Verbose Output:
If you want more details about the request and response, add the -v flag:

curl -v -X PATCH "https://huggingface.co/api/inference-endpoints/your_endpoint_id" \
     -H "Authorization: Bearer your_api_token" \
     -H "Content-Type: application/json" \
     -d '{"min_replicas": 0, "max_replicas": 0}'

Check Status:
Once you execute the command, the API should respond with a status message indicating whether the update was successful. In case of errors, the response will detail what went wrong, which can help with troubleshooting.

Using these commands from the terminal provides a quick and direct way to test and control your endpoint’s state programmatically without needing a separate script.

The endpoint ID is a unique identifier for the specific inference endpoint that you’ve deployed on Hugging Face. You can find it in your Hugging Face account dashboard:

In the URL:
When you navigate to your endpoint’s page (for example, via https://huggingface.co/inference-endpoints), the endpoint ID is usually included as part of the URL. If your endpoint URL looks like https://huggingface.co/inference-endpoints/my-endpoint, then my-endpoint is your endpoint ID.
In the Dashboard:
Log in to your Hugging Face account, go to the Inference Endpoints section, and select the endpoint you’re interested in. The details or the overview page for that endpoint should display the endpoint ID.

If you’re still unsure, you might check the API documentation for your account or ask Hugging Face support for clarification regarding your endpoint’s identification.

Host endpoints.huggingface.co:443 was resolved.
IPv6: 2a04:4e42:70::796
IPv4: 146.75.3.28
Trying [2a04:4e42:70::796]:443…
Connected to endpoints.huggingface.co (2a04:4e42:70::796) port 443
schannel: disabled automatic use of client certificate
ALPN: curl offers http/1.1
ALPN: server accepted http/1.1
using HTTP/1.x

PATCH /george-adamon/endpoints/deepseek-r1-distill-qwen-32b-isu HTTP/1.1 Host: endpoints.huggingface.co User-Agent: curl/8.9.1 Accept: / Authorization: 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 Content-Type: application/json Content-Length: 38

upload completely sent off: 38 bytes < HTTP/1.1 405 < Connection: keep-alive < Content-Length: 24 < allow: GET, OPTIONS, HEAD, HEAD < cache-control: public, max-age=0, must-revalidate < x-edgee: proxy-only(non-html) < content-type: text/plain; charset=utf-8 < date: Thu, 10 Apr 2025 08:06:13 GMT < server: Vercel < set-cookie: __Host-authjs.csrf-token=e116d930777d63e2d9e305fd01ab4692332f64b20c83ae514e9fee0df5531158%7Caa17e76b63a061b62886a76c3a936b8240de586901c91303ac77f624236ef8e5; Path=/; HttpOnly; Secure; SameSite=Lax < set-cookie: __Secure-authjs.callback-url=https%3A%2F%2Fendpoints.huggingface.co; Path=/; HttpOnly; Secure; SameSite=Lax < strict-transport-security: max-age=63072000 < x-vercel-cache: MISS < x-vercel-id: fra1::iad1::p6h9f-1744272373788-bdd0b6d0c626 < x-cache: MISS < x-cache-hits: 0 < x-served-by: cache-sof1510025-SOF < PATCH method not allowed* Connection #0 to host endpoints.huggingface.co left intact

It looks like you’re seeing an HTTP 405 “Method Not Allowed” error because the PATCH request isn’t hitting the management API endpoint as expected. The key points to check are:

1. Use the Correct API URL

According to Hugging Face’s documentation, management API calls for pausing/resuming an endpoint should be made to a URL that begins with:

https://huggingface.co/api/inference-endpoints/<endpoint_id>

In your command, you used:

https://endpoints.huggingface.co/george-adamon/endpoints/deepseek-r1-distill-qwen-32b-isu

The extra subdomain (endpoints.huggingface.co) and the URL path structure (including “george-adamon/endpoints/…”) may be causing the request to hit the wrong service. Instead, try using the API URL as follows (replace <endpoint_id> with the proper identifier):

curl -X PATCH "https://huggingface.co/api/inference-endpoints/deepseek-r1-distill-qwen-32b-isu" \
     -H "Authorization: Bearer your_api_token" \
     -H "Content-Type: application/json" \
     -d '{"min_replicas": 0, "max_replicas": 0}'

2. Confirm Your Endpoint ID

Your endpoint ID is the unique identifier shown in your Hugging Face dashboard. Often, if your URL in the dashboard is something like:

https://huggingface.co/inference-endpoints/deepseek-r1-distill-qwen-32b-isu

then the endpoint ID is deepseek-r1-distill-qwen-32b-isu. If your account or project structure shows your username as part of the URL, note that the management API endpoint typically expects just the endpoint identifier without the username prefix in the API path.

3. Verify with Documentation

Make sure to review the most recent Hugging Face Inference Endpoints documentation for any updates on the API URL structure and allowed methods. There might be nuances or changes in the API paths between the user-facing endpoint and the management API.

In Summary

Change the URL: Use https://huggingface.co/api/inference-endpoints/<endpoint_id> rather than https://endpoints.huggingface.co/...
Confirm the Endpoint ID: Ensure you’re using the proper identifier from your dashboard.
Retry with cURL: Once you’ve updated the URL and confirmed the endpoint ID, your PATCH request should hit the correct endpoint where PATCH is allowed.

Give that a try, and you should be able to pause or resume your endpoint from the terminal. If you continue to receive an error, double-check your API token and review the latest management API documentation, as the allowed methods and URL structure may have been updated.

Below is an example Python application that uses a simple Tkinter UI. It lets you:

Click buttons to “Resume,” “Pause,” and “Check Status” of your endpoint.
Enter your API token.
Specify daily pause and resume times (in HH:MM format) and enable an automatic timer that will run the corresponding API calls once each day.

Before running the code, make sure you have the required packages installed (Tkinter is part of Python’s standard library and you can install requests via pip):

pip install requests

Then save the code below (for example, as hf_endpoint_control.py) and run it.

import tkinter as tk
from tkinter import messagebox
from datetime import datetime
import threading
import time
import requests

# Constants for the API endpoints (replace the endpoint id if needed)
RESUME_URL = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/resume"
PAUSE_URL  = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/pause"
STATUS_URL = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/replica"

# Global variables to prevent multiple triggers in the same day
last_pause_day = None
last_resume_day = None

def call_api(url, token, method="POST"):
    headers = {
        'accept': 'application/json',
        'authorization': f'Bearer {token}'
    }
    try:
        if method.upper() == "POST":
            response = requests.post(url, headers=headers)
        elif method.upper() == "GET":
            response = requests.get(url, headers=headers)
        else:
            return {"error": f"Unsupported method {method}"}
        return response.json() if response.text else {"status": response.status_code}
    except Exception as e:
        return {"error": str(e)}

def resume_endpoint(token, log_func):
    def task():
        log_func("Resuming endpoint...")
        result = call_api(RESUME_URL, token, "POST")
        log_func(f"Resume result: {result}")
    threading.Thread(target=task, daemon=True).start()

def pause_endpoint(token, log_func):
    def task():
        log_func("Pausing endpoint...")
        result = call_api(PAUSE_URL, token, "POST")
        log_func(f"Pause result: {result}")
    threading.Thread(target=task, daemon=True).start()

def check_status(token, log_func):
    def task():
        log_func("Checking status...")
        result = call_api(STATUS_URL, token, "GET")
        log_func(f"Status: {result}")
    threading.Thread(target=task, daemon=True).start()

# Automatic schedule check: every 30 seconds, check if the current time matches the scheduled times.
def schedule_check(token, pause_time_str, resume_time_str, auto_enabled, log_func, root):
    global last_pause_day, last_resume_day
    if auto_enabled.get():
        now = datetime.now()
        current_time = now.strftime("%H:%M")
        today = now.date()

        # Check pause time
        if pause_time_str.get().strip() == current_time:
            if last_pause_day != today:
                log_func(f"Automatic pause triggered at {current_time}")
                pause_endpoint(token.get(), log_func)
                last_pause_day = today

        # Check resume time
        if resume_time_str.get().strip() == current_time:
            if last_resume_day != today:
                log_func(f"Automatic resume triggered at {current_time}")
                resume_endpoint(token.get(), log_func)
                last_resume_day = today

    # Schedule this function to run again in 30 seconds (30000 milliseconds)
    root.after(30000, schedule_check, token, pause_time_str, resume_time_str, auto_enabled, log_func, root)

# Function to log messages in the UI
def log_message(log_widget, message):
    timestamp = datetime.now().strftime("[%Y-%m-%d %H:%M:%S]")
    log_widget.insert(tk.END, f"{timestamp} {message}\n")
    log_widget.see(tk.END)

def main():
    root = tk.Tk()
    root.title("Hugging Face Endpoint Control")

    # API Token input
    tk.Label(root, text="API Token:").grid(row=0, column=0, sticky="e")
    token_var = tk.StringVar()
    tk.Entry(root, textvariable=token_var, width=50, show="*").grid(row=0, column=1, padx=5, pady=5)

    # Buttons for resume, pause, check status
    button_frame = tk.Frame(root)
    button_frame.grid(row=1, column=0, columnspan=2, pady=5)

    log_text = tk.Text(root, height=15, width=70)
    log_text.grid(row=3, column=0, columnspan=2, padx=5, pady=5)

    # Helper: lambda to log messages in the text widget
    log_func = lambda msg: log_message(log_text, msg)

    resume_btn = tk.Button(button_frame, text="Resume", width=12,
                           command=lambda: resume_endpoint(token_var.get(), log_func))
    resume_btn.grid(row=0, column=0, padx=5)

    pause_btn = tk.Button(button_frame, text="Pause", width=12,
                          command=lambda: pause_endpoint(token_var.get(), log_func))
    pause_btn.grid(row=0, column=1, padx=5)

    status_btn = tk.Button(button_frame, text="Check Status", width=12,
                           command=lambda: check_status(token_var.get(), log_func))
    status_btn.grid(row=0, column=2, padx=5)

    # Automatic scheduling controls
    auto_enabled = tk.BooleanVar()
    auto_enabled.set(False)
    tk.Checkbutton(root, text="Enable Auto Pause/Resume", variable=auto_enabled).grid(row=2, column=0, columnspan=2, sticky="w", padx=5)

    tk.Label(root, text="Pause Time (HH:MM):").grid(row=4, column=0, sticky="e")
    pause_time_str = tk.StringVar(value="23:00")
    tk.Entry(root, textvariable=pause_time_str, width=10).grid(row=4, column=1, sticky="w", padx=5, pady=2)

    tk.Label(root, text="Resume Time (HH:MM):").grid(row=5, column=0, sticky="e")
    resume_time_str = tk.StringVar(value="08:00")
    tk.Entry(root, textvariable=resume_time_str, width=10).grid(row=5, column=1, sticky="w", padx=5, pady=2)

    # Start the automatic schedule check loop
    root.after(30000, schedule_check, token_var, pause_time_str, resume_time_str, auto_enabled, log_func, root)

    # Start Tkinter main loop
    root.mainloop()

if __name__ == "__main__":
    main()

How It Works

API Calls:
Each of the functions (resume_endpoint, pause_endpoint, and check_status) makes an HTTP call to the corresponding Hugging Face endpoint using the provided API token. They run on a separate thread so the UI remains responsive.
User Interface:
The UI provides an input field for your API token (masked for privacy). Three buttons allow you to manually issue resume, pause, and status-check commands. A text area logs responses and messages.
Automatic Scheduler:
A checkbox enables/disables the automatic schedule. You set the pause and resume times (in HH:MM format). Every 30 seconds the app checks whether the current time matches either scheduled time and, if it does (and the action hasn’t been triggered yet on that day), calls the appropriate endpoint.
Logging:
Actions and responses are logged with timestamps in the text widget at the bottom of the UI.

This example provides a basic framework; you can refine the UI design, error handling, and scheduling logic as needed for your use case. Enjoy testing and automating your Hugging Face endpoint control!

Below is an updated version of the Python application that not only provides buttons to resume, pause, and check the status of your endpoint but also pretty prints and syntax highlights the JSON responses in the log area using the Pygments library. (If you haven’t already installed Pygments, run pip install pygments.)

When a response is a dictionary or list, it is pretty printed (using 4‐space indent) and then “lexed” by Pygments so that different token types are inserted in different colors in the text widget. You can adjust the color mapping as desired.

Save the code (for example, as hf_endpoint_control.py) and run it.

import tkinter as tk
from datetime import datetime
import threading
import requests
import json
from pygments import lex
from pygments.lexers import JsonLexer

# API endpoints for your Hugging Face endpoint actions
RESUME_URL = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/resume"
PAUSE_URL  = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/pause"
STATUS_URL = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/replica"

# Global variables to avoid re-triggering the same action multiple times in one day
last_pause_day = None
last_resume_day = None

def call_api(url, token, method="POST"):
    headers = {
        'accept': 'application/json',
        'authorization': f'Bearer {token}'
    }
    try:
        if method.upper() == "POST":
            response = requests.post(url, headers=headers)
        elif method.upper() == "GET":
            response = requests.get(url, headers=headers)
        else:
            return {"error": f"Unsupported method {method}"}
        return response.json() if response.text else {"status": response.status_code}
    except Exception as e:
        return {"error": str(e)}

def resume_endpoint(token, log_func):
    def task():
        log_func("Resuming endpoint...")
        result = call_api(RESUME_URL, token, "POST")
        log_func(result)
    threading.Thread(target=task, daemon=True).start()

def pause_endpoint(token, log_func):
    def task():
        log_func("Pausing endpoint...")
        result = call_api(PAUSE_URL, token, "POST")
        log_func(result)
    threading.Thread(target=task, daemon=True).start()

def check_status(token, log_func):
    def task():
        log_func("Checking status...")
        result = call_api(STATUS_URL, token, "GET")
        log_func(result)
    threading.Thread(target=task, daemon=True).start()

def schedule_check(token, pause_time_str, resume_time_str, auto_enabled, log_func, root):
    """Automatically checks every 30 seconds if it is time to pause or resume."""
    global last_pause_day, last_resume_day
    if auto_enabled.get():
        now = datetime.now()
        current_time = now.strftime("%H:%M")
        today = now.date()
        if pause_time_str.get().strip() == current_time:
            if last_pause_day != today:
                log_func(f"Automatic pause triggered at {current_time}")
                pause_endpoint(token.get(), log_func)
                last_pause_day = today
        if resume_time_str.get().strip() == current_time:
            if last_resume_day != today:
                log_func(f"Automatic resume triggered at {current_time}")
                resume_endpoint(token.get(), log_func)
                last_resume_day = today
    root.after(30000, schedule_check, token, pause_time_str, resume_time_str, auto_enabled, log_func, root)

def highlight_json(text_widget, json_text):
    """
    Uses Pygments to lex a JSON string and insert it into the text_widget with syntax highlighting.
    A simple color scheme is applied according to the token type.
    """
    # Ensure widget is editable for insertion
    text_widget.configure(state="normal")
    lexer = JsonLexer()
    for ttype, value in lex(json_text, lexer):
         tag = str(ttype)
         # Create a tag for a token type if not already defined
         if tag not in text_widget.tag_names():
              # Define a simple color mapping for some token types
              if 'String' in tag:
                  color = "green"
              elif 'Number' in tag:
                  color = "blue"
              elif 'Keyword' in tag:
                  color = "orange"
              elif 'Name' in tag:
                  color = "darkblue"
              elif 'Operator' in tag:
                  color = "red"
              else:
                  color = "black"
              text_widget.tag_configure(tag, foreground=color)
         text_widget.insert(tk.END, value, tag)
    # Lock the widget
    text_widget.configure(state="disabled")

def log_message(log_widget, message):
    """
    Logs messages to the text widget.
    If the message is a dict or list, it pretty prints the JSON with syntax highlighting.
    """
    timestamp = datetime.now().strftime("[%Y-%m-%d %H:%M:%S] ")
    log_widget.configure(state="normal")
    log_widget.insert(tk.END, timestamp)
    if isinstance(message, (dict, list)):
         pretty = json.dumps(message, indent=4)
         highlight_json(log_widget, pretty)
         log_widget.insert(tk.END, "\n")
    else:
         log_widget.insert(tk.END, str(message) + "\n")
    log_widget.see(tk.END)
    log_widget.configure(state="disabled")

def main():
    root = tk.Tk()
    root.title("Hugging Face Endpoint Control")

    # API Token entry
    tk.Label(root, text="API Token:").grid(row=0, column=0, sticky="e")
    token_var = tk.StringVar()
    tk.Entry(root, textvariable=token_var, width=50, show="*").grid(row=0, column=1, padx=5, pady=5)

    # Control Buttons
    button_frame = tk.Frame(root)
    button_frame.grid(row=1, column=0, columnspan=2, pady=5)

    # Logging text widget
    log_text = tk.Text(root, height=20, width=80)
    log_text.grid(row=3, column=0, columnspan=2, padx=5, pady=5)

    # Lambda helper to log messages
    log_func = lambda msg: log_message(log_text, msg)

    resume_btn = tk.Button(button_frame, text="Resume", width=12,
                           command=lambda: resume_endpoint(token_var.get(), log_func))
    resume_btn.grid(row=0, column=0, padx=5)

    pause_btn = tk.Button(button_frame, text="Pause", width=12,
                          command=lambda: pause_endpoint(token_var.get(), log_func))
    pause_btn.grid(row=0, column=1, padx=5)

    status_btn = tk.Button(button_frame, text="Check Status", width=12,
                           command=lambda: check_status(token_var.get(), log_func))
    status_btn.grid(row=0, column=2, padx=5)

    # Automatic scheduling for pause/resume
    auto_enabled = tk.BooleanVar()
    auto_enabled.set(False)
    tk.Checkbutton(root, text="Enable Auto Pause/Resume", variable=auto_enabled).grid(row=2, column=0, columnspan=2, sticky="w", padx=5)

    tk.Label(root, text="Pause Time (HH:MM):").grid(row=4, column=0, sticky="e")
    pause_time_str = tk.StringVar(value="23:00")
    tk.Entry(root, textvariable=pause_time_str, width=10).grid(row=4, column=1, sticky="w", padx=5, pady=2)

    tk.Label(root, text="Resume Time (HH:MM):").grid(row=5, column=0, sticky="e")
    resume_time_str = tk.StringVar(value="08:00")
    tk.Entry(root, textvariable=resume_time_str, width=10).grid(row=5, column=1, sticky="w", padx=5, pady=2)

    # Start the automatic scheduler loop (runs every 30 seconds)
    root.after(30000, schedule_check, token_var, pause_time_str, resume_time_str, auto_enabled, log_func, root)

    root.mainloop()

if __name__ == "__main__":
    main()

How It Works

API Calls:
Functions like resume_endpoint, pause_endpoint, and check_status call the corresponding Hugging Face endpoints on separate threads so that the user interface remains responsive.
Pretty Print & Highlight:
The log_message() function checks if the response is a JSON object (dict or list) and pretty prints it using json.dumps with an indent of 4 spaces. The highlight_json() function uses the Pygments lexer for JSON to insert parts of the response in different colors (for example, strings appear in green, numbers in blue, etc.).
User Interface & Automatic Scheduling:
The Tkinter UI includes an entry for your API token, buttons for manual actions, and additional fields (with a checkbox) to schedule automatic pause/resume actions daily based on the provided times.

Enjoy testing and using your new Python tool to manage your Hugging Face endpoint!

Below is an updated version of the Python app that sets a monospace font for the log widget and includes some minor modifications to ensure that the pretty-printed JSON is inserted with its newlines and spaces intact. In this version, the log widget is configured to use a monospace font (Courier), and the highlight function handles the tokens properly so that your JSON formatting isn’t lost.

Make sure you have installed the required packages (Tkinter is built‑in; install requests and pygments with pip):

pip install requests pygments

Then try the updated code below:

import tkinter as tk
from datetime import datetime
import threading
import requests
import json
from pygments import lex
from pygments.lexers import JsonLexer

# API endpoints for your Hugging Face endpoint actions
RESUME_URL = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/resume"
PAUSE_URL  = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/pause"
STATUS_URL = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/replica"

last_pause_day = None
last_resume_day = None

def call_api(url, token, method="POST"):
    headers = {
        'accept': 'application/json',
        'authorization': f'Bearer {token}'
    }
    try:
        if method.upper() == "POST":
            response = requests.post(url, headers=headers)
        elif method.upper() == "GET":
            response = requests.get(url, headers=headers)
        else:
            return {"error": f"Unsupported method {method}"}
        return response.json() if response.text else {"status": response.status_code}
    except Exception as e:
        return {"error": str(e)}

def resume_endpoint(token, log_func):
    def task():
        log_func("Resuming endpoint...")
        result = call_api(RESUME_URL, token, "POST")
        log_func(result)
    threading.Thread(target=task, daemon=True).start()

def pause_endpoint(token, log_func):
    def task():
        log_func("Pausing endpoint...")
        result = call_api(PAUSE_URL, token, "POST")
        log_func(result)
    threading.Thread(target=task, daemon=True).start()

def check_status(token, log_func):
    def task():
        log_func("Checking status...")
        result = call_api(STATUS_URL, token, "GET")
        log_func(result)
    threading.Thread(target=task, daemon=True).start()

def schedule_check(token, pause_time_str, resume_time_str, auto_enabled, log_func, root):
    """Automatically checks every 30 seconds if it is time to pause or resume."""
    global last_pause_day, last_resume_day
    if auto_enabled.get():
        now = datetime.now()
        current_time = now.strftime("%H:%M")
        today = now.date()
        if pause_time_str.get().strip() == current_time:
            if last_pause_day != today:
                log_func(f"Automatic pause triggered at {current_time}")
                pause_endpoint(token.get(), log_func)
                last_pause_day = today
        if resume_time_str.get().strip() == current_time:
            if last_resume_day != today:
                log_func(f"Automatic resume triggered at {current_time}")
                resume_endpoint(token.get(), log_func)
                last_resume_day = today
    root.after(30000, schedule_check, token, pause_time_str, resume_time_str, auto_enabled, log_func, root)

def highlight_json(text_widget, json_text):
    """
    Uses Pygments to lex a JSON string and insert it into the text_widget with syntax highlighting.
    This implementation assumes the text_widget is already set to a monospace font.
    """
    # Make sure the widget is editable while inserting text
    text_widget.configure(state="normal")
    lexer = JsonLexer()
    for ttype, value in lex(json_text, lexer):
         tag = str(ttype)
         # Define a tag for a token type if not already defined
         if tag not in text_widget.tag_names():
              if 'String' in tag:
                  color = "green"
              elif 'Number' in tag:
                  color = "blue"
              elif 'Keyword' in tag:
                  color = "orange"
              elif 'Name' in tag:
                  color = "darkblue"
              elif 'Operator' in tag:
                  color = "red"
              else:
                  color = "black"
              text_widget.tag_configure(tag, foreground=color)
         text_widget.insert(tk.END, value, tag)
    text_widget.configure(state="disabled")

def log_message(log_widget, message):
    """
    Logs messages to the text widget.
    If the message is a dict or list, it is pretty printed and then syntax highlighted.
    """
    timestamp = datetime.now().strftime("[%Y-%m-%d %H:%M:%S] ")
    log_widget.configure(state="normal")
    log_widget.insert(tk.END, timestamp)
    if isinstance(message, (dict, list)):
         pretty = json.dumps(message, indent=4)
         highlight_json(log_widget, pretty)
         log_widget.insert(tk.END, "\n")
    else:
         log_widget.insert(tk.END, str(message) + "\n")
    log_widget.see(tk.END)
    log_widget.configure(state="disabled")

def main():
    root = tk.Tk()
    root.title("Hugging Face Endpoint Control")

    # API Token entry
    tk.Label(root, text="API Token:").grid(row=0, column=0, sticky="e")
    token_var = tk.StringVar()
    tk.Entry(root, textvariable=token_var, width=50, show="*").grid(row=0, column=1, padx=5, pady=5)

    # Control Buttons
    button_frame = tk.Frame(root)
    button_frame.grid(row=1, column=0, columnspan=2, pady=5)

    # Logging text widget with a monospace font
    log_text = tk.Text(root, height=20, width=80, font=("Courier", 10))
    log_text.grid(row=3, column=0, columnspan=2, padx=5, pady=5)

    # Lambda helper to log messages
    log_func = lambda msg: log_message(log_text, msg)

    resume_btn = tk.Button(button_frame, text="Resume", width=12,
                           command=lambda: resume_endpoint(token_var.get(), log_func))
    resume_btn.grid(row=0, column=0, padx=5)

    pause_btn = tk.Button(button_frame, text="Pause", width=12,
                          command=lambda: pause_endpoint(token_var.get(), log_func))
    pause_btn.grid(row=0, column=1, padx=5)

    status_btn = tk.Button(button_frame, text="Check Status", width=12,
                           command=lambda: check_status(token_var.get(), log_func))
    status_btn.grid(row=0, column=2, padx=5)

    # Automatic scheduling for pause/resume
    auto_enabled = tk.BooleanVar()
    auto_enabled.set(False)
    tk.Checkbutton(root, text="Enable Auto Pause/Resume", variable=auto_enabled).grid(row=2, column=0, columnspan=2, sticky="w", padx=5)

    tk.Label(root, text="Pause Time (HH:MM):").grid(row=4, column=0, sticky="e")
    pause_time_str = tk.StringVar(value="23:00")
    tk.Entry(root, textvariable=pause_time_str, width=10).grid(row=4, column=1, sticky="w", padx=5, pady=2)

    tk.Label(root, text="Resume Time (HH:MM):").grid(row=5, column=0, sticky="e")
    resume_time_str = tk.StringVar(value="08:00")
    tk.Entry(root, textvariable=resume_time_str, width=10).grid(row=5, column=1, sticky="w", padx=5, pady=2)

    # Start the automatic scheduler loop (runs every 30 seconds)
    root.after(30000, schedule_check, token_var, pause_time_str, resume_time_str, auto_enabled, log_func, root)

    root.mainloop()

if __name__ == "__main__":
    main()

Key Changes

Monospace Font:
The log_text widget is now explicitly set to use Courier (a monospace font), which helps preserve the indentation and spacing from your pretty-printed JSON.
Highlighting Function:
The highlight_json function processes each token from Pygments and inserts it into the text widget. Any token that represents a string, number, keyword, etc., is given a color tag. Newline characters are preserved as part of the token values.
State Management:
The text widget is made editable during insertion and then set back to read-only at the end of each log message to prevent accidental modifications.

This updated version should help maintain the structure and syntax highlighting of your JSON responses in the UI. Let me know if you run into any further issues!

although i do the json.dumps with indent=4, that’s what gets pritned

“Status: {‘items’: [{‘id’: ‘hy83m5q6-2goka’, ‘versionId’: ‘hy83m5q6’, ‘status’: {‘startedAt’: ‘2025-04-10T08:34:36Z’, ‘stoppedAt’: None, ‘stage’: ‘running’, ‘live’: True, ‘updating’: False}}, {‘id’: ‘92bxzzsa-1x8n7’, ‘versionId’: ‘92bxzzsa’, ‘status’: {‘startedAt’: ‘2025-04-10T08:16:48Z’, ‘stoppedAt’: ‘2025-04-10T08:19:38Z’, ‘stage’: ‘stopped’, ‘live’: False, ‘updating’: False}}, {‘id’: ‘3d7xkyv4-p0zsk’, ‘versionId’: ‘3d7xkyv4’, ‘status’: {‘startedAt’: ‘2025-04-10T08:12:57Z’, ‘stoppedAt’: ‘2025-04-10T08:16:01Z’, ‘stage’: ‘stopped’, ‘live’: False, ‘updating’: False}}, {‘id’: ‘p819oj6v-b4vpe’, ‘versionId’: ‘p819oj6v’, ‘status’: {‘startedAt’: ‘2025-04-09T13:21:34Z’, ‘stoppedAt’: ‘2025-04-09T13:43:38Z’, ‘stage’: ‘stopped’, ‘live’: False, ‘updating’: False}}, {‘id’: ‘adnx09rt-6u8xl’, ‘versionId’: ‘adnx09rt’, ‘status’: {‘startedAt’: ‘2025-04-07T13:33:18Z’, ‘stoppedAt’: ‘2025-04-07T13:55:21Z’, ‘stage’: ‘stopped’, ‘live’: False, ‘updating’: False}}, {‘id’: ‘kg53jwdj-nh4f7’, ‘versionId’: ‘kg53jwdj’, ‘status’: {‘startedAt’: ‘2025-04-07T11:33:12Z’, ‘stoppedAt’: ‘2025-04-07T11:39:07Z’, ‘stage’: ‘stopped’, ‘live’: False, ‘updating’: False}}, {‘id’: ‘g25gd115-bv91u’, ‘versionId’: ‘g25gd115’, ‘status’: {‘startedAt’: ‘2025-04-07T09:40:09Z’, ‘stoppedAt’: ‘2025-04-07T09:49:32Z’, ‘stage’: ‘stopped’, ‘live’: False, ‘updating’: False}}, {‘id’: ‘4ob1w5yi-8amtj’, ‘versionId’: ‘4ob1w5yi’, ‘status’: {‘startedAt’: ‘2025-04-07T09:08:27Z’, ‘stoppedAt’: ‘2025-04-07T09:42:22Z’, ‘stage’: ‘stopped’, ‘live’: False, ‘updating’: False}}]}”

Below is an updated version of the code. In this version, the log function first checks whether it received a dictionary/list (in which case it pretty‑prints it) and otherwise it will try to detect a JSON-like substring (using Python’s safe literal evaluation with ast.literal_eval). This way, if you accidentally pass a string such as

"Status: {'items': [{'id': 'hy83m5q6-2goka', ...}]}"

the code will search for the first curly brace, try to parse the substring into a Python object, and if successful, pretty print that object. (If it fails, it simply prints the string as is.)

Make sure you have the required packages installed (Tkinter is part of Python’s standard library; install requests and pygments via pip):

pip install requests pygments

Then use the code below (for example, saved as hf_endpoint_control.py):

import tkinter as tk
from datetime import datetime
import threading
import requests
import json
import ast
from pygments import lex
from pygments.lexers import JsonLexer

# API endpoints for your Hugging Face endpoint actions
RESUME_URL = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/resume"
PAUSE_URL  = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/pause"
STATUS_URL = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/replica"

last_pause_day = None
last_resume_day = None

def call_api(url, token, method="POST"):
    headers = {
        'accept': 'application/json',
        'authorization': f'Bearer {token}'
    }
    try:
        if method.upper() == "POST":
            response = requests.post(url, headers=headers)
        elif method.upper() == "GET":
            response = requests.get(url, headers=headers)
        else:
            return {"error": f"Unsupported method {method}"}
        return response.json() if response.text else {"status": response.status_code}
    except Exception as e:
        return {"error": str(e)}

def resume_endpoint(token, log_func):
    def task():
        log_func("Resuming endpoint...")
        result = call_api(RESUME_URL, token, "POST")
        log_func(result)
    threading.Thread(target=task, daemon=True).start()

def pause_endpoint(token, log_func):
    def task():
        log_func("Pausing endpoint...")
        result = call_api(PAUSE_URL, token, "POST")
        log_func(result)
    threading.Thread(target=task, daemon=True).start()

def check_status(token, log_func):
    def task():
        log_func("Checking status...")
        result = call_api(STATUS_URL, token, "GET")
        log_func(result)
    threading.Thread(target=task, daemon=True).start()

def schedule_check(token, pause_time_str, resume_time_str, auto_enabled, log_func, root):
    """Automatically checks every 30 seconds if it is time to pause or resume."""
    global last_pause_day, last_resume_day
    if auto_enabled.get():
        now = datetime.now()
        current_time = now.strftime("%H:%M")
        today = now.date()
        if pause_time_str.get().strip() == current_time:
            if last_pause_day != today:
                log_func(f"Automatic pause triggered at {current_time}")
                pause_endpoint(token.get(), log_func)
                last_pause_day = today
        if resume_time_str.get().strip() == current_time:
            if last_resume_day != today:
                log_func(f"Automatic resume triggered at {current_time}")
                resume_endpoint(token.get(), log_func)
                last_resume_day = today
    root.after(30000, schedule_check, token, pause_time_str, resume_time_str, auto_enabled, log_func, root)

def highlight_json(text_widget, json_text):
    """
    Uses Pygments to lex a JSON string and insert it into the text_widget with syntax highlighting.
    Assumes the text_widget uses a monospace font.
    """
    text_widget.configure(state="normal")
    lexer = JsonLexer()
    for ttype, value in lex(json_text, lexer):
        tag = str(ttype)
        if tag not in text_widget.tag_names():
            # Define a color mapping for some token types.
            if 'String' in tag:
                color = "green"
            elif 'Number' in tag:
                color = "blue"
            elif 'Keyword' in tag:
                color = "orange"
            elif 'Name' in tag:
                color = "darkblue"
            elif 'Operator' in tag:
                color = "red"
            else:
                color = "black"
            text_widget.tag_configure(tag, foreground=color)
        text_widget.insert(tk.END, value, tag)
    text_widget.configure(state="disabled")

def log_message(log_widget, message):
    """
    Logs messages to the text widget.
    If the message is a dict or list, it pretty prints it.
    Otherwise, it tries to detect JSON in a string (using ast.literal_eval)
    and pretty prints that if possible.
    """
    timestamp = datetime.now().strftime("[%Y-%m-%d %H:%M:%S] ")
    log_widget.configure(state="normal")
    log_widget.insert(tk.END, timestamp)

    data = None
    if isinstance(message, (dict, list)):
        data = message
    elif isinstance(message, str):
        # If the string contains a JSON-like substring, try to extract and parse it.
        idx = message.find('{')
        if idx != -1:
            try:
                possible_json = message[idx:]
                data = ast.literal_eval(possible_json)
            except Exception:
                pass
    if data is not None:
        # Pretty print the data with indent=4 and syntax highlight it.
        pretty = json.dumps(data, indent=4)
        highlight_json(log_widget, pretty)
        log_widget.insert(tk.END, "\n")
    else:
        log_widget.insert(tk.END, str(message) + "\n")
    log_widget.see(tk.END)
    log_widget.configure(state="disabled")

def main():
    root = tk.Tk()
    root.title("Hugging Face Endpoint Control")

    # API Token entry
    tk.Label(root, text="API Token:").grid(row=0, column=0, sticky="e")
    token_var = tk.StringVar()
    tk.Entry(root, textvariable=token_var, width=50, show="*").grid(row=0, column=1, padx=5, pady=5)

    # Control Buttons
    button_frame = tk.Frame(root)
    button_frame.grid(row=1, column=0, columnspan=2, pady=5)

    # Logging text widget with a monospace font (Courier)
    log_text = tk.Text(root, height=20, width=80, font=("Courier", 10))
    log_text.grid(row=3, column=0, columnspan=2, padx=5, pady=5)

    # Lambda helper to log messages
    log_func = lambda msg: log_message(log_text, msg)

    resume_btn = tk.Button(button_frame, text="Resume", width=12,
                           command=lambda: resume_endpoint(token_var.get(), log_func))
    resume_btn.grid(row=0, column=0, padx=5)

    pause_btn = tk.Button(button_frame, text="Pause", width=12,
                          command=lambda: pause_endpoint(token_var.get(), log_func))
    pause_btn.grid(row=0, column=1, padx=5)

    status_btn = tk.Button(button_frame, text="Check Status", width=12,
                           command=lambda: check_status(token_var.get(), log_func))
    status_btn.grid(row=0, column=2, padx=5)

    # Automatic scheduling for pause/resume
    auto_enabled = tk.BooleanVar()
    auto_enabled.set(False)
    tk.Checkbutton(root, text="Enable Auto Pause/Resume", variable=auto_enabled).grid(row=2, column=0, columnspan=2, sticky="w", padx=5)

    tk.Label(root, text="Pause Time (HH:MM):").grid(row=4, column=0, sticky="e")
    pause_time_str = tk.StringVar(value="23:00")
    tk.Entry(root, textvariable=pause_time_str, width=10).grid(row=4, column=1, sticky="w", padx=5, pady=2)

    tk.Label(root, text="Resume Time (HH:MM):").grid(row=5, column=0, sticky="e")
    resume_time_str = tk.StringVar(value="08:00")
    tk.Entry(root, textvariable=resume_time_str, width=10).grid(row=5, column=1, sticky="w", padx=5, pady=2)

    # Start the automatic scheduler loop (runs every 30 seconds)
    root.after(30000, schedule_check, token_var, pause_time_str, resume_time_str, auto_enabled, log_func, root)

    root.mainloop()

if __name__ == "__main__":
    main()

How It Works

log_message() Function:
- If the message is a dictionary or list, it uses json.dumps with an indent of 4.
- If the message is a string, it looks for the first "{" character and attempts to safely evaluate the remainder using ast.literal_eval.
- If successful, it pretty prints the parsed data with syntax highlighting. Otherwise, it simply prints the string.
highlight_json() Function:
Uses Pygments’ JsonLexer to parse the pretty‑printed JSON and insert it token‑by‑token into the text widget, applying a color tag for various token types.
Monospace Font:
The text widget uses the “Courier” font so that indentation is preserved.

This updated version should help preserve the formatting you expect. If you still see the JSON on one line, ensure you’re not pre-concatenating text (like adding "Status: " before the dictionary) before passing it to the log function. Instead, pass the raw JSON data so that the pretty-printing can occur properly.

Below is an updated version of your Python app that now does two things in addition to the previous behavior:

Autoscaling Text Widget:
The text field now uses grid weight configuration (with an attached vertical scrollbar) so that it automatically resizes as you change the window size.
State‐of‐the‐Art Search:
A search bar (with “Search” and “Clear” buttons) is added above the log widget. When you enter a term and press “Search,” the app uses the Tkinter text widget’s search method to find and highlight all occurrences in the log. Matching text is highlighted (with a yellow background), and you can clear the highlights using the “Clear” button.

Save the code (for example, as hf_endpoint_control.py), install the required packages if needed (pip install requests pygments), and run it.

import tkinter as tk
from tkinter import ttk
from datetime import datetime
import threading
import requests
import json
import ast
from pygments import lex
from pygments.lexers import JsonLexer

# API endpoints for your Hugging Face endpoint actions
RESUME_URL = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/resume"
PAUSE_URL  = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/pause"
STATUS_URL = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/replica"

# Global variables to avoid re-triggering the same action multiple times in a day
last_pause_day = None
last_resume_day = None

def call_api(url, token, method="POST"):
    headers = {
        'accept': 'application/json',
        'authorization': f'Bearer {token}'
    }
    try:
        if method.upper() == "POST":
            response = requests.post(url, headers=headers)
        elif method.upper() == "GET":
            response = requests.get(url, headers=headers)
        else:
            return {"error": f"Unsupported method {method}"}
        return response.json() if response.text else {"status": response.status_code}
    except Exception as e:
        return {"error": str(e)}

def resume_endpoint(token, log_func):
    def task():
        log_func("Resuming endpoint...")
        result = call_api(RESUME_URL, token, "POST")
        log_func(result)
    threading.Thread(target=task, daemon=True).start()

def pause_endpoint(token, log_func):
    def task():
        log_func("Pausing endpoint...")
        result = call_api(PAUSE_URL, token, "POST")
        log_func(result)
    threading.Thread(target=task, daemon=True).start()

def check_status(token, log_func):
    def task():
        log_func("Checking status...")
        result = call_api(STATUS_URL, token, "GET")
        log_func(result)
    threading.Thread(target=task, daemon=True).start()

def schedule_check(token, pause_time_str, resume_time_str, auto_enabled, log_func, root):
    """Automatically checks every 30 seconds if it is time to pause or resume."""
    global last_pause_day, last_resume_day
    if auto_enabled.get():
        now = datetime.now()
        current_time = now.strftime("%H:%M")
        today = now.date()
        if pause_time_str.get().strip() == current_time:
            if last_pause_day != today:
                log_func(f"Automatic pause triggered at {current_time}")
                pause_endpoint(token.get(), log_func)
                last_pause_day = today
        if resume_time_str.get().strip() == current_time:
            if last_resume_day != today:
                log_func(f"Automatic resume triggered at {current_time}")
                resume_endpoint(token.get(), log_func)
                last_resume_day = today
    root.after(30000, schedule_check, token, pause_time_str, resume_time_str, auto_enabled, log_func, root)

def highlight_json(text_widget, json_text):
    """
    Uses Pygments to lex a JSON string and insert it into the text_widget with syntax highlighting.
    Assumes the text_widget uses a monospace font.
    """
    text_widget.configure(state="normal")
    lexer = JsonLexer()
    for ttype, value in lex(json_text, lexer):
        tag = str(ttype)
        if tag not in text_widget.tag_names():
            # Define a color mapping for a few token types.
            if 'String' in tag:
                color = "green"
            elif 'Number' in tag:
                color = "blue"
            elif 'Keyword' in tag:
                color = "orange"
            elif 'Name' in tag:
                color = "darkblue"
            elif 'Operator' in tag:
                color = "red"
            else:
                color = "black"
            text_widget.tag_configure(tag, foreground=color)
        text_widget.insert(tk.END, value, tag)
    text_widget.configure(state="disabled")

def log_message(log_widget, message):
    """
    Logs messages to the text widget.
    If the message is a dict or list, it pretty prints it.
    Otherwise, attempts to detect and pretty print a JSON-like substring.
    """
    timestamp = datetime.now().strftime("[%Y-%m-%d %H:%M:%S] ")
    log_widget.configure(state="normal")
    log_widget.insert(tk.END, timestamp)

    data = None
    if isinstance(message, (dict, list)):
        data = message
    elif isinstance(message, str):
        idx = message.find('{')
        if idx != -1:
            try:
                possible_json = message[idx:]
                data = ast.literal_eval(possible_json)
            except Exception:
                pass
    if data is not None:
        pretty = json.dumps(data, indent=4)
        highlight_json(log_widget, pretty)
        log_widget.insert(tk.END, "\n")
    else:
        log_widget.insert(tk.END, str(message) + "\n")
    log_widget.see(tk.END)
    log_widget.configure(state="disabled")

def search_text(text_widget, query):
    """Search the text widget for the query and highlight all matches."""
    # Remove previous search highlights.
    text_widget.tag_remove("found", "1.0", tk.END)
    if not query:
        return
    start_pos = "1.0"
    while True:
        start_pos = text_widget.search(query, start_pos, stopindex=tk.END, nocase=True)
        if not start_pos:
            break
        end_pos = f"{start_pos}+{len(query)}c"
        text_widget.tag_add("found", start_pos, end_pos)
        start_pos = end_pos
    # Configure the "found" tag to have a yellow background.
    text_widget.tag_config("found", background="yellow")

def clear_search(text_widget):
    """Clear search highlights."""
    text_widget.tag_remove("found", "1.0", tk.END)

def main():
    root = tk.Tk()
    root.title("Hugging Face Endpoint Control")

    # Configure grid to have resizable rows and columns.
    root.rowconfigure(3, weight=1)
    root.columnconfigure(1, weight=1)

    # API Token entry
    tk.Label(root, text="API Token:").grid(row=0, column=0, sticky="e", padx=5, pady=5)
    token_var = tk.StringVar()
    tk.Entry(root, textvariable=token_var, width=50, show="*").grid(row=0, column=1, sticky="ew", padx=5, pady=5)

    # Control Buttons
    button_frame = tk.Frame(root)
    button_frame.grid(row=1, column=0, columnspan=2, pady=5)

    resume_btn = tk.Button(button_frame, text="Resume", width=12,
                           command=lambda: resume_endpoint(token_var.get(), log_func))
    resume_btn.grid(row=0, column=0, padx=5)

    pause_btn = tk.Button(button_frame, text="Pause", width=12,
                          command=lambda: pause_endpoint(token_var.get(), log_func))
    pause_btn.grid(row=0, column=1, padx=5)

    status_btn = tk.Button(button_frame, text="Check Status", width=12,
                           command=lambda: check_status(token_var.get(), log_func))
    status_btn.grid(row=0, column=2, padx=5)

    # Automatic scheduling for pause/resume
    auto_enabled = tk.BooleanVar()
    auto_enabled.set(False)
    tk.Checkbutton(root, text="Enable Auto Pause/Resume", variable=auto_enabled).grid(row=2, column=0, columnspan=2, sticky="w", padx=5)

    tk.Label(root, text="Pause Time (HH:MM):").grid(row=4, column=0, sticky="e", padx=5)
    pause_time_str = tk.StringVar(value="23:00")
    tk.Entry(root, textvariable=pause_time_str, width=10).grid(row=4, column=1, sticky="w", padx=5, pady=2)

    tk.Label(root, text="Resume Time (HH:MM):").grid(row=5, column=0, sticky="e", padx=5)
    resume_time_str = tk.StringVar(value="08:00")
    tk.Entry(root, textvariable=resume_time_str, width=10).grid(row=5, column=1, sticky="w", padx=5, pady=2)

    # Search frame: Entry and buttons for searching log text.
    search_frame = tk.Frame(root)
    search_frame.grid(row=6, column=0, columnspan=2, sticky="ew", padx=5, pady=5)
    tk.Label(search_frame, text="Search:").grid(row=0, column=0, sticky="w")
    search_var = tk.StringVar()
    search_entry = tk.Entry(search_frame, textvariable=search_var, width=30)
    search_entry.grid(row=0, column=1, sticky="ew", padx=5)
    search_btn = tk.Button(search_frame, text="Search", command=lambda: search_text(log_text, search_var.get()))
    search_btn.grid(row=0, column=2, padx=5)
    clear_btn = tk.Button(search_frame, text="Clear", command=lambda: clear_search(log_text))
    clear_btn.grid(row=0, column=3, padx=5)
    search_frame.columnconfigure(1, weight=1)

    # Logging text widget with a monospace font and a vertical scrollbar.
    log_text = tk.Text(root, height=20, width=80, font=("Courier", 10))
    log_text.grid(row=3, column=0, columnspan=2, sticky="nsew", padx=5, pady=5)
    scrollbar = tk.Scrollbar(root, command=log_text.yview)
    scrollbar.grid(row=3, column=2, sticky="ns")
    log_text.configure(yscrollcommand=scrollbar.set)

    # Helper lambda for logging.
    log_func = lambda msg: log_message(log_text, msg)

    # Start the automatic scheduler loop (runs every 30 seconds)
    root.after(30000, schedule_check, token_var, pause_time_str, resume_time_str, auto_enabled, log_func, root)

    root.mainloop()

if __name__ == "__main__":
    main()

How It Works

Autoscaling Text Widget:
The text widget is placed with sticky="nsew" in grid row 3, and the root window’s row and column weights have been set so that the widget expands as you resize the window. A vertical scrollbar is attached to the text widget.
Search Functionality:
A search frame is added with an entry field and “Search” and “Clear” buttons. The search_text function highlights every occurrence (using a yellow background) of the search query, while clear_search removes these highlights.

This setup provides a robust, easy-to-navigate log view so you can quickly locate the responses among all the data dumps. Enjoy using your enhanced control app!

What you’re seeing is a historical record of all the replicas that have been launched for your endpoint—not just the single currently active one. Even if you’ve set up your endpoint to run with one replica, the status endpoint reports every instance that has been started (and then stopped) over time. In other words:

Historical Entries: Each time the endpoint is resumed or even restarted due to other events, a new replica instance is created. These instances remain in the status list as historical entries.
Lifecycle Tracking: The API is designed to show the full lifecycle of the endpoint’s replica(s) so that you can track when a replica was started, stopped, and its status at that time.
Active vs. Inactive: Even if only one replica is running at a given moment, the list includes previous instances that are now stopped or inactive. Look at the “live” flag and the “stage” field to tell which (if any) is the current, active one.

This behavior allows you to review the operational history (for example, to debug issues or monitor performance over time) even though, for billing and performance, only one instance is active at a time.

Below is one way to do it: after you receive the full status response (which is a JSON object with an “items” list), you can extract the latest replica by comparing their start times. For example, if each replica object has a timestamp in its "status" dictionary (like "startedAt"), you can sort (or use the max function) based on that field. One sample approach is shown below.

In this updated snippet, we add a helper function, get_latest_replica(), which takes the JSON response and returns the replica with the most recent "startedAt" value. Then we modify the check_status() function to only log that latest entry.

Below is the modified code:

import tkinter as tk
from datetime import datetime
import threading
import requests
import json
import ast
from pygments import lex
from pygments.lexers import JsonLexer

# API endpoints for your Hugging Face endpoint actions
RESUME_URL = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/resume"
PAUSE_URL  = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/pause"
STATUS_URL = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/replica"

last_pause_day = None
last_resume_day = None

def call_api(url, token, method="POST"):
    headers = {
        'accept': 'application/json',
        'authorization': f'Bearer {token}'
    }
    try:
        if method.upper() == "POST":
            response = requests.post(url, headers=headers)
        elif method.upper() == "GET":
            response = requests.get(url, headers=headers)
        else:
            return {"error": f"Unsupported method {method}"}
        return response.json() if response.text else {"status": response.status_code}
    except Exception as e:
        return {"error": str(e)}

def get_latest_replica(data):
    """
    Given a response dictionary with an 'items' list, extract the replica
    with the most recent 'startedAt' value.
    """
    items = data.get('items', [])
    if not items:
        return None
    # Parse the "startedAt" timestamps (which are assumed to be in ISO 8601 format)
    def parse_time(item):
        time_str = item.get("status", {}).get("startedAt", "")
        try:
            # Convert to a datetime object (handling the trailing 'Z')
            return datetime.fromisoformat(time_str.replace("Z", "+00:00"))
        except Exception:
            return datetime.min
    # Use max() to select the item with the latest start time.
    return max(items, key=parse_time)

def resume_endpoint(token, log_func):
    def task():
        log_func("Resuming endpoint...")
        result = call_api(RESUME_URL, token, "POST")
        log_func(result)
    threading.Thread(target=task, daemon=True).start()

def pause_endpoint(token, log_func):
    def task():
        log_func("Pausing endpoint...")
        result = call_api(PAUSE_URL, token, "POST")
        log_func(result)
    threading.Thread(target=task, daemon=True).start()

def check_status(token, log_func):
    def task():
        log_func("Checking status...")
        result = call_api(STATUS_URL, token, "GET")
        # If we received an object with an "items" list, try to extract the latest replica.
        if isinstance(result, dict) and "items" in result:
            latest = get_latest_replica(result)
            if latest:
                log_func(f"Latest replica:\n{json.dumps(latest, indent=4)}")
            else:
                log_func("No replicas found.")
        else:
            log_func(result)
    threading.Thread(target=task, daemon=True).start()

def schedule_check(token, pause_time_str, resume_time_str, auto_enabled, log_func, root):
    global last_pause_day, last_resume_day
    if auto_enabled.get():
        now = datetime.now()
        current_time = now.strftime("%H:%M")
        today = now.date()
        if pause_time_str.get().strip() == current_time:
            if last_pause_day != today:
                log_func(f"Automatic pause triggered at {current_time}")
                pause_endpoint(token.get(), log_func)
                last_pause_day = today
        if resume_time_str.get().strip() == current_time:
            if last_resume_day != today:
                log_func(f"Automatic resume triggered at {current_time}")
                resume_endpoint(token.get(), log_func)
                last_resume_day = today
    root.after(30000, schedule_check, token, pause_time_str, resume_time_str, auto_enabled, log_func, root)

def highlight_json(text_widget, json_text):
    """
    Uses Pygments to lex a JSON string and insert it into the text_widget with syntax highlighting.
    Assumes the text_widget uses a monospace font.
    """
    text_widget.configure(state="normal")
    lexer = JsonLexer()
    for ttype, value in lex(json_text, lexer):
        tag = str(ttype)
        if tag not in text_widget.tag_names():
            # Define a color mapping for some token types.
            if 'String' in tag:
                color = "green"
            elif 'Number' in tag:
                color = "blue"
            elif 'Keyword' in tag:
                color = "orange"
            elif 'Name' in tag:
                color = "darkblue"
            elif 'Operator' in tag:
                color = "red"
            else:
                color = "black"
            text_widget.tag_configure(tag, foreground=color)
        text_widget.insert(tk.END, value, tag)
    text_widget.configure(state="disabled")

def log_message(log_widget, message):
    """
    Logs messages to the text widget.
    If the message is a dict or list, it pretty prints it.
    Otherwise, attempts to detect and pretty print a JSON-like substring.
    """
    timestamp = datetime.now().strftime("[%Y-%m-%d %H:%M:%S] ")
    log_widget.configure(state="normal")
    log_widget.insert(tk.END, timestamp)

    data = None
    if isinstance(message, (dict, list)):
        data = message
    elif isinstance(message, str):
        idx = message.find('{')
        if idx != -1:
            try:
                possible_json = message[idx:]
                data = ast.literal_eval(possible_json)
            except Exception:
                pass
    if data is not None:
        pretty = json.dumps(data, indent=4)
        highlight_json(log_widget, pretty)
        log_widget.insert(tk.END, "\n")
    else:
        log_widget.insert(tk.END, str(message) + "\n")
    log_widget.see(tk.END)
    log_widget.configure(state="disabled")

# (The rest of your UI code goes here, e.g., resume/pause buttons, autoscaling text widget, search functionality, etc.)

def main():
    root = tk.Tk()
    root.title("Hugging Face Endpoint Control")

    # Configuration for autoscaling the text widget in the grid.
    root.rowconfigure(3, weight=1)
    root.columnconfigure(1, weight=1)

    # API Token entry
    tk.Label(root, text="API Token:").grid(row=0, column=0, sticky="e", padx=5, pady=5)
    token_var = tk.StringVar()
    tk.Entry(root, textvariable=token_var, width=50, show="*").grid(row=0, column=1, sticky="ew", padx=5, pady=5)

    # Control Buttons
    button_frame = tk.Frame(root)
    button_frame.grid(row=1, column=0, columnspan=2, pady=5)

    resume_btn = tk.Button(button_frame, text="Resume", width=12,
                           command=lambda: resume_endpoint(token_var.get(), log_func))
    resume_btn.grid(row=0, column=0, padx=5)

    pause_btn = tk.Button(button_frame, text="Pause", width=12,
                          command=lambda: pause_endpoint(token_var.get(), log_func))
    pause_btn.grid(row=0, column=1, padx=5)

    status_btn = tk.Button(button_frame, text="Check Status", width=12,
                           command=lambda: check_status(token_var.get(), log_func))
    status_btn.grid(row=0, column=2, padx=5)

    # Automatic scheduling for pause/resume
    auto_enabled = tk.BooleanVar()
    auto_enabled.set(False)
    tk.Checkbutton(root, text="Enable Auto Pause/Resume", variable=auto_enabled).grid(row=2, column=0, columnspan=2, sticky="w", padx=5)

    tk.Label(root, text="Pause Time (HH:MM):").grid(row=4, column=0, sticky="e", padx=5)
    pause_time_str = tk.StringVar(value="23:00")
    tk.Entry(root, textvariable=pause_time_str, width=10).grid(row=4, column=1, sticky="w", padx=5, pady=2)

    tk.Label(root, text="Resume Time (HH:MM):").grid(row=5, column=0, sticky="e", padx=5)
    resume_time_str = tk.StringVar(value="08:00")
    tk.Entry(root, textvariable=resume_time_str, width=10).grid(row=5, column=1, sticky="w", padx=5, pady=2)

    # Search frame for the log widget.
    search_frame = tk.Frame(root)
    search_frame.grid(row=6, column=0, columnspan=2, sticky="ew", padx=5, pady=5)
    tk.Label(search_frame, text="Search:").grid(row=0, column=0, sticky="w")
    search_var = tk.StringVar()
    search_entry = tk.Entry(search_frame, textvariable=search_var, width=30)
    search_entry.grid(row=0, column=1, sticky="ew", padx=5)
    search_btn = tk.Button(search_frame, text="Search", command=lambda: search_text(log_text, search_var.get()))
    search_btn.grid(row=0, column=2, padx=5)
    clear_btn = tk.Button(search_frame, text="Clear", command=lambda: clear_search(log_text))
    clear_btn.grid(row=0, column=3, padx=5)
    search_frame.columnconfigure(1, weight=1)

    # Logging text widget with a monospace font and a vertical scrollbar.
    log_text = tk.Text(root, height=20, width=80, font=("Courier", 10))
    log_text.grid(row=3, column=0, columnspan=2, sticky="nsew", padx=5, pady=5)
    scrollbar = tk.Scrollbar(root, command=log_text.yview)
    scrollbar.grid(row=3, column=2, sticky="ns")
    log_text.configure(yscrollcommand=scrollbar.set)

    # Helper lambda for logging.
    global log_func
    log_func = lambda msg: log_message(log_text, msg)

    root.after(30000, schedule_check, token_var, pause_time_str, resume_time_str, auto_enabled, log_func, root)
    root.mainloop()

# Search functionality helpers.
def search_text(text_widget, query):
    text_widget.tag_remove("found", "1.0", tk.END)
    if not query:
        return
    start_pos = "1.0"
    while True:
        start_pos = text_widget.search(query, start_pos, stopindex=tk.END, nocase=True)
        if not start_pos:
            break
        end_pos = f"{start_pos}+{len(query)}c"
        text_widget.tag_add("found", start_pos, end_pos)
        start_pos = end_pos
    text_widget.tag_config("found", background="yellow")

def clear_search(text_widget):
    text_widget.tag_remove("found", "1.0", tk.END)

if __name__ == "__main__":
    main()

Explanation

get_latest_replica() Function:
This helper function looks at the "items" list in the status JSON and returns the replica with the most recent "startedAt" field. (It uses Python’s built‑in max() with a custom key function based on parsing the ISO timestamp.)
Modified check_status():
Once the status JSON is fetched, we call get_latest_replica() to extract just the latest replica and then pretty-print that (using json.dumps with indentation).
UI and Autoscaling:
The remainder of the code keeps your autoscaling text widget, search bar, and control buttons intact.

Now, when you click “Check Status,” only the latest replica’s information will be extracted and displayed. Feel free to adjust the logic (for example, filtering by a "live": True flag if that better defines the “active” replica) as needed for your use case.

Below is an example of a complete Python application that not only provides Resume/Pause controls but also displays a “dashboard” of the latest replica’s properties in a well‐organized UI. In this design:

• The status panel shows key properties (“id”, “versionId”, “startedAt”, “stoppedAt”, “stage”, “live”, “updating”) as labels.
• A small Canvas is used as a LED indicator that paints green if the replica is live (ON) and red if not.
• Once you either resume or pause the replica, the app automatically polls the status every 30 seconds (using Tkinter’s after method) to update the display.

To run this example, be sure you have installed the required packages (Tkinter is built in; install requests via pip):

pip install requests

Then save the code (for example, as hf_endpoint_dashboard.py) and run it.

import tkinter as tk
from tkinter import ttk
from datetime import datetime
import threading
import requests
import json

# API endpoints for your Hugging Face endpoint actions
RESUME_URL = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/resume"
PAUSE_URL  = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/pause"
STATUS_URL = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/replica"

# ----------------------------------------
# API helper functions
# ----------------------------------------
def call_api(url, token, method="POST"):
    headers = {
        'accept': 'application/json',
        'authorization': f'Bearer {token}'
    }
    try:
        if method.upper() == "POST":
            response = requests.post(url, headers=headers)
        elif method.upper() == "GET":
            response = requests.get(url, headers=headers)
        else:
            return {"error": f"Unsupported method {method}"}
        return response.json() if response.text else {"status": response.status_code}
    except Exception as e:
        return {"error": str(e)}

def get_latest_replica(data):
    """
    Given a response dictionary with an 'items' list, return the replica
    with the most recent 'startedAt' timestamp.
    """
    items = data.get('items', [])
    if not items:
        return None

    # Parse ISO timestamps and select the item with the highest (latest) value
    def parse_time(item):
        time_str = item.get("status", {}).get("startedAt", "")
        try:
            return datetime.fromisoformat(time_str.replace("Z", "+00:00"))
        except Exception:
            return datetime.min

    return max(items, key=parse_time)

# ----------------------------------------
# UI update functions
# ----------------------------------------
def update_status_ui(replica):
    """Updates the status panel with the latest replica info."""
    if replica is None:
        label_replica_id.config(text="N/A")
        label_version_id.config(text="N/A")
        label_started_at.config(text="N/A")
        label_stopped_at.config(text="N/A")
        label_stage.config(text="N/A")
        label_updating.config(text="N/A")
        update_led(False)
    else:
        label_replica_id.config(text=replica.get('id', ''))
        label_version_id.config(text=replica.get('versionId', ''))
        status = replica.get('status', {})
        label_started_at.config(text=status.get('startedAt', 'N/A'))
        label_stopped_at.config(text=status.get('stoppedAt', 'N/A'))
        label_stage.config(text=status.get('stage', 'N/A'))
        label_updating.config(text=str(status.get('updating', 'N/A')))
        update_led(status.get('live', False))

def update_led(is_live):
    """Draws a green circle if live; otherwise red."""
    canvas_led.delete("all")
    color = "green" if is_live else "red"
    # Draw a circle (20x20)
    canvas_led.create_oval(2, 2, 22, 22, fill=color, outline=color)

# ----------------------------------------
# Functions to interact with the endpoint
# ----------------------------------------
def resume_endpoint(token):
    def task():
        result = call_api(RESUME_URL, token, "POST")
        # After resuming, trigger status update
        poll_status()
    threading.Thread(target=task, daemon=True).start()

def pause_endpoint(token):
    def task():
        result = call_api(PAUSE_URL, token, "POST")
        # After pausing, trigger status update
        poll_status()
    threading.Thread(target=task, daemon=True).start()

def fetch_status(token, callback):
    """Fetches the status and passes the latest replica to the callback."""
    result = call_api(STATUS_URL, token, "GET")
    if isinstance(result, dict) and "items" in result:
        latest = get_latest_replica(result)
        callback(latest)
    else:
        callback(None)

def poll_status():
    """Polls the status every 30 seconds and updates the UI."""
    token = token_var.get().strip()
    if token:  # Only if a token is provided
        threading.Thread(target=lambda: fetch_status(token,
                                   lambda replica: root.after(0, update_status_ui, replica)),
                         daemon=True).start()
    # Schedule next poll in 30 seconds (30000 milliseconds)
    root.after(30000, poll_status)

# ----------------------------------------
# Main UI
# ----------------------------------------
root = tk.Tk()
root.title("Hugging Face Endpoint Control Dashboard")

# Configure grid so the status frame expands with window resizing
root.columnconfigure(0, weight=1)
root.rowconfigure(3, weight=1)

# --------------------------
# Top: API Token and Buttons
# --------------------------
frame_top = ttk.Frame(root, padding="10")
frame_top.grid(row=0, column=0, sticky="ew")

ttk.Label(frame_top, text="API Token:").grid(row=0, column=0, padx=5, pady=5, sticky="e")
token_var = tk.StringVar()
token_entry = ttk.Entry(frame_top, textvariable=token_var, width=50, show="*")
token_entry.grid(row=0, column=1, padx=5, pady=5, sticky="ew")
frame_top.columnconfigure(1, weight=1)

button_frame = ttk.Frame(root, padding="10")
button_frame.grid(row=1, column=0, sticky="ew")
resume_btn = ttk.Button(button_frame, text="Resume Endpoint", command=lambda: resume_endpoint(token_var.get()))
resume_btn.grid(row=0, column=0, padx=5)
pause_btn = ttk.Button(button_frame, text="Pause Endpoint", command=lambda: pause_endpoint(token_var.get()))
pause_btn.grid(row=0, column=1, padx=5)

# --------------------------
# Status Panel: Display the latest replica properties
# --------------------------
status_frame = ttk.LabelFrame(root, text="Replica Status", padding="10")
status_frame.grid(row=2, column=0, padx=10, pady=10, sticky="nsew")
status_frame.columnconfigure(1, weight=1)

# LED indicator canvas (for live status)
ttk.Label(status_frame, text="Live:").grid(row=0, column=0, sticky="w", padx=5, pady=5)
canvas_led = tk.Canvas(status_frame, width=24, height=24)
canvas_led.grid(row=0, column=1, sticky="w", padx=5, pady=5)

# Replica ID
ttk.Label(status_frame, text="Replica ID:").grid(row=1, column=0, sticky="e", padx=5, pady=5)
label_replica_id = ttk.Label(status_frame, text="N/A")
label_replica_id.grid(row=1, column=1, sticky="w", padx=5, pady=5)

# Version ID
ttk.Label(status_frame, text="Version ID:").grid(row=2, column=0, sticky="e", padx=5, pady=5)
label_version_id = ttk.Label(status_frame, text="N/A")
label_version_id.grid(row=2, column=1, sticky="w", padx=5, pady=5)

# Started At
ttk.Label(status_frame, text="Started At:").grid(row=3, column=0, sticky="e", padx=5, pady=5)
label_started_at = ttk.Label(status_frame, text="N/A")
label_started_at.grid(row=3, column=1, sticky="w", padx=5, pady=5)

# Stopped At
ttk.Label(status_frame, text="Stopped At:").grid(row=4, column=0, sticky="e", padx=5, pady=5)
label_stopped_at = ttk.Label(status_frame, text="N/A")
label_stopped_at.grid(row=4, column=1, sticky="w", padx=5, pady=5)

# Stage
ttk.Label(status_frame, text="Stage:").grid(row=5, column=0, sticky="e", padx=5, pady=5)
label_stage = ttk.Label(status_frame, text="N/A")
label_stage.grid(row=5, column=1, sticky="w", padx=5, pady=5)

# Updating
ttk.Label(status_frame, text="Updating:").grid(row=6, column=0, sticky="e", padx=5, pady=5)
label_updating = ttk.Label(status_frame, text="N/A")
label_updating.grid(row=6, column=1, sticky="w", padx=5, pady=5)

# --------------------------
# Start automatic polling of status every 30 seconds.
# --------------------------
# Immediately poll status and then every 30 seconds will refresh.
poll_status()

root.mainloop()

How It Works

Dashboard UI:
The top section contains an API token entry and two buttons (“Resume Endpoint” and “Pause Endpoint”).
The middle section (in a Labelframe) is dedicated to displaying the replica’s properties. Each field is updated with the latest values from the API.
A small Canvas is used as an LED indicator; it draws a green circle when the "live" property is true and a red circle when false.
Automatic Polling:
The function poll_status() is scheduled with root.after(30000, poll_status) to run every 30 seconds. It fetches the status in a background thread and then updates the UI (using root.after(0, update_status_ui, replica)) on the main thread.
Button Actions:
When you press the Resume or Pause button, the respective API call is invoked in a background thread. After that call completes, the status is updated immediately (and then again on the 30‑second polling schedule).

Feel free to adjust colors, fonts, or layout further to suit your aesthetic preferences. Enjoy your new dashboard UI!

Below is an updated version of the dashboard code that starts polling for status updates as soon as the API token field is non‐empty. In this version, we add a callback (using a trace on the token variable) so that once a token is entered the first time, polling begins immediately (and runs every 30 seconds thereafter).

The LED indicator (green for “live”, red otherwise) and the status panel are updated based on the latest replica data from the API.

Save the code (e.g. as hf_endpoint_dashboard.py), and run it once you have installed the required packages:

pip install requests

Below is the complete code:

import tkinter as tk
from tkinter import ttk
from datetime import datetime
import threading
import requests
import json

# API endpoints for your Hugging Face endpoint actions
RESUME_URL = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/resume"
PAUSE_URL  = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/pause"
STATUS_URL = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/replica"

# Global flag to ensure that polling only starts once a token is provided.
polling_started = False

# ----------------------------------------
# API helper functions
# ----------------------------------------
def call_api(url, token, method="POST"):
    headers = {
        'accept': 'application/json',
        'authorization': f'Bearer {token}'
    }
    try:
        if method.upper() == "POST":
            response = requests.post(url, headers=headers)
        elif method.upper() == "GET":
            response = requests.get(url, headers=headers)
        else:
            return {"error": f"Unsupported method {method}"}
        return response.json() if response.text else {"status": response.status_code}
    except Exception as e:
        return {"error": str(e)}

def get_latest_replica(data):
    """
    Given a response dictionary with an 'items' list, return the replica
    with the most recent 'startedAt' timestamp.
    """
    items = data.get('items', [])
    if not items:
        return None

    def parse_time(item):
        time_str = item.get("status", {}).get("startedAt", "")
        try:
            return datetime.fromisoformat(time_str.replace("Z", "+00:00"))
        except Exception:
            return datetime.min

    return max(items, key=parse_time)

# ----------------------------------------
# UI update functions
# ----------------------------------------
def update_status_ui(replica):
    """Updates the status panel with the latest replica info."""
    if replica is None:
        label_replica_id.config(text="N/A")
        label_version_id.config(text="N/A")
        label_started_at.config(text="N/A")
        label_stopped_at.config(text="N/A")
        label_stage.config(text="N/A")
        label_updating.config(text="N/A")
        update_led(False)
    else:
        label_replica_id.config(text=replica.get('id', ''))
        label_version_id.config(text=replica.get('versionId', ''))
        status = replica.get('status', {})
        label_started_at.config(text=status.get('startedAt', 'N/A'))
        label_stopped_at.config(text=status.get('stoppedAt', 'N/A'))
        label_stage.config(text=status.get('stage', 'N/A'))
        label_updating.config(text=str(status.get('updating', 'N/A')))
        update_led(status.get('live', False))

def update_led(is_live):
    """Draws a green circle if live; otherwise red."""
    canvas_led.delete("all")
    color = "green" if is_live else "red"
    canvas_led.create_oval(2, 2, 22, 22, fill=color, outline=color)

# ----------------------------------------
# Functions to interact with the endpoint
# ----------------------------------------
def resume_endpoint(token):
    def task():
        call_api(RESUME_URL, token, "POST")
        poll_status()  # Immediately update status afterward.
    threading.Thread(target=task, daemon=True).start()

def pause_endpoint(token):
    def task():
        call_api(PAUSE_URL, token, "POST")
        poll_status()  # Immediately update status afterward.
    threading.Thread(target=task, daemon=True).start()

def fetch_status(token, callback):
    """Fetches the status and passes the latest replica to the callback."""
    result = call_api(STATUS_URL, token, "GET")
    if isinstance(result, dict) and "items" in result:
        latest = get_latest_replica(result)
        callback(latest)
    else:
        callback(None)

def poll_status():
    """Polls the status every 30 seconds and updates the UI."""
    token = token_var.get().strip()
    if token:  # Only if a token is provided.
        threading.Thread(
            target=lambda: fetch_status(token, lambda replica: root.after(0, update_status_ui, replica)),
            daemon=True
        ).start()
    # Schedule next poll in 30 seconds.
    root.after(30000, poll_status)

# ----------------------------------------
# Callback for token changes
# ----------------------------------------
def on_token_change(*args):
    global polling_started
    token = token_var.get().strip()
    if token and not polling_started:
        polling_started = True
        poll_status()

# ----------------------------------------
# Build the UI
# ----------------------------------------
root = tk.Tk()
root.title("Hugging Face Endpoint Control Dashboard")

# Configure grid so the status frame expands.
root.columnconfigure(0, weight=1)
root.rowconfigure(2, weight=1)

# --------------------------
# Top: API Token and Buttons
# --------------------------
frame_top = ttk.Frame(root, padding="10")
frame_top.grid(row=0, column=0, sticky="ew")

ttk.Label(frame_top, text="API Token:").grid(row=0, column=0, padx=5, pady=5, sticky="e")
token_var = tk.StringVar()
token_entry = ttk.Entry(frame_top, textvariable=token_var, width=50, show="*")
token_entry.grid(row=0, column=1, padx=5, pady=5, sticky="ew")
frame_top.columnconfigure(1, weight=1)

# Add a trace callback to token_var so polling starts as soon as a token is entered.
token_var.trace("w", on_token_change)

button_frame = ttk.Frame(root, padding="10")
button_frame.grid(row=1, column=0, sticky="ew")
resume_btn = ttk.Button(button_frame, text="Resume Endpoint", command=lambda: resume_endpoint(token_var.get()))
resume_btn.grid(row=0, column=0, padx=5)
pause_btn = ttk.Button(button_frame, text="Pause Endpoint", command=lambda: pause_endpoint(token_var.get()))
pause_btn.grid(row=0, column=1, padx=5)

# --------------------------
# Status Panel: Display the latest replica properties
# --------------------------
status_frame = ttk.LabelFrame(root, text="Replica Status", padding="10")
status_frame.grid(row=2, column=0, padx=10, pady=10, sticky="nsew")
status_frame.columnconfigure(1, weight=1)

ttk.Label(status_frame, text="Live:").grid(row=0, column=0, sticky="w", padx=5, pady=5)
canvas_led = tk.Canvas(status_frame, width=24, height=24)
canvas_led.grid(row=0, column=1, sticky="w", padx=5, pady=5)

ttk.Label(status_frame, text="Replica ID:").grid(row=1, column=0, sticky="e", padx=5, pady=5)
label_replica_id = ttk.Label(status_frame, text="N/A")
label_replica_id.grid(row=1, column=1, sticky="w", padx=5, pady=5)

ttk.Label(status_frame, text="Version ID:").grid(row=2, column=0, sticky="e", padx=5, pady=5)
label_version_id = ttk.Label(status_frame, text="N/A")
label_version_id.grid(row=2, column=1, sticky="w", padx=5, pady=5)

ttk.Label(status_frame, text="Started At:").grid(row=3, column=0, sticky="e", padx=5, pady=5)
label_started_at = ttk.Label(status_frame, text="N/A")
label_started_at.grid(row=3, column=1, sticky="w", padx=5, pady=5)

ttk.Label(status_frame, text="Stopped At:").grid(row=4, column=0, sticky="e", padx=5, pady=5)
label_stopped_at = ttk.Label(status_frame, text="N/A")
label_stopped_at.grid(row=4, column=1, sticky="w", padx=5, pady=5)

ttk.Label(status_frame, text="Stage:").grid(row=5, column=0, sticky="e", padx=5, pady=5)
label_stage = ttk.Label(status_frame, text="N/A")
label_stage.grid(row=5, column=1, sticky="w", padx=5, pady=5)

ttk.Label(status_frame, text="Updating:").grid(row=6, column=0, sticky="e", padx=5, pady=5)
label_updating = ttk.Label(status_frame, text="N/A")
label_updating.grid(row=6, column=1, sticky="w", padx=5, pady=5)

# Start the main loop.
root.mainloop()

How It Works

Token Trace Callback:
A trace is added on the token_var (the API token variable). When the token value is updated (and becomes non-empty), the function on_token_change() is called. If polling hasn’t started already, it will set a flag and start the polling process immediately.
Polling Status:
The poll_status() function is scheduled to run every 30 seconds via root.after(30000, poll_status) and fetches the latest replica using the provided token.
Once the status is fetched, update_status_ui() updates all labels and the LED indicator.
Endpoint Controls:
The “Resume” and “Pause” buttons invoke their corresponding API calls and immediately trigger a status poll afterward.

Feel free to further adjust the UI layout or color scheme as needed. Enjoy your responsive dashboard that starts polling right away when you provide your token!

Below is an updated version of the dashboard code that now includes a timer feature. When you press the Resume button, the app records the current time. Then, as the status is polled every 30 seconds, if the latest replica’s "stage" becomes "running", the app calculates and displays the elapsed time between pressing Resume and reaching the running state.

In this example, the dashboard UI now includes an extra field labeled “Time to Running” that shows the elapsed time in a human‑readable format.

Save the code (for example, as hf_endpoint_dashboard_timer.py), install the required packages (pip install requests), and run it.

import tkinter as tk
from tkinter import ttk
from datetime import datetime
import threading
import requests
import json

# API endpoints for your Hugging Face endpoint actions
RESUME_URL = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/resume"
PAUSE_URL  = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/pause"
STATUS_URL = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/replica"

# Global flag to ensure that polling only starts once a token is provided.
polling_started = False
# Global variable to record when resume was pressed
resume_start_time = None

# ----------------------------------------
# API helper functions
# ----------------------------------------
def call_api(url, token, method="POST"):
    headers = {
        'accept': 'application/json',
        'authorization': f'Bearer {token}'
    }
    try:
        if method.upper() == "POST":
            response = requests.post(url, headers=headers)
        elif method.upper() == "GET":
            response = requests.get(url, headers=headers)
        else:
            return {"error": f"Unsupported method {method}"}
        return response.json() if response.text else {"status": response.status_code}
    except Exception as e:
        return {"error": str(e)}

def get_latest_replica(data):
    """
    Given a response dictionary with an 'items' list,
    return the replica with the most recent 'startedAt' timestamp.
    """
    items = data.get('items', [])
    if not items:
        return None

    def parse_time(item):
        time_str = item.get("status", {}).get("startedAt", "")
        try:
            return datetime.fromisoformat(time_str.replace("Z", "+00:00"))
        except Exception:
            return datetime.min

    return max(items, key=parse_time)

# ----------------------------------------
# UI update functions
# ----------------------------------------
def update_status_ui(replica):
    """Updates the status panel with the latest replica info."""
    global resume_start_time
    if replica is None:
        label_replica_id.config(text="N/A")
        label_version_id.config(text="N/A")
        label_started_at.config(text="N/A")
        label_stopped_at.config(text="N/A")
        label_stage.config(text="N/A")
        label_updating.config(text="N/A")
        label_time_to_running.config(text="N/A")
        update_led(False)
    else:
        label_replica_id.config(text=replica.get('id', ''))
        label_version_id.config(text=replica.get('versionId', ''))
        status = replica.get('status', {})
        label_started_at.config(text=status.get('startedAt', 'N/A'))
        label_stopped_at.config(text=status.get('stoppedAt', 'N/A'))
        label_stage.config(text=status.get('stage', 'N/A'))
        label_updating.config(text=str(status.get('updating', 'N/A')))
        update_led(status.get('live', False))
        # If the stage is "running", compute the elapsed time from the moment resume was pressed.
        if status.get("stage", "").lower() == "running" and resume_start_time is not None:
            elapsed = datetime.now() - resume_start_time
            # Format elapsed as H:MM:SS
            elapsed_str = str(elapsed).split('.')[0]  # remove microseconds
            label_time_to_running.config(text=elapsed_str)
            # Optionally, reset the timer so that it's only calculated once per resume.
            resume_start_time = None
        else:
            label_time_to_running.config(text="Waiting...")

def update_led(is_live):
    """Draws a green circle if live; otherwise red."""
    canvas_led.delete("all")
    color = "green" if is_live else "red"
    canvas_led.create_oval(2, 2, 22, 22, fill=color, outline=color)

# ----------------------------------------
# Functions to interact with the endpoint
# ----------------------------------------
def resume_endpoint(token):
    global resume_start_time
    # Record the time when resume was pressed.
    resume_start_time = datetime.now()
    def task():
        call_api(RESUME_URL, token, "POST")
        poll_status()  # Immediately update status afterward.
    threading.Thread(target=task, daemon=True).start()

def pause_endpoint(token):
    def task():
        call_api(PAUSE_URL, token, "POST")
        poll_status()  # Immediately update status afterward.
    threading.Thread(target=task, daemon=True).start()

def fetch_status(token, callback):
    """Fetches the status and passes the latest replica to the callback."""
    result = call_api(STATUS_URL, token, "GET")
    if isinstance(result, dict) and "items" in result:
        latest = get_latest_replica(result)
        callback(latest)
    else:
        callback(None)

def poll_status():
    """Polls the status every 30 seconds and updates the UI."""
    token = token_var.get().strip()
    if token:  # Only if a token is provided.
        threading.Thread(
            target=lambda: fetch_status(token, lambda replica: root.after(0, update_status_ui, replica)),
            daemon=True
        ).start()
    # Schedule the next poll in 30 seconds.
    root.after(30000, poll_status)

# ----------------------------------------
# Callback for token changes
# ----------------------------------------
def on_token_change(*args):
    global polling_started
    token = token_var.get().strip()
    if token and not polling_started:
        polling_started = True
        poll_status()

# ----------------------------------------
# Build the UI
# ----------------------------------------
root = tk.Tk()
root.title("Hugging Face Endpoint Control Dashboard")

# Configure grid so the main frame expands.
root.columnconfigure(0, weight=1)
root.rowconfigure(2, weight=1)

# --------------------------
# Top: API Token and Buttons
# --------------------------
frame_top = ttk.Frame(root, padding="10")
frame_top.grid(row=0, column=0, sticky="ew")

ttk.Label(frame_top, text="API Token:").grid(row=0, column=0, padx=5, pady=5, sticky="e")
token_var = tk.StringVar()
token_entry = ttk.Entry(frame_top, textvariable=token_var, width=50, show="*")
token_entry.grid(row=0, column=1, padx=5, pady=5, sticky="ew")
frame_top.columnconfigure(1, weight=1)

# Trace changes on token_var so that polling starts as soon as a token is entered.
token_var.trace("w", on_token_change)

button_frame = ttk.Frame(root, padding="10")
button_frame.grid(row=1, column=0, sticky="ew")
resume_btn = ttk.Button(button_frame, text="Resume Endpoint", command=lambda: resume_endpoint(token_var.get()))
resume_btn.grid(row=0, column=0, padx=5)
pause_btn = ttk.Button(button_frame, text="Pause Endpoint", command=lambda: pause_endpoint(token_var.get()))
pause_btn.grid(row=0, column=1, padx=5)

# --------------------------
# Status Panel: Display the latest replica properties
# --------------------------
status_frame = ttk.LabelFrame(root, text="Replica Status", padding="10")
status_frame.grid(row=2, column=0, padx=10, pady=10, sticky="nsew")
status_frame.columnconfigure(1, weight=1)

# LED indicator for live status.
ttk.Label(status_frame, text="Live:").grid(row=0, column=0, sticky="w", padx=5, pady=5)
canvas_led = tk.Canvas(status_frame, width=24, height=24)
canvas_led.grid(row=0, column=1, sticky="w", padx=5, pady=5)

# Replica ID
ttk.Label(status_frame, text="Replica ID:").grid(row=1, column=0, sticky="e", padx=5, pady=5)
label_replica_id = ttk.Label(status_frame, text="N/A")
label_replica_id.grid(row=1, column=1, sticky="w", padx=5, pady=5)

# Version ID
ttk.Label(status_frame, text="Version ID:").grid(row=2, column=0, sticky="e", padx=5, pady=5)
label_version_id = ttk.Label(status_frame, text="N/A")
label_version_id.grid(row=2, column=1, sticky="w", padx=5, pady=5)

# Started At
ttk.Label(status_frame, text="Started At:").grid(row=3, column=0, sticky="e", padx=5, pady=5)
label_started_at = ttk.Label(status_frame, text="N/A")
label_started_at.grid(row=3, column=1, sticky="w", padx=5, pady=5)

# Stopped At
ttk.Label(status_frame, text="Stopped At:").grid(row=4, column=0, sticky="e", padx=5, pady=5)
label_stopped_at = ttk.Label(status_frame, text="N/A")
label_stopped_at.grid(row=4, column=1, sticky="w", padx=5, pady=5)

# Stage
ttk.Label(status_frame, text="Stage:").grid(row=5, column=0, sticky="e", padx=5, pady=5)
label_stage = ttk.Label(status_frame, text="N/A")
label_stage.grid(row=5, column=1, sticky="w", padx=5, pady=5)

# Updating
ttk.Label(status_frame, text="Updating:").grid(row=6, column=0, sticky="e", padx=5, pady=5)
label_updating = ttk.Label(status_frame, text="N/A")
label_updating.grid(row=6, column=1, sticky="w", padx=5, pady=5)

# Time to Running Timer
ttk.Label(status_frame, text="Time to Running:").grid(row=7, column=0, sticky="e", padx=5, pady=5)
label_time_to_running = ttk.Label(status_frame, text="N/A")
label_time_to_running.grid(row=7, column=1, sticky="w", padx=5, pady=5)

# Start polling immediately once token is provided.
root.mainloop()

How It Works

Recording the Start Time:
When the Resume button is pressed, the global variable resume_start_time is set to the current time (using datetime.now()).
Polling Status:
The app polls the endpoint status every 30 seconds. The fetch_status function extracts the latest replica based on the "startedAt" field.
Calculating Elapsed Time:
In update_status_ui(), if the replica’s status "stage" is "running" and resume_start_time is set, the code calculates the time difference between now and resume_start_time, formats it (as H:MM:SS), and displays it in the new “Time to Running” field. Once calculated, the timer resets so that it only reflects the latest resume event.
Immediate Polling on Token Entry:
A trace on the token variable starts the polling immediately as soon as a token is entered.

This updated version allows you to monitor not only the status of the replica with a colored LED (green for live, red otherwise) but also shows how long it took for the resumed endpoint to change its state to "running". Enjoy your enhanced dashboard!

Porting a Python Tkinter application to a native SwiftUI app isn’t a one-to-one process, but the underlying logic is similar. In other words, the same features (network polling, timers, and dynamic UI updates) exist in SwiftUI; you just need to re‐implement them in a way that fits the Swift and SwiftUI paradigms. Here’s what to expect:

What Remains Similar

Core Logic & Features:
You still need to:
- Make HTTP requests to your API endpoints (using URLSession instead of Python’s requests).
- Parse JSON responses (using Swift’s Codable protocols).
- Poll the API on a timer (using a Timer publisher, async/await, or DispatchQueue).
- Update the UI based on the current state (e.g., show a green or red LED indicator, display properties such as replica id, version, status, etc.).
- Measure elapsed time from when Resume is pressed until the replica’s stage becomes “running.”
Data Flow:
Your application’s fundamental logic (fetching data, updating the UI) is the same. In SwiftUI, you would likely encapsulate this in an ObservableObject (a view model) with @Published properties that drive the UI.

What Changes

Language and Framework Differences:
- UI Framework:
  Instead of Tkinter, you’ll use SwiftUI’s declarative syntax to build your views. For example, buttons, text fields, and indicators are constructed using views like Button, TextField, and Circle.
- Networking:
  You’ll use URLSession for network calls and the Codable system to decode JSON into Swift structures.
- Timers and Concurrency:
  With Swift 5.5 and newer, you can use async/await or Combine’s Timer publishers. SwiftUI’s declarative nature means you might write something like:
```
Timer.publish(every: 30, on: .main, in: .common)
     .autoconnect()
     .sink { _ in
         // Poll the status and update @Published properties
     }
     .store(in: &cancellables)
```
  Or, if you prefer async/await, you can wrap your polling in a task that sleeps for 30 seconds between calls.
State Management:
Instead of using Tkinter’s callbacks and manual thread scheduling, you’d use @State and @ObservedObject to automatically update your UI when your data changes. For example, your view might automatically re-render when a published replicaStatus property in your view model changes.
UI Responsiveness:
SwiftUI allows you to define a “dashboard” view with dynamic layout, and it’s straightforward to add a Circle view that changes color (green for running, red for stopped), just like your LED indicator.

How Simple Is It?

For an Experienced iOS Developer:
If you’re already familiar with SwiftUI, Combine (or async/await), and URLSession, porting the functionality should be quite straightforward. The conceptual tasks (network calls, timers, data binding) have direct equivalents in the Swift ecosystem.
For a Developer New to SwiftUI:
There will be a learning curve as you learn the declarative paradigm, data binding with @State and @ObservedObject, and Swift’s asynchronous programming patterns. However, once you’re up to speed, the tasks are well-documented and there are many examples of similar dashboard apps.

High-Level Outline of a SwiftUI Implementation

Create Data Models:
Define structs that conform to Codable for your API responses:

struct ReplicaStatus: Codable, Identifiable {
    let id: String
    let versionId: String
    let status: StatusInfo

    struct StatusInfo: Codable {
        let startedAt: String
        let stoppedAt: String?
        let stage: String
        let live: Bool
        let updating: Bool
    }
}

Build a ViewModel:
Create an ObservableObject that:
- Fetches data from the API using URLSession.
- Uses a Timer (or async loop) to poll every 30 seconds.
- Records the time when Resume is pressed and calculates elapsed time when the status reaches "running".
- Exposes @Published properties for the replica status and elapsed time.
Design the SwiftUI View:
Using SwiftUI’s declarative syntax, build your interface:
- A text field for the API token.
- Buttons for Resume and Pause that invoke the view model’s methods.
- A status panel showing labels for replica properties.
- A Circle view that changes color based on the live property.
- A timer display that shows how long it took for the instance to become “running.”
Integration:
Bind the view’s components to the view model. As the view model updates its properties, the UI refreshes automatically.

Conclusion

Porting the functionality to SwiftUI is very feasible. While you’ll need to rewrite the code to adhere to SwiftUI’s declarative style and use Swift’s networking and concurrency mechanisms, the overall complexity is similar to the Python version once you’re familiar with the SwiftUI ecosystem. If you have experience with Swift and iOS development, you could set up a basic version in a matter of hours, and refine it over time.

Let me know if you’d like a more detailed code sample or further guidance on specific parts of the porting process!

In many cloud inference environments the GPU resources and the host’s system memory are decoupled. That means even when you’re paying for a 4 GPU L4 cluster, the underlying instance might be configured to provide only 24GB of system (CPU) RAM regardless of how many GPUs are attached. Here are a few reasons why this happens:

• Instance Configuration:
The hardware configuration for the L4 cluster is preset. In many cases, the instance type is designed for inference workloads where the heavy lifting is offloaded to the GPUs and the CPU does not require a large amount of memory. So, while your GPUs have their own dedicated VRAM for handling large models and inference tasks, the system memory remains capped at 24GB according to the instance specification.

• Cost Optimization:
Providers often optimize for cost efficiency. If your workload is primarily GPU‐bound, the extra system memory might add unnecessary cost. The available 24GB is likely deemed sufficient for handling input data, pre‑ and post‑processing, and other minor CPU tasks required by the inference workload.

• Hardware Trade-offs:
When designing instances optimized for deep learning inference, hardware vendors balance CPU performance, system memory, and GPU capacity. Even with multiple GPUs, if your primary workload isn’t memory‑intensive on the CPU side, the host memory may be purposely limited.

In summary, even if you have a cluster with 4 GPUs, the instance’s CPU memory remains at 24GB because that’s the pre‐configured allocation for that type of instance. If you need more host memory, you might need to look at a different instance type or configuration that offers a higher CPU memory-to-GPU ratio.

NVIDIA’s L4 GPUs each come with 24GB of dedicated VRAM, so even if you’re paying for a “4‑GPU L4 cluster,” each individual GPU still has 24GB of memory. The graph you’re seeing reflects how much memory a single GPU can utilize (up to 24GB), rather than an aggregated total across all four cards. Here’s why:

One L4 GPU = 24GB VRAM
Each L4 GPU in your cluster has a fixed 24GB of onboard graphics memory. That 24GB isn’t multiplied by the number of GPUs unless your workload explicitly spans multiple GPUs in parallel.
Typical GPU Utilization Reporting
The metrics dashboard usually plots usage per GPU, so if your inference workload is only using one GPU at a time, you’ll see the memory cap near 24GB. Even with four GPUs available, each is capped at 24GB individually.
Multi-GPU Workloads
If your model or framework can distribute the workload across multiple GPUs—using data parallelism or model parallelism—you could potentially leverage the total VRAM of all four (for a combined 96GB). However, this requires configuring your environment and model scripts to scale across GPUs.
Compare with Other GPU Types
If you need more VRAM on a single GPU (rather than across multiple GPUs), you’d look at GPUs like the A100 (40GB or 80GB) or similar. But in a multi-GPU setup, each card’s VRAM remains fixed to what that specific GPU offers (24GB for L4, 40GB or 80GB for certain A100 SKUs, etc.).

In short, the “24GB maximum” in your dashboard refers to the per‑card limit on the L4 GPU. Even though you have four of them, each one is still capped at 24GB VRAM unless you explicitly scale your workload across multiple GPUs.

Quantization often comes down to balancing memory footprint (and inference speed) against model accuracy. While there’s no absolute “one-size-fits-all” winner for every large language model, GPTQ and Bitsandbytes NF4 (4-bit) have emerged as popular choices for 30B+ parameter models. Here’s a quick rundown of your options:

None
- No quantization; highest memory usage but preserves all floating‐point precision.
- Likely to exceed VRAM constraints on a single GPU for very large models (32B+ parameters).
AWQ
- A relatively new post‐training quantization method that attempts to minimize accuracy loss by automatically figuring out optimal weight scales.
- Promising results in some benchmarks, but less widely used (and benchmarked) for 30B+ models than GPTQ.
Bitsandbytes (8-bit or 4-bit)
- The Bitsandbytes library from Tim Dettmers (and the Hugging Face ecosystem) has become a go‐to for quantizing large LLMs.
- A typical scenario is running a model in “8-bit” or “4-bit” to reduce GPU memory usage while retaining decent performance.
- “Bitsandbytes NF4” is a newer 4-bit quantization format (“NormalFloat4”) aimed at improving numerical stability over older 4-bit approaches.
EETQ
- Another quantization approach, but generally less commonly used or documented for big LLMs compared to GPTQ or Bitsandbytes.
- May be worth experimenting with if you already have a pipeline for it, but references/benchmarks for 30B+ models are not as widespread.
GPTQ
- Currently one of the most popular post‐training quantization methods for large LLMs (30B, 65B, etc.) because it can offer:
  - Significantly reduced VRAM usage (often 4-bit).
  - Minimal accuracy/perplexity degradation compared to many other 4-bit methods.
- Many community LLM repositories (e.g., Hugging Face Transformers) provide GPTQ‐quantized weights for big models.

Which One Is “Best” for a 32B Model?

If your priority is maximum compression with minimal accuracy loss:
GPTQ is widely recognized to provide excellent accuracy retention at 4-bit precision. For many open‐source LLMs (30B–70B), GPTQ quantization has become a “standard” for single‐GPU or multi‐GPU inference.
If you need a simpler toolchain or already rely on Bitsandbytes:
Bitsandbytes NF4 is also a strong contender and is quite seamless to enable in the Hugging Face ecosystem. You might see a slightly higher accuracy drop than GPTQ in some scenarios, but it’s well‐tested and easy to integrate.
If you want a fairly new approach and don’t mind experimentation:
AWQ can produce very good results too, but it’s somewhat newer and might require more experimentation to dial in the best configuration for your 32B model.

Overall, GPTQ or Bitsandbytes NF4 (4-bit) are the two most common quantization picks for a 30B+ LLM. If you’re looking for a straightforward recommendation for Deepseek 32B, GPTQ is likely the sweet spot—both in terms of memory savings and preserving model quality.

Yes, you can store the token securely so that it persists across sessions. One common way to do this in Python is to use the keyring library, which interacts with the operating system’s secure credential store (for example, Windows Credential Manager, macOS Keychain, or the Secret Service API on Linux). This approach is more secure than saving the token in a plain text file.

Below is an example of how you might modify your Tkinter app to use keyring for storing and retrieving the API token:

Install the keyring package (if you haven’t already):
Terminal window
```
pip install keyring
```

Integrate keyring in your code:
Use keyring’s functions to save and retrieve the token. For example, you could create two helper functions:

import keyring

SERVICE_NAME = "DeepseekDashboardApp"
TOKEN_KEY = "user_api_token"

def save_token(token: str):
    """Store the token securely in the OS credential store."""
    keyring.set_password(SERVICE_NAME, TOKEN_KEY, token)

def load_token() -> str:
    """Retrieve the token securely from the OS credential store."""
    token = keyring.get_password(SERVICE_NAME, TOKEN_KEY)
    return token if token is not None else ""

Load the token when your app starts:
When building your Tkinter UI, set the token entry field’s text to the stored token (if it exists).

For example, when you create your token entry widget, you can prepopulate it like this:
```
token_var = tk.StringVar(value=load_token())
token_entry = ttk.Entry(frame_top, textvariable=token_var, width=50, show="*")
token_entry.grid(row=0, column=1, padx=5, pady=5, sticky="ew")
```
Save the token when it changes or when the user logs out/in:
You might decide to save the token automatically when the token variable changes or when the user clicks a “Save” button. One easy method is to add a trace callback to the token variable. For instance, update the keyring whenever the token changes:
```
def on_token_change(*args):
    token = token_var.get().strip()
    if token:
        save_token(token)
        # Optionally, you can also start polling here.
        if not polling_started:
            polling_started = True
            poll_status()

token_var.trace("w", on_token_change)
```
In this example, every time the user modifies the token field, it is saved to the OS store.
Security Considerations:
- The keyring library leverages the OS-level security so your token isn’t stored in plaintext on disk by your application.
- Be aware that the security of the stored token is only as strong as the underlying operating system’s credential store.
- If you want to give your users additional control, you might offer a “forget token” option that clears the stored credential.

Putting It All Together

Below is a simplified excerpt integrating keyring into your Tkinter example:

import tkinter as tk
from tkinter import ttk
import keyring
from datetime import datetime
import threading
import requests
import json

# Constants for keyring
SERVICE_NAME = "DeepseekDashboardApp"
TOKEN_KEY = "user_api_token"

def save_token(token: str):
    keyring.set_password(SERVICE_NAME, TOKEN_KEY, token)

def load_token() -> str:
    token = keyring.get_password(SERVICE_NAME, TOKEN_KEY)
    return token if token is not None else ""

# Example API endpoint (replace with your actual endpoints)
STATUS_URL = "https://api.endpoints.huggingface.cloud/v2/endpoint/your-endpoint/replica"

# Global flag
polling_started = False

# -----------------------------
# API Helper Functions
# -----------------------------
def call_api(url, token, method="GET"):
    headers = {
        'accept': 'application/json',
        'authorization': f'Bearer {token}'
    }
    try:
        if method.upper() == "POST":
            response = requests.post(url, headers=headers)
        elif method.upper() == "GET":
            response = requests.get(url, headers=headers)
        return response.json() if response.text else {"status": response.status_code}
    except Exception as e:
        return {"error": str(e)}

def fetch_status(token, callback):
    result = call_api(STATUS_URL, token, "GET")
    # For demonstration, we just pass the whole result to the callback.
    callback(result)

def poll_status():
    token = token_var.get().strip()
    if token:
        threading.Thread(
            target=lambda: fetch_status(token, lambda resp: root.after(0, lambda: label_status.config(text=json.dumps(resp, indent=4)))),
            daemon=True
        ).start()
    root.after(30000, poll_status)

# -----------------------------
# Token change callback
# -----------------------------
def on_token_change(*args):
    global polling_started
    token = token_var.get().strip()
    if token:
        # Save token securely
        save_token(token)
        if not polling_started:
            polling_started = True
            poll_status()

# -----------------------------
# Building the UI
# -----------------------------
root = tk.Tk()
root.title("Deepseek Dashboard")

frame_top = ttk.Frame(root, padding="10")
frame_top.grid(row=0, column=0, sticky="ew")
ttk.Label(frame_top, text="API Token:").grid(row=0, column=0, padx=5, pady=5, sticky="e")

# Prepopulate the token from keyring
token_var = tk.StringVar(value=load_token())
token_entry = ttk.Entry(frame_top, textvariable=token_var, width=50, show="*")
token_entry.grid(row=0, column=1, padx=5, pady=5, sticky="ew")
frame_top.columnconfigure(1, weight=1)

# Set up token trace to save token and start polling when token is entered.
token_var.trace("w", on_token_change)

# A simple label to display the fetched status for demonstration
label_status = ttk.Label(root, text="Status will appear here", padding="10")
label_status.grid(row=1, column=0, sticky="ew")

root.mainloop()

Summary

Keyring is an effective and secure solution to store tokens (or any credentials) in a way similar to how web browsers store cookies.
With keyring, your token is stored securely in the OS’s credential store, and you can load it automatically when the app starts so that the user doesn’t have to re-enter it every time.

This method allows your Tkinter application to “remember” the token securely across sessions. Let me know if you need additional details or further enhancements!

Below is one approach: instead of saving the token as soon as it’s typed, you can add a “Validate & Save Token” button that calls the API with the given token. Only if the response shows that the token is accepted does the code save it (using keyring, for example) and then start polling. This prevents an invalid token from being stored.

Below is a modified version of the previous Tkinter app snippet that adds a validation button. (Make sure you have installed keyring, e.g., via pip install keyring.)

import tkinter as tk
from tkinter import ttk, messagebox
from datetime import datetime
import threading
import requests
import keyring
import json

# Constants for keyring
SERVICE_NAME = "DeepseekDashboardApp"
TOKEN_KEY = "user_api_token"

# API endpoints for your Hugging Face endpoint actions
RESUME_URL = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/resume"
PAUSE_URL  = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/pause"
STATUS_URL = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/replica"

# Global flags
polling_started = False
resume_start_time = None

def save_token(token: str):
    keyring.set_password(SERVICE_NAME, TOKEN_KEY, token)

def load_token() -> str:
    stored = keyring.get_password(SERVICE_NAME, TOKEN_KEY)
    return stored if stored is not None else ""

# -----------------------------
# API Helper Functions
# -----------------------------
def call_api(url, token, method="POST"):
    headers = {
        'accept': 'application/json',
        'authorization': f'Bearer {token}'
    }
    try:
        if method.upper() == "POST":
            response = requests.post(url, headers=headers)
        elif method.upper() == "GET":
            response = requests.get(url, headers=headers)
        return response.json() if response.text else {"status": response.status_code}
    except Exception as e:
        return {"error": str(e)}

def get_latest_replica(data):
    items = data.get('items', [])
    if not items:
        return None
    def parse_time(item):
        time_str = item.get("status", {}).get("startedAt", "")
        try:
            return datetime.fromisoformat(time_str.replace("Z", "+00:00"))
        except Exception:
            return datetime.min
    return max(items, key=parse_time)

# -----------------------------
# UI Update Functions
# -----------------------------
def update_status_ui(replica):
    global resume_start_time
    if replica is None:
        label_replica_id.config(text="N/A")
        label_version_id.config(text="N/A")
        label_started_at.config(text="N/A")
        label_stopped_at.config(text="N/A")
        label_stage.config(text="N/A")
        label_updating.config(text="N/A")
        label_time_to_running.config(text="N/A")
        update_led(False)
    else:
        label_replica_id.config(text=replica.get('id', ''))
        label_version_id.config(text=replica.get('versionId', ''))
        status = replica.get('status', {})
        label_started_at.config(text=status.get('startedAt', 'N/A'))
        label_stopped_at.config(text=status.get('stoppedAt', 'N/A'))
        label_stage.config(text=status.get('stage', 'N/A'))
        label_updating.config(text=str(status.get('updating', 'N/A')))
        update_led(status.get('live', False))
        if status.get("stage", "").lower() == "running" and resume_start_time is not None:
            elapsed = datetime.now() - resume_start_time
            elapsed_str = str(elapsed).split('.')[0]  # Remove microseconds
            label_time_to_running.config(text=elapsed_str)
            resume_start_time = None  # Reset timer after success
        else:
            label_time_to_running.config(text="Waiting...")

def update_led(is_live):
    canvas_led.delete("all")
    color = "green" if is_live else "red"
    canvas_led.create_oval(2, 2, 22, 22, fill=color, outline=color)

# -----------------------------
# Functions to Interact with the Endpoint
# -----------------------------
def resume_endpoint(token):
    global resume_start_time
    resume_start_time = datetime.now()
    def task():
        call_api(RESUME_URL, token, "POST")
        poll_status()  # Poll status immediately after resume.
    threading.Thread(target=task, daemon=True).start()

def pause_endpoint(token):
    def task():
        call_api(PAUSE_URL, token, "POST")
        poll_status()
    threading.Thread(target=task, daemon=True).start()

def fetch_status(token, callback):
    result = call_api(STATUS_URL, token, "GET")
    if isinstance(result, dict) and "items" in result:
        latest = get_latest_replica(result)
        callback(latest)
    else:
        callback(None)

def poll_status():
    token = token_var.get().strip()
    if token:
        threading.Thread(
            target=lambda: fetch_status(token, lambda replica: root.after(0, update_status_ui, replica)),
            daemon=True
        ).start()
    root.after(30000, poll_status)

# -----------------------------
# Token Validation
# -----------------------------
def validate_and_save_token():
    token = token_var.get().strip()
    if not token:
        messagebox.showwarning("Token Validation", "Please enter a token first.")
        return
    result = call_api(STATUS_URL, token, "GET")
    # You might get an error message if the token is invalid.
    if "error" in result:
        messagebox.showerror("Token Validation", f"Token not accepted: {result['error']}")
    elif isinstance(result, dict) and "items" in result:
        save_token(token)
        messagebox.showinfo("Token Validation", "Token accepted and saved.")
        global polling_started
        if not polling_started:
            polling_started = True
            poll_status()
    else:
        messagebox.showerror("Token Validation", "Token not accepted by the API.")

# -----------------------------
# Build the UI
# -----------------------------
root = tk.Tk()
root.title("Hugging Face Endpoint Control Dashboard")

root.columnconfigure(0, weight=1)
root.rowconfigure(2, weight=1)

# Top Frame: Token Entry and Validation
frame_top = ttk.Frame(root, padding="10")
frame_top.grid(row=0, column=0, sticky="ew")
ttk.Label(frame_top, text="API Token:").grid(row=0, column=0, padx=5, pady=5, sticky="e")
token_var = tk.StringVar(value=load_token())
token_entry = ttk.Entry(frame_top, textvariable=token_var, width=50, show="*")
token_entry.grid(row=0, column=1, padx=5, pady=5, sticky="ew")
frame_top.columnconfigure(1, weight=1)
# Add a button for validating and saving the token
validate_btn = ttk.Button(frame_top, text="Validate & Save Token", command=validate_and_save_token)
validate_btn.grid(row=0, column=2, padx=5, pady=5)

# Button Frame: Resume/Pause Controls
button_frame = ttk.Frame(root, padding="10")
button_frame.grid(row=1, column=0, sticky="ew")
resume_btn = ttk.Button(button_frame, text="Resume Endpoint", command=lambda: resume_endpoint(token_var.get()))
resume_btn.grid(row=0, column=0, padx=5)
pause_btn = ttk.Button(button_frame, text="Pause Endpoint", command=lambda: pause_endpoint(token_var.get()))
pause_btn.grid(row=0, column=1, padx=5)

# Status Panel: Replica Properties
status_frame = ttk.LabelFrame(root, text="Replica Status", padding="10")
status_frame.grid(row=2, column=0, padx=10, pady=10, sticky="nsew")
status_frame.columnconfigure(1, weight=1)

ttk.Label(status_frame, text="Live:").grid(row=0, column=0, sticky="w", padx=5, pady=5)
canvas_led = tk.Canvas(status_frame, width=24, height=24)
canvas_led.grid(row=0, column=1, sticky="w", padx=5, pady=5)

ttk.Label(status_frame, text="Replica ID:").grid(row=1, column=0, sticky="e", padx=5, pady=5)
label_replica_id = ttk.Label(status_frame, text="N/A")
label_replica_id.grid(row=1, column=1, sticky="w", padx=5, pady=5)

ttk.Label(status_frame, text="Version ID:").grid(row=2, column=0, sticky="e", padx=5, pady=5)
label_version_id = ttk.Label(status_frame, text="N/A")
label_version_id.grid(row=2, column=1, sticky="w", padx=5, pady=5)

ttk.Label(status_frame, text="Started At:").grid(row=3, column=0, sticky="e", padx=5, pady=5)
label_started_at = ttk.Label(status_frame, text="N/A")
label_started_at.grid(row=3, column=1, sticky="w", padx=5, pady=5)

ttk.Label(status_frame, text="Stopped At:").grid(row=4, column=0, sticky="e", padx=5, pady=5)
label_stopped_at = ttk.Label(status_frame, text="N/A")
label_stopped_at.grid(row=4, column=1, sticky="w", padx=5, pady=5)

ttk.Label(status_frame, text="Stage:").grid(row=5, column=0, sticky="e", padx=5, pady=5)
label_stage = ttk.Label(status_frame, text="N/A")
label_stage.grid(row=5, column=1, sticky="w", padx=5, pady=5)

ttk.Label(status_frame, text="Updating:").grid(row=6, column=0, sticky="e", padx=5, pady=5)
label_updating = ttk.Label(status_frame, text="N/A")
label_updating.grid(row=6, column=1, sticky="w", padx=5, pady=5)

ttk.Label(status_frame, text="Time to Running:").grid(row=7, column=0, sticky="e", padx=5, pady=5)
label_time_to_running = ttk.Label(status_frame, text="N/A")
label_time_to_running.grid(row=7, column=1, sticky="w", padx=5, pady=5)

root.mainloop()

How This Works

Token Validation:
When the user clicks the “Validate & Save Token” button, the function validate_and_save_token() runs. It makes an API call (here using the status endpoint) with the current token value.
- If the call returns an error (e.g. unauthorized), an error message is shown and the token is not saved.
- If the call is successful (i.e. the token is accepted), the token is saved securely using keyring and polling starts.
Polling & UI Updates:
Once validated and saved, the app begins polling the API every 30 seconds. The replica status is updated in the UI and, if applicable, the elapsed time until the replica reached "running" is calculated and shown.

This design ensures that you only store a token if it’s actually valid with the API—providing an extra layer of security (and avoiding storing bogus tokens).

Below is an updated version of the Tkinter dashboard code. In this version, two clickable URLs have been added at the very top of the window. When clicked, they’ll open in your default browser.

Key changes include:

Importing Python’s built‑in webbrowser module.
Creating a “links frame” at the top with two labels styled like hyperlinks.
Binding a mouse click event (Button‑1) to each label so that it calls webbrowser.open_new(url).

You can combine this with the previous token-validation, polling, and status UI code.

import tkinter as tk
from tkinter import ttk, messagebox
from datetime import datetime
import threading
import requests
import keyring
import webbrowser
import json

# Constants for keyring
SERVICE_NAME = "DeepseekDashboardApp"
TOKEN_KEY = "user_api_token"

# API endpoints for your Hugging Face endpoint actions
RESUME_URL = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/resume"
PAUSE_URL  = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/pause"
STATUS_URL = "https://api.endpoints.huggingface.cloud/v2/endpoint/george-adamon/deepseek-r1-distill-qwen-32b-isu/replica"

# Global flags
polling_started = False
resume_start_time = None

def save_token(token: str):
    keyring.set_password(SERVICE_NAME, TOKEN_KEY, token)

def load_token() -> str:
    stored = keyring.get_password(SERVICE_NAME, TOKEN_KEY)
    return stored if stored is not None else ""

# Open URL in browser
def open_url(url):
    webbrowser.open_new(url)

# -----------------------------
# API Helper Functions
# -----------------------------
def call_api(url, token, method="POST"):
    headers = {
        'accept': 'application/json',
        'authorization': f'Bearer {token}'
    }
    try:
        if method.upper() == "POST":
            response = requests.post(url, headers=headers)
        elif method.upper() == "GET":
            response = requests.get(url, headers=headers)
        return response.json() if response.text else {"status": response.status_code}
    except Exception as e:
        return {"error": str(e)}

def get_latest_replica(data):
    items = data.get('items', [])
    if not items:
        return None
    def parse_time(item):
        time_str = item.get("status", {}).get("startedAt", "")
        try:
            return datetime.fromisoformat(time_str.replace("Z", "+00:00"))
        except Exception:
            return datetime.min
    return max(items, key=parse_time)

# -----------------------------
# UI Update Functions
# -----------------------------
def update_status_ui(replica):
    global resume_start_time
    if replica is None:
        label_replica_id.config(text="N/A")
        label_version_id.config(text="N/A")
        label_started_at.config(text="N/A")
        label_stopped_at.config(text="N/A")
        label_stage.config(text="N/A")
        label_updating.config(text="N/A")
        label_time_to_running.config(text="N/A")
        update_led(False)
    else:
        label_replica_id.config(text=replica.get('id', ''))
        label_version_id.config(text=replica.get('versionId', ''))
        status = replica.get('status', {})
        label_started_at.config(text=status.get('startedAt', 'N/A'))
        label_stopped_at.config(text=status.get('stoppedAt', 'N/A'))
        label_stage.config(text=status.get('stage', 'N/A'))
        label_updating.config(text=str(status.get('updating', 'N/A')))
        update_led(status.get('live', False))
        # Calculate elapsed time if stage becomes "running".
        if status.get("stage", "").lower() == "running" and resume_start_time is not None:
            elapsed = datetime.now() - resume_start_time
            elapsed_str = str(elapsed).split('.')[0]  # Format H:MM:SS (remove microseconds)
            label_time_to_running.config(text=elapsed_str)
            resume_start_time = None  # Reset timer after success.
        else:
            label_time_to_running.config(text="Waiting...")

def update_led(is_live):
    canvas_led.delete("all")
    color = "green" if is_live else "red"
    canvas_led.create_oval(2, 2, 22, 22, fill=color, outline=color)

# -----------------------------
# Functions to Interact with the Endpoint
# -----------------------------
def resume_endpoint(token):
    global resume_start_time
    resume_start_time = datetime.now()
    def task():
        call_api(RESUME_URL, token, "POST")
        poll_status()  # Immediately update status after resume.
    threading.Thread(target=task, daemon=True).start()

def pause_endpoint(token):
    def task():
        call_api(PAUSE_URL, token, "POST")
        poll_status()  # Immediately update status after pause.
    threading.Thread(target=task, daemon=True).start()

def fetch_status(token, callback):
    result = call_api(STATUS_URL, token, "GET")
    if isinstance(result, dict) and "items" in result:
        latest = get_latest_replica(result)
        callback(latest)
    else:
        callback(None)

def poll_status():
    token = token_var.get().strip()
    if token:
        threading.Thread(
            target=lambda: fetch_status(token, lambda replica: root.after(0, update_status_ui, replica)),
            daemon=True
        ).start()
    root.after(30000, poll_status)

# -----------------------------
# Token Validation
# -----------------------------
def validate_and_save_token():
    token = token_var.get().strip()
    if not token:
        messagebox.showwarning("Token Validation", "Please enter a token first.")
        return
    result = call_api(STATUS_URL, token, "GET")
    if "error" in result:
        messagebox.showerror("Token Validation", f"Token not accepted: {result['error']}")
    elif isinstance(result, dict) and "items" in result:
        save_token(token)
        messagebox.showinfo("Token Validation", "Token accepted and saved.")
        global polling_started
        if not polling_started:
            polling_started = True
            poll_status()
    else:
        messagebox.showerror("Token Validation", "Token not accepted by the API.")

# -----------------------------
# Build the UI
# -----------------------------
root = tk.Tk()
root.title("Hugging Face Endpoint Control Dashboard")

# Configure grid so the main frame expands.
root.columnconfigure(0, weight=1)
root.rowconfigure(3, weight=1)

# -------------- Links Frame --------------
# This frame contains clickable URLs for easy access.
frame_links = ttk.Frame(root, padding="10")
frame_links.grid(row=0, column=0, sticky="ew")
# Inference Endpoint link
link_label1 = ttk.Label(frame_links, text="Visit Inference Endpoint", foreground="blue", cursor="hand2")
link_label1.grid(row=0, column=0, padx=5)
link_label1.bind("<Button-1>", lambda e: open_url("https://endpoints.huggingface.co/george-adamon/endpoints/deepseek-r1-distill-qwen-32b-isu"))
# Interactive LLM link
link_label2 = ttk.Label(frame_links, text="Visit Interactive LLM", foreground="blue", cursor="hand2")
link_label2.grid(row=0, column=1, padx=5)
link_label2.bind("<Button-1>", lambda e: open_url("https://interactivellm.tree-correspondents.com/"))

# -------------- Token & Validation Frame --------------
frame_top = ttk.Frame(root, padding="10")
frame_top.grid(row=1, column=0, sticky="ew")
ttk.Label(frame_top, text="API Token:").grid(row=0, column=0, padx=5, pady=5, sticky="e")
token_var = tk.StringVar(value=load_token())
token_entry = ttk.Entry(frame_top, textvariable=token_var, width=50, show="*")
token_entry.grid(row=0, column=1, padx=5, pady=5, sticky="ew")
frame_top.columnconfigure(1, weight=1)
validate_btn = ttk.Button(frame_top, text="Validate & Save Token", command=validate_and_save_token)
validate_btn.grid(row=0, column=2, padx=5, pady=5)

# -------------- Controls Frame --------------
button_frame = ttk.Frame(root, padding="10")
button_frame.grid(row=2, column=0, sticky="ew")
resume_btn = ttk.Button(button_frame, text="Resume Endpoint", command=lambda: resume_endpoint(token_var.get()))
resume_btn.grid(row=0, column=0, padx=5)
pause_btn = ttk.Button(button_frame, text="Pause Endpoint", command=lambda: pause_endpoint(token_var.get()))
pause_btn.grid(row=0, column=1, padx=5)

# -------------- Status Panel --------------
status_frame = ttk.LabelFrame(root, text="Replica Status", padding="10")
status_frame.grid(row=3, column=0, padx=10, pady=10, sticky="nsew")
status_frame.columnconfigure(1, weight=1)

ttk.Label(status_frame, text="Live:").grid(row=0, column=0, sticky="w", padx=5, pady=5)
canvas_led = tk.Canvas(status_frame, width=24, height=24)
canvas_led.grid(row=0, column=1, sticky="w", padx=5, pady=5)

ttk.Label(status_frame, text="Replica ID:").grid(row=1, column=0, sticky="e", padx=5, pady=5)
label_replica_id = ttk.Label(status_frame, text="N/A")
label_replica_id.grid(row=1, column=1, sticky="w", padx=5, pady=5)

ttk.Label(status_frame, text="Version ID:").grid(row=2, column=0, sticky="e", padx=5, pady=5)
label_version_id = ttk.Label(status_frame, text="N/A")
label_version_id.grid(row=2, column=1, sticky="w", padx=5, pady=5)

ttk.Label(status_frame, text="Started At:").grid(row=3, column=0, sticky="e", padx=5, pady=5)
label_started_at = ttk.Label(status_frame, text="N/A")
label_started_at.grid(row=3, column=1, sticky="w", padx=5, pady=5)

ttk.Label(status_frame, text="Stopped At:").grid(row=4, column=0, sticky="e", padx=5, pady=5)
label_stopped_at = ttk.Label(status_frame, text="N/A")
label_stopped_at.grid(row=4, column=1, sticky="w", padx=5, pady=5)

ttk.Label(status_frame, text="Stage:").grid(row=5, column=0, sticky="e", padx=5, pady=5)
label_stage = ttk.Label(status_frame, text="N/A")
label_stage.grid(row=5, column=1, sticky="w", padx=5, pady=5)

ttk.Label(status_frame, text="Updating:").grid(row=6, column=0, sticky="e", padx=5, pady=5)
label_updating = ttk.Label(status_frame, text="N/A")
label_updating.grid(row=6, column=1, sticky="w", padx=5, pady=5)

ttk.Label(status_frame, text="Time to Running:").grid(row=7, column=0, sticky="e", padx=5, pady=5)
label_time_to_running = ttk.Label(status_frame, text="N/A")
label_time_to_running.grid(row=7, column=1, sticky="w", padx=5, pady=5)

root.mainloop()

Explanation

Clickable URLs:
Two labels (“Visit Inference Endpoint” and “Visit Interactive LLM”) are added at the very top in a frame_links. They are styled with blue text and a hand cursor to mimic hyperlinks. Clicking on each label invokes the open_url(url) function.
Remaining Code:
The rest of the code remains essentially the same (token validation, polling, and status UI). Once a valid token is entered and validated, the application will start polling the API every 30 seconds.

This approach gives you a simple way to include external links at the top of your application while preserving all the functionality you already have. Let me know if you have further questions or modifications!