Spaces:
Running
Running
| """Pattern registry for managing and matching patterns.""" | |
| from typing import Any, Dict, List, Match, Optional, Tuple | |
| from .base import Pattern, PatternPhase | |
| class PatternRegistry: | |
| """Central registry for all parsing patterns. | |
| Patterns are organized by phase and sorted by priority. | |
| Lower priority number = higher precedence. | |
| """ | |
| def __init__(self): | |
| self._patterns: Dict[PatternPhase, List[Pattern]] = {phase: [] for phase in PatternPhase} | |
| self._frozen = False | |
| def register(self, pattern: Pattern) -> "PatternRegistry": | |
| """Register a pattern. Returns self for chaining.""" | |
| if self._frozen: | |
| raise RuntimeError("Cannot register patterns after registry is frozen") | |
| self._patterns[pattern.phase].append(pattern) | |
| return self | |
| def register_all(self, patterns: List[Pattern]) -> "PatternRegistry": | |
| """Register multiple patterns. Returns self for chaining.""" | |
| for p in patterns: | |
| self.register(p) | |
| return self | |
| def freeze(self) -> "PatternRegistry": | |
| """Freeze the registry and sort patterns by priority.""" | |
| for phase in PatternPhase: | |
| self._patterns[phase].sort(key=lambda p: p.priority) | |
| self._frozen = True | |
| return self | |
| def match_all( | |
| self, text: str, phase: PatternPhase, context: Optional[str] = None | |
| ) -> List[Tuple[Pattern, Match, Dict[str, Any]]]: | |
| """Find all matching patterns for a phase. | |
| Args: | |
| text: Text to match against | |
| phase: Which parsing phase to match | |
| context: Full sentence context for requires/excludes | |
| Returns: | |
| List of (pattern, match, extracted_data) tuples | |
| """ | |
| results = [] | |
| working_text = text | |
| for pattern in self._patterns[phase]: | |
| # Loop to find all matches for this pattern | |
| while True: | |
| # Always check against original context but match against working_text | |
| m = pattern.matches(working_text, context or text) | |
| if not m: | |
| break | |
| data = pattern.extract(working_text, m) | |
| results.append((pattern, m, data)) | |
| # If pattern consumes, mask out the matched text | |
| if pattern.consumes: | |
| start, end = m.start(), m.end() | |
| working_text = working_text[:start] + " " * (end - start) + working_text[end:] | |
| # Continue looking for more matches of this pattern | |
| continue | |
| # If pattern doesn't consume, we stop after first match to prevent infinite loop | |
| break | |
| if pattern.exclusive and any(r[0] == pattern for r in results): | |
| break | |
| # Sort results by match position to maintain text order | |
| results.sort(key=lambda x: x[1].start()) | |
| return results | |
| def match_first( | |
| self, text: str, phase: PatternPhase, context: Optional[str] = None | |
| ) -> Optional[Tuple[Pattern, Match, Dict[str, Any]]]: | |
| """Find the first (highest priority) matching pattern. | |
| Returns: | |
| (pattern, match, extracted_data) tuple or None | |
| """ | |
| for pattern in self._patterns[phase]: | |
| if m := pattern.matches(text, context): | |
| data = pattern.extract(text, m) | |
| return (pattern, m, data) | |
| return None | |
| def get_patterns(self, phase: PatternPhase) -> List[Pattern]: | |
| """Get all patterns for a phase (for debugging/testing).""" | |
| return list(self._patterns[phase]) | |
| def stats(self) -> Dict[str, int]: | |
| """Get pattern counts per phase.""" | |
| return {phase.name: len(patterns) for phase, patterns in self._patterns.items()} | |
| # Global registry instance | |
| _global_registry: Optional[PatternRegistry] = None | |
| def get_registry() -> PatternRegistry: | |
| """Get the global pattern registry, creating if needed.""" | |
| global _global_registry | |
| if _global_registry is None: | |
| _global_registry = PatternRegistry() | |
| _load_all_patterns(_global_registry) | |
| _global_registry.freeze() | |
| return _global_registry | |
| def _load_all_patterns(registry: PatternRegistry): | |
| """Load all patterns from pattern definition modules.""" | |
| from .conditions import CONDITION_PATTERNS | |
| from .effects import EFFECT_PATTERNS | |
| from .modifiers import MODIFIER_PATTERNS | |
| from .triggers import TRIGGER_PATTERNS | |
| registry.register_all(TRIGGER_PATTERNS) | |
| registry.register_all(CONDITION_PATTERNS) | |
| registry.register_all(EFFECT_PATTERNS) | |
| registry.register_all(MODIFIER_PATTERNS) | |