DoLuEdu-WordFileParse/scripts/omml_mathml2latex/core/cache.py

"""
Caching utilities for performance optimization.

Provides singleton pattern for expensive-to-create objects like
XSL transformers, and LRU caching for frequently-used operations.
"""

import os
import threading
from typing import Dict, Optional
from lxml import etree as ET
from functools import lru_cache


class XSLTransformerCache:
    """
    Thread-safe singleton cache for XSL transformers.

    The OMML2MML.xsl file is expensive to parse (lxml XSLT compilation).
    This cache ensures it's only parsed once and reused across all conversions.

    Example:
        >>> cache = XSLTransformerCache.get_instance()
        >>> transformer = cache.get_transformer('path/to/OMML2MML.xsl')
        >>> # Subsequent calls return cached transformer
        >>> transformer2 = cache.get_transformer('path/to/OMML2MML.xsl')
        >>> assert transformer is transformer2
    """

    _instance: Optional['XSLTransformerCache'] = None
    _lock = threading.Lock()

    def __init__(self):
        """Initialize cache storage."""
        self._transformers: Dict[str, ET.XSLT] = {}
        self._cache_lock = threading.Lock()

    @classmethod
    def get_instance(cls) -> 'XSLTransformerCache':
        """
        Get singleton instance of the cache.

        Thread-safe singleton pattern using double-checked locking.

        Returns:
            Singleton XSLTransformerCache instance
        """
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = cls()
        return cls._instance

    def get_transformer(self, xsl_path: str) -> ET.XSLT:
        """
        Get cached XSL transformer or create and cache new one.

        Args:
            xsl_path: Path to XSL stylesheet file

        Returns:
            Compiled XSLT transformer

        Raises:
            FileNotFoundError: If XSL file doesn't exist
            ET.XSLTParseError: If XSL file is malformed

        Example:
            >>> cache = XSLTransformerCache.get_instance()
            >>> transformer = cache.get_transformer('OMML2MML.xsl')
            >>> result = transformer(omml_doc)
        """
        # Normalize path for consistent caching
        xsl_path = os.path.abspath(xsl_path)

        # Check cache (read lock)
        if xsl_path in self._transformers:
            return self._transformers[xsl_path]

        # Not in cache - create with write lock
        with self._cache_lock:
            # Double-check after acquiring lock
            if xsl_path in self._transformers:
                return self._transformers[xsl_path]

            # Validate file exists
            if not os.path.exists(xsl_path):
                raise FileNotFoundError(f"XSL file not found: {xsl_path}")

            # Parse and compile XSL
            try:
                xsl_doc = ET.parse(xsl_path)
                transformer = ET.XSLT(xsl_doc)
                self._transformers[xsl_path] = transformer
                return transformer

            except (ET.XSLTParseError, ET.XMLSyntaxError) as e:
                raise ET.XSLTParseError(f"Failed to parse XSL file {xsl_path}: {e}") from e

    def clear_cache(self) -> None:
        """
        Clear all cached transformers.

        Useful for testing or when XSL files have been updated.

        Example:
            >>> cache = XSLTransformerCache.get_instance()
            >>> cache.clear_cache()
        """
        with self._cache_lock:
            self._transformers.clear()

    def get_cache_size(self) -> int:
        """
        Get number of cached transformers.

        Returns:
            Number of cached XSL transformers

        Example:
            >>> cache = XSLTransformerCache.get_instance()
            >>> size = cache.get_cache_size()
            >>> print(f"Cached {size} transformers")
        """
        return len(self._transformers)


class ConverterObjectPool:
    """
    Object pool for visitor registries and strategies.

    Maintains a pool of pre-initialized conversion objects to avoid
    recreation overhead for each equation.

    Thread-safe implementation using locks.

    Example:
        >>> pool = ConverterObjectPool(pool_size=4)
        >>> obj = pool.acquire_converter('math', wrap_identifiers=False)
        >>> # Use converter...
        >>> pool.release_converter(obj)
    """

    def __init__(self, pool_size: int = 4):
        """
        Initialize object pool.

        Args:
            pool_size: Maximum number of pooled objects
        """
        self.pool_size = pool_size
        self._available = []
        self._in_use = set()
        self._lock = threading.Lock()

    def acquire_converter(
        self,
        subject_mode: str = 'auto',
        wrap_identifiers: bool = False,
    ):
        """
        Acquire a converter from the pool.

        If pool is exhausted, creates a new converter.

        Args:
            subject_mode: Subject formatting mode
            wrap_identifiers: Wrap identifiers in \\mathrm{}

        Returns:
            Converter instance

        Example:
            >>> pool = ConverterObjectPool()
            >>> converter = pool.acquire_converter('math')
        """
        with self._lock:
            # Try to reuse from pool
            for i, (cached_mode, cached_wrap, obj) in enumerate(self._available):
                if cached_mode == subject_mode and cached_wrap == wrap_identifiers:
                    self._available.pop(i)
                    self._in_use.add(id(obj))
                    return obj

            # Pool miss - create new converter
            from ..compat import LegacyMathMLToTeXConverter
            converter = LegacyMathMLToTeXConverter(
                wrap_identifiers_in_mathrm=wrap_identifiers,
                subject_mode=subject_mode,
            )
            self._in_use.add(id(converter))
            return converter

    def release_converter(
        self,
        converter,
        subject_mode: str = 'auto',
        wrap_identifiers: bool = False,
    ) -> None:
        """
        Release converter back to pool.

        Args:
            converter: Converter instance to release
            subject_mode: Subject mode of the converter
            wrap_identifiers: Wrap identifiers setting

        Example:
            >>> pool.release_converter(converter, 'math', False)
        """
        with self._lock:
            obj_id = id(converter)
            if obj_id in self._in_use:
                self._in_use.remove(obj_id)

                # Only pool if under limit
                if len(self._available) < self.pool_size:
                    self._available.append((subject_mode, wrap_identifiers, converter))

    def clear_pool(self) -> None:
        """Clear the object pool."""
        with self._lock:
            self._available.clear()
            self._in_use.clear()


# LRU cache for expensive pattern matching operations
@lru_cache(maxsize=256)
def cached_function_detection(text: str) -> Optional[str]:
    """
    Cached function name detection for fragmented functions.

    Args:
        text: Text to check for function names

    Returns:
        Function name if detected, None otherwise

    Example:
        >>> cached_function_detection('sin')
        'sin'
        >>> cached_function_detection('xyz')
        None
    """
    from ..data import MATH_FUNCTIONS

    text_lower = text.lower()
    if text_lower in MATH_FUNCTIONS:
        return text_lower
    return None


@lru_cache(maxsize=128)
def cached_chemical_element_detection(symbol: str) -> bool:
    """
    Cached chemical element symbol detection.

    Args:
        symbol: Symbol to check

    Returns:
        True if symbol is a chemical element

    Example:
        >>> cached_chemical_element_detection('Fe')
        True
        >>> cached_chemical_element_detection('Xy')
        False
    """
    from ..data import CHEMICAL_ELEMENTS

    return symbol in CHEMICAL_ELEMENTS


def clear_all_caches() -> None:
    """
    Clear all caches (XSL, LRU caches, object pools).

    Useful for testing or when memory needs to be reclaimed.

    Example:
        >>> clear_all_caches()
    """
    # Clear XSL cache
    cache = XSLTransformerCache.get_instance()
    cache.clear_cache()

    # Clear LRU caches
    cached_function_detection.cache_clear()
    cached_chemical_element_detection.cache_clear()