""" 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()