""" Batch JSON converter for omml2tex with key preservation. Handles conversion of multiple equations in JSON format while preserving unique identifiers. Supports both sequential and parallel processing. """ from typing import Dict, Any import sys from .compat import LegacyMathMLToTeXConverter from .parsers.omml_parser import OmmlParser class BatchJSONConverter: """ Batch converter for JSON format with key preservation. Converts a dictionary of equations (key -> XML) to LaTeX while preserving the original keys. Supports both OMML and MathML input formats with automatic detection. Example: >>> converter = BatchJSONConverter(subject_mode="chemistry") >>> equations = {"eq1": "x", "eq2": "..."} >>> results = converter.convert_batch(equations) >>> print(results["eq1"]) # LaTeX output """ def __init__( self, subject_mode: str = "math", wrap_identifiers: bool = True, prefer_xlongequal: bool = True, universal_delimiter_scaling: bool = True, verbose: bool = False ): """ Initialize batch converter. Args: subject_mode: Subject mode for conversion (math, chemistry, physics, statistics) wrap_identifiers: Whether to wrap identifiers in \\mathrm{} prefer_xlongequal: Whether to prefer \\xlongequal for overset equations universal_delimiter_scaling: Whether to add \\left\\right to all delimiters verbose: Enable verbose output """ self.subject_mode = subject_mode self.wrap_identifiers = wrap_identifiers self.prefer_xlongequal = prefer_xlongequal self.universal_delimiter_scaling = universal_delimiter_scaling self.verbose = verbose # Initialize converter self.converter = LegacyMathMLToTeXConverter( wrap_identifiers_in_mathrm=wrap_identifiers, prefer_xlongequal=prefer_xlongequal, subject_mode=subject_mode, universal_delimiter_scaling=universal_delimiter_scaling, ) # OMML parser for OMML input self.omml_parser = OmmlParser() def detect_input_type(self, content: str) -> str: """ Detect if input is OMML or MathML. Args: content: XML content string Returns: "omml" or "mathml" """ content_stripped = content.strip() # Check for OMML namespace if 'm:oMath' in content_stripped or 'm:oMathPara' in content_stripped: return "omml" # Check for MathML namespace if ' Dict[str, Any]: """ Convert a single equation while preserving the key. Args: key: Unique identifier for the equation equation_xml: OMML or MathML XML string Returns: Dict with 'key', 'latex', 'success', and optional 'error' """ try: # Detect input type input_type = self.detect_input_type(equation_xml) if self.verbose: print(f" {key}: {input_type} -> ", end="", file=sys.stderr) # Convert OMML to MathML if needed if input_type == "omml": mathml_list = self.omml_parser.parse(equation_xml) # OmmlParser.parse() returns a list - take the first element if not mathml_list: raise ValueError("OMML parser returned empty result") mathml = mathml_list[0] else: mathml = equation_xml # Convert MathML to LaTeX latex = self.converter.convert(mathml) if self.verbose: print(f"LaTeX ({len(latex)} chars)", file=sys.stderr) return { "key": key, "latex": latex, "success": True, } except Exception as e: if self.verbose: print(f"FAILED: {e}", file=sys.stderr) return { "key": key, "latex": "", "success": False, "error": str(e), } def convert_batch(self, input_dict: Dict[str, str], quiet: bool = False) -> Dict[str, str]: """ Convert a batch of equations from JSON dict (sequential). Args: input_dict: Dict mapping unique keys to equation XML strings quiet: Suppress progress messages Returns: Dict mapping same keys to LaTeX strings (failed conversions get empty string) """ if self.verbose and not quiet: print(f"Converting {len(input_dict)} equations...", file=sys.stderr) results = {} successful = 0 failed = 0 for key, equation_xml in input_dict.items(): equation_xml = equation_xml.replace("JBYSIMJBY", "~") result = self.convert_single(key, equation_xml) # Store LaTeX string (or empty string if failed) results[key] = result["latex"] if result["success"]: successful += 1 else: failed += 1 if not self.verbose and not quiet: print( f"Warning: Failed to convert {key}: {result.get('error', 'Unknown error')}", file=sys.stderr ) if (self.verbose or failed > 0) and not quiet: print(f"\nConversion complete:", file=sys.stderr) print(f" Total: {len(input_dict)}", file=sys.stderr) print(f" Successful: {successful}", file=sys.stderr) print(f" Failed: {failed}", file=sys.stderr) print(f" Success rate: {successful/len(input_dict)*100:.1f}%", file=sys.stderr) return results def convert_batch_parallel( self, input_dict: Dict[str, str], workers: int = None, quiet: bool = False ) -> Dict[str, str]: """ Convert batch using parallel processing (for large batches). Args: input_dict: Dict mapping unique keys to equation XML strings workers: Number of worker processes (default: CPU count, max 8) quiet: Suppress progress messages Returns: Dict mapping same keys to LaTeX strings """ from .parallel import ParallelConverter if self.verbose and not quiet: print(f"Converting {len(input_dict)} equations in parallel...", file=sys.stderr) # Create parallel converter parallel_converter = ParallelConverter( workers=workers, subject_mode=self.subject_mode, wrap_identifiers=self.wrap_identifiers, prefer_xlongequal=self.prefer_xlongequal, show_progress=self.verbose and not quiet, ) # Prepare input list (keys and equations) keys = list(input_dict.keys()) equations = list(input_dict.values()) # Detect input types and convert OMML to MathML if needed mathml_equations = [] for equation_xml in equations: equation_xml = equation_xml.replace("JBYSIMJBY", "~") input_type = self.detect_input_type(equation_xml) if input_type == "omml": mathml_list = self.omml_parser.parse(equation_xml) # OmmlParser.parse() returns a list - take the first element if not mathml_list: raise ValueError("OMML parser returned empty result") mathml = mathml_list[0] else: mathml = equation_xml mathml_equations.append(mathml) # Convert in parallel conversion_results = parallel_converter.convert_batch(mathml_equations, quiet=quiet) # Build result dict results = {} successful = 0 failed = 0 for key, result in zip(keys, conversion_results): if result['success']: results[key] = result['latex'] successful += 1 else: results[key] = "" failed += 1 if not self.verbose and not quiet: print( f"Warning: Failed to convert {key}: {result.get('error', 'Unknown error')}", file=sys.stderr ) if (self.verbose or failed > 0) and not quiet: print(f"\nConversion complete:", file=sys.stderr) print(f" Total: {len(input_dict)}", file=sys.stderr) print(f" Successful: {successful}", file=sys.stderr) print(f" Failed: {failed}", file=sys.stderr) print(f" Success rate: {successful/len(input_dict)*100:.1f}%", file=sys.stderr) return results