| 12 | |
| 13 | @comfy_node(name="LTXVLoadConditioning") |
| 14 | class LTXVLoadConditioning(io.ComfyNode): |
| 15 | @classmethod |
| 16 | def define_schema(cls) -> io.Schema: |
| 17 | files = folder_paths.get_filename_list("embeddings") |
| 18 | if not files: |
| 19 | files = [""] |
| 20 | return io.Schema( |
| 21 | node_id="LTXVLoadConditioning", |
| 22 | display_name="🅛🅣🅧 LTXV Load Conditioning", |
| 23 | category="lightricks/LTXV", |
| 24 | inputs=[ |
| 25 | io.Combo.Input("file_name", options=sorted(files)), |
| 26 | io.Combo.Input("device", options=["cpu", "gpu"]), |
| 27 | ], |
| 28 | outputs=[ |
| 29 | io.Conditioning.Output(), |
| 30 | ], |
| 31 | ) |
| 32 | |
| 33 | @classmethod |
| 34 | def execute(cls, file_name: str, device: str) -> io.NodeOutput: |
| 35 | file_path = folder_paths.get_full_path("embeddings", file_name) |
| 36 | if not Path(file_path).exists(): |
| 37 | raise FileNotFoundError(f"Conditioning file not found: {file_path}") |
| 38 | |
| 39 | target_device = "cpu" |
| 40 | if device == "gpu": |
| 41 | target_device = "cuda" if torch.cuda.is_available() else "cpu" |
| 42 | |
| 43 | conditioning: list[list[Any]] = [] |
| 44 | |
| 45 | with safetensors.safe_open( |
| 46 | file_path, framework="pt", device=target_device |
| 47 | ) as f: |
| 48 | tensor_keys = [k for k in f.keys() if k.startswith("conditioning_data_")] |
| 49 | |
| 50 | for tensor_key in sorted(tensor_keys): |
| 51 | idx = tensor_key.replace("conditioning_data_", "") |
| 52 | tensor = f.get_tensor(tensor_key) |
| 53 | |
| 54 | options: dict[str, Any] = {} |
| 55 | mask_key = f"attention_mask_{idx}" |
| 56 | if mask_key in f.keys(): |
| 57 | options["attention_mask"] = f.get_tensor(mask_key) |
| 58 | |
| 59 | conditioning.append([tensor, options]) |
| 60 | |
| 61 | if not conditioning: |
| 62 | raise ValueError(f"No conditioning data found in file: {file_name}") |
| 63 | |
| 64 | return io.NodeOutput(conditioning) |
| 65 | |
| 66 | @classmethod |
| 67 | def fingerprint_inputs(cls, file_name: str, device: str) -> str: |
| 68 | file_path = folder_paths.get_full_path("embeddings", file_name) |
| 69 | with open(file_path, "rb") as f: |
| 70 | return hashlib.sha256(f.read()).hexdigest() |
| 71 |
nothing calls this directly
no outgoing calls
no test coverage detected