
    qi6                     B   U d dl Z d dlZd dlZd dlmZ ddlmZmZ ddlm	Z	  e       rd dl
mZ  ej                  e      Zi aeeeej&                     f   ed<   i Zeee j,                  f   ed<    ej.                         Zded	e j,                  dz  fd
Zdedeeeej&                     f   d	eej&                     dz  fdZddeeeej&                     f   ded	dfdZdee   d	dfdZd	eeeej&                     f   fdZddZ ed        Z!dej&                  d	dfdZ"y)    N)contextmanager   )is_torch_availablelogging)OutputRecorder_monkey_patch_mapping_cache_compiled_patterns_cachepatternreturnc                     | t         v r	t         |    S 	 t        j                  |       }|t         | <   |S # t        j                  $ r&}t        j                  d|  d| d       Y d}~yd}~ww xY w)z
    Compile a regex pattern and cache it. Returns None if pattern is invalid.

    Args:
        pattern: The regex pattern string to compile

    Returns:
        Compiled regex pattern or None if invalid
    zInvalid regex pattern 'z': z. Treating as non-pattern.N)r	   recompileerrorloggerwarning)r
   compiledes      N/opt/pipecat/venv/lib/python3.12/site-packages/transformers/monkey_patching.py_compile_patternr   "   so     **'00::g&,4 )88 0	QC?YZ[s   3 A,A''A,
class_namemappingc                     | |v r||    S |j                         D ].  \  }}|| k(  rt        |      }||j                  |       s,|c S  y)a=  
    Find replacement class for a given class name, checking exact matches first, then regex patterns.

    Args:
        class_name: The class name to find a replacement for
        mapping: Dictionary of patterns/names to replacement classes

    Returns:
        The replacement class if found, None otherwise
    N)itemsr   search)r   r   r
   replacement_classcompiled_patterns        r   _find_replacement_classr   8   sk     Wz"" '.mmo %""j  ,G4',<,C,CJ,O$$%     	overwritec                    t         5  | j                         D ]  \  }}t        |t              s$t	        d| dt        |      j
                         t        |t        j                        sCt	        d| d|j
                   d|j                  dd D cg c]  }|j
                   c}       |t        v r|st        d| d	      |t        |<    	 ddd       yc c}w # 1 sw Y   yxY w)
u,	  
    Register patch mappings to enable automatic patching during model creation using `from_pretrained`,
    `from_config` or within the `apply_patches` context manager.

    Use this to register class replacements that will be automatically applied when loading any model.
    This is useful for quantization library compatibility, structural optimizations, and architectural
    experimentation. The mapping is global, can grow with multiple calls, and can be cleared entirely.

    Args:
        mapping (`Dict[str, type[nn.Module]]`):
            Mapping from original class names (or regex patterns) to replacement classes. Supports:
            - Exact class names: `"Qwen2MoeExperts"` → `CustomExperts`
            - Regex patterns: `".*Attention"` matches `LlamaAttention`, `MistralAttention`, etc.,
            or `"^Llama\d+Attention$"` matches `Llama2Attention`, `Llama3Attention`, etc.

            Exact matches take precedence over patterns. Patterns are matched using `re.search()`,
            so they can match anywhere in the class name unless you use anchors (`^` for start, `$` for end).
        overwrite (`bool`, *optional*, defaults to `False`):
            Whether to overwrite existing mappings for class names that are already registered.

    Example:
        ```python
        from transformers import AutoModelForCausalLM
        from transformers.monkey_patching import register_patch_mapping

        # Define custom expert implementation
        class SequentialExperts(nn.Module):
            ...

        # Register exact class name
        register_patch_mapping(
            mapping={"Qwen2MoeExperts": SequentialExperts}
        )

        # Register with regex pattern to match multiple classes
        register_patch_mapping(
            mapping={".*Attention": CustomAttention}  # Matches LlamaAttention, MistralAttention, etc.
        )

        # Match specific model versions
        register_patch_mapping(
            mapping={"^Llama\d+Attention$": CustomLlamaAttention}  # Matches Llama2Attention, Llama3Attention
        )

        # The patch will be automatically applied during loading
        model = AutoModelForCausalLM.from_pretrained("meta-llama/Llama-3.2-1B")
        ```

    Note:
        For weight conversions, use [`~transformers.register_checkpoint_conversion_mapping`] instead.
    zReplacement for 'z' must be a class, got zReplacement class for 'z'' must be a subclass of nn.Module, got z which inherits from r   NzClass 'zK' already has a patch mapping registered. Use overwrite=True to replace it.)_monkey_patch_lockr   
isinstancetype	TypeError__name__
issubclassnnModule__mro__r   
ValueError)r   r   r   r   cs        r   register_patch_mappingr,   U   s   j 
 H-4]]_ 	H)J)/6'
|3J4PaKbKkKkJlm  /;-j\ :,5566Karazaz{|{}a~L\]QZZL  LAB 
 88 j\)tu  7H'
3!	HH H M@H Hs   BC"C&-C"C""C+keysc                     t         5  | D ]   }|t        vrt        d| d      t        |= " 	 ddd       y# 1 sw Y   yxY w)a  
    Unregister patch mappings to disable automatic patching.

    This removes specified mappings from the global registry, preventing them from being applied
    during model loading. You must provide the exact same name or pattern that was used during registration.

    Args:
        keys (`List[str]`):
            List of mapping keys (class names or regex patterns) to remove from the patch mapping
            (e.g., `["Qwen2MoeExperts"]` or `[".*Attention"]`).

    Example:
        ```python
        from transformers import AutoModelForCausalLM
        from transformers.monkey_patching import register_patch_mapping, unregister_patch_mapping

        # Register a patch
        register_patch_mapping(
            mapping={"Qwen2MoeExperts": CustomExperts}
        )

        # Unregister the patch
        unregister_patch_mapping(["Qwen2MoeExperts"])

        # The patch will no longer be applied during loading
        model = AutoModelForCausalLM.from_pretrained("Qwen/Qwen1.5-MoE-A2.7B")
        ```
    zClass or pattern 'z\' not found in monkey patch mapping cache. Cannot unregister a class that is not registered.N)r!   r   r*   )r-   keys     r   unregister_patch_mappingr0      s^    < 
 1 	1C55 ( .H I  ,C0	11 1 1s	   &7A c                  b    t         5  t        j                         cddd       S # 1 sw Y   yxY w)z
    Get all registered patch mappings.

    Returns:
        `Dict[str, type[nn.Module]]`: Dictionary mapping class names or patterns to replacement classes.
    N)r!   r   copy r   r   get_patch_mappingr4      s(     
 2*//12 2 2s   %.c                  b    t         5  t        j                          ddd       y# 1 sw Y   yxY w)a  
    Clear all registered patch mappings.

    This removes all registered mappings from the global registry.

    Example:
        ```python
        from transformers.monkey_patching import register_patch_mapping, clear_patch_mapping

        # Register some patches
        register_patch_mapping(
            mapping={"Qwen2MoeExperts": CustomExperts}
        )

        # Clear all patches
        clear_patch_mapping()
        ```
    N)r!   r   clearr3   r   r   clear_patch_mappingr7      s(    ( 
 ,#))+, , ,s   %.c               #   H  K   t               } | sd yi }t        t        j                  j	                               D ]  }|t        |d      s|j                  j                  d      s.t        |      D ]M  }t        ||       }|	 t        ||      }t        |t              s0|||j                  |f<   t        |||       O  d |j#                         D ](  \  \  }}}t        j                  |   }t        |||       * y# t        t        t         f$ r Y w xY ww)a  
    Context manager to apply registered monkey patches within a block of code.

    This temporarily replaces original classes with their registered replacements during the execution of the block, and restores the original classes afterward.

    Example:
        ```python
        from transformers import Qwen2MoeModel, Qwen2MoeConfig
        from transformers.monkey_patching import register_patch_mapping, apply_patches

        # Register a patch
        register_patch_mapping(
            mapping={"Qwen2MoeExperts": CustomExperts}
        )

        # Apply patches within the context
        with apply_patches():
            # The model will use CustomExperts instead of Qwen2MoeExperts
            model = Qwen2MoeModel(Qwen2MoeConfig())

        # Outside the context, original classes are restored
        # The model will use Qwen2MoeExperts again
        model = Qwen2MoeModel(Qwen2MoeConfig())
        ```
    Nr%   transformers)r4   listsysmodulesvalueshasattrr%   
startswithdirr   getattrr"   r#   setattrAttributeErrorr$   ImportErrorr   )	r   original_classesmodule	attr_namer   attrmodule_namer   original_classs	            r   apply_patchesrK      s-    6  !G s{{))+, >!<)).9 V 	I 7	7 K (
vy1!$-AE &//9!=>	+<=	2 
5E5K5K5M 41!j>[)
N34 #I{; s7   BD"	D%D"&DAD"DD"DD"modelc                    t               }|sy| j                         D ]  }t        |d      s|j                  |j                  j	                         D ]y  \  }}t        |t              r+t        |j                  j                  |      }|9||_        At        |t              sRt        |j                  |      }|k||j                  |<   {  y)a  
    Patch the model instance's output recorders to use the registered replacement classes.

    This function updates output recorders in a model's submodules to use monkey-patched replacement
    classes. Output recorders are used by the transformers library to track intermediate outputs during
    forward passes (via the `_can_record_outputs` attribute). When classes are monkey-patched, these
    recorders need to be updated to reference the new classes.

    This is automatically called during model initialization when loading with `from_pretrained` or
    `from_config`. You typically don't need to call this manually unless you're constructing models
    in custom ways.

    Note:
        The `_can_record_outputs` attribute is a class-level attribute that maps output names to either:
        - `OutputRecorder` instances that have a `target_class` attribute
        - Class types directly

        This function patches both cases to use the replacement classes from the monkey patch registry.

    Args:
        model (`nn.Module`):
            The model instance whose output recorders should be patched. All submodules will be
            traversed to find and patch their `_can_record_outputs` attributes.

    Example:
        ```python
        from transformers import AutoModelForCausalLM
        from transformers.monkey_patching import register_patch_mapping, patch_output_recorders

        # Register a patch
        register_patch_mapping(mapping={"Qwen2MoeExperts": CustomExperts})

        # If you construct a model manually (without from_pretrained), patch recorders
        model = Qwen2MoeModel(config)
        patch_output_recorders(model)  # Updates output recorders to use CustomExperts
        ```
    N_can_record_outputs)r4   r<   r>   rN   r   r"   r   r   target_classr%   r#   )rL   r   	submoduleoutputrecorderr   s         r   patch_output_recordersrS   .  s    N  !G]]_ R	9349V9V9b$-$A$A$G$G$I 
R h7(?@U@U@^@^`g(h%(40A-$/(?@Q@QSZ([%(4@Q	55f=
RRr   )F)r   N)#r   r;   	threading
contextlibr   utilsr   r   utils.output_capturingr   torch.nnr'   
get_loggerr%   r   r   dictstrr#   r(   __annotations__r	   PatternLockr!   r   r   boolr,   r:   r0   r4   r7   rK   rS   r3   r   r   <module>r`      su   
 
  % . 2 			H	%:< T#tBII"67 <24 $sBJJ/ 4#Y^^% c bjj4&7 , d3RYY;O6P UYZ\ZcZcUdgkUk :FHDd299o)=$> FH4 FH\` FHR%149 %1 %1P24T"))_ 45 2,0 ?4 ?4H7R")) 7R 7Rr   