activation_dump_node/capture_activations_v2.py

"""star_capture_activations_v2.py"""
import os
import torch
import folder_paths

CAPTURE_DIR = os.path.join(folder_paths.get_output_directory(), "star_activations")
os.makedirs(CAPTURE_DIR, exist_ok=True)

TARGET_SUFFIXES = (
    "attn.wq", "attn.wk", "attn.wv", "attn.wo", "attn.gate",
    "mlp.gate", "mlp.up", "mlp.down",
)

class StarCaptureActivationsV2:
    @classmethod
    def INPUT_TYPES(s):
        return {
            "required": {
                "model": ("MODEL",),
                "run_tag": ("STRING", {"default": "capture1"}),
                "max_calls_per_layer": ("INT", {"default": 3, "min": 1, "max": 50}),
            }
        }

    RETURN_TYPES = ("MODEL",)
    FUNCTION = "wrap"
    CATEGORY = "⭐StarNodes/Model Tools"

    def wrap(self, model, run_tag, max_calls_per_layer):
        real_model = model.model
        target = real_model.diffusion_model if hasattr(real_model, "diffusion_model") else real_model
        out_dir = os.path.join(CAPTURE_DIR, run_tag)
        os.makedirs(out_dir, exist_ok=True)

        counters = {}
        handles = []

        def make_hook(name):
            safe_name = name.replace(".", "_")
            def hook(module, inputs, output):
                c = counters.get(name, 0)
                if c >= max_calls_per_layer:
                    return
                counters[name] = c + 1
                x = inputs[0].detach().to("cpu", dtype=torch.float16).contiguous()
                torch.save(x, os.path.join(out_dir, f"{safe_name}__call{c}.pt"))
            return hook

        matched = 0
        for name, module in target.named_modules():
            # match by name suffix + presence of a 2D weight param, not isinstance
            has_2d_weight = hasattr(module, "weight") and getattr(module.weight, "ndim", 0) == 2
            if has_2d_weight and any(name.endswith(s) for s in TARGET_SUFFIXES):
                h = module.register_forward_hook(make_hook(name))
                handles.append(h)
                matched += 1

        print(f"⭐ StarCaptureActivationsV2: hooked {matched} layers -> {out_dir}")
        model._star_capture_handles = handles
        return (model,)


NODE_CLASS_MAPPINGS = {"StarCaptureActivationsV2": StarCaptureActivationsV2}
NODE_DISPLAY_NAME_MAPPINGS = {"StarCaptureActivationsV2": "⭐ Star Capture Activations V2"}