quantum_ml.py

"""

Quantum Machine Learning Module

================================



This module implements quantum ML algorithms for catalyst discovery and scoring:



Feature 1 (Discovery): QGAN + VQC + VQE

- Generate new catalyst candidates with QGAN

- Classify effectiveness with VQC

- Validate with VQE energy calculations



Feature 2 (Education): QSVM + VQC + VQE

- Score user's catalyst guess against ideal

- Provide feedback on chemical suitability

- Compare energy profiles



Key Algorithms:

- QSVM: Quantum Support Vector Machine for classification

- VQC: Variational Quantum Classifier (quantum neural network)

- QGAN: Quantum Generative Adversarial Network

"""

import numpy as np
import random
from qiskit import QuantumCircuit
from qiskit.circuit.library import ZZFeatureMap, RealAmplitudes
from qiskit_algorithms import VQE
from qiskit_algorithms.optimizers import COBYLA, SPSA, SLSQP
from qiskit.primitives import StatevectorSampler, StatevectorEstimator
from qiskit.quantum_info import Statevector, state_fidelity
from typing import Dict, List, Tuple, Optional
from rdkit import Chem
from rdkit.Chem import Descriptors
from modules.molecule_generation import mutate_catalyst

# NOTE: qiskit_machine_learning is optional - we implement core functionality without it

FEATURE_DIMENSION = 16


def _is_degenerate_feature_vector(features: np.ndarray, tol: float = 1e-12) -> bool:
    """Return True if features carry effectively no signal."""
    return bool(np.all(np.abs(features) <= tol))


def _validate_feature_vector(features: np.ndarray) -> Tuple[bool, str]:
    """

    Validate shape and numerical quality of a molecular feature vector.



    Returns:

        (is_valid, reason)

    """
    if features is None:
        return False, "feature vector is None"

    if not isinstance(features, np.ndarray):
        return False, f"feature vector must be numpy.ndarray, got {type(features).__name__}"

    if features.ndim != 1:
        return False, f"feature vector must be 1D, got {features.ndim}D"

    if len(features) != FEATURE_DIMENSION:
        return False, f"expected {FEATURE_DIMENSION} features, got {len(features)}"

    if not np.all(np.isfinite(features)):
        return False, "feature vector contains non-finite values"

    if _is_degenerate_feature_vector(features):
        return False, "feature vector is degenerate (all near zero)"

    return True, ""


# ========================================================================
# MOLECULAR FEATURE EXTRACTION
# ========================================================================

def extract_molecular_features(smiles: str) -> np.ndarray:
    """

    Extract numerical features from a molecule for ML algorithms.



    Features extracted (16D):

    - 8 baseline physicochemical descriptors

    - 8 additional descriptors for kinetics and catalyst chemistry



    Args:

        smiles: SMILES string



    Returns:

        16D feature vector (normalized to [0, 1])

    """
    try:
        mol = Chem.MolFromSmiles(smiles)
        if mol is None:
            return np.zeros(FEATURE_DIMENSION)

        # Baseline 8 descriptors
        baseline = np.array([
            Descriptors.MolWt(mol),                    # Molecular weight
            mol.GetNumHeavyAtoms(),                     # Heavy atoms
            sum(1 for atom in mol.GetAtoms() if atom.GetAtomicNum() not in [1, 6]),  # Heteroatoms
            Descriptors.NumRotatableBonds(mol),         # Rotatable bonds
            Descriptors.MolLogP(mol),                   # LogP
            Descriptors.TPSA(mol),                      # TPSA
            sum(1 for atom in mol.GetAtoms() if atom.GetIsAromatic()),  # Aromatic atoms
            Descriptors.NumValenceElectrons(mol),       # Valence electrons
        ], dtype=float)

        # Normalize to [0, 1] range
        baseline_scales = np.array([200.0, 20.0, 10.0, 10.0, 6.0, 200.0, 20.0, 120.0])
        baseline = np.clip(baseline / baseline_scales, 0, 1)

        # Additional 8 descriptors requested for 16D alignment.
        carbon_count = sum(1 for atom in mol.GetAtoms() if atom.GetAtomicNum() == 6)
        heavy_atoms = max(1, mol.GetNumHeavyAtoms())
        formal_charge = float(sum(atom.GetFormalCharge() for atom in mol.GetAtoms()))
        heavy_to_carbon_ratio = float(heavy_atoms / max(1, carbon_count))

        extra = np.array([
            Descriptors.NumHDonors(mol),
            Descriptors.NumHAcceptors(mol),
            Descriptors.RingCount(mol),
            Descriptors.FractionCSP3(mol),
            Descriptors.NumAliphaticRings(mol),
            Descriptors.NumAromaticRings(mol),
            formal_charge,
            heavy_to_carbon_ratio,
        ], dtype=float)

        extra_scales = np.array([10.0, 15.0, 8.0, 1.0, 8.0, 8.0, 5.0, 10.0])
        extra_shift = np.array([0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 2.5, 0.0])
        extra = np.clip((extra + extra_shift) / extra_scales, 0, 1)

        features = np.concatenate([baseline, extra])
        return np.clip(features, 0, 1)

    except Exception:
        return np.zeros(FEATURE_DIMENSION)


