import gradio as gr import json import os import time import threading from gradio_client import Client # Configuration for HF Persistent Storage STORAGE_DIR = "/data" STORAGE_PATH = os.path.join(STORAGE_DIR, "evolution_state.json") if os.path.exists(STORAGE_DIR) else "evolution_state.json" class ArchitectState: def __init__(self): self.code = """

SYSTEM READY

Waiting for evolution loop to start...

""" self.gen = 0 self.is_running = False self.current_stream = "" self.logs = [] self.load() def save(self): try: data = {"code": self.code, "gen": self.gen, "is_running": self.is_running} with open(STORAGE_PATH, "w") as f: json.dump(data, f) except Exception as e: print(f"Save error: {e}") def load(self): if os.path.exists(STORAGE_PATH): try: with open(STORAGE_PATH, "r") as f: data = json.load(f) self.code = data.get("code", self.code) self.gen = data.get("gen", self.gen) self.is_running = data.get("is_running", False) except Exception as e: print(f"Load error: {e}") def add_log(self, text): timestamp = time.strftime("%H:%M:%S") self.logs.append(f"[{timestamp}] {text}") if len(self.logs) > 15: self.logs.pop(0) state = ArchitectState() SPACE_ID = "Sachin5112/Bihgfgh" def extract_html(text): start = text.lower().find("") if start != -1 and end != -1: return text[start:end+7] return text def evolution_worker(): """Background thread using the corrected Gradio iterator for live streaming.""" while True: if state.is_running: try: state.add_log(f"Gen {state.gen + 1}: Connecting...") client = Client(SPACE_ID) prompt = ( f"Task: Improve the 3D Minecraft Mobile Clone using Three.js. " f"Current Generation: {state.gen}. " f"STRICT: Output ONLY the full source code. No conversational text. " f"Current Code: {state.code[:1000]}..." ) # Use submit to get a job job = client.submit(message=prompt, api_name="/chat") state.add_log("Stream started...") # The correct way to stream in recent Gradio versions is iterating over the job for output in job: if not state.active_check(): break # Gradio chat usually returns a list or tuple; we want the string part if isinstance(output, (list, tuple)): chunk = output[0] else: chunk = output state.current_stream = chunk # Final processing final_raw = state.current_stream processed_code = extract_html(final_raw) if len(processed_code) > 200: state.code = processed_code state.gen += 1 state.current_stream = "" state.add_log(f"Success! Gen {state.gen} saved to bucket.") state.save() else: state.add_log("Warning: Invalid code received. Retrying...") except Exception as e: state.add_log(f"Stream Error: {str(e)}") time.sleep(12) # Cooldown else: time.sleep(2) state.active_check = lambda: state.is_running # Start worker worker = threading.Thread(target=evolution_worker, daemon=True) worker.start() # --- UI --- with gr.Blocks(theme=gr.themes.Monochrome(), css=".log-box textarea { font-family: monospace; font-size: 11px; }") as demo: gr.Markdown("# 🧱 ARCHITECT V17 - FIXED STREAMER") with gr.Row(): with gr.Column(scale=3): preview = gr.HTML(value=state.code) with gr.Column(scale=1): status = gr.Markdown(f"### STATUS: {'RUNNING' if state.is_running else 'STOPPED'}\n**Generation:** {state.gen}") start_btn = gr.Button("🚀 START", variant="primary") stop_btn = gr.Button("🛑 STOP") logs = gr.Textbox(label="Logs", value="\n".join(state.logs), lines=8, interactive=False, elem_classes="log-box") with gr.Tabs(): with gr.Tab("Live Stream"): live_code = gr.Code(label="Streaming...", value=state.current_stream, language="html", interactive=False) with gr.Tab("Last Stable"): full_code = gr.Code(label="Current Saved Code", value=state.code, language="html", interactive=False) def toggle_start(): state.is_running = True state.save() return "### STATUS: RUNNING", "\n".join(state.logs) def toggle_stop(): state.is_running = False state.save() return "### STATUS: STOPPED", "\n".join(state.logs) def refresh_ui(): return ( state.code, f"### STATUS: {'RUNNING' if state.is_running else 'STOPPED'}\n**Generation:** {state.gen}", "\n".join(state.logs), state.current_stream, state.code ) start_btn.click(toggle_start, None, [status, logs]) stop_btn.click(toggle_stop, None, [status, logs]) timer = gr.Timer(2) timer.tick(refresh_ui, None, [preview, status, logs, live_code, full_code]) demo.launch()