feat(optimizer): masked MSE, strategy seeds, adversarial IFRA guard; add accords/materials/importer
fd5ba83 | 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") | |