(cls, file_name: str, device: str)
| 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: |
nothing calls this directly
no outgoing calls
no test coverage detected