import gradio as gr from urllib.parse import urlparse import subprocess import threading import sys LOG_FILE = "output.log" def read_logs(): sys.stdout.flush() with open(LOG_FILE, "r") as f: return f.read() previous_url = "" ml_worker = None def run_ml_worker(url, api_key, hf_token): global ml_worker, previous_url previous_url = url ml_worker = subprocess.Popen( [ "giskard", "worker", "start", "-u", f"{url}", "-k", f"{api_key}", "-t", f"{hf_token}" ], stdout=open(LOG_FILE, "w"), stderr=subprocess.STDOUT ) args = ml_worker.args print(f"Process {args} exited with {ml_worker.wait()}") def stop_ml_worker(): global ml_worker, previous_url if ml_worker is not None: print(f"Stopping ML worker for {previous_url}") ml_worker.terminate() print("ML worker stopped") return "ML worker stopped" return "ML worker not started" def start_ml_worker(url, api_key, hf_token): # Always run an external ML worker stop_ml_worker() print(f"Starting ML worker for {url}") thread = threading.Thread(target=run_ml_worker, args=(url, api_key, hf_token)) thread.start() return f"ML worker running for {url}" with gr.Blocks() as iface: with gr.Row(): with gr.Column(): url_input = gr.Textbox(label="Giskard Hub URL") api_key_input = gr.Textbox(label="Giskard Hub API Key", placeholder="gsk-xxxxxxxxxxxxxxxxxxxxxxxxxxxx") hf_token_input = gr.Textbox(label="Hugging Face Spaces Token") output = gr.Textbox(label="Status") with gr.Row(): run_btn = gr.Button("Run") run_btn.click(start_ml_worker, [url_input, api_key_input, hf_token_input], output) stop_btn = gr.Button("Stop") stop_btn.click(stop_ml_worker, None, output) logs = gr.Textbox(label="Giskard ML worker log:") iface.load(read_logs, None, logs, every=0.5) iface.queue() iface.launch()