| import os as _os |
| HERE = _os.path.dirname(_os.path.abspath(__file__)) |
| ROOT = _os.path.dirname(HERE) |
| """Stand-ins for the parts of ComfyUI the pack imports, so its code can run here. Not ComfyUI: they do the minimum.""" |
| import sys, types, os, importlib.util, contextlib |
| import torch, torch.nn as nn |
| from safetensors.torch import load_file, safe_open |
|
|
| def mod(name, **attrs): |
| m = types.ModuleType(name); m.__dict__.update(attrs); sys.modules[name] = m; return m |
|
|
| class Any: |
| def __init__(self, *a, **k): self.a, self.k = a, k |
| class Inp(Any): |
| def __init__(self, kind, *a, **k): super().__init__(*a, **k); self.kind = kind |
| class Custom: |
| def __init__(self, name): self.name = name |
| def Output(self, *a, **k): return Any() |
| def Input(self, *a, **k): return Inp("custom:" + self.name, *a, **k) |
| def kind(name): return types.SimpleNamespace(Input=lambda *a, **k: Inp(name, *a, **k), Output=Any) |
| class NodeOutput: |
| def __init__(self, *args, ui=None): self.args, self.ui = args, ui |
| class ComfyNode: pass |
| io = types.SimpleNamespace(Custom=Custom, Schema=Any, ComfyNode=ComfyNode, NodeOutput=NodeOutput, Combo=kind("COMBO"), Int=kind("INT"), |
| Float=kind("FLOAT"), String=kind("STRING"), Boolean=kind("BOOLEAN"), Audio=kind("AUDIO")) |
| mod("comfy_api"); mod("comfy_api.latest", ComfyExtension=object, io=io) |
|
|
| state = {"progress": [], "loaded": 0} |
| class ProgressBar: |
| def __init__(self, total): self.total = total |
| def update_absolute(self, v): state["progress"].append((v, self.total)) |
| def load_torch_file(path, return_metadata=False): |
| t = load_file(path) |
| if not return_metadata: return t |
| with safe_open(path, "pt") as f: return t, f.metadata() |
| class ModelPatcher: |
| def __init__(self, model, load_device, offload_device): self.model, self.load_device, self.offload_device = model, load_device, offload_device |
| cpu = torch.device("cpu") |
| mm = mod("comfy.model_management", get_torch_device=lambda: cpu, unet_offload_device=lambda: cpu, vae_offload_device=lambda: cpu, |
| load_models_gpu=lambda ps, **k: state.__setitem__("loaded", state["loaded"] + 1), |
| throw_exception_if_processing_interrupted=lambda: None, archive_model_dtypes=lambda m: None, |
| cuda_device_context=lambda d: contextlib.nullcontext()) |
| mp = mod("comfy.model_patcher", ModelPatcher=ModelPatcher) |
| cu = mod("comfy.utils", ProgressBar=ProgressBar, load_torch_file=load_torch_file) |
| def resample(w, sr, to): return torch.nn.functional.interpolate(w[None], size=int(w.shape[-1] * to / sr), mode="linear")[0] |
| ca = mod("comfy.audio", resample=resample) |
|
|
| class Embedding(nn.Embedding): |
| def forward(self, x, out_dtype=None): |
| y = super().forward(x); return y if out_dtype is None else y.to(out_dtype) |
| ops = mod("comfy.ops", manual_cast=types.SimpleNamespace(Conv1d=nn.Conv1d, ConvTranspose1d=nn.ConvTranspose1d, Linear=nn.Linear, |
| LayerNorm=nn.LayerNorm, Embedding=Embedding)) |
| def apply_rope1(x, freqs_cis): |
| x_ = x.to(dtype=freqs_cis.dtype).reshape(*x.shape[:-1], -1, 1, 2) |
| out = freqs_cis[..., 0] * x_[..., 0] |
| out.addcmul_(freqs_cis[..., 1], x_[..., 1]) |
| return out.reshape(*x.shape).type_as(x) |
| class Snake1d(nn.Module): |
| def __init__(self, channels, dtype=None, device=None): |
| super().__init__(); self.alpha = nn.Parameter(torch.ones(1, channels, 1, dtype=dtype, device=device)) |
| def forward(self, x): return x + (self.alpha + 1e-9).reciprocal() * torch.sin(self.alpha * x).pow(2) |
| mod("comfy.ldm"); mod("comfy.ldm.flux"); mod("comfy.ldm.flux.math", apply_rope1=apply_rope1) |
| mod("comfy.ldm.minimax_music"); mod("comfy.ldm.minimax_music.dav", Snake1d=Snake1d) |
| mod("comfy.ldm.modules"); mod("comfy.ldm.modules.attention", optimized_attention_for_device=lambda *a, **k: None) |
| mod("comfy.text_encoders"); |
| llama = mod("comfy.text_encoders.llama", Qwen3_06B=object, Qwen3_06BConfig=object, __file__="/nonexistent/llama.py") |
| sys.modules["comfy.text_encoders"].llama = llama |
| comfy = mod("comfy", model_management=mm, model_patcher=mp, utils=cu, audio=ca, ops=ops) |
| OUT = HERE + "/out"; os.makedirs(OUT + "/temp", exist_ok=True) |
| fp = mod("folder_paths", models_dir=HERE + "/models", folder_names_and_paths={}, supported_pt_extensions={".safetensors"}, |
| add_model_folder_path=lambda n, p: fp.folder_names_and_paths.__setitem__(n, ([p], set())), |
| get_filename_list=lambda n: [], get_full_path_or_raise=lambda n, f: f, get_output_directory=lambda: OUT, |
| get_temp_directory=lambda: OUT + "/temp", |
| get_save_image_path=lambda prefix, folder: (os.makedirs(os.path.dirname(os.path.join(folder, prefix)) or folder, exist_ok=True) or |
| os.path.dirname(os.path.join(folder, prefix)), os.path.basename(prefix), 1, "", prefix)) |
| class _Web: |
| @staticmethod |
| def json_response(x): return x |
| mod("aiohttp", web=_Web); sys.modules["aiohttp"].web = _Web |
| mod("server", PromptServer=types.SimpleNamespace(instance=None)) |
|
|
| def load_pack(): |
| spec = importlib.util.spec_from_file_location("ogg_pack", ROOT + "/__init__.py", submodule_search_locations=[ROOT]) |
| pack = importlib.util.module_from_spec(spec); sys.modules["ogg_pack"] = pack; spec.loader.exec_module(pack) |
| return pack, sys.modules["ogg_pack.nodes"] |
|
|
