from __future__ import annotations import logging from typing import Any from rdkit import Chem from rdkit.Chem import Descriptors from .materials import resolve_material from .models import Ingredient, IngredientResolutionError from .registry import AromaRegistry logger = logging.getLogger("pino.structures") class IngredientResolver: """ High-level resolver that combines a pre-computed SQLite registry with a live fallback to RDKit / PubChem / ugropy. Resolution order: 1. AromaRegistry lookup (fast, no network) 2. Live PubChem CAS -> SMILES + MW 3. Live ugropy Dortmund-UNIFAC fragmentation """ def __init__(self, registry: AromaRegistry | None = None) -> None: self.registry = registry or AromaRegistry() def resolve(self, identifier: str, weight_fraction: float, *, name: str | None = None) -> Ingredient: """Resolve a single raw formula component into a validated Ingredient.""" # Registry lookup first record = self.registry.get(identifier) if record is None and name: record = self.registry.get(name) if record: logger.debug("Resolved %s from registry", identifier) return self._ingredient_from_record(record, weight_fraction) # Known standard-material fallback: trade names in materials.py provide # a canonical SMILES, so we can build a record without a network round-trip. for lookup in (name, identifier): if not lookup: continue material = resolve_material(lookup) if material and not material.get("accord"): smiles = material.get("smiles") if smiles: logger.debug("Resolved %s from standard materials library", lookup) record = AromaRegistry.validate_smiles(smiles) if record is None: raise IngredientResolutionError( f"Standard material {lookup} has an invalid SMILES", cas=identifier ) record["name"] = material.get("name", lookup) record["cas"] = material.get("cas", identifier) if material.get("vapor_pressure_pa") is not None: record["vapor_pressure_pa"] = material["vapor_pressure_pa"] if material.get("boiling_point_k") is not None: record["boiling_point_k"] = material["boiling_point_k"] if material.get("odor_threshold_ug_m3") is not None: record["odor_threshold_ug_m3"] = material["odor_threshold_ug_m3"] self.registry.add(record) return self._ingredient_from_record(record, weight_fraction) # Live fallback logger.info("Registry miss for %s; falling back to PubChem/ugropy", identifier) try: live = AromaRegistry.build_record(identifier, "cas") except Exception as exc: raise IngredientResolutionError( f"Unable to resolve {identifier} from registry or live sources: {exc}", cas=identifier, ) from exc # Add to registry so future lookups are fast self.registry.add(live) return self._ingredient_from_record(live, weight_fraction) def resolve_formula( self, formula: list[dict[str, Any]], *, skip_unresolved: bool = False, ) -> tuple[list[Ingredient], list[IngredientResolutionError]]: """ Resolve a list of raw formula components. Returns (ingredients, errors). If skip_unresolved is False, the first error is raised immediately. """ ingredients: list[Ingredient] = [] errors: list[IngredientResolutionError] = [] for comp in formula: if "weight_fraction" not in comp: raise ValueError(f"Formula component missing weight_fraction: {comp}") wf = float(comp["weight_fraction"]) ident = str(comp.get("cas", comp.get("smiles", comp.get("name", "")))) try: ing = self.resolve(ident, wf, name=comp.get("name")) ingredients.append(ing) except IngredientResolutionError as exc: logger.warning("Out of applicability domain for %s: %s", ident, exc) errors.append(exc) if not skip_unresolved: raise return ingredients, errors def resolve_formula_pairs( self, formula: list[dict[str, Any]], *, skip_unresolved: bool = False, ) -> tuple[list[tuple[dict[str, Any], Ingredient]], list[IngredientResolutionError]]: """ Resolve formula components and return (record, ingredient) pairs. This is useful when the caller needs to keep the original record alongside the resolved ingredient so that weight fractions stay aligned after filtering out unresolved materials. """ pairs: list[tuple[dict[str, Any], Ingredient]] = [] errors: list[IngredientResolutionError] = [] for comp in formula: if "weight_fraction" not in comp: raise ValueError(f"Formula component missing weight_fraction: {comp}") wf = float(comp["weight_fraction"]) ident = str(comp.get("cas", comp.get("smiles", comp.get("name", "")))) try: ing = self.resolve(ident, wf, name=comp.get("name")) pairs.append((comp, ing)) except IngredientResolutionError as exc: logger.warning("Out of applicability domain for %s: %s", ident, exc) errors.append(exc) if not skip_unresolved: raise return pairs, errors def _ingredient_from_record(self, record: dict[str, Any], weight_fraction: float) -> Ingredient: """Construct an Ingredient from a registry/live record.""" smiles = record["smiles"] mw = record.get("molecular_weight") if mw is None: mw = molecular_weight_from_smiles(smiles) logp = record.get("logp") if logp is None: try: mol = Chem.MolFromSmiles(smiles) logp = float(Descriptors.MolLogP(mol)) if mol else None except Exception: logp = None return Ingredient( name=record.get("name", smiles), cas=record.get("cas", ""), smiles=smiles, molecular_weight=float(mw), vapor_pressure_pa=float(record.get("vapor_pressure_pa") or 1.0), boiling_point_k=float(record["boiling_point_k"]) if record.get("boiling_point_k") is not None else None, odor_threshold_ug_m3=float(record["odor_threshold_ug_m3"]) if record.get("odor_threshold_ug_m3") is not None else None, logp=logp, unifac_groups={str(k): int(v) for k, v in record.get("unifac_groups", {}).items()}, ) def canonicalize_smiles(smiles: str) -> str: mol = Chem.MolFromSmiles(smiles) if mol is None: raise IngredientResolutionError(f"Invalid SMILES: {smiles}", smiles=smiles) return Chem.MolToSmiles(mol, canonical=True) def molecular_weight_from_smiles(smiles: str) -> float: mol = Chem.MolFromSmiles(smiles) if mol is None: raise IngredientResolutionError(f"Invalid SMILES: {smiles}", smiles=smiles) return Descriptors.MolWt(mol) def ingredient_from_smiles( name: str, cas: str, smiles: str, vapor_pressure_pa: float, boiling_point_k: float | None = None, odor_threshold_ug_m3: float | None = None, ) -> Ingredient: """Build an Ingredient directly from a SMILES string (with live fragmentation).""" resolver = IngredientResolver() record = AromaRegistry.build_record(smiles, "smiles", vapor_pressure_pa, boiling_point_k, odor_threshold_ug_m3) record["name"] = name record["cas"] = cas return resolver._ingredient_from_record(record, 0.0) def ingredient_from_data(data: dict[str, Any]) -> Ingredient: """Build an Ingredient from a dict (with optional precomputed registry groups).""" resolver = IngredientResolver() if "cas" in data or "smiles" in data or "name" in data: return resolver.resolve( str(data.get("cas", data.get("smiles", data.get("name", "")))), float(data.get("weight_fraction", 0.0)), name=data.get("name"), ) raise IngredientResolutionError("No identifier (cas, smiles, name) provided in data")