comfyui

1#!/usr/bin/env python32"""3health_check.py — One-stop verification that the ComfyUI environment is ready.4 5Runs through the verification checklist:6  1. comfy-cli on PATH7  2. server reachable (/system_stats)8  3. at least one checkpoint installed9  4. (optional) a specific workflow's deps are met10  5. (optional) actually submit a tiny test workflow and verify round-trip11 12Usage:13    python3 health_check.py14    python3 health_check.py --host https://cloud.comfy.org15    python3 health_check.py --workflow my.json16    python3 health_check.py --smoke-test    # actually submit a tiny workflow17"""18 19from __future__ import annotations20 21import argparse22import json23import shutil24import sys25from pathlib import Path26 27sys.path.insert(0, str(Path(__file__).resolve().parent))28from _common import (  # noqa: E40229    DEFAULT_LOCAL_HOST, ENV_API_KEY, emit_json, http_get, parse_model_list,30    resolve_api_key, resolve_url, unwrap_workflow,31)32 33 34def comfy_cli_status() -> dict:35    if shutil.which("comfy"):36        return {"available": True, "method": "comfy", "path": shutil.which("comfy")}37    if shutil.which("uvx"):38        return {"available": True, "method": "uvx",39                "hint": "Invoke as `uvx --from comfy-cli comfy ...`"}40    return {41        "available": False,42        "hint": "Install with: pipx install comfy-cli (or `pip install comfy-cli`)",43    }44 45 46def server_status(host: str, headers: dict) -> dict:47    url = resolve_url(host, "/system_stats")48    try:49        r = http_get(url, headers=headers, retries=2, timeout=10)50        if r.status == 200:51            try:52                stats = r.json() or {}53            except Exception:54                stats = {}55            return {"reachable": True, "url": url, "stats": stats}56        return {"reachable": False, "url": url, "http_status": r.status, "body": r.text()[:200]}57    except Exception as e:58        return {"reachable": False, "url": url, "error": str(e)}59 60 61def checkpoint_status(host: str, headers: dict) -> dict:62    url = resolve_url(host, "/models/checkpoints")63    try:64        r = http_get(url, headers=headers, retries=2, timeout=15)65    except Exception as e:66        return {"queryable": False, "error": str(e)}67    if r.status != 200:68        return {"queryable": False, "http_status": r.status, "url": url, "body": r.text()[:200]}69    try:70        models = parse_model_list(r.json())71    except Exception:72        models = set()73    return {"queryable": True, "count": len(models),74            "first_few": sorted(models)[:5]}75 76 77SMOKE_WORKFLOW = {78    # Minimal SD1.5 workflow that doesn't depend on rare nodes.79    # 256x256 + 1 step is the smallest config that doesn't trigger SDXL/Flux80    # validation errors while still executing fast.81    "3": {82        "class_type": "KSampler",83        "inputs": {84            "seed": 1, "steps": 1, "cfg": 7.0,85            "sampler_name": "euler", "scheduler": "normal", "denoise": 1.0,86            "model": ["4", 0], "positive": ["6", 0], "negative": ["7", 0],87            "latent_image": ["5", 0],88        },89    },90    "4": {"class_type": "CheckpointLoaderSimple",91          "inputs": {"ckpt_name": "REPLACE_ME"}},92    "5": {"class_type": "EmptyLatentImage",93          "inputs": {"width": 256, "height": 256, "batch_size": 1}},94    "6": {"class_type": "CLIPTextEncode",95          "inputs": {"text": "test", "clip": ["4", 1]}},96    "7": {"class_type": "CLIPTextEncode",97          "inputs": {"text": "", "clip": ["4", 1]}},98    "9": {"class_type": "SaveImage",99          "inputs": {"filename_prefix": "smoke", "images": ["3", 0]}},100}101 102 103def smoke_test(host: str, headers: dict, ckpt_name: str | None) -> dict:104    """Submit a tiny workflow and verify the server accepts it.105 106    Cancels the job immediately after acceptance so we don't burn GPU107    time / cloud minutes on a smoke test.108    """109    if not ckpt_name:110        return {"ran": False, "reason": "no checkpoint available"}111    wf = json.loads(json.dumps(SMOKE_WORKFLOW))112    wf["4"]["inputs"]["ckpt_name"] = ckpt_name113 114    # Lazy import to avoid circular issues115    from run_workflow import ComfyRunner116    api_key = headers.get("X-API-Key")117    runner = ComfyRunner(host=host, api_key=api_key)118    sub = runner.submit(wf)119    if "_http_error" in sub:120        return {"ran": True, "submitted": False,121                "http_status": sub["_http_error"], "body": sub.get("body")}122    pid = sub.get("prompt_id")123    if not pid:124        return {"ran": True, "submitted": False, "response": sub}125 126    # Cancel so we don't actually waste compute on the smoke test.127    cancelled = False128    try:129        cancelled = runner.cancel(pid)130    except Exception:131        pass132 133    return {134        "ran": True, "submitted": True, "prompt_id": pid,135        "cancelled_after_submit": cancelled,136        "note": "Submission accepted; cancelled to avoid running the full pipeline.",137    }138 139 140def main(argv: list[str] | None = None) -> int:141    p = argparse.ArgumentParser(description="One-stop ComfyUI health check")142    p.add_argument("--host", default=DEFAULT_LOCAL_HOST)143    p.add_argument("--api-key", help=f"or set ${ENV_API_KEY}")144    p.add_argument("--workflow", help="Optional: also run check_deps on this workflow")145    p.add_argument("--smoke-test", action="store_true",146                   help="Submit a tiny test workflow and verify round-trip")147    p.add_argument("--strict", action="store_true",148                   help="Exit non-zero on any non-pass condition (including warnings)")149    args = p.parse_args(argv)150 151    api_key = resolve_api_key(args.api_key)152    headers = {"X-API-Key": api_key} if api_key else {}153 154    cli = comfy_cli_status()155    server = server_status(args.host, headers)156    ckpts = checkpoint_status(args.host, headers) if server.get("reachable") else None157 158    # ---- workflow check ----159    workflow_check: dict | None = None160    if args.workflow:161        wf_path = Path(args.workflow).expanduser()162        if not wf_path.exists():163            workflow_check = {"error": "workflow file not found"}164        else:165            try:166                with wf_path.open(encoding="utf-8-sig") as f:167                    workflow = unwrap_workflow(json.load(f))168                from check_deps import check_deps169                workflow_check = check_deps(workflow, host=args.host, api_key=api_key)170            except (ValueError, json.JSONDecodeError) as e:171                workflow_check = {"error": str(e)}172 173    smoke = None174    if args.smoke_test and server.get("reachable"):175        first_ckpt = ckpts["first_few"][0] if ckpts and ckpts.get("first_few") else None176        smoke = smoke_test(args.host, headers, first_ckpt)177 178    # ---- verdict ----179    verdict = "pass"180    reasons: list[str] = []181    if not server.get("reachable"):182        verdict = "fail"183        reasons.append("server unreachable")184    if ckpts and ckpts.get("queryable") and ckpts.get("count", 0) == 0:185        verdict = "warn" if verdict == "pass" else verdict186        reasons.append("no checkpoints installed")187    if workflow_check and workflow_check.get("error"):188        verdict = "fail"189        reasons.append(f"workflow check failed: {workflow_check['error']}")190    elif workflow_check and not workflow_check.get("is_ready"):191        if workflow_check.get("node_check_skipped"):192            reasons.append("node check skipped (cloud free tier)")193        else:194            verdict = "fail"195            reasons.append("workflow has missing deps")196    if smoke and smoke.get("ran") and not smoke.get("submitted"):197        verdict = "fail"198        reasons.append("smoke-test submission failed")199    if not cli.get("available"):200        verdict = "warn" if verdict == "pass" else verdict201        reasons.append("comfy-cli not on PATH (lifecycle commands won't work)")202 203    report = {204        "verdict": verdict,205        "reasons": reasons,206        "host": args.host,207        "comfy_cli": cli,208        "server": server,209        "checkpoints": ckpts,210        "workflow_check": workflow_check,211        "smoke_test": smoke,212    }213    emit_json(report)214 215    if verdict == "pass":216        return 0217    if verdict == "warn":218        return 1 if args.strict else 0219    return 1220 221 222if __name__ == "__main__":223    sys.exit(main())224