File size: 9,579 Bytes
b233cf7 0e7a065 2671b56 b233cf7 2671b56 b233cf7 2671b56 fd5ba83 b233cf7 fd5ba83 b233cf7 0e7a065 b233cf7 75c66ff b233cf7 75c66ff b233cf7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 | from __future__ import annotations
import json
import logging
from pathlib import Path
from typing import Any
import numpy as np
from .evaporation import EvaporationModel
from .ifra import check_ifra_restrictions
from .models import Formula, Ingredient, IngredientResolutionError, Status, Trajectory
from .structures import IngredientResolver
from .vle import VLECalculator
logger = logging.getLogger("pino.verifier")
class FragrancePipelineVerifier:
"""
High-level end-to-end verifier for synthetic fragrance formulations.
Ingests a list of {cas, weight_fraction} records, resolves chemical
attributes, checks safety constraints, and runs the stiff ODE evaporation
model to produce a structured trajectory result.
"""
def __init__(
self,
temperature_k: float = 298.15,
ambient_pressure_pa: float = 101325.0,
headspace_volume_m3: float = 1e-3,
liquid_volume_m3: float = 1e-6,
density_g_ml: float = 0.9,
surface_area_m2: float = 1e-4,
mass_transfer_coefficient: float = 1e-4,
) -> None:
self.temperature_k = temperature_k
self.ambient_pressure_pa = ambient_pressure_pa
self.headspace_volume_m3 = headspace_volume_m3
self.liquid_volume_m3 = liquid_volume_m3
self.density_g_ml = density_g_ml
self.surface_area_m2 = surface_area_m2
self.mass_transfer_coefficient = mass_transfer_coefficient
self.resolver = IngredientResolver()
def run_sim(
self,
formula: list[dict[str, Any]],
duration_seconds: float = 3600.0,
interval_seconds: float = 600.0,
method: str = "BDF",
skip_ifra: bool = False,
skip_unresolved: bool = False,
) -> dict[str, Any]:
"""
Run the full verification pipeline on a draft formula.
Returns a dict with:
- status: "passed" or "rejected"
- trajectory: list of per-time-step records
- depletion_rates: dict of CAS -> fractional depletion
- ifra_report: IFRA restriction check
- applicability_errors: list of out-of-domain errors
- message: human-readable summary
"""
# Resolve ingredients. By default, a verifier pass means every supplied
# component was inside the applicability domain. Dataset builders may
# opt into partial simulation after preserving unresolved rows upstream.
try:
resolved_pairs, errors = self.resolver.resolve_formula_pairs(formula, skip_unresolved=skip_unresolved)
if errors:
for exc in errors:
logger.warning("Skipping unresolved ingredient: %s", exc)
except IngredientResolutionError as exc:
logger.warning("Formula rejected: out of applicability domain for %s", exc.cas)
return {
"status": "rejected",
"trajectory": [],
"depletion_rates": {},
"ifra_report": {"passed": False, "violations": [], "max_usage": []},
"applicability_errors": [
{"cas": exc.cas, "smiles": exc.smiles, "reason": str(exc)}
],
"message": f"Out of applicability domain: {exc}",
}
if not resolved_pairs:
return {
"status": "rejected",
"trajectory": [],
"depletion_rates": {},
"ifra_report": {"passed": False, "violations": [], "max_usage": []},
"applicability_errors": [],
"message": "No resolvable ingredients in formula",
}
ingredients = [ing for _, ing in resolved_pairs]
weights = [float(rec.get("weight_fraction", 0.0)) for rec, _ in resolved_pairs]
# Re-normalize in case skipped ingredients changed total
total_w = sum(weights)
if total_w <= 0:
return {
"status": "rejected",
"trajectory": [],
"depletion_rates": {},
"ifra_report": {"passed": False, "violations": [], "max_usage": []},
"applicability_errors": [],
"message": "Formula has no positive weight fractions",
}
weights = [w / total_w for w in weights]
resolved_formula = Formula(
ingredients=tuple(ingredients),
weight_fractions=tuple(weights),
temperature_k=self.temperature_k,
ambient_pressure_pa=self.ambient_pressure_pa,
headspace_volume_m3=self.headspace_volume_m3,
liquid_volume_m3=self.liquid_volume_m3,
metadata={"density_g_ml": self.density_g_ml},
)
# IFRA check
ifra_report = check_ifra_restrictions(resolved_formula)
if not skip_ifra and not ifra_report["passed"]:
logger.warning("Formula rejected by IFRA restrictions: %s", ifra_report["violations"])
return {
"status": "rejected",
"trajectory": [],
"depletion_rates": {},
"ifra_report": ifra_report,
"applicability_errors": [],
"message": f"IFRA restriction violation: {ifra_report['violations']}",
}
# Run evaporation model
vle = VLECalculator(resolved_formula)
model = EvaporationModel(
resolved_formula,
vle=vle,
surface_area_m2=self.surface_area_m2,
mass_transfer_coefficient=self.mass_transfer_coefficient,
)
n_steps = max(2, int(duration_seconds / interval_seconds) + 1)
t_eval = np.linspace(0.0, duration_seconds, n_steps)
trajectory = model.solve(
t_span=(0.0, duration_seconds),
t_eval=t_eval,
method=method,
)
# Compute depletion rates per CAS
depletion_rates = self._depletion_rates(trajectory)
# Convert trajectory to list of dicts
trajectory_records = self._trajectory_to_records(trajectory)
# Detect film-depletion terminal event
if trajectory.t[-1] < duration_seconds:
status = Status.DEPLETED
message = f"Film depleted at t={trajectory.t[-1]:.1f}s before horizon {duration_seconds:.1f}s"
else:
status = Status.PASSED
message = "Formula verified successfully"
return {
"status": status,
"trajectory": trajectory_records,
"depletion_rates": depletion_rates,
"ifra_report": ifra_report,
"applicability_errors": [],
"message": message,
}
@staticmethod
def _depletion_rates(trajectory: Trajectory) -> dict[str, float]:
"""True fractional liquid molar depletion."""
rates: dict[str, float] = {}
n0 = trajectory.n_liquid[:, 0]
n1 = trajectory.n_liquid[:, -1]
for idx, ing in enumerate(trajectory.formula.ingredients):
if n0[idx] > 0:
rates[ing.cas] = float((n0[idx] - n1[idx]) / n0[idx])
else:
rates[ing.cas] = 0.0
return rates
@staticmethod
def _trajectory_to_records(trajectory: Trajectory) -> list[dict[str, Any]]:
"""Convert a Trajectory object into a list of JSON-serializable records keyed by CAS."""
records: list[dict[str, Any]] = []
species = [ing.cas for ing in trajectory.formula.ingredients]
for i, t in enumerate(trajectory.t):
records.append({
"t_s": float(t),
"x_liquid": {s: float(trajectory.x_liquid[j, i]) for j, s in enumerate(species)},
"C_gas_mg_m3": {s: float(trajectory.C_gas[j, i]) for j, s in enumerate(species)},
"y_gas": {s: float(trajectory.y_gas[j, i]) for j, s in enumerate(species)},
"OAV": {s: float(trajectory.OAV[j, i]) for j, s in enumerate(species)},
})
return records
class FormulaVerifier:
"""Original high-level verifier: intake a draft formula and emit a JSON trajectory."""
def __init__(self, formula: Formula) -> None:
self.formula = formula
self.vle = VLECalculator(formula)
self.model = EvaporationModel(formula, vle=self.vle)
def verify(self, **solve_kwargs: Any) -> Trajectory:
return self.model.solve(**solve_kwargs)
def verify_and_write(
self,
output_path: str | Path = "trajectory_output.json",
**solve_kwargs: Any,
) -> Path:
trajectory = self.verify(**solve_kwargs)
path = Path(output_path)
path.write_text(json.dumps(trajectory.to_dict(), indent=2))
return path
def verify_formula_dict(data: dict[str, Any], **solve_kwargs: Any) -> Trajectory:
"""Convenience entrypoint: build a Formula from a dict and run the verifier."""
from .structures import ingredient_from_data
ingredients = [ingredient_from_data(i) for i in data["ingredients"]]
formula = Formula(
ingredients=tuple(ingredients),
weight_fractions=tuple(data["weight_fractions"]),
temperature_k=float(data.get("temperature_k", 298.15)),
ambient_pressure_pa=float(data.get("ambient_pressure_pa", 101325.0)),
headspace_volume_m3=float(data.get("headspace_volume_m3", 1e-3)),
liquid_volume_m3=float(data.get("liquid_volume_m3", 1e-6)),
metadata=data.get("metadata", {}),
)
verifier = FormulaVerifier(formula)
return verifier.verify(**solve_kwargs)
|