def get_catalyst_properties(smiles: str) -> Dict:
    """

    Get catalyst-specific chemical properties.



    Returns:

        Dictionary with catalyst properties

    """
    mol = Chem.MolFromSmiles(smiles)
    if mol is None:
        return {"valid": False}

    atoms = [atom.GetSymbol() for atom in mol.GetAtoms()]

    # Catalyst type classification
    catalyst_metals = {'Fe', 'Pt', 'Pd', 'Ni', 'Cu', 'Rh', 'Ru', 'Co', 'Au', 'Ag', 'Ti', 'Zn', 'Cr', 'Mn', 'V', 'Mo', 'W'}
    is_metal_catalyst = bool(set(atoms) & catalyst_metals)

    # Electronic properties (simplified)
    valence_electrons = Descriptors.NumValenceElectrons(mol)
    atomic_number = mol.GetAtomWithIdx(0).GetAtomicNum() if mol.GetNumAtoms() > 0 else 0

    return {
        "valid": True,
        "is_metal": is_metal_catalyst,
        "metal_type": list(set(atoms) & catalyst_metals)[0] if is_metal_catalyst else None,
        "valence_electrons": valence_electrons,
        "atomic_number": atomic_number,
        "atoms": atoms
    }


# ========================================================================
# QUANTUM SUPPORT VECTOR MACHINE (QSVM)
# ========================================================================

