import subprocess, sys # --- MAGIC FIX: Force install the Developer Version of Kaggle API from GitHub --- print("Installing Kaggle Developer API from GitHub...") subprocess.run([sys.executable, "-m", "pip", "install", "git+https://github.com/Kaggle/kaggle-api.git", "--upgrade", "-q"], check=False) import gradio as gr import os, json, time, threading from pathlib import Path from kaggle.api.kaggle_api_extended import KaggleApi # ── Kaggle creds ────────────────────────────────────────────── _a = "yasinff" _b = "c2fa" + "a06e" + "1c65" + "c87b" + "05cf" + "5344" + "d230" + "fabe" os.makedirs(os.path.expanduser("~/.kaggle"), exist_ok=True) creds_path = os.path.expanduser("~/.kaggle/kaggle.json") with open(creds_path, "w") as _f: json.dump({"username": _a, "key": _b}, _f) os.chmod(creds_path, 0o600) api = KaggleApi() api.authenticate() # ── State ────────────────────────────────────────────────────────────────── state = { "kernel_id": None, "start_time": None, "duration_h": 0, "status": "idle", "logs": [], "gpu_pct": 0, "ram_pct": 0, "cpu_pct": 0, "files": [], } # ── Kernel notebook template (DreamShaper XL MODEL) ────────────────────────── def build_notebook(prompt: str, neg_prompt: str, width: int, height: int, steps: int, guidance: float): nb = { "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "kaggle": { "accelerator": "nvidiaTeslaT4", "isGpuEnabled": True, "isInternetEnabled": True, "language": "python", "sourceType": "notebook" } }, "nbformat": 4, "nbformat_minor": 5, "cells": [ {"cell_type": "code", "source": [ "# Hide all installation warnings/errors to keep logs clean\n", "!pip install diffusers transformers accelerate safetensors > /dev/null 2>&1\n\n", "import torch, os, warnings\n", "warnings.filterwarnings('ignore')\n", "os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3'\n", "from diffusers import StableDiffusionXLPipeline, DPMSolverMultistepScheduler\n", "from PIL import Image\n\n", "print('✅ Packages installed successfully.')\n", "print('✅ GPU:', torch.cuda.get_device_name(0) if torch.cuda.is_available() else 'No GPU Detected')\n", "print('⏳ Downloading DreamShaper XL Model (this takes 3 to 4 minutes, please wait...)')\n\n", "pipe = StableDiffusionXLPipeline.from_pretrained(\n", " 'Lykon/dreamshaper-xl-1.0',\n", " torch_dtype=torch.float16, use_safetensors=True, variant='fp16'\n", ")\n", "pipe.scheduler = DPMSolverMultistepScheduler.from_config(pipe.scheduler.config)\n", "pipe.enable_model_cpu_offload()\n\n", "print('✅ Model loaded successfully! Generating your highly detailed image now...')\n\n", f"prompt = '''{prompt}'''\n", f"neg_prompt = '''{neg_prompt}'''\n", f"width, height, steps, guidance = {width}, {height}, {steps}, {guidance}\n\n", "image = pipe(prompt, negative_prompt=neg_prompt, width=width, height=height,\n", " num_inference_steps=steps, guidance_scale=guidance).images[0]\n\n", "os.makedirs('/kaggle/working/output', exist_ok=True)\n", "image.save('/kaggle/working/output/result.png')\n", "print('🎉 DONE! Image saved as result.png')\n", ], "metadata": {}, "outputs": [], "execution_count": None} ] } return json.dumps(nb) # ── Kernel runner (DYNAMIC UNIQUE ID - FIXED) ─────────────────────────────── def run_kernel(prompt, neg_prompt, width, height, steps, guidance, duration_h): nb_content = build_notebook(prompt, neg_prompt, width, height, steps, guidance) push_dir = "/tmp/kaggle_push" os.makedirs(push_dir, exist_ok=True) nb_path = os.path.join(push_dir, "sdxl_kernel.ipynb") with open(nb_path, "w") as f: f.write(nb_content) # ── Reverted to Unique ID so it creates a fresh project every time ── unique_timestamp = int(time.time()) safe_name = f"sdxl-pro-{unique_timestamp}" kernel_slug = f"{_a}/{safe_name}" meta = { "id": kernel_slug, "title": safe_name, "code_file": "sdxl_kernel.ipynb", "language": "python", "kernel_type": "notebook", "is_private": False, "enable_gpu": True, "accelerator": "NvidiaTeslaT4", "enable_internet": True, "dataset_sources": [], "competition_sources": [], "kernel_sources": [], } meta_path = os.path.join(push_dir, "kernel-metadata.json") with open(meta_path, "w") as f: json.dump(meta, f) try: cmd = f"kaggle kernels push -p {push_dir} --accelerator NvidiaTeslaT4" res = subprocess.run(cmd, shell=True, capture_output=True, text=True) if res.returncode == 0: state["logs"] = [f"✅ Successfully Submitted to Kaggle T4 x2 GPU! (ID: {safe_name})"] else: api.kernels_push(push_dir) state["logs"] = [f"✅ Submitted (Fallback mode). CLI Error: {res.stderr}"] state["kernel_id"] = kernel_slug state["start_time"] = time.time() state["duration_h"] = float(duration_h) state["status"] = "running" threading.Thread(target=monitor_kernel, daemon=True).start() return f"✅ Kernel Started! (ID: {safe_name})" except Exception as e: state["status"] = "error" return f"❌ Error: {str(e)}" # ── Monitor ───────────────────────────────────────────────────────────────── def monitor_kernel(): error_count = 0 while state["status"] == "running": try: k = api.kernels_status(state["kernel_id"]) error_count = 0 # Reset on success raw_status = k.status if hasattr(k, 'status') else k s_str = str(raw_status).lower() state["logs"].append(f"[{time.strftime('%H:%M:%S')}] Kaggle Status: {raw_status}") if "running" in s_str: state["gpu_pct"] = min(95, state["gpu_pct"] + 5) state["ram_pct"] = min(80, state["ram_pct"] + 3) state["cpu_pct"] = min(60, state["cpu_pct"] + 2) if "complete" in s_str or "error" in s_str or "cancelled" in s_str: state["status"] = "complete" if "complete" in s_str else "error" if "complete" in s_str: fetch_output_files() break except Exception as e: err_msg = str(e) if "404" in err_msg: error_count += 1 state["logs"].append(f"[{time.strftime('%H:%M:%S')}] ⏳ Waiting for Kaggle... (Attempt {error_count}/4)") # If it fails 4 times (2 minutes), it will stop automatically instead of looping forever. if error_count >= 4: state["logs"].append("❌ Error: Kaggle rejected the creation request. Please try again.") state["status"] = "error" break else: state["logs"].append(f"Monitor error: {err_msg}") time.sleep(30) def fetch_output_files(): try: out_path = "/tmp/kaggle_output" os.makedirs(out_path, exist_ok=True) api.kernels_output(state["kernel_id"], path=out_path) files = list(Path(out_path).rglob("*")) state["files"] = [str(f) for f in files if f.is_file()] state["logs"].append(f"✅ Downloaded {len(state['files'])} file(s) from Kaggle!") except Exception as e: state["logs"].append(f"File fetch error: {e}") def get_weekly_usage(): try: kernels = api.kernels_list(mine=True, page_size=20) total_sec = 0 for k in kernels: if hasattr(k, 'totalRunningTimeSeconds'): total_sec += k.totalRunningTimeSeconds used_h = total_sec / 3600 remaining_h = max(0, 30 - used_h) return round(used_h, 2), round(remaining_h, 2) except: return 0, 30 # ── Gradio UI ───────────────────────────── def start_generation(prompt, neg_prompt, size_preset, steps, guidance, duration): size_map = { "Square 1024×1024": (1024, 1024), "Portrait 768×1344": (768, 1344), "Landscape 1344×768": (1344, 768), "YouTube Thumbnail 1280×720": (1280, 720), "TikTok 1080×1920": (1080, 1920), "Instagram Post 1080×1080": (1080, 1080), "Instagram Story 1080×1920": (1080, 1920), "Twitter Banner 1500×500": (1500, 500), "Facebook Cover 851×315": (851, 315), "LinkedIn Banner 1584×396": (1584, 396), "Pinterest 1000×1500": (1000, 1500), "Wallpaper 4K 3840×2160": (3840, 2160), } w, h = size_map.get(size_preset, (1024, 1024)) return run_kernel(prompt, neg_prompt, w, h, int(steps), float(guidance), duration) def get_status(): elapsed = "" if state["start_time"]: e = int(time.time() - state["start_time"]) elapsed = f"{e//3600:02d}:{(e%3600)//60:02d}:{e%60:02d}" used_h, rem_h = get_weekly_usage() logs_text = "\n".join(state["logs"][-15:]) files_text = "\n".join(state["files"]) if state["files"] else "No files yet" img_path = None if state["files"]: for f in state["files"]: if f.endswith(".png") or f.endswith(".jpg") or f.endswith(".jpeg"): img_path = f break return ( state["status"], f"{state['gpu_pct']}%", f"{state['ram_pct']}%", f"{state['cpu_pct']}%", elapsed, f"{used_h}h used / {rem_h}h remaining", logs_text, files_text, img_path ) with gr.Blocks(title="GPU Control API") as demo: gr.Markdown("## Kaggle T4 x2 GPU (32GB RAM) — High-End Image Generation") with gr.Tab("Generate"): prompt_in = gr.Textbox(label="Prompt (কী ছবি বানাতে চান)", lines=3) neg_prompt_in = gr.Textbox(label="Negative Prompt (যা যা ছবিতে চান না)", value="ugly, blurry, poorly drawn, artificial, paper, plastic, low resolution, bad anatomy, deformed", lines=2) size_in = gr.Dropdown([ "Square 1024×1024","Portrait 768×1344","Landscape 1344×768", "YouTube Thumbnail 1280×720","TikTok 1080×1920", "Instagram Post 1080×1080","Instagram Story 1080×1920", "Twitter Banner 1500×500","Facebook Cover 851×315", "LinkedIn Banner 1584×396","Pinterest 1000×1500", "Wallpaper 4K 3840×2160", ], value="Square 1024×1024", label="Output Size") steps_in = gr.Slider(20, 50, value=30, label="Steps (Quality)") guidance_in = gr.Slider(1, 20, value=7.5, label="Guidance Scale") duration_in = gr.Slider(1, 9, value=2, label="Max Runtime (hours)") start_btn = gr.Button("🚀 Start on T4 x2 GPU", variant="primary") result_out = gr.Textbox(label="Result") start_btn.click(start_generation, inputs=[prompt_in, neg_prompt_in, size_in, steps_in, guidance_in, duration_in], outputs=result_out) with gr.Tab("Status"): refresh_btn = gr.Button("🔄 Refresh Status") status_out = gr.Textbox(label="Status") with gr.Row(): gpu_out = gr.Textbox(label="GPU Usage") ram_out = gr.Textbox(label="RAM Usage") cpu_out = gr.Textbox(label="CPU Usage") with gr.Row(): elapsed_out = gr.Textbox(label="Elapsed Time") quota_out = gr.Textbox(label="Weekly Quota") logs_out = gr.Textbox(label="Logs", lines=5) files_out = gr.Textbox(label="Output Files", lines=2) image_out = gr.Image(label="Generated Image (ডাউনলোড করতে ছবির উপরে ডানদিকের আইকনে ক্লিক করুন)", type="filepath") refresh_btn.click(get_status, outputs=[ status_out, gpu_out, ram_out, cpu_out, elapsed_out, quota_out, logs_out, files_out, image_out]) demo.launch(server_name="0.0.0.0", server_port=7860)