Create app.py

app.py

@@ -0,0 +1,121 @@
+import os
+import torch
+from diffusers import ZImagePipeline
+import gradio as gr
+import threading
+import queue
+import psutil
+
+# =================== CPU GOD MODE SETTINGS ===================
+torch.set_num_threads(torch.get_num_threads())
+torch.inference_mode()
+
+MODEL_ID = "Tongyi-MAI/Z-Image-Turbo"
+
+pipe = ZImagePipeline.from_pretrained(
+    MODEL_ID,
+    torch_dtype=torch.bfloat16,
+    low_cpu_mem_usage=True
+)
+pipe.to("cpu")
+pipe.enable_attention_slicing()
+pipe.enable_model_cpu_offload()
+
+try:
+    pipe.transformer.compile(fullgraph=True, dynamic=True)
+except Exception:
+    pass
+
+try:
+    pipe.enable_attention_slicing(slice_size="auto")
+except Exception:
+    pass
+
+MAX_THREADS = min(torch.get_num_threads(), os.cpu_count() or 4)
+
+# =================== QUEUE & WORKERS ===================
+job_queue = queue.Queue()
+status_dict = {}
+
+def worker(worker_id):
+    while True:
+        job = job_queue.get()
+        if job is None:
+            break
+        job_id, prompt, width, height, steps, seed, batch, out_folder = job
+        status_dict[job_id] = f"Worker {worker_id}: Processing..."
+        for i in range(batch):
+            img_seed = seed + i
+            image = pipe(
+                prompt=prompt,
+                height=height,
+                width=width,
+                num_inference_steps=steps,
+                guidance_scale=0.0,
+                generator=torch.Generator("cpu").manual_seed(img_seed),
+            ).images[0]
+            out_path = os.path.join(out_folder, f"{job_id}_{i}.png")
+            image.save(out_path)
+        status_dict[job_id] = f"Worker {worker_id}: Done ({batch} images)"
+        job_queue.task_done()
+
+workers = []
+for i in range(MAX_THREADS):
+    t = threading.Thread(target=worker, args=(i+1,), daemon=True)
+    t.start()
+    workers.append(t)
+
+# =================== JOB MANAGEMENT ===================
+job_counter = 0
+OUTPUT_DIR = "outputs"
+os.makedirs(OUTPUT_DIR, exist_ok=True)
+
+def enqueue_job(prompt, width, height, steps, seed, batch):
+    global job_counter
+    job_counter += 1
+    job_id = f"job_{job_counter}"
+    job_queue.put((job_id, prompt, width, height, steps, seed, batch, OUTPUT_DIR))
+    status_dict[job_id] = "Queued"
+    return job_id
+
+# =================== GRADIO INTERFACE ===================
+with gr.Blocks() as demo:
+    gr.Markdown("# ⚡ CPU God Mode Z-Image + Gradio Ultimate")
+
+    with gr.Row():
+        prompt_input = gr.Textbox(label="Prompt", placeholder="Type prompt here...")
+        seed_input = gr.Number(label="Seed", value=42, precision=0)
+
+    with gr.Row():
+        width_input = gr.Dropdown(["256","512","768","1024"], value="512", label="Width")
+        height_input = gr.Dropdown(["256","512","768","1024"], value="512", label="Height")
+        batch_input = gr.Slider(1, 5, value=1, step=1, label="Batch Size")
+
+    steps_input = gr.Slider(1, 25, value=9, step=1, label="Inference Steps")
+    output_gallery = gr.Gallery(label="Generated Images").style(grid=[2], height="auto")
+    status_box = gr.Textbox(label="Queue Status", interactive=False)
+
+    generate_btn = gr.Button("Generate")
+
+    def on_generate(prompt, width, height, steps, seed, batch):
+        width = int(width)
+        height = int(height)
+        seed = int(seed)
+        batch = int(batch)
+        job_id = enqueue_job(prompt, width, height, steps, seed, batch)
+        return [], f"Job {job_id} queued ({batch} images)"
+
+    def poll_status():
+        if status_dict:
+            return "\n".join([f"{k}: {v}" for k,v in status_dict.items()])
+        return "No jobs in queue."
+
+    generate_btn.click(
+        on_generate,
+        inputs=[prompt_input, width_input, height_input, steps_input, seed_input, batch_input],
+        outputs=[output_gallery, status_box]
+    )
+
+    status_box.change(fn=lambda: poll_status(), inputs=[], outputs=status_box)
+
+demo.launch(server_name="0.0.0.0", server_port=int(os.getenv("PORT", 7860)))