""" Pipeline orchestrator for the conversion process. This module implements the pipeline pattern, coordinating the execution of multiple stages in sequence to convert inputs to LaTeX. """ import time import logging from typing import List, Optional from .interfaces import Stage, ConversionContext logger = logging.getLogger(__name__) class ConversionPipeline: """ Orchestrates conversion through multiple stages. The pipeline executes stages in sequence, passing the context from one stage to the next. Each stage can read from and write to the shared context object. Example: >>> pipeline = (ConversionPipeline() ... .add_stage(InputStage()) ... .add_stage(ValidationStage()) ... .add_stage(ConversionStage()) ... .add_stage(OutputStage())) >>> context = ConversionContext(input_type='mathml', mathml='...') >>> result = pipeline.process(context) """ def __init__(self, enable_validation: bool = True, enable_timing: bool = True): """ Initialize the pipeline. Args: enable_validation: Whether to validate stage preconditions before execution enable_timing: Whether to record stage execution times """ self.stages: List[Stage] = [] self.enable_validation = enable_validation self.enable_timing = enable_timing def add_stage(self, stage: Stage) -> 'ConversionPipeline': """ Add a stage to the pipeline (fluent interface). Args: stage: The stage to add to the pipeline Returns: Self for method chaining Raises: TypeError: If stage is not an instance of Stage """ if not isinstance(stage, Stage): raise TypeError(f"Stage must be instance of Stage, got {type(stage)}") self.stages.append(stage) logger.debug(f"Added stage: {stage.name}") return self def process(self, context: ConversionContext) -> ConversionContext: """ Execute all stages in order. Args: context: The initial conversion context Returns: Updated conversion context after all stages Raises: ValueError: If any stage validation fails RuntimeError: If any stage processing fails """ if not self.stages: raise RuntimeError("Pipeline has no stages configured") logger.info(f"Starting pipeline with {len(self.stages)} stages") pipeline_start_time = time.time() try: for stage in self.stages: context = self._execute_stage(stage, context) # Record total pipeline time pipeline_end_time = time.time() context.total_time = pipeline_end_time - pipeline_start_time logger.info( f"Pipeline completed successfully in {context.total_time:.4f}s " f"({len(self.stages)} stages)" ) return context except Exception as e: logger.error(f"Pipeline failed: {e}", exc_info=True) raise def _execute_stage(self, stage: Stage, context: ConversionContext) -> ConversionContext: """ Execute a single stage with validation and timing. Args: stage: The stage to execute context: The current context Returns: Updated context after stage execution Raises: ValueError: If stage validation fails RuntimeError: If stage execution fails """ logger.debug(f"Executing stage: {stage.name}") # Validate preconditions if enabled if self.enable_validation: if not stage.validate_input(context): error_msg = f"Stage '{stage.name}' validation failed" logger.error(error_msg) raise ValueError(error_msg) # Execute stage with timing if self.enable_timing: stage_start_time = time.time() context = stage.process(context) stage_end_time = time.time() stage_time = stage_end_time - stage_start_time context.record_stage_time(stage.name, stage_time) logger.debug(f"Stage '{stage.name}' completed in {stage_time:.4f}s") else: context = stage.process(context) return context def clear(self) -> 'ConversionPipeline': """ Remove all stages from the pipeline. Returns: Self for method chaining """ self.stages.clear() logger.debug("Pipeline cleared") return self def get_stages(self) -> List[Stage]: """Get a copy of the stages list.""" return self.stages.copy() def get_stage_count(self) -> int: """Get the number of stages in the pipeline.""" return len(self.stages) def __repr__(self) -> str: """String representation for debugging.""" stage_names = [stage.name for stage in self.stages] return f"ConversionPipeline(stages={stage_names})" def __str__(self) -> str: """Human-readable representation.""" if not self.stages: return "ConversionPipeline (empty)" stage_list = "\n ".join([f"{i+1}. {stage.name}" for i, stage in enumerate(self.stages)]) return f"ConversionPipeline ({len(self.stages)} stages):\n {stage_list}"