class QuantumCatalystScorer:
    """

    QSVM-based catalyst scoring system (simplified implementation).



    Uses quantum feature maps and classical SVM for classification.

    """

    def __init__(self, reaction_type: str):
        """

        Initialize QSVM scorer for a specific reaction.



        Args:

            reaction_type: Type of reaction (e.g., "H2_O2", "CO2_reduction")

        """
        self.reaction_type = reaction_type
        self.feature_map = ZZFeatureMap(feature_dimension=FEATURE_DIMENSION, reps=2)
        self.trained = False
        self.training_data = self._get_training_data()

        # Use quantum feature encoding with classical SVM
        self.sampler = StatevectorSampler()

    def _get_training_data(self) -> Tuple[np.ndarray, np.ndarray]:
        """

        Get training data for the reaction type.



        Returns:

            Tuple of (features, labels)

            labels: 1 = good catalyst, -1 = poor catalyst

        """
        # Training data based on known catalyst performance
        reaction_hint = (self.reaction_type or "").lower()

        if self.reaction_type == "H2_O2":
            # H2 + O2 -> H2O reaction
            good_catalysts = ["[Pt]", "[Pd]", "[Ni]=O", "[Fe]=O"]
            poor_catalysts = ["[Au]", "[Ag]", "[Cu]", "[Zn]"]

        elif self.reaction_type == "N2_H2":
            # N2 + H2 -> NH3 (Haber process)
            good_catalysts = ["[Fe]", "[Ru]", "[Fe]=O"]
            poor_catalysts = ["[Pt]", "[Au]", "[Cu]", "[Ag]"]

        elif self.reaction_type == "CO2_reduction":
            # CO2 -> CO/CH4
            good_catalysts = ["[Cu]", "[Ag]", "[Au]", "[Pd]"]
            poor_catalysts = ["[Fe]", "[Ni]", "[Zn]"]

        elif ("o2" in reaction_hint) or ("o=" in reaction_hint):
            # Dynamic oxidation-like custom reaction pool.
            good_catalysts = ["[Pt]", "[Pd]", "[Ni]=O", "[Fe]=O"]
            poor_catalysts = ["[Au]", "[Ag]", "[Cu]", "[Zn]"]

        elif ("h2" in reaction_hint) or ("n2" in reaction_hint):
            # Dynamic reduction/hydrogenation custom reaction pool.
            good_catalysts = ["[Fe]", "[Ru]", "[Ni]", "[Co]"]
            poor_catalysts = ["[Au]", "[Ag]", "[Zn]"]

        else:
            # Default generic training
            good_catalysts = ["[Pt]", "[Pd]", "[Ni]"]
            poor_catalysts = ["[Au]", "[Ag]", "[Zn]"]

        # Extract features
        good_features = [extract_molecular_features(s) for s in good_catalysts]
        poor_features = [extract_molecular_features(s) for s in poor_catalysts]

        X = np.array(good_features + poor_features)
        y = np.array([1] * len(good_catalysts) + [-1] * len(poor_catalysts))

        return X, y

    def train(self):
        """Train the quantum classifier (simplified)."""
        try:
            X, y = self.training_data
            # Store training data for distance-based classification
            self.X_train = X
            self.y_train = y
            self.trained = True
            return {"success": True, "error": ""}
        except Exception as e:
            return {"success": False, "error": str(e)}

    def _quantum_similarity(self, x1: np.ndarray, x2: np.ndarray) -> float:
        """

        Calculate quantum kernel similarity between two feature vectors.



        Uses quantum feature map overlap |<ψ(x1)|ψ(x2)>|²

        """
        try:
            valid_x1, _ = _validate_feature_vector(x1)
            valid_x2, _ = _validate_feature_vector(x2)
            if not valid_x1 or not valid_x2:
                return 0.0

            # Create and evaluate actual quantum states for the two feature vectors.
            qc1 = self.feature_map.assign_parameters(np.clip(x1, 0.0, 1.0))
            qc2 = self.feature_map.assign_parameters(np.clip(x2, 0.0, 1.0))

            state1 = Statevector(qc1)
            state2 = Statevector(qc2)
            fidelity = float(state_fidelity(state1, state2))
            return float(np.clip(fidelity, 0.0, 1.0))
        except Exception:
            return 0.5

    def score_catalyst(self, smiles: str) -> Dict:
        """

        Score a catalyst for the reaction.



        Args:

            smiles: Catalyst SMILES string



        Returns:

            Dictionary with:

            - score: 0-100 score

            - classification: "excellent", "good", "fair", "poor"

            - confidence: prediction confidence

            - feedback: text explanation

        """
        if not self.trained:
            self.train()

        try:
            features = extract_molecular_features(smiles)
            is_valid, reason = _validate_feature_vector(features)
            if not is_valid:
                return {
                    "score": 0,
                    "classification": "error",
                    "confidence": 0,
                    "feedback": f"Feature extraction failed for {smiles}: {reason}.",
                    "method": "QSVM (Quantum Kernel)",
                    "error": reason
                }

            if not hasattr(self, "X_train") or not hasattr(self, "y_train"):
                return {
                    "score": 0,
                    "classification": "error",
                    "confidence": 0,
                    "feedback": "Model training data unavailable.",
                    "method": "QSVM (Quantum Kernel)",
                    "error": "missing training state"
                }

            # Calculate similarities to training examples
            good_indices = np.where(self.y_train == 1)[0]
            poor_indices = np.where(self.y_train == -1)[0]

            if len(good_indices) == 0 or len(poor_indices) == 0:
                return {
                    "score": 0,
                    "classification": "error",
                    "confidence": 0,
                    "feedback": "Training data is incomplete for this reaction class.",
                    "method": "QSVM (Quantum Kernel)",
                    "error": "invalid class split"
                }

            # Average similarity to good catalysts
            good_sim = np.mean([
                self._quantum_similarity(features, self.X_train[i])
                for i in good_indices
            ])

            # Average similarity to poor catalysts
            poor_sim = np.mean([
                self._quantum_similarity(features, self.X_train[i])
                for i in poor_indices
            ])

            # Decision based on relative similarity
            decision = good_sim - poor_sim
            confidence = min(abs(decision) / 1.0, 1.0) * 100

            # Map to score
            if decision > 0:
                prediction = 1
                score = 70 + (confidence * 0.3)
                classification = "good" if score < 85 else "excellent"
            else:
                prediction = -1
                score = 50 - (confidence * 0.5)
                classification = "poor" if score < 30 else "fair"

            # Get catalyst properties for feedback
            props = get_catalyst_properties(smiles)
            feedback = self._generate_feedback(score, props)

            return {
                "score": float(score),
                "classification": classification,
                "confidence": float(confidence),
                "feedback": feedback,
                "method": "QSVM (Quantum Kernel)",
                "error": ""
            }

        except Exception as e:
            return {
                "score": 0,
                "classification": "error",
                "confidence": 0,
                "feedback": f"Error: {str(e)}",
                "method": "QSVM",
                "error": str(e)
            }

    def _generate_feedback(self, score: float, props: Dict) -> str:
        """Generate human-readable feedback."""
        feedback = []

        if score >= 85:
            feedback.append("Excellent catalyst choice!")
        elif score >= 70:
            feedback.append("Good catalyst for this reaction.")
        elif score >= 50:
            feedback.append("Fair catalyst, but better options exist.")
        else:
            feedback.append("Poor catalyst choice for this reaction.")

        if props.get("is_metal"):
            metal = props.get("metal_type")
            if self.reaction_type == "H2_O2" and metal in ["Pt", "Pd"]:
                feedback.append(f"{metal} is excellent for oxidation reactions.")
            elif self.reaction_type == "N2_H2" and metal in ["Fe", "Ru"]:
                feedback.append(f"{metal} is great for nitrogen fixation.")

        return " ".join(feedback)


