Corrosion Engineering MCP Server

""" Tier 1 Tool: Run PHREEQC Aqueous Speciation Calculates equilibrium speciation of aqueous solutions including: - pH from charge balance or user-specified - Major species concentrations (mol/L) - Saturation indices for minerals (calcite, gypsum, etc.) - Ionic strength - Alkalinity Performance: ~1 second Accuracy: ±0.1 pH units, ±10% for species concentrations """ from typing import Dict, Optional import json import logging from core.chemistry_backend import PHREEQCBackend, validate_water_chemistry logger = logging.getLogger(__name__) def run_phreeqc_speciation( ions_json: str, temperature_C: float = 25.0, pH: Optional[float] = None, pe: float = 4.0, validate_charge_balance: bool = True, max_imbalance: float = 5.0, ) -> Dict: """ Run PHREEQC aqueous speciation calculation. Args: ions_json: JSON string of ion concentrations in mg/L Example: '{"Na+": 1000.0, "Cl-": 1500.0, "Ca2+": 100.0, "HCO3-": 200.0}' temperature_C: Water temperature in degrees Celsius (default 25.0) pH: Initial pH (if None, PHREEQC calculates from charge balance) pe: Redox potential, dimensionless (default 4.0 for oxic conditions) validate_charge_balance: Check charge balance before running (default True) max_imbalance: Maximum acceptable charge imbalance in % (default 5.0) Returns: Dictionary containing: - pH: Calculated pH - pe: Redox potential - temperature_C: Temperature - ionic_strength_M: Ionic strength (mol/L) - alkalinity_mg_L_CaCO3: Total alkalinity as mg/L CaCO₃ - species: Dict of major species concentrations (mol/L) - saturation_indices: Dict of mineral saturation indices - charge_balance_percent: Charge imbalance (%) - interpretation: Text summary of results Example: >>> result = run_phreeqc_speciation( ... ions_json='{"Na+": 1000, "Cl-": 1500, "Ca2+": 100, "HCO3-": 200}', ... temperature_C=25.0 ... ) >>> print(result["pH"]) 7.45 >>> print(result["saturation_indices"]["Calcite"]) -0.15 Raises: ValueError: If charge imbalance exceeds max_imbalance RuntimeError: If PHREEQC calculation fails """ # Parse ion concentrations try: ions = json.loads(ions_json) except json.JSONDecodeError as e: raise ValueError(f"Invalid JSON for ions: {e}") from e if not isinstance(ions, dict): raise ValueError("ions_json must be a JSON object (dictionary)") # Validate charge balance if requested if validate_charge_balance: try: validate_water_chemistry(ions, max_imbalance=max_imbalance) except ValueError as e: logger.warning(f"Charge balance validation failed: {e}") # Continue anyway, but log the warning # Run PHREEQC speciation backend = PHREEQCBackend() result = backend.run_speciation( ions=ions, temperature_C=temperature_C, pH=pH, pe=pe, ) # Format results for MCP tool output output = { "pH": round(result.pH, 3), "pe": round(result.pe, 3), "temperature_C": temperature_C, "ionic_strength_M": round(result.ionic_strength_M, 6), "alkalinity_mg_L_CaCO3": round(result.alkalinity_mg_L_CaCO3, 2), "species": { species: round(conc, 9) for species, conc in result.species.items() }, "saturation_indices": { mineral: round(si, 3) for mineral, si in result.saturation_indices.items() }, "charge_balance_percent": round(result.charge_balance_percent, 2), } # Add interpretation interpretation_parts = [] # pH interpretation if result.pH < 4.0: interpretation_parts.append("Highly acidic (pH < 4.0) - severe corrosion risk") elif result.pH < 7.0: interpretation_parts.append("Acidic (pH < 7.0) - corrosive") elif result.pH > 10.0: interpretation_parts.append("Highly alkaline (pH > 10.0) - scaling risk") elif result.pH > 8.5: interpretation_parts.append("Alkaline (pH > 8.5) - moderate scaling risk") else: interpretation_parts.append(f"Near-neutral pH ({result.pH:.2f})") # Saturation index interpretation si_calcite = result.saturation_indices.get("Calcite", 0.0) if si_calcite > 0.5: interpretation_parts.append(f"Supersaturated with calcite (SI = {si_calcite:.2f}) - scaling likely") elif si_calcite < -0.5: interpretation_parts.append(f"Undersaturated with calcite (SI = {si_calcite:.2f}) - corrosive to carbonates") # Ionic strength interpretation if result.ionic_strength_M > 0.5: interpretation_parts.append(f"High ionic strength ({result.ionic_strength_M:.3f} M) - saline water") elif result.ionic_strength_M < 0.001: interpretation_parts.append(f"Low ionic strength ({result.ionic_strength_M:.6f} M) - freshwater") output["interpretation"] = ". ".join(interpretation_parts) return output