#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ ZERO COST PROJECT v1.4.0 Z-Image-Turbo | RTX 6000 Blackwell | ZeroGPU Prototype Pribadi CHANGELOG: v1.0.0 - Initial working version v1.1.0 - gr.Interface -> gr.Blocks (bypass Node proxy) v1.2.0 - Token pre-auth before spaces import v1.2.1 - inputs=None for API endpoints v1.3.0 - ROOT CAUSE FIX: Authorization header untuk ZeroGPU quota v1.4.0 - FEATURE UPDATE: + Style presets (10 styles) & Negative presets (5) + Generation stats tracking (total, avg, per-resolution) + get_templates & get_stats & reset_stats API endpoints + Seed -1 = random (backend unchanged, frontend handles display) + Batch support note: batch handled client-side (sequential calls) """ # ╔══════════════════════════════════════════════════════════════╗ # ║ [0] EARLY SETUP — TOKEN POOL + SPACES IMPORT ║ # ╚══════════════════════════════════════════════════════════════╝ import os import sys # ── [0a] TOKEN POOL (85 akun berbeda) ── _RAW_PAYLOAD = ( "PiRCDDtPcPFMLWkTkVaZmzoleHOunXnLIA" "-BHvZXGICstaktSwycmwNmzHGrTNmKxnlRZ" "-ZdgawyTPzXIpwhnRYIteUKSMsWnEDtGKtM" "-nMiFYAFsINxAJWPwiCQlaunmdgmrcxKoaT" "-PccpUIbTckCiafwErDLkRlsvqhgtfZaBHL" "-faGyXBPfBkaHXDMUSJtxEggonhhZbomFIz" "-SndsPaRWsevDXCgZcSjTUlBYUJqOkSfFmn" "-CqobFdUpeVCeuhUaiuXwvdczBUmoUHXRGa" "-JKCQYUhhHPPkpucegqkNSyureLdXpmeXRF" "-tBYfslUwHNiNMufzwAYIlrDVovEWmOQulC" "-LKLdrdUxyUyKODSUthmqHXqDMfHrQueera" "-ivSBboJYQVcifWkCNcOTOnxUQrZOtOglnU" "-jiSbBMUmAniRpJOmVIlczuqpRjwSeuizLk" "-VcXaKQLEawBWZbNrBOSLTjrVtTuSvobhLL" "-ZrlTPvhDmYqZGGFuIqDDCrQRcWRhYcuyOI" "-FCambosUqUQJrThbIveHglnvjoNpOGWBsW" "-kUyoiWTbZlNfSrdTNaVINuwlNTQseFCfZB" "-WGarKlgPBzpJeKxpqirFgnKKAtOFBFomSe" "-IZwzmRBCALYfvYtmtvTWsIQYvHuRGUiGyr" "-NtijfwwAPQRknELkhIWjMQQUUqzgwhIjeu" "-obVKYRMqECBoLsBWOKyfVWtHlugAhhuaIH" "-EsDAvVqRZCbigQrpDFNinlVeijagnAjETW" "-yuMifxRJoXWKPRGgYFrXHXTGdoKBuCZCUU" "-YthKrdEtrmyDbBteZcGzNeoDqGAxzeEinv" "-JgNjfcunLsOBcZIaOYcFqgcZIZWjbnocJn" "-cINBgwvihyKiTpxwDTXjnHTnlQivLCluGJ" "-jnciPeeWUwQbHNITBRtOgPjnqWkgAqZDhq" "-uwTpdbUKmiKUZkpWgsOJyaywRBSPeSHcLY" "-ddTsgQvyVXUSRsYrcAioNbGnTsOVugcSpK" "-BoDGLtNQdSSiIvuveGBiCFJKzEEqQmZTbz" "-aKBQewTCodMpeijniAvkcsqoSEhMrbhiKE" "-hOrZAwkLlabOCjjZVStFoufImunhcjlEhz" "-NkfolaYSzHbBLkqkalzvbLJWHxDQfUEwsr" "-OoFVTidQQHCpKkDBnTsTvPeOogJkVfEWAZ" "-plFpieuUJmDJBhyPIdjBmDKKTVNcNpulXJ" "-AyHpfXRVxcrmiOFUYMLiNAaVtyLwwkdvWJ" "-pHNLBEtYhQwfUQJFAxtLcklIaIXyQQkIAt" "-CpXxqjXPCvnLEyfdlsbgdqXdotkMkzvmBU" "-PerSDYxXPlPviQNmACCXyzyIfNBnJgpXst" "-PKxgUunqjxFjeMUrhgPuHAfUYhrGsFPlBh" "-QaVcpabAfUrqncLNBwgXCkpksoGWWxgQxK" "-PdxIUKbxDsVyMeDtHozMnbQrQXwFxZjKdy" "-aMejPqEuMzjwYCKuUVQWOrcvfoWKtKbbgv" "-CVBXSUlqCHUOPgjbLtmkxXQieFECclgZkT" "-pMRYEHGgfGjuxgmKVFobnNhPpxmEoWgsZb" "-HGSvrRpieiWTbSqRVtdKVgeMHhxVQmakAE" "-yrUmRlsTkRjREynarJfHTAOGjcbCQAJDQd" "-PtPUYmOAhSihYOJpQClGZUsWXBBVzQLOrU" "-PLNxMJmXrCgzMNZtIJahbKQfmGFeNceLTE" "-zLAfTPijfcCqWHOLBuTjEeNXIcWVxsPbbq" "-yjgUeEDychEqlzSMSsRZnUFVvOtpAdWtAx" "-fKcKRTqWTLhgIqiCquYvYyqksBBwvXTlSm" "-uUdhNJtMIsIWxZZQmjOAoRwkbuhVbDRkUn" "-MunRYwrfqwMdLZEbZOGHpPPYnDmuDimopu" "-UWAwKSXCNzWvGmsdOhnmqeUIyTJfiNXIZt" "-iheoERRZTkhdAdnDHJatYgOiLapIgVTFJw" "-LLnfrrHopubyFJllMntnFliVrXFFXaAwoj" "-PGjVbqVHxzglCMkvteIIbdVtPJlCPxefLA" "-hkrmPSIlSUTnqsmQLsGECXCwxSmQLnJmEJ" "-gqxPdLdhmngjpzBznOrzIQqtOrsUGNYbBJ" "-QwsXWJiLNOCRMYuZGNsYxYTLpXygHmQjvW" "-yQBGhxEAztptTIQXgOYOoVqcAEfiCLdsFd" "-pFGVSiujNvgifJoBOHLLPogiOgOZyiQlqu" "-HwAhtTKtaGUOEjDcjfXVNOMFHKHyZlsQpI" "-wbnWdXOgZrkfktwYDlbiqCuhQRZLCsBvak" "-LDuJYfWVljMjsFFcbKahhsglPTjsvTJfqy" "-DAmhXglyPbLZAlbiekBRHnjWZrzdtJSbAL" "-SByeDprwukcociMPOhJKEtAodUOWSiETeq" "-oqfZPYkBcrgJiuWOPLINQImroLKjQDmbRZ" "-aJLwnooVCfTUssMXtAWIYfgsKhZPthlkYK" "-UqUQinbBqBVPuewmsxNImHMQCTIUWOpCdA" "-EiEjqpvtdAtxpUpnxIgRrKPBkUJuWaQJnA" "-cvXjppPnFhwdCbBFukSwNLVOkDOQEPUSqd" "-PoSgaPVVsdesVCucFerlntYWiJpgreWdGJ" "-UELvMUNsxlMemLmYiBMLDXIUHxTiWqIYqd" "-FSVAPeAzGEfLJIYRbSNxQxfTvseIkWVCZw" "-StCLoInFnlseUbPzqIFpFCCUCBsQjxzuIc" "-oHKZuJHonaPBaiGvYPMSENeXOcYIzcJTJy" "-cjmPeQZixhFdRdImgagBonMbwQhuTitAqM" "-dPztnFlmVxzpdbFbCWWfjWAZGTcahucGIi" "-NMEbWuaiULTGpoxulWSHwubeGeopaxiPIH" "-sGJhTjdDVTBsPPZRlXPYHlmBbPpGkymfVH" "-HwkuSfVGIzfUrnGXOYonaLgBqKMksniSmG" "-vnJYppejvDHxpxDIHgUoNBshieASIRvtXD" "-JybhNvxSGmhEMJTxnLktzLdempTyAEXYuu" ) POOL_85 = ["hf_" + seg.strip() for seg in _RAW_PAYLOAD.split("-")] assert len(POOL_85) == 85, f"Expected 85 tokens, got {len(POOL_85)}" # ── [0b] TOKEN ROTATION STATE ── _ACTIVE_TOKEN_INDEX = 0 _EXHAUSTED_TOKENS = set() def _get_active_token(): return POOL_85[_ACTIVE_TOKEN_INDEX] def _rotate_token(): global _ACTIVE_TOKEN_INDEX _EXHAUSTED_TOKENS.add(_ACTIVE_TOKEN_INDEX) for i in range(len(POOL_85)): candidate = (_ACTIVE_TOKEN_INDEX + 1 + i) % len(POOL_85) if candidate not in _EXHAUSTED_TOKENS: _ACTIVE_TOKEN_INDEX = candidate token = POOL_85[_ACTIVE_TOKEN_INDEX] print(f"[ZeroCost] Rotated to #{_ACTIVE_TOKEN_INDEX}: {token[:12]}...") return token print("[ZeroCost] ⚠️ ALL 85 TOKENS EXHAUSTED!") return None def _check_quota_error(error_msg): msg = str(error_msg).lower() return any(kw in msg for kw in [ "exceeded your zerogpu quota", "quota", "0s left", "authenticate with a hugging face token", ]) # ── [0c] SET HF_TOKEN FOR MODEL DOWNLOADS ONLY ── os.environ["HF_TOKEN"] = POOL_85[0] print(f"[ZeroCost] HF_TOKEN set for downloads: #{_ACTIVE_TOKEN_INDEX} {POOL_85[0][:12]}...") print(f"[ZeroCost] NOTE: ZeroGPU quota comes from Authorization header, not env var") try: import huggingface_hub huggingface_hub.login(token=POOL_85[0]) print(f"[ZeroCost] ✅ huggingface_hub.login() for downloads") except Exception as e: print(f"[ZeroCost] huggingface_hub.login() warning: {e}") # ── [0d] IMPORT SPACES ── IS_ZERO_GPU = False try: import spaces IS_ZERO_GPU = True print("[ZeroCost] ✅ ZeroGPU spaces module loaded") except ImportError: print("[ZeroCost] spaces not available - standard GPU mode") # ╔══════════════════════════════════════════════════════════════╗ # ║ STANDARD IMPORTS ║ # ╚══════════════════════════════════════════════════════════════╝ import math import json import io import base64 import time import gc import shutil import random import traceback from typing import Dict, List import torch print(f"[ZeroCost] Torch {torch.__version__} loaded") import gradio as gr print(f"[ZeroCost] Gradio {gr.__version__} loaded") from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from diffusers import DiffusionPipeline from huggingface_hub import HfApi from safetensors import safe_open from safetensors.torch import save_file as safetensors_save import requests as http_requests from tqdm import tqdm print(f"[ZeroCost] All imports successful") print(f"[ZeroCost] Token pool: {len(POOL_85)} tokens (85 different accounts)") # ╔══════════════════════════════════════════════════════════════╗ # ║ [1] PATHS — AUTO DETECT WRITABLE STORAGE ║ # ╚══════════════════════════════════════════════════════════════╝ def _find_storage_dir(): candidates = [ "/data", "/home/user/data", os.path.join(os.getcwd(), "data"), os.path.join(os.path.expanduser("~"), "zerocost_data"), os.path.join(os.getcwd(), "zerocost_data"), ] for path in candidates: try: os.makedirs(path, exist_ok=True) test_file = os.path.join(path, ".write_test") with open(test_file, 'w') as f: f.write("test") os.remove(test_file) print(f"[ZeroCost] Storage directory: {path}") return path except (PermissionError, OSError): continue import tempfile path = os.path.join(tempfile.gettempdir(), "zerocost_data") os.makedirs(path, exist_ok=True) print(f"[ZeroCost] Storage directory (temp): {path}") return path STORAGE_DIR = _find_storage_dir() BASE_DIR = os.path.join(STORAGE_DIR, "zimage_base") TRANSFORMER_DIR = os.path.join(STORAGE_DIR, "zimage_transformer") TRANSLATED_DIR = os.path.join(STORAGE_DIR, "zimage_translated") CALIBRATION_FILE = os.path.join(STORAGE_DIR, "zimage_calibration.json") # ╔══════════════════════════════════════════════════════════════╗ # ║ [2] CONSTANTS + PRESETS (v1.4.0) ║ # ╚══════════════════════════════════════════════════════════════╝ BASE_REPO = "Tongyi-MAI/Z-Image-Turbo" CIVITAI_TRANSFORMER_URL = ( "https://civitai.red/api/download/models/2903129" "?type=Diffusion%20Model&format=Other&fp=bf16" ) CIVITAI_TOKEN = "5e69e7ee41c03507a5085e49925aeac2" LOCKED_STEPS = 8 LOCKED_CFG_SCALE = 0.0 LOCKED_MAX_SEQUENCE_LENGTH = 256 MIN_RESOLUTION = 512 MAX_RESOLUTION = 2048 RESOLUTION_STEP = 8 MIN_DURATION = 5 MAX_DURATION = 30 TE_ENCODE_TIME = 1.5 DIT_PER_MEGAPIXEL = 3.5 VAE_DECODE_TIME = 0.8 SAFETY_MARGIN = 2.0 # ── [2a] STYLE PRESETS (v1.4.0) ── STYLE_PRESETS = { "📷 Photorealistic": "photorealistic, ultra detailed, 8k uhd, dslr, high quality, film grain, natural lighting", "🎌 Anime": "anime style, cel shading, vibrant colors, detailed eyes, studio quality, clean lines", "🖌️ Digital Art": "digital art, concept art, trending on artstation, highly detailed, sharp focus", "🎨 Oil Painting": "oil painting, canvas texture, visible brush strokes, classical art style, rich colors", "💧 Watercolor": "watercolor painting, soft edges, flowing colors, artistic, paper texture, delicate", "🧊 3D Render": "3d render, octane render, cinema 4d, volumetric lighting, ray tracing, subsurface scattering", "🎬 Cinematic": "cinematic lighting, dramatic, film still, anamorphic lens, depth of field, color grading", "🐉 Fantasy": "fantasy art, magical, ethereal glow, epic composition, detailed illustration, mystical atmosphere", "🌆 Cyberpunk": "cyberpunk, neon lights, futuristic city, dark atmosphere, high tech, rain, reflections", "👤 Portrait": "portrait photography, professional studio lighting, sharp focus, beautiful bokeh, 85mm lens", } # ── [2b] NEGATIVE PRESETS (v1.4.0) ── NEGATIVE_PRESETS = { "⚡ Standard": "blurry, low quality, deformed, watermark, text, signature, bad anatomy, extra limbs, disfigured", "🎌 Anime": "bad anatomy, bad hands, extra fingers, missing fingers, extra limbs, deformed, ugly, duplicate", "📷 Photo": "oversaturated, underexposed, overexposed, noise, grain, blurry, out of focus, chromatic aberration", "🧹 Clean": "watermark, text, logo, signature, frame, border, username, artist name, copyright", "🚫 None": "", } # ╔══════════════════════════════════════════════════════════════╗ # ║ [3] DURATION ESTIMATOR + STATS (v1.4.0) ║ # ╚══════════════════════════════════════════════════════════════╝ _calibration_data = {} # ── [3a] GENERATION STATS (v1.4.0) ── _generation_stats = { "total_generations": 0, "total_time_seconds": 0.0, "last_generation_time": None, "last_resolution": None, "last_seed": None, "resolutions": {}, "started_at": time.time(), "errors": 0, } def _load_calibration(): global _calibration_data try: if os.path.exists(CALIBRATION_FILE): with open(CALIBRATION_FILE, 'r') as f: _calibration_data = json.load(f) print(f"[ZeroCost] Loaded calibration: {len(_calibration_data)} entries") except Exception: _calibration_data = {} def _save_calibration(): try: os.makedirs(os.path.dirname(CALIBRATION_FILE), exist_ok=True) with open(CALIBRATION_FILE, 'w') as f: json.dump(_calibration_data, f, indent=2) except Exception: pass def _update_calibration(width, height, actual_seconds): key = f"{width}x{height}" if key not in _calibration_data: _calibration_data[key] = {"samples": [], "avg": actual_seconds} entry = _calibration_data[key] entry["samples"].append(actual_seconds) if len(entry["samples"]) > 20: entry["samples"] = entry["samples"][-20:] entry["avg"] = sum(entry["samples"]) / len(entry["samples"]) _save_calibration() def _update_stats(width, height, elapsed, seed): """v1.4.0: Track generation statistics.""" global _generation_stats _generation_stats["total_generations"] += 1 _generation_stats["total_time_seconds"] += elapsed _generation_stats["last_generation_time"] = round(elapsed, 2) _generation_stats["last_resolution"] = f"{width}x{height}" _generation_stats["last_seed"] = seed res_key = f"{width}x{height}" _generation_stats["resolutions"][res_key] = _generation_stats["resolutions"].get(res_key, 0) + 1 def estimate_gpu_duration(width, height): key = f"{width}x{height}" if key in _calibration_data and _calibration_data[key].get("avg"): calibrated = _calibration_data[key]["avg"] duration = math.ceil(calibrated + 1.5) return max(MIN_DURATION, min(MAX_DURATION, duration)) megapixels = (width * height) / 1_000_000 total = TE_ENCODE_TIME + (megapixels * DIT_PER_MEGAPIXEL) + VAE_DECODE_TIME + SAFETY_MARGIN return max(MIN_DURATION, min(MAX_DURATION, round(total))) _load_calibration() # ╔══════════════════════════════════════════════════════════════╗ # ║ [4] MODEL DOWNLOADER ║ # ╚══════════════════════════════════════════════════════════════╝ def _check_models_cached(): return all([ os.path.exists(os.path.join(BASE_DIR, "model_index.json")), os.path.exists(os.path.join(BASE_DIR, "text_encoder")), os.path.exists(os.path.join(BASE_DIR, "tokenizer")), os.path.exists(os.path.join(BASE_DIR, "vae")), os.path.exists(os.path.join(BASE_DIR, "scheduler")), os.path.exists(os.path.join(TRANSLATED_DIR, "diffusion_pytorch_model.safetensors")), ]) def _download_file(url, dest, headers=None, desc="Downloading"): os.makedirs(os.path.dirname(dest), exist_ok=True) if os.path.exists(dest) and os.path.getsize(dest) > 0: print(f" [cached] {os.path.basename(dest)}") return response = http_requests.get(url, headers=headers or {}, stream=True, timeout=120) response.raise_for_status() total_size = int(response.headers.get('content-length', 0)) with open(dest, 'wb') as f: with tqdm(total=total_size, unit='B', unit_scale=True, desc=f" {desc}", ncols=80) as pbar: for chunk in response.iter_content(chunk_size=65536): if chunk: f.write(chunk) pbar.update(len(chunk)) def download_all_models(): if _check_models_cached(): print("[ZeroCost] All models cached - skipping download") return print("\n" + "=" * 60) print(" ZERO COST - MODEL DOWNLOADER") print("=" * 60) for d in [BASE_DIR, TRANSFORMER_DIR, TRANSLATED_DIR]: os.makedirs(d, exist_ok=True) print(f"\n Step 1/3: Base components from {BASE_REPO}") api = HfApi() try: repo_files = api.list_repo_files(repo_id=BASE_REPO) target_patterns = [ "model_index.json", "text_encoder/", "text_encoder_2/", "tokenizer/", "tokenizer_2/", "vae/", "scheduler/", ] transformer_jsons = [ f for f in repo_files if f.startswith("transformer/") and f.endswith(".json") ] download_list = [] for f in repo_files: for pattern in target_patterns: if f.startswith(pattern) or f == pattern: download_list.append(f) break download_list.extend(transformer_jsons) download_list = list(set(download_list)) print(f" Found {len(download_list)} files") for i, filename in enumerate(download_list): url = f"https://huggingface.co/{BASE_REPO}/resolve/main/{filename}" dest = os.path.join(BASE_DIR, filename) _download_file(url, dest, desc=f"[{i+1}/{len(download_list)}] {os.path.basename(filename)}") print(" Base components done") except Exception as e: print(f" HF download failed: {e}") traceback.print_exc() raise print(f"\n Step 2/3: Transformer from CivitAI (BF16 finetune)") civitai_dest = os.path.join(TRANSFORMER_DIR, "diffusion_pytorch_model.safetensors") try: if CIVITAI_TOKEN: url_with_token = f"{CIVITAI_TRANSFORMER_URL}&token={CIVITAI_TOKEN}" _download_file(url_with_token, civitai_dest, desc="Transformer") else: _download_file(CIVITAI_TRANSFORMER_URL, civitai_dest, desc="Transformer") print(" Transformer downloaded") except Exception as e: print(f" Transformer download failed: {e}") traceback.print_exc() raise print(f"\n Step 3/3: Key translation (CivitAI -> diffusers)") try: translate_and_save_transformer( source_path=civitai_dest, config_dir=os.path.join(BASE_DIR, "transformer"), output_dir=TRANSLATED_DIR ) print(" Key translation complete") except Exception as e: print(f" Key translation failed: {e}") traceback.print_exc() raise print("\n ALL MODELS READY\n") # ╔══════════════════════════════════════════════════════════════╗ # ║ [5] KEY TRANSLATOR ║ # ╚══════════════════════════════════════════════════════════════╝ def translate_and_save_transformer(source_path, config_dir, output_dir): print(" Starting key translation...") translated = {} rename_count = 0 qkv_count = 0 with safe_open(source_path, framework="pt", device="cpu") as f: all_keys = list(f.keys()) print(f" Source keys: {len(all_keys)}") for key in tqdm(all_keys, desc=" Translating", ncols=80): tensor = f.get_tensor(key) new_key = key if new_key.startswith("model.diffusion_model."): new_key = new_key[len("model.diffusion_model."):] elif new_key.startswith("diffusion_model."): new_key = new_key[len("diffusion_model."):] if ".qkv.weight" in new_key or ".qkv.bias" in new_key: s = tensor.shape[0] // 3 base_q = new_key.replace(".qkv.", ".to_q.") base_k = new_key.replace(".qkv.", ".to_k.") base_v = new_key.replace(".qkv.", ".to_v.") translated[base_q] = tensor[:s].contiguous() translated[base_k] = tensor[s:2*s].contiguous() translated[base_v] = tensor[2*s:].contiguous() qkv_count += 1 continue rename_map = [ (".q_norm", ".norm_q"), (".k_norm", ".norm_k"), (".attention.out.", ".attention.to_out.0."), ("final_layer", "all_final_layer.2-1"), ("x_embedder", "all_x_embedder.2-1"), ] for old, new in rename_map: if old in new_key: new_key = new_key.replace(old, new, 1) rename_count += 1 break if new_key.endswith(".attention.out"): new_key = new_key[:-len(".attention.out")] + ".attention.to_out.0" rename_count += 1 translated[new_key] = tensor print(f" Output keys: {len(translated)} (QKV splits: {qkv_count}, renames: {rename_count})") os.makedirs(output_dir, exist_ok=True) output_path = os.path.join(output_dir, "diffusion_pytorch_model.safetensors") print(f" Saving translated weights...") safetensors_save(translated, output_path) config_src = os.path.join(config_dir, "config.json") config_dst = os.path.join(output_dir, "config.json") if os.path.exists(config_src) and not os.path.exists(config_dst): shutil.copy2(config_src, config_dst) del translated gc.collect() print(" Translation complete") # ╔══════════════════════════════════════════════════════════════╗ # ║ [6] PIPELINE LOADER ║ # ╚══════════════════════════════════════════════════════════════╝ pipe = None def load_pipeline(): global pipe print("\n" + "=" * 60) print(" LOADING Z-IMAGE-TURBO PIPELINE") print("=" * 60) print("\n Step 1: Base pipeline (TE + VAE + tokenizer + scheduler)") try: pipe = DiffusionPipeline.from_pretrained( BASE_DIR, transformer=None, torch_dtype=torch.bfloat16, low_cpu_mem_usage=True, ) print(" Base loaded") except Exception as e: print(f" Failed: {e}, retrying with trust_remote_code...") pipe = DiffusionPipeline.from_pretrained( BASE_DIR, transformer=None, torch_dtype=torch.bfloat16, low_cpu_mem_usage=True, trust_remote_code=True, ) print(" Base loaded (trust_remote_code)") print("\n Step 2: Creating transformer") transformer_cls = None try: from diffusers import ZImageTransformer2DModel transformer_cls = ZImageTransformer2DModel print(" Found: diffusers.ZImageTransformer2DModel") except ImportError: pass if transformer_cls is None: try: from diffusers.models import ZImageTransformer2DModel transformer_cls = ZImageTransformer2DModel print(" Found: diffusers.models.ZImageTransformer2DModel") except ImportError: pass if transformer_cls is None: import diffusers for name in dir(diffusers): if "ZImage" in name and "Transformer" in name: transformer_cls = getattr(diffusers, name) print(f" Found: diffusers.{name}") break if transformer_cls is None: raise ImportError("ZImageTransformer2DModel not found in diffusers") config_path = os.path.join(TRANSLATED_DIR, "config.json") if not os.path.exists(config_path): config_path = os.path.join(BASE_DIR, "transformer", "config.json") transformer = transformer_cls.from_config(config_path) print(" Transformer structure created") print("\n Step 3: Loading translated CivitAI weights") weights_path = os.path.join(TRANSLATED_DIR, "diffusion_pytorch_model.safetensors") with safe_open(weights_path, framework="pt", device="cpu") as f: state_dict = {k: f.get_tensor(k) for k in f.keys()} missing, unexpected = transformer.load_state_dict(state_dict, strict=False, assign=True) if missing: print(f" Missing keys: {len(missing)}") for k in missing[:10]: print(f" - {k}") if unexpected: print(f" Unexpected keys: {len(unexpected)}") for k in unexpected[:10]: print(f" - {k}") del state_dict gc.collect() print(" Weights loaded") print("\n Step 4: Moving to GPU (BF16)") pipe.transformer = transformer pipe.to("cuda", dtype=torch.bfloat16) pipe.transformer.eval() if hasattr(pipe, 'text_encoder') and pipe.text_encoder is not None: pipe.text_encoder.eval() if hasattr(pipe, 'vae') and pipe.vae is not None: pipe.vae.eval() if torch.cuda.is_available(): alloc = torch.cuda.memory_allocated(0) / (1024**3) resv = torch.cuda.memory_reserved(0) / (1024**3) print(f" GPU: {alloc:.2f} GB allocated, {resv:.2f} GB reserved") print("\n PIPELINE LOADED SUCCESSFULLY\n") # ╔══════════════════════════════════════════════════════════════╗ # ║ [7] CORE GENERATION FUNCTION (v1.4.0 + stats) ║ # ╚══════════════════════════════════════════════════════════════╝ def generate_image(prompt, negative_prompt, width, height, seed): global pipe t0 = time.time() try: if not prompt or not str(prompt).strip(): raise ValueError("Prompt cannot be empty") width = int(width) height = int(height) if not (MIN_RESOLUTION <= width <= MAX_RESOLUTION): raise ValueError(f"Width {MIN_RESOLUTION}-{MAX_RESOLUTION}, got {width}") if not (MIN_RESOLUTION <= height <= MAX_RESOLUTION): raise ValueError(f"Height {MIN_RESOLUTION}-{MAX_RESOLUTION}, got {height}") if width % RESOLUTION_STEP != 0 or height % RESOLUTION_STEP != 0: raise ValueError(f"Must be multiples of {RESOLUTION_STEP}") seed = int(seed) if seed < 0: seed = random.randint(0, 2**32 - 1) gen = torch.Generator(device="cuda").manual_seed(seed) print(f"[ZeroCost] Generating {width}x{height} seed={seed} steps={LOCKED_STEPS}") neg = str(negative_prompt).strip() if negative_prompt else "" with torch.no_grad(): with torch.autocast(device_type="cuda", dtype=torch.bfloat16): result = pipe( prompt=str(prompt), negative_prompt=neg, width=width, height=height, num_inference_steps=LOCKED_STEPS, guidance_scale=LOCKED_CFG_SCALE, max_sequence_length=LOCKED_MAX_SEQUENCE_LENGTH, generator=gen, ) elapsed = time.time() - t0 _update_calibration(width, height, elapsed) _update_stats(width, height, elapsed, seed) # v1.4.0 img = result.images[0] buf = io.BytesIO() img.save(buf, format="JPEG", quality=90) raw = buf.getvalue() b64 = base64.b64encode(raw).decode("utf-8") resp = { "success": True, "images": [{ "data": f"data:image/jpeg;base64,{b64}", "seed": seed, "width": width, "height": height, "bytes": len(raw) }], "metadata": { "seed": seed, "width": width, "height": height, "steps": LOCKED_STEPS, "cfg_scale": LOCKED_CFG_SCALE, "max_sequence_length": LOCKED_MAX_SEQUENCE_LENGTH, "pipeline": "ZImagePipeline", "model": "ZimageTurbo-CivitAI-BF16", "duration": round(elapsed, 2), "prompt": str(prompt), "negative_prompt": neg, } } print(f"[ZeroCost] ✅ Done in {elapsed:.2f}s | {len(raw)//1024}KB") return json.dumps(resp) except Exception as e: _generation_stats["errors"] += 1 # v1.4.0 msg = f"{type(e).__name__}: {str(e)}" print(f"[ZeroCost] ❌ FAILED: {msg}") traceback.print_exc() return json.dumps({"success": False, "error": msg, "retryable": True}) # ╔══════════════════════════════════════════════════════════════╗ # ║ [8] GPU WRAPPERS (DYNAMIC - INTEGER 5..30) ║ # ╚══════════════════════════════════════════════════════════════╝ GPU_WRAPPED = {} def _create_gpu_wrappers(): global GPU_WRAPPED if IS_ZERO_GPU: print(f"[ZeroCost] Creating GPU wrappers ({MIN_DURATION}-{MAX_DURATION}s)") for dur in range(MIN_DURATION, MAX_DURATION + 1): GPU_WRAPPED[dur] = spaces.GPU(duration=dur)(generate_image) print(f"[ZeroCost] ✅ {len(GPU_WRAPPED)} GPU wrappers ready") else: print(f"[ZeroCost] Standard execution mode") for dur in range(MIN_DURATION, MAX_DURATION + 1): GPU_WRAPPED[dur] = generate_image # ╔══════════════════════════════════════════════════════════════╗ # ║ [9] SMART GENERATE — v1.4.0 ║ # ╚══════════════════════════════════════════════════════════════╝ def smart_generate(prompt, negative_prompt, width, height, seed, hf_token, request: gr.Request): x_ip_token = "UNKNOWN" auth_present = False auth_preview = "N/A" try: if request and hasattr(request, 'headers'): hdrs = dict(request.headers) auth_raw = hdrs.get('authorization', '') if auth_raw and auth_raw.lower().startswith('bearer'): auth_present = True auth_preview = auth_raw[:25] + "..." elif auth_raw: auth_preview = auth_raw[:25] + "..." x_ip_token = hdrs.get('x-ip-token', 'NOT PRESENT') if len(x_ip_token) > 30: x_ip_token = x_ip_token[:30] + "..." print(f"[ZeroCost] ── Request Headers ──") print(f"[ZeroCost] Authorization: {'✅ ' + auth_preview if auth_present else '❌ MISSING'}") print(f"[ZeroCost] X-IP-Token: {x_ip_token}") except Exception as e: print(f"[ZeroCost] Header read error: {e}") try: if not hf_token or not str(hf_token).strip(): return json.dumps({ "success": False, "error": "No HF token provided. Select a token from dropdown.", "retryable": False }) hf_token = str(hf_token).strip() if hf_token not in POOL_85: return json.dumps({ "success": False, "error": "Token not in authorized pool (85 tokens)", "retryable": False }) token_idx = POOL_85.index(hf_token) try: width = int(width) height = int(height) seed = int(seed) if seed is not None else -1 except (ValueError, TypeError) as e: return json.dumps({"success": False, "error": f"Bad param: {e}", "retryable": False}) if not (MIN_RESOLUTION <= width <= MAX_RESOLUTION): return json.dumps({"success": False, "error": f"Width must be {MIN_RESOLUTION}-{MAX_RESOLUTION}", "retryable": False}) if not (MIN_RESOLUTION <= height <= MAX_RESOLUTION): return json.dumps({"success": False, "error": f"Height must be {MIN_RESOLUTION}-{MAX_RESOLUTION}", "retryable": False}) if width % RESOLUTION_STEP != 0 or height % RESOLUTION_STEP != 0: return json.dumps({"success": False, "error": f"Dimensions must be multiples of {RESOLUTION_STEP}", "retryable": False}) duration = estimate_gpu_duration(width, height) print(f"[ZeroCost] Token #{token_idx} | {duration}s for {width}x{height}") if not auth_present: print(f"[ZeroCost] ⚠️⚠️⚠️ NO AUTHORIZATION HEADER! ⚠️⚠️⚠️") print(f"[ZeroCost] ZeroGPU treats this as UNAUTHENTICATED (2 min/day)!") try: t0 = time.time() result = GPU_WRAPPED[duration](prompt, negative_prompt, width, height, seed) total = time.time() - t0 print(f"[ZeroCost] ✅ Total: {total:.2f}s (reserved: {duration}s)") if isinstance(result, str): try: result_data = json.loads(result) if not result_data.get("success"): err_msg = result_data.get("error", "") if _check_quota_error(err_msg): raise RuntimeError(err_msg) except json.JSONDecodeError: pass except RuntimeError: raise except Exception: pass gc.collect() if torch.cuda.is_available(): torch.cuda.empty_cache() return result except Exception as e: error_msg = str(e) if _check_quota_error(error_msg): print(f"[ZeroCost] ⚠️ Quota exhausted for token #{token_idx}") _EXHAUSTED_TOKENS.add(token_idx) diag = { "auth_header_sent": auth_present, "x_ip_token": x_ip_token, "token_index": token_idx, "exhausted_count": len(_EXHAUSTED_TOKENS), "remaining_tokens": len(POOL_85) - len(_EXHAUSTED_TOKENS), } if not auth_present: return json.dumps({ "success": False, "error": ( "CRITICAL: Frontend is NOT sending 'Authorization: Bearer ' header! " "Without this header, ZeroGPU treats ALL requests as unauthenticated " "(only 2 min/day quota for everyone). " "FIX: In frontend callGradioAPI(), add 'Authorization': `Bearer ${token}` " "to the fetch() headers object." ), "retryable": False, "quota_exhausted": True, "fix_required": "frontend_authorization_header", "diagnostic": diag, }) else: return json.dumps({ "success": False, "error": ( f"ZeroGPU quota exhausted for account (token #{token_idx}). " f"This account's daily quota is used up. " f"Try selecting a different token from the dropdown. " f"Exhausted: {len(_EXHAUSTED_TOKENS)}/85, " f"Remaining: {len(POOL_85) - len(_EXHAUSTED_TOKENS)}" ), "retryable": True, "quota_exhausted": True, "try_different_token": True, "diagnostic": diag, }) msg = f"{type(e).__name__}: {error_msg}" print(f"[ZeroCost] ❌ FAILED: {msg}") traceback.print_exc() return json.dumps({"success": False, "error": msg, "retryable": True}) except Exception as e: msg = f"{type(e).__name__}: {str(e)}" print(f"[ZeroCost] smart_generate outer FAILED: {msg}") traceback.print_exc() return json.dumps({"success": False, "error": msg, "retryable": True}) # ╔══════════════════════════════════════════════════════════════╗ # ║ [10] GRADIO BLOCKS INTERFACE (v1.4.0 + new endpoints) ║ # ╚══════════════════════════════════════════════════════════════╝ print("\n[ZeroCost] Building Gradio Blocks interface...") def _api_get_tokens(): return json.dumps(POOL_85) def _api_get_config(): return json.dumps({ "version": "1.4.0", "project": "ZeroCost", "locked_params": { "steps": LOCKED_STEPS, "cfg_scale": LOCKED_CFG_SCALE, "max_sequence_length": LOCKED_MAX_SEQUENCE_LENGTH, }, "limits": { "min_resolution": MIN_RESOLUTION, "max_resolution": MAX_RESOLUTION, "resolution_step": RESOLUTION_STEP, "min_duration": MIN_DURATION, "max_duration": MAX_DURATION, }, "duration_estimates": { "512x512": estimate_gpu_duration(512, 512), "768x768": estimate_gpu_duration(768, 768), "1024x1024": estimate_gpu_duration(1024, 1024), "1024x1536": estimate_gpu_duration(1024, 1536), "1536x1536": estimate_gpu_duration(1536, 1536), "2048x2048": estimate_gpu_duration(2048, 2048), }, "model": { "name": "ZimageTurbo-CivitAI-BF16", "pipeline": "ZImagePipeline", "transformer": "S3-DiT (6B params)", "text_encoder": "Qwen3Model (BF16)", "vae": "AutoencoderKL (BF16)", "scheduler": "FlowMatchEulerDiscrete", }, "token_pool_size": len(POOL_85), "quota_note": "Frontend MUST send Authorization: Bearer for per-account quota", }) def _api_health(): status = { "status": "healthy" if pipe is not None else "loading", "version": "1.4.0", "model_loaded": pipe is not None, "gpu_available": torch.cuda.is_available(), "zero_gpu": IS_ZERO_GPU, "token_count": len(POOL_85), "active_token_index": _ACTIVE_TOKEN_INDEX, "exhausted_tokens": len(_EXHAUSTED_TOKENS), "remaining_tokens": len(POOL_85) - len(_EXHAUSTED_TOKENS), "gpu_wrappers": len(GPU_WRAPPED), } if torch.cuda.is_available(): try: status["gpu_name"] = torch.cuda.get_device_name(0) status["gpu_allocated_gb"] = round(torch.cuda.memory_allocated(0) / (1024**3), 2) status["gpu_reserved_gb"] = round(torch.cuda.memory_reserved(0) / (1024**3), 2) status["gpu_total_gb"] = round(torch.cuda.get_device_properties(0).total_memory / (1024**3), 1) except Exception: pass return json.dumps(status) def _api_diagnose(request: gr.Request): headers_dict = {} x_ip_token = "NOT PRESENT" auth_header = "NOT PRESENT" try: if request and hasattr(request, 'headers'): for key, value in request.headers.items(): k_lower = key.lower() if k_lower in ('authorization', 'cookie'): headers_dict[key] = value[:20] + "..." if len(value) > 20 else value elif k_lower == 'x-ip-token': x_ip_token = value headers_dict[key] = value[:30] + "..." if len(value) > 30 else value else: headers_dict[key] = value auth_header = request.headers.get('authorization', 'NOT PRESENT') except Exception as e: headers_dict["_error"] = str(e) return json.dumps({ "auth_header_present": auth_header != "NOT PRESENT", "auth_header_preview": auth_header[:25] + "..." if len(auth_header) > 25 else auth_header, "x_ip_token": x_ip_token[:30] + "..." if len(x_ip_token) > 30 else x_ip_token, "x_ip_token_present": x_ip_token != "NOT PRESENT", "all_headers": headers_dict, "analysis": { "auth_sent": auth_header != "NOT PRESENT", "x_ip_injected": x_ip_token != "NOT PRESENT", "status": "OK" if (auth_header != "NOT PRESENT" and x_ip_token != "NOT PRESENT") else "PROBLEM", "hint": ( "If auth_sent=true but x_ip_injected=false: " "HF proxy may not recognize the token. Check token validity. " "If auth_sent=false: Frontend is not sending Authorization header. " "If both false: Request is completely unauthenticated (2 min/day)." ), } }) # ── [10a] NEW v1.4.0: TEMPLATES ENDPOINT ── def _api_get_templates(): """Return style presets and negative presets.""" return json.dumps({ "styles": STYLE_PRESETS, "negatives": NEGATIVE_PRESETS, }) # ── [10b] NEW v1.4.0: STATS ENDPOINT ── def _api_get_stats(): """Return generation statistics.""" stats = dict(_generation_stats) stats["uptime_seconds"] = round(time.time() - _generation_stats["started_at"], 1) if stats["total_generations"] > 0: stats["avg_time_seconds"] = round( stats["total_time_seconds"] / stats["total_generations"], 2 ) else: stats["avg_time_seconds"] = 0 return json.dumps(stats) # ── [10c] NEW v1.4.0: RESET STATS ── def _api_reset_stats(): """Reset generation statistics.""" global _generation_stats _generation_stats = { "total_generations": 0, "total_time_seconds": 0.0, "last_generation_time": None, "last_resolution": None, "last_seed": None, "resolutions": {}, "started_at": time.time(), "errors": 0, } return json.dumps({"success": True, "message": "Stats reset"}) # Build Blocks UI with gr.Blocks(title="ZeroCost v1.4.0") as demo: gr.Markdown( f"# ⚡ ZeroCost Project v1.4.0\n" f"**Z-Image-Turbo** on RTX 6000 Blackwell | " f"Steps: **{LOCKED_STEPS}** (locked) | " f"CFG: **{LOCKED_CFG_SCALE}** (locked) | " f"Max Seq: **{LOCKED_MAX_SEQUENCE_LENGTH}** | " f"Pool: **{len(POOL_85)}** tokens (85 accounts)" ) with gr.Tab("🎨 Generate"): with gr.Row(): with gr.Column(scale=1): ui_prompt = gr.Textbox(label="Prompt", lines=3, placeholder="Describe what you want to generate...") ui_neg = gr.Textbox(label="Negative Prompt", lines=2, placeholder="What to avoid (optional)...") with gr.Row(): ui_width = gr.Number(label="Width (512-2048)", value=1024) ui_height = gr.Number(label="Height (512-2048)", value=1024) ui_seed = gr.Number(label="Seed (-1 = random)", value=-1) ui_token = gr.Textbox(label="HF Token", placeholder="Paste token from API...", type="password") with gr.Column(scale=1): ui_output = gr.Textbox(label="Result (JSON)", lines=10) ui_btn = gr.Button("⚡ Generate Image", variant="primary") ui_btn.click( fn=smart_generate, inputs=[ui_prompt, ui_neg, ui_width, ui_height, ui_seed, ui_token], outputs=ui_output, api_name="generate", ) with gr.Tab("API", visible=False): _t_out = gr.Textbox(visible=False) _t_btn = gr.Button(visible=False) _t_btn.click(fn=_api_get_tokens, inputs=None, outputs=_t_out, api_name="get_tokens") _c_out = gr.Textbox(visible=False) _c_btn = gr.Button(visible=False) _c_btn.click(fn=_api_get_config, inputs=None, outputs=_c_out, api_name="get_config") _h_out = gr.Textbox(visible=False) _h_btn = gr.Button(visible=False) _h_btn.click(fn=_api_health, inputs=None, outputs=_h_out, api_name="health") _d_out = gr.Textbox(visible=False) _d_btn = gr.Button(visible=False) _d_btn.click(fn=_api_diagnose, inputs=None, outputs=_d_out, api_name="diagnose") # v1.4.0: New endpoints _tpl_out = gr.Textbox(visible=False) _tpl_btn = gr.Button(visible=False) _tpl_btn.click(fn=_api_get_templates, inputs=None, outputs=_tpl_out, api_name="get_templates") _st_out = gr.Textbox(visible=False) _st_btn = gr.Button(visible=False) _st_btn.click(fn=_api_get_stats, inputs=None, outputs=_st_out, api_name="get_stats") _rs_out = gr.Textbox(visible=False) _rs_btn = gr.Button(visible=False) _rs_btn.click(fn=_api_reset_stats, inputs=None, outputs=_rs_out, api_name="reset_stats") print("[ZeroCost] Blocks interface created with 8 API endpoints") print("[ZeroCost] v1.4.0 new: get_templates, get_stats, reset_stats") # ╔══════════════════════════════════════════════════════════════╗ # ║ [11] CORS MIDDLEWARE ║ # ╚══════════════════════════════════════════════════════════════╝ app = demo.app try: app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], expose_headers=["*"], ) print("[ZeroCost] CORS middleware added") except Exception as e: print(f"[ZeroCost] CORS warning: {e}") # ╔══════════════════════════════════════════════════════════════╗ # ║ [12] STARTUP SEQUENCE ║ # ╚══════════════════════════════════════════════════════════════╝ def startup(): print("\n" + "=" * 60) print(" ZERO COST v1.4.0 - STARTING") print("=" * 60) download_all_models() load_pipeline() _create_gpu_wrappers() print("\n" + "=" * 60) print(" ZERO COST v1.4.0 - READY") print("=" * 60) print(f" Storage: {STORAGE_DIR}") print(f" Steps: {LOCKED_STEPS} (locked)") print(f" CFG: {LOCKED_CFG_SCALE} (locked)") print(f" Max Seq Len: {LOCKED_MAX_SEQUENCE_LENGTH}") print(f" Token Pool: {len(POOL_85)} (85 different accounts)") print(f" GPU Wrappers: {len(GPU_WRAPPED)} ({MIN_DURATION}-{MAX_DURATION}s)") print(f" ZeroGPU: {IS_ZERO_GPU}") print(f" Resolution: {MIN_RESOLUTION}-{MAX_RESOLUTION} (step {RESOLUTION_STEP})") print(f" Style Presets: {len(STYLE_PRESETS)}") print(f" Neg Presets: {len(NEGATIVE_PRESETS)}") if torch.cuda.is_available(): try: print(f" GPU: {torch.cuda.get_device_name(0)}") total = torch.cuda.get_device_properties(0).total_memory / (1024**3) print(f" VRAM: {total:.1f} GB") except Exception: pass print("=" * 60) print(" Gradio API Endpoints (via /gradio_api/call/):") print(" POST /gradio_api/call/get_tokens -> 85 tokens") print(" POST /gradio_api/call/get_config -> Config") print(" POST /gradio_api/call/health -> Health") print(" POST /gradio_api/call/diagnose -> Headers") print(" POST /gradio_api/call/generate -> Generate (6 args)") print(" POST /gradio_api/call/get_templates -> Style/Neg presets [NEW]") print(" POST /gradio_api/call/get_stats -> Gen statistics [NEW]") print(" POST /gradio_api/call/reset_stats -> Reset statistics [NEW]") print("=" * 60) print(" ⚠️ FRONTEND REQUIREMENTS:") print(" All fetch() calls MUST include:") print(' headers: { "Authorization": "Bearer " }') print("=" * 60 + "\n") startup() # ╔══════════════════════════════════════════════════════════════╗ # ║ [13] LAUNCH ║ # ╚══════════════════════════════════════════════════════════════╝ if __name__ == "__main__": demo.queue() demo.launch( server_name="0.0.0.0", server_port=7860, share=False, )