model/nn/model_modifier_utils.py

# This file is a part of the `nequip` package. Please see LICENSE and README at the root for information on using it.
import torch
from typing import Final, Callable, Optional, List


# NOTE: persistent modifiers are modifiers that fundamentally change the behavior of the model (same input will lead to different outputs)
# non-persistent modifiers generally refer to accelerations that should preserve similar model behavior, with the only difference being speed
_MODEL_MODIFIER_PERSISTENT_ATTR_NAME: Final[str] = (
    "_nequip_model_modifier_is_persistent"
)
_MODEL_MODIFIER_PRIVATE_ATTR_NAME: Final[str] = "_nequip_model_modifier_is_private"

# these latter two attributes (unsupported devices and supported compile modes) are meant for acceleration modifiers
_MODEL_MODIFIER_UNSUPPORTED_DEVICES_ATTR_NAME: Final[str] = (
    "_nequip_model_modifier_unsupported_devices"
)
_MODEL_MODIFIER_SUPPORTED_COMPILE_MODES_ATTR_NAME: Final[str] = (
    "_nequip_model_modifier_supported_compile_modes"
)


def model_modifier(
    persistent: bool,
    private: Optional[bool] = None,
    unsupported_devices: List[str] = [],
    supported_compile_modes: Optional[List[str]] = None,
):
    """
    Mark a ``@classmethod`` of an ``nn.Module`` as a "model modifier" that can be applied by the user to modify a packaged or other loaded model on-the-fly. Model modifiers must be a ``@classmethod`` of one of the ``nn.Module`` objects in the model.

    Args:
        persistent (bool): Whether the modifier should be applied when building the model for packaging.
        private (bool, optional): Whether the modifier is private and should not be exposed in public interfaces. Defaults to None.
        unsupported_devices (List[str], optional): List of device types that this modifier does not support. Defaults to [].
        supported_compile_modes (List[str], optional): List of compile modes that this modifier supports. Defaults to None.
    """

    def decorator(func):
        assert isinstance(func, classmethod), (
            "@model_modifier must be applied after @classmethod"
        )
        assert not hasattr(func.__func__, _MODEL_MODIFIER_PERSISTENT_ATTR_NAME)

        setattr(func.__func__, _MODEL_MODIFIER_PERSISTENT_ATTR_NAME, persistent)

        if private is not None:
            setattr(func.__func__, _MODEL_MODIFIER_PRIVATE_ATTR_NAME, private)

        setattr(
            func.__func__,
            _MODEL_MODIFIER_UNSUPPORTED_DEVICES_ATTR_NAME,
            unsupported_devices,
        )

        setattr(
            func.__func__,
            _MODEL_MODIFIER_SUPPORTED_COMPILE_MODES_ATTR_NAME,
            supported_compile_modes,
        )

        return func

    return decorator


def is_model_modifier(func: callable) -> bool:
    # for backwards compatibility, we use the "persistent" flag as a marker for whether the method is a model modifier
    return hasattr(func, _MODEL_MODIFIER_PERSISTENT_ATTR_NAME)


def is_persistent_model_modifier(func: callable) -> bool:
    return getattr(func, _MODEL_MODIFIER_PERSISTENT_ATTR_NAME)


def is_private_model_modifier(func: callable) -> Optional[bool]:
    # for backwards compatibility of packaged models whose modifier would not have this metadata entry,
    # we just default to making it public for convenience of clients
    # should be ok since this mechanism is not safety critical and more just a convenience for documenting modifiers
    return getattr(func, _MODEL_MODIFIER_PRIVATE_ATTR_NAME, False)


def get_model_modifier_unsupported_devices(func: callable) -> List[str]:
    """Get the list of unsupported devices for a model modifier. Returns empty list for backwards compatibility."""
    return getattr(func, _MODEL_MODIFIER_UNSUPPORTED_DEVICES_ATTR_NAME, [])


def get_model_modifier_supported_compile_modes(func: callable) -> Optional[List[str]]:
    """Get the list of supported compile modes for a model modifier. Returns None if not set."""
    return getattr(func, _MODEL_MODIFIER_SUPPORTED_COMPILE_MODES_ATTR_NAME, None)


def replace_submodules(
    model: torch.nn.Module,
    target_cls: type,
    factory: Callable[[torch.nn.Module], torch.nn.Module],
) -> torch.nn.Module:
    """
    Recursively walk the children of ``model``, and whenever we see an instance of ``target_cls``, replace it (in-place) with ``factory(old_module)`` by mutating ``model._modules[name]``.
    """
    for name, child in list(model.named_children()):
        if isinstance(child, target_cls):
            # build a brand-new one based on `factory`
            model._modules[name] = factory(child)
        else:
            # recurse down
            replace_submodules(child, target_cls, factory)
    return model