Add reaction experiment handling and tests

chemistry_lab/chemistry_lab.py

@@ -7,7 +7,8 @@
 
 import json
 import numpy as np
-from dataclasses import dataclass
+from dataclasses import dataclass, asdict
+from numbers import Real
 from typing import Dict, List, Tuple, Optional, TYPE_CHECKING, Any
 
 # Import all sub-modules
@@ -119,9 +120,41 @@ def run_experiment(self, experiment_spec: Dict[str, Any]) -> Dict[str, Any]:
             return {"status": "completed", "frames": len(trajectory)}
 
         elif exp_type == "reaction_simulation":
-            # This requires complex reactant/product molecule objects
-            # For now, we'll just indicate it's a valid experiment type
-             raise NotImplementedError("Reaction simulation via run_experiment is not fully implemented yet.")
+            reactants_spec = experiment_spec.get("reactants")
+            products_spec = experiment_spec.get("products")
+            conditions_spec = experiment_spec.get("conditions", {})
+
+            self._validate_reaction_inputs(reactants_spec, products_spec, conditions_spec)
+
+            reactants = self._build_reactant_molecules(reactants_spec)
+            products = self._build_reactant_molecules(products_spec)
+            conditions = self._build_reaction_conditions(conditions_spec)
+            reaction_name = experiment_spec.get("reaction_name")
+
+            result = self.simulate_reaction(
+                reactants,
+                products,
+                conditions,
+                reaction_name=reaction_name
+            )
+
+            kinetics = result.get("kinetics")
+            profiles = {
+                "time": result["time"].tolist(),
+                "reactant_concentration": result["reactant_concentration"].tolist(),
+                "product_concentration": result["product_concentration"].tolist(),
+                "product_distribution": {
+                    key: value.tolist() for key, value in result["product_distribution"].items()
+                }
+            }
+
+            return {
+                "status": "completed",
+                "reaction_name": reaction_name or "custom_reaction",
+                "kinetics": asdict(kinetics) if kinetics else {},
+                "selectivity": kinetics.product_selectivity if kinetics else {},
+                "profiles": profiles,
+            }
 
         elif exp_type == "spectroscopy":
             molecule = experiment_spec.get("molecule")
@@ -484,6 +517,93 @@ def complete_molecule_characterization(self, molecule: Dict) -> Dict:
 
         return results
 
+    def _validate_reaction_inputs(
+        self,
+        reactants: Any,
+        products: Any,
+        conditions: Any
+    ) -> None:
+        """Minimal validation for reaction experiment inputs."""
+        if not isinstance(reactants, list) or not reactants:
+            raise ValueError("'reactants' must be a non-empty list of molecule specifications.")
+        if not isinstance(products, list) or not products:
+            raise ValueError("'products' must be a non-empty list of molecule specifications.")
+        if not isinstance(conditions, dict):
+            raise ValueError("'conditions' must be provided as a dictionary.")
+
+        for label, collection in (("reactant", reactants), ("product", products)):
+            for idx, entry in enumerate(collection):
+                if not isinstance(entry, dict):
+                    raise ValueError(f"Each {label} specification must be a dictionary (item {idx}).")
+                if not (entry.get("formula") or entry.get("smiles")):
+                    raise ValueError(f"{label.capitalize()} {idx + 1} requires 'formula' or 'smiles'.")
+                for field in ("energy", "enthalpy", "entropy"):
+                    if field not in entry:
+                        raise ValueError(f"{label.capitalize()} {idx + 1} missing required field '{field}'.")
+                    if not isinstance(entry[field], Real):
+                        raise ValueError(f"{label.capitalize()} {idx + 1} field '{field}' must be numeric.")
+
+        self._validate_conditions(
+            conditions.get("temperature", self.config.temperature),
+            conditions.get("pressure", self.config.pressure),
+            conditions.get("pH")
+        )
+
+    def _validate_conditions(self, temperature: Any, pressure: Any, pH: Any = None) -> None:
+        """Validate basic reaction conditions."""
+        if not isinstance(temperature, Real) or temperature <= 0:
+            raise ValueError("'temperature' must be a positive number in Kelvin.")
+        if not isinstance(pressure, Real) or pressure <= 0:
+            raise ValueError("'pressure' must be a positive number in bar.")
+        if pH is not None and (not isinstance(pH, Real) or pH < 0 or pH > 14):
+            raise ValueError("'pH' must be between 0 and 14 if provided.")
+
+    def _build_reaction_conditions(self, conditions_spec: Dict[str, Any]) -> ReactionConditions:
+        """Construct ReactionConditions from a dictionary specification."""
+        temperature = conditions_spec.get("temperature", self.config.temperature)
+        pressure = conditions_spec.get("pressure", self.config.pressure)
+        pH = conditions_spec.get("pH")
+        catalyst = conditions_spec.get("catalyst")
+
+        if catalyst is not None and not isinstance(catalyst, Catalyst):
+            raise ValueError("'catalyst' must be a Catalyst instance if provided.")
+
+        self._validate_conditions(temperature, pressure, pH)
+
+        return ReactionConditions(
+            temperature=temperature,
+            pressure=pressure,
+            solvent=conditions_spec.get("solvent"),
+            pH=pH,
+            catalyst=catalyst
+        )
+
+    def _build_reactant_molecules(self, molecules_spec: List[Dict[str, Any]]) -> List[ReactMolecule]:
+        """Convert dictionary specifications into ReactMolecule instances."""
+        molecules: List[ReactMolecule] = []
+        for index, spec in enumerate(molecules_spec):
+            formula = spec.get("formula", "")
+            smiles = spec.get("smiles", "")
+            geometry = spec.get("geometry")
+
+            try:
+                molecule = ReactMolecule(
+                    formula=formula,
+                    smiles=smiles,
+                    energy=float(spec["energy"]),
+                    enthalpy=float(spec["enthalpy"]),
+                    entropy=float(spec["entropy"]),
+                    geometry=geometry
+                )
+            except KeyError as exc:
+                raise ValueError(f"Molecule {index + 1} missing required key: {exc.args[0]}") from exc
+            except (TypeError, ValueError) as exc:
+                raise ValueError(f"Molecule {index + 1} has invalid numeric values: {exc}") from exc
+
+            molecules.append(molecule)
+
+        return molecules
+
     def reaction_optimization_workflow(
         self,
         reactants: List[ReactMolecule],

tests/test_chemistry_reaction.py

@@ -0,0 +1,51 @@
+"""
+Unit test for chemistry reaction experiment runner.
+"""
+
+import sys
+from pathlib import Path
+
+import numpy as np
+
+# Ensure project root import resolution
+PROJECT_ROOT = Path(__file__).resolve().parent.parent
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+from chemistry_lab.chemistry_lab import ChemistryLaboratory  # noqa: E402
+
+
+def test_reaction_experiment_happy_path():
+    """Reaction experiments should build molecules, run simulation, and return kinetics + profiles."""
+    lab = ChemistryLaboratory()
+
+    experiment = {
+        "experiment_type": "reaction_simulation",
+        "reactants": [
+            {"formula": "A", "smiles": "A", "energy": 0.0, "enthalpy": 0.0, "entropy": 50.0},
+            {"formula": "B", "smiles": "B", "energy": 0.0, "enthalpy": 0.0, "entropy": 60.0},
+        ],
+        "products": [
+            {"formula": "AB", "smiles": "AB", "energy": -10.0, "enthalpy": -10.0, "entropy": 80.0},
+        ],
+        "conditions": {
+            "temperature": 298.15,
+            "pressure": 1.0,
+            "solvent": "water",
+        },
+        "reaction_name": "test_reaction",
+    }
+
+    result = lab.run_experiment(experiment)
+
+    assert result["status"] == "completed"
+    assert "kinetics" in result and result["kinetics"]
+    assert result["kinetics"]["rate_constant"] > 0
+    assert result["profiles"]["time"][0] > 0
+
+    reactant_curve = np.array(result["profiles"]["reactant_concentration"])
+    product_curve = np.array(result["profiles"]["product_concentration"])
+
+    assert reactant_curve[0] > reactant_curve[-1]
+    assert product_curve[-1] > product_curve[0]
+    assert "total" in result["profiles"]["product_distribution"]