/ludwig/model_inspector.py

"""Model introspection utilities.

Extracted from LudwigModel to reduce the god object. Provides:
- Weight collection
- Activation collection
- Schema generation
- Model summary
"""

import logging
from typing import Any

import torch

logger = logging.getLogger(__name__)


class ModelInspector:
    """Inspect and analyze a trained Ludwig model."""

    def __init__(self, model: torch.nn.Module, config: dict, training_set_metadata: dict):
        self.model = model
        self.config = config
        self.training_set_metadata = training_set_metadata

    def collect_weights(self, tensor_names: list[str] | None = None) -> list[dict[str, Any]]:
        """Collect model weight tensors.

        Args:
            tensor_names: Specific parameter names to collect. None for all.

        Returns:
            List of dicts with 'name', 'shape', 'dtype', 'values' keys.
        """
        results = []
        for name, param in self.model.named_parameters():
            if tensor_names is None or name in tensor_names:
                results.append(
                    {
                        "name": name,
                        "shape": list(param.shape),
                        "dtype": str(param.dtype),
                        "requires_grad": param.requires_grad,
                        "num_elements": param.numel(),
                    }
                )
        return results

    def model_summary(self) -> dict[str, Any]:
        """Generate a summary of the model architecture.

        Returns:
            Dict with parameter counts, layer info, and feature details.
        """
        total_params = sum(p.numel() for p in self.model.parameters())
        trainable_params = sum(p.numel() for p in self.model.parameters() if p.requires_grad)
        frozen_params = total_params - trainable_params

        # Model size in MB
        model_size_mb = sum(p.numel() * p.element_size() for p in self.model.parameters()) / (1024 * 1024)

        # Count layers by type
        layer_counts = {}
        for module in self.model.modules():
            class_name = type(module).__name__
            layer_counts[class_name] = layer_counts.get(class_name, 0) + 1

        return {
            "total_parameters": total_params,
            "trainable_parameters": trainable_params,
            "frozen_parameters": frozen_params,
            "model_size_mb": round(model_size_mb, 2),
            "layer_counts": layer_counts,
            "model_type": self.config.get("model_type", "ecd"),
            "combiner_type": self.config.get("combiner", {}).get("type", "concat"),
            "num_input_features": len(self.config.get("input_features", [])),
            "num_output_features": len(self.config.get("output_features", [])),
        }

    def feature_importance_proxy(self) -> dict[str, float]:
        """Estimate feature importance from encoder weight magnitudes.

        This is a rough proxy, not a rigorous importance measure. For proper
        feature importance, use SHAP or Captum via Ludwig's explain module.

        Returns:
            Dict mapping feature names to relative importance scores.
        """
        importance = {}
        if hasattr(self.model, "input_features"):
            for name, feature in self.model.input_features.items():
                total_weight_magnitude = 0.0
                for param in feature.parameters():
                    total_weight_magnitude += param.abs().mean().item()
                importance[name] = total_weight_magnitude

        # Normalize to 0-1 range
        if importance:
            max_imp = max(importance.values())
            if max_imp > 0:
                importance = {k: v / max_imp for k, v in importance.items()}

        return importance