#!/usr/bin/env python3 """ comfy_apps.py — the "apps around the workflows" layer. Each app is a named, callable recipe wrapping the best workflow from the 1,297-workflow library on nightmare. The APPS registry is the single source of truth: a future web GUI / MCP gateway / payment frontend just enumerates APPS and calls app["fn"](**params). Models live on nightmare (~/ComfyUI/models). See comfy_workflows_lib.py for the low-level builders (flux_inpaint, flux_outpaint, sdxl_txt2img, ...). """ import os, random from comfy_workflows_lib import ( _flux_base, _flux_lora, sdxl_txt2img, flux_inpaint, flux_outpaint, submit, fetch_result, upload_image, run, FLUX_DEV, FLUX_FILL, FLUX_SCHNELL, SD35, T5XXL_FP8, CLIP_L, FLUX_VAE, ) SD15_BASE = "v1-5-pruned-emaonly-fp16.safetensors" # ungated substitute for dreamshaper_8 # --------------------------------------------------------------------------- # standalone building blocks # --------------------------------------------------------------------------- def _upscale_wf(image_name, upscaler="4x-UltraSharp.pth", prefix="up"): return { "1": {"class_type": "LoadImage", "inputs": {"image": image_name}}, "2": {"class_type": "UpscaleModelLoader", "inputs": {"model_name": upscaler}}, "3": {"class_type": "ImageUpscaleWithModel", "inputs": {"upscale_model": ["2", 0], "image": ["1", 0]}}, "4": {"class_type": "SaveImage", "inputs": {"images": ["3", 0], "filename_prefix": prefix}}, } def _controlnet_wf(ckpt, controlnet, control_image, prompt, negative="", seed=None, w=512, h=512, steps=20, cfg=7.0, strength=1.0): if seed is None: seed = random.randint(0, 2**63) return { "1": {"class_type": "CheckpointLoaderSimple", "inputs": {"ckpt_name": ckpt}}, "2": {"class_type": "ControlNetLoader", "inputs": {"control_net_name": controlnet}}, "3": {"class_type": "LoadImage", "inputs": {"image": control_image}}, "4": {"class_type": "CLIPTextEncode", "inputs": {"text": prompt, "clip": ["1", 1]}}, "5": {"class_type": "CLIPTextEncode", "inputs": {"text": negative, "clip": ["1", 1]}}, "6": {"class_type": "EmptyLatentImage", "inputs": {"width": w, "height": h, "batch_size": 1}}, "7": {"class_type": "ControlNetApply", "inputs": {"conditioning": ["4", 0], "control_net": ["2", 0], "image": ["3", 0], "strength": strength}}, "8": {"class_type": "KSampler", "inputs": {"model": ["1", 0], "positive": ["7", 0], "negative": ["5", 0], "latent_image": ["6", 0], "seed": seed, "steps": steps, "cfg": cfg, "sampler_name": "euler", "scheduler": "normal", "denoise": 1.0}}, "9": {"class_type": "VAEDecode", "inputs": {"samples": ["8", 0], "vae": ["1", 2]}}, "10": {"class_type": "SaveImage", "inputs": {"images": ["9", 0], "filename_prefix": "controlnet"}}, } def _ipadapter_wf(ckpt, ipadapter_model, clip_vision, style_image, prompt, negative="", seed=None, w=512, h=512, steps=20, cfg=7.0, weight=1.0): if seed is None: seed = random.randint(0, 2**63) # modern ComfyUI_IPAdapter_plus (0.33.x) uses IPAdapterUnifiedLoader + IPAdapter node # (the old IPAdapterApply / separate CLIPVisionLoader path is gone) preset = "PLUS (high strength)" if "plus" in ipadapter_model else "STANDARD (medium strength)" return { "1": {"class_type": "CheckpointLoaderSimple", "inputs": {"ckpt_name": ckpt}}, "2": {"class_type": "IPAdapterUnifiedLoader", "inputs": {"model": ["1", 0], "preset": preset}}, "3": {"class_type": "LoadImage", "inputs": {"image": style_image}}, "4": {"class_type": "CLIPTextEncode", "inputs": {"text": prompt, "clip": ["1", 1]}}, "5": {"class_type": "CLIPTextEncode", "inputs": {"text": negative, "clip": ["1", 1]}}, "6": {"class_type": "IPAdapter", "inputs": {"model": ["2", 0], "ipadapter": ["2", 1], "image": ["3", 0], "weight": weight, "weight_type": "style transfer", "start_at": 0.0, "end_at": 1.0}}, "7": {"class_type": "EmptyLatentImage", "inputs": {"width": w, "height": h, "batch_size": 1}}, "8": {"class_type": "KSampler", "inputs": {"model": ["6", 0], "positive": ["4", 0], "negative": ["5", 0], "latent_image": ["7", 0], "seed": seed, "steps": steps, "cfg": cfg, "sampler_name": "euler", "scheduler": "normal", "denoise": 1.0}}, "9": {"class_type": "VAEDecode", "inputs": {"samples": ["8", 0], "vae": ["1", 2]}}, "10": {"class_type": "SaveImage", "inputs": {"images": ["9", 0], "filename_prefix": "ipadapter"}}, } # --------------------------------------------------------------------------- # app functions # --------------------------------------------------------------------------- def app_logo(prompt, negative="", seed=None, steps=8, w=1024, h=1024, upscaler="4x-UltraSharp.pth"): """Brand mark / logo (SDXL DreamShaper -> 4x UltraSharp).""" return sdxl_txt2img("dreamshaper", prompt, negative, seed, w, h, steps, 2.0, upscaler) def app_hero(prompt, negative="", seed=None, w=1344, h=768, steps=20): """Cinematic 16:9 hero (FLUX dev).""" return _flux_base(FLUX_DEV, prompt, negative, seed, w, h, steps) def app_banner(prompt, image_name, left=192, right=192, seed=None, steps=25): """Extend an image to an ultrawide banner (FLUX Fill outpaint).""" return flux_outpaint(image_name, prompt, seed, steps, 0.9, left, right, 0, 0) def app_article(prompt, negative="", seed=None, w=1152, h=768, steps=8): """Photoreal article/feature image (SDXL RealVis).""" return sdxl_txt2img("realvis", prompt, negative, seed, w, h, steps, 2.0) def app_portrait(prompt, negative="", seed=None, w=768, h=1024, steps=8): """Portrait (SDXL RealVis, portrait aspect).""" return sdxl_txt2img("realvis", prompt, negative, seed, w, h, steps, 2.0) def app_upscale(image_name, upscaler="4x-UltraSharp.pth"): """4x/8x upscale any image (RealESRGAN / UltraSharp / Remacri / NMKD).""" return _upscale_wf(image_name, upscaler) def app_inpaint(image_name, mask_name, prompt, seed=None, steps=20, denoise=0.7): """Edit a region (FLUX Fill).""" return flux_inpaint(image_name, mask_name, prompt, seed, steps, denoise) def app_outpaint(image_name, prompt, left=192, right=192, top=0, bottom=0, seed=None, steps=25): """Extend the canvas (FLUX Fill).""" return flux_outpaint(image_name, prompt, seed, steps, 0.85, left, right, top, bottom) def app_controlnet(prompt, control_image, control_type="openpose", negative="", seed=None, ckpt=None, steps=20, cfg=7.0, strength=1.0): """Structure-guided generation (SD15 controlnet).""" ckpt = ckpt or SD15_BASE cns = {"openpose": "control_v11p_sd15_openpose.pth", "canny": "control_v11p_sd15_canny.pth", "depth": "control_v11f1p_sd15_depth.pth", "lineart": "control_v11p_sd15_lineart.pth", "tile": "control_v11f1e_sd15_tile.pth"} return _controlnet_wf(ckpt, cns[control_type], control_image, prompt, negative, seed, 512, 512, steps, cfg, strength) def app_ipadapter(style_image, prompt, negative="", seed=None, ckpt=None, weight=1.0, steps=20, cfg=7.0): """Style transfer (IPAdapter-plus SD15).""" ckpt = ckpt or SD15_BASE return _ipadapter_wf(ckpt, "ip-adapter-plus_sd15.safetensors", "CLIP-ViT-H-14.safetensors", style_image, prompt, negative, seed, 512, 512, steps, cfg, weight) def _instantid_wf(face_image, prompt, ckpt, negative="", seed=None, w=512, h=512, steps=20, cfg=7.0, ip_weight=0.8): if seed is None: seed = random.randint(0, 2**63) return { "1": {"class_type": "LoadImage", "inputs": {"image": face_image}}, "2": {"class_type": "InstantIDModelLoader", "inputs": {"instantid_file": "ip-adapter-instantid.bin"}}, "3": {"class_type": "InstantIDFaceAnalysis", "inputs": {"provider": "CPU"}}, "4": {"class_type": "ControlNetLoader", "inputs": {"control_net_name": "instantid-controlnet.safetensors"}}, "5": {"class_type": "CheckpointLoaderSimple", "inputs": {"ckpt_name": ckpt}}, "6": {"class_type": "CLIPTextEncode", "inputs": {"text": prompt, "clip": ["5", 1]}}, "7": {"class_type": "CLIPTextEncode", "inputs": {"text": negative, "clip": ["5", 1]}}, "8": {"class_type": "ApplyInstantID", "inputs": {"instantid": ["2", 0], "insightface": ["3", 0], "control_net": ["4", 0], "image": ["1", 0], "model": ["5", 0], "positive": ["6", 0], "negative": ["7", 0], "weight": ip_weight, "start_at": 0.0, "end_at": 1.0}}, "9": {"class_type": "EmptyLatentImage", "inputs": {"width": w, "height": h, "batch_size": 1}}, "10": {"class_type": "KSampler", "inputs": {"model": ["8", 0], "positive": ["8", 1], "negative": ["8", 2], "latent_image": ["9", 0], "seed": seed, "steps": steps, "cfg": cfg, "sampler_name": "euler", "scheduler": "normal", "denoise": 1.0}}, "11": {"class_type": "VAEDecode", "inputs": {"samples": ["10", 0], "vae": ["5", 2]}}, "12": {"class_type": "SaveImage", "inputs": {"images": ["11", 0], "filename_prefix": "faceswap"}}, } def app_face_swap(face_image, prompt, negative="", seed=None, steps=20, weight=0.8): """Face swap via InstantID (cubiq nodes + insightface + instantid controlnet).""" return _instantid_wf(face_image, prompt, "sd_xl_base_1.0.safetensors", negative, seed, 512, 512, steps, 7.0, weight) def app_restore_face(image_name, upscaler="4x-UltraSharp.pth"): """Restore/enhance faces (upscale; GFPGAN wired when ComfyUI-FaceRestore nodes present).""" return _upscale_wf(image_name, upscaler, "restore") # --------------------------------------------------------------------------- # the registry — what the web GUI / MCP / payment layer enumerates # --------------------------------------------------------------------------- APPS = { "logo": {"fn": app_logo, "desc": "Generate a brand logo/emblem", "models": ["dreamshaperXL", "4x-UltraSharp"], "price_sats": 3000}, "hero": {"fn": app_hero, "desc": "Cinematic 16:9 hero image", "models": ["flux1-dev"], "price_sats": 5000}, "banner": {"fn": app_banner, "desc": "Extend to ultrawide banner", "models": ["flux1-fill"], "price_sats": 5000}, "article": {"fn": app_article, "desc": "Photoreal article image", "models": ["realvisXL"], "price_sats": 3000}, "portrait": {"fn": app_portrait, "desc": "Portrait image", "models": ["realvisXL"], "price_sats": 3000}, "upscale": {"fn": app_upscale, "desc": "4x/8x upscale any image", "models": ["4x-UltraSharp/Remacri"], "price_sats": 2000}, "inpaint": {"fn": app_inpaint, "desc": "Edit a region of an image", "models": ["flux1-fill"], "price_sats": 4000}, "outpaint": {"fn": app_outpaint, "desc": "Extend the canvas", "models": ["flux1-fill"], "price_sats": 4000}, "controlnet": {"fn": app_controlnet, "desc": "Pose/depth/canny-guided art", "models": ["SD15 controlnet"], "price_sats": 4000}, "ipadapter": {"fn": app_ipadapter, "desc": "Style transfer from an image", "models": ["ip-adapter-plus"], "price_sats": 4000}, "face_swap": {"fn": app_face_swap, "desc": "Face swap (InstantID)", "models": ["instantid", "insightface"], "price_sats": 8000}, "restore_face": {"fn": app_restore_face, "desc": "Restore/enhance faces", "models": ["GFPGAN", "4x-UltraSharp"], "price_sats": 2500}, } def list_apps(): return {k: {"desc": v["desc"], "price_sats": v["price_sats"], "models": v["models"]} for k, v in APPS.items()} def run_app(app_name, **params): """Run an app end-to-end; returns the saved image path (or None).""" if app_name not in APPS: return None, f"unknown app: {app_name} (available: {', '.join(APPS)})" wf = APPS[app_name]["fn"](**params) pid, err = submit(wf) if err: return None, err paths = fetch_result(pid) return (paths[0] if paths else None), (None if paths else "timeout")