(workflow: str, *, host: str, timeout: float)
| 451 | |
| 452 | |
| 453 | def _load_comfy_workflow(workflow: str, *, host: str, timeout: float) -> dict[str, Any]: |
| 454 | path = Path(workflow).expanduser() |
| 455 | if path.exists(): |
| 456 | try: |
| 457 | data = json.loads(path.read_text(encoding="utf-8")) |
| 458 | except json.JSONDecodeError as exc: |
| 459 | raise ModlyCliError(f"workflow must be valid JSON: {path}: {exc}") from exc |
| 460 | if not isinstance(data, dict): |
| 461 | raise ModlyCliError(f"workflow JSON must be an object: {path}") |
| 462 | return data |
| 463 | |
| 464 | candidates = [workflow, f"{workflow}.json"] if not workflow.endswith(".json") else [workflow] |
| 465 | for name in candidates: |
| 466 | quoted = urllib.parse.quote(name.lstrip("/"), safe="/") |
| 467 | for prefix in ("/userdata/workflows/", "/api/userdata/workflows/", "/userdata/", "/api/userdata/"): |
| 468 | try: |
| 469 | data = _request_json("GET", f"{host.rstrip('/')}{prefix}{quoted}", timeout=timeout) |
| 470 | except ModlyCliError: |
| 471 | continue |
| 472 | if isinstance(data, dict): |
| 473 | return data |
| 474 | |
| 475 | search_roots: list[Path] = [] |
| 476 | for value in [os.environ.get("COMFYUI_WORKFLOW_DIR"), os.environ.get("COMFYUI_USER_DIR")]: |
| 477 | if value: |
| 478 | search_roots.append(Path(value).expanduser()) |
| 479 | search_roots.extend([ |
| 480 | Path.home() / "ComfyUI" / "user" / "default" / "workflows", |
| 481 | Path.home() / "Documents" / "ComfyUI" / "user" / "default" / "workflows", |
| 482 | ]) |
| 483 | for appdata in _windows_env_paths("APPDATA"): |
| 484 | search_roots.extend([ |
| 485 | appdata / "ComfyUI" / "user" / "default" / "workflows", |
| 486 | appdata / "comfyui" / "user" / "default" / "workflows", |
| 487 | ]) |
| 488 | for root in search_roots: |
| 489 | for name in candidates: |
| 490 | candidate = root / name |
| 491 | if candidate.exists(): |
| 492 | try: |
| 493 | data = json.loads(candidate.read_text(encoding="utf-8")) |
| 494 | except json.JSONDecodeError as exc: |
| 495 | raise ModlyCliError(f"workflow must be valid JSON: {candidate}: {exc}") from exc |
| 496 | if isinstance(data, dict): |
| 497 | return data |
| 498 | raise ModlyCliError(f"Could not find ComfyUI workflow '{workflow}'. Pass a JSON path or set COMFYUI_WORKFLOW_DIR.") |
| 499 | |
| 500 | |
| 501 | def _patch_comfy_workflow(workflow: dict[str, Any], *, prompt: str | None, seed: int | None) -> dict[str, Any]: |
no test coverage detected