# ========================================================================
# VARIATIONAL QUANTUM CLASSIFIER (VQC)
# ========================================================================

class VariationalCatalystClassifier:
    """

    VQC for multi-class catalyst classification.



    Uses parameterized quantum circuits to classify catalysts into categories:

    - Oxidation catalysts

    - Reduction catalysts

    - Hydrogenation catalysts

    - Inert/Poor catalysts

    """

    def __init__(self):
        self.num_features = FEATURE_DIMENSION
        self.feature_map = ZZFeatureMap(self.num_features, reps=1)
        self.ansatz = RealAmplitudes(self.num_features, reps=3)
        self.trained = False

    def train(self, X: np.ndarray, y: np.ndarray):
        """

        Train VQC classifier.



        Args:

            X: Feature matrix (N x 8)

            y: Labels (N,) with values 0-3

        """
        try:
            sampler = StatevectorSampler()
            optimizer = COBYLA(maxiter=50)

            # Note: VQC implementation requires qiskit-machine-learning
            # Simplified version for this demo
            self.training_data = (X, y)
            self.trained = True

            return {"success": True, "error": ""}
        except Exception as e:
            return {"success": False, "error": str(e)}

    def classify(self, smiles: str) -> Dict:
        """

        Classify catalyst into categories.



        Returns:

            Dictionary with classification results

        """
        features = extract_molecular_features(smiles)
        is_valid_features, reason = _validate_feature_vector(features)
        props = get_catalyst_properties(smiles)

        # Simplified classification logic
        # In a full implementation, this would use the trained VQC
        if not props["valid"]:
            return {"category": "invalid", "confidence": 0, "error": "Invalid molecule"}

        if not is_valid_features:
            return {
                "category": "invalid_features",
                "confidence": 0,
                "method": "VQC (Variational Quantum Classifier)",
                "error": reason
            }

        if props["is_metal"]:
            metal = props["metal_type"]
            if metal in ["Pt", "Pd", "Ni"]:
                category = "oxidation"
                confidence = 85
            elif metal in ["Fe", "Ru", "Co"]:
                category = "reduction"
                confidence = 80
            elif metal in ["Cu", "Ag"]:
                category = "hydrogenation"
                confidence = 75
            else:
                category = "general"
                confidence = 60
        else:
            category = "poor"
            confidence = 40

        return {
            "category": category,
            "confidence": confidence,
            "method": "VQC (Variational Quantum Classifier)",
            "error": ""
        }


# ========================================================================
# QUANTUM GENERATIVE ADVERSARIAL NETWORK (QGAN)
# ========================================================================

class CatalystGenerator:
    """

    QGAN for generating new catalyst candidates.



    Uses quantum circuits to generate feature vectors that represent

    potentially good catalysts.

    """

    def __init__(self, target_reaction: str):
        """

        Initialize QGAN for catalyst generation.



        Args:

            target_reaction: Target reaction type

        """
        self.target_reaction = target_reaction
        self.num_qubits = 8
        self.generator = self._build_generator()

    def _build_generator(self) -> QuantumCircuit:
        """

        Build generator circuit.



        The generator creates quantum states that map to catalyst features.

        """
        qc = QuantumCircuit(self.num_qubits)

        # Stochastic latent-space initialization (QGAN-style exploration).
        for i in range(self.num_qubits):
            qc.ry(random.uniform(0.0, 2 * np.pi), i)
            qc.rz(random.uniform(0.0, 2 * np.pi), i)

        # Entanglement
        for i in range(self.num_qubits - 1):
            qc.cx(i, i + 1)

        # More parameterization
        for i in range(self.num_qubits):
            qc.ry(random.uniform(0.0, 2 * np.pi), i)

        return qc

    def _bitstring_to_action(self, bitstring: str) -> str:
        """Map measured quantum bitstrings to mutation actions."""
        action_map = {
            "00": "metal_swap",
            "01": "ligand_oxo",
            "10": "ligand_hydroxyl",
            "11": "doping",
        }
        return action_map.get(bitstring[:2], "metal_swap")

    def _sample_quantum_actions(self, num_actions: int) -> List[str]:
        """Generate mutation actions by probabilistic statevector sampling."""
        state = Statevector(self.generator)
        probabilities = state.probabilities_dict()

        bitstrings = list(probabilities.keys())
        probs = np.array(list(probabilities.values()), dtype=float)
        if probs.sum() <= 0:
            probs = np.ones(len(bitstrings), dtype=float) / max(1, len(bitstrings))
        else:
            probs = probs / probs.sum()

        sampled = np.random.choice(bitstrings, size=max(1, num_actions), p=probs, replace=True)
        return [self._bitstring_to_action(bits) for bits in sampled.tolist()]

    def generate_candidates(self, num_candidates: int = 5) -> List[Dict]:
        """

        Generate new catalyst candidates.



        Args:

            num_candidates: Number of candidates to generate



        Returns:

            List of candidate dictionaries with properties

        """
        candidates: List[Dict] = []

        reaction_bases = {
            "H2_O2": "[Pt]",
            "N2_H2": "[Fe]",
            "CO2_reduction": "[Cu]",
        }
        base_catalyst = reaction_bases.get(self.target_reaction, "[Pd]")

        action_sequence = self._sample_quantum_actions(max(8, num_candidates * 4))

        generated_smiles: List[str] = []
        seen = set()
        for action in action_sequence:
            for smiles in mutate_catalyst(base_catalyst, num_variations=4, mutation_mode=action):
                if smiles not in seen:
                    seen.add(smiles)
                    generated_smiles.append(smiles)
                if len(generated_smiles) >= max(8, num_candidates * 4):
                    break
            if len(generated_smiles) >= max(8, num_candidates * 4):
                break

        # Backstop: if stochastic draws under-sample action diversity, sweep all modes.
        if len(generated_smiles) < num_candidates:
            for action in ["metal_swap", "ligand_oxo", "ligand_hydroxyl", "ligand_methyl", "doping"]:
                for smiles in mutate_catalyst(base_catalyst, num_variations=6, mutation_mode=action):
                    if smiles not in seen:
                        seen.add(smiles)
                        generated_smiles.append(smiles)
                    if len(generated_smiles) >= max(8, num_candidates * 4):
                        break
                if len(generated_smiles) >= max(8, num_candidates * 4):
                    break

        if not generated_smiles:
            generated_smiles = [base_catalyst]

        for smiles in generated_smiles:
            features = extract_molecular_features(smiles)
            is_valid_features, _ = _validate_feature_vector(features)
            if not is_valid_features:
                continue

            props = get_catalyst_properties(smiles)
            candidates.append({
                "smiles": smiles,
                "features": features.tolist(),
                "metal_type": props.get("metal_type"),
                "generation_score": float(0.65 + 0.3 * np.mean(features)),
                "method": "Quantum Statevector-Driven Mutation"
            })

        # Ensure minimum output size for downstream UI.
        if not candidates:
            features = extract_molecular_features(base_catalyst)
            props = get_catalyst_properties(base_catalyst)
            candidates.append({
                "smiles": base_catalyst,
                "features": features.tolist(),
                "metal_type": props.get("metal_type"),
                "generation_score": float(0.65 + 0.3 * np.mean(features)),
                "method": "Quantum Statevector-Driven Mutation"
            })

        candidates.sort(key=lambda x: x["generation_score"], reverse=True)
        return candidates[:max(1, num_candidates)]


# ========================================================================
# MAIN API FUNCTIONS
# ========================================================================

def score_user_catalyst(user_smiles: str, ideal_smiles: str, reaction_type: str) -> Dict:
    """

    Score user's catalyst guess against the ideal catalyst.



    Uses QSVM + VQC for comprehensive scoring.



    Args:

        user_smiles: User's catalyst guess

        ideal_smiles: Ideal catalyst for comparison

        reaction_type: Type of reaction



    Returns:

        Comprehensive scoring report

    """
    # QSVM scoring
    scorer = QuantumCatalystScorer(reaction_type)
    qsvm_result = scorer.score_catalyst(user_smiles)

    # VQC classification
    classifier = VariationalCatalystClassifier()
    vqc_result = classifier.classify(user_smiles)

    # Compare features
    user_features = extract_molecular_features(user_smiles)
    ideal_features = extract_molecular_features(ideal_smiles)

    user_valid, user_reason = _validate_feature_vector(user_features)
    if not user_valid:
        return {
            "overall_score": 0,
            "qsvm_score": 0,
            "qsvm_feedback": f"Feature extraction failed for user catalyst: {user_reason}",
            "vqc_category": "invalid_features",
            "feature_similarity": 0,
            "classification": "error",
            "error": user_reason
        }

    ideal_valid, ideal_reason = _validate_feature_vector(ideal_features)
    if not ideal_valid:
        return {
            "overall_score": 0,
            "qsvm_score": qsvm_result.get("score", 0),
            "qsvm_feedback": "Ideal catalyst reference features are invalid.",
            "vqc_category": "invalid_features",
            "feature_similarity": 0,
            "classification": "error",
            "error": ideal_reason
        }

    distance = np.linalg.norm(user_features - ideal_features)
    feature_similarity = float(np.clip(1.0 - distance / np.sqrt(FEATURE_DIMENSION), 0.0, 1.0))

    # Overall score
    overall_score = (qsvm_result["score"] * 0.6 + feature_similarity * 40)

    return {
        "overall_score": round(overall_score, 2),
        "qsvm_score": qsvm_result["score"],
        "qsvm_feedback": qsvm_result["feedback"],
        "vqc_category": vqc_result["category"],
        "feature_similarity": round(feature_similarity * 100, 2),
        "classification": qsvm_result["classification"],
        "error": ""
    }


def discover_catalysts(reaction_type: str, num_candidates: int = 5) -> List[Dict]:
    """

    Discover new catalyst candidates using QGAN + VQC + VQE.



    Args:

        reaction_type: Target reaction

        num_candidates: Number of candidates to generate



    Returns:

        List of evaluated catalyst candidates

    """
    # Generate candidates with QGAN
    generator = CatalystGenerator(reaction_type)
    candidates = generator.generate_candidates(num_candidates)

    # Score each candidate
    scorer = QuantumCatalystScorer(reaction_type)

    for candidate in candidates:
        score_result = scorer.score_catalyst(candidate["smiles"])
        candidate.update({
            "catalyst_score": score_result["score"],
            "classification": score_result["classification"],
            "feedback": score_result["feedback"]
        })

    # Sort by score
    candidates.sort(key=lambda x: x["catalyst_score"], reverse=True)

    return candidates