"""Constraints and candidate generation.
A recommendation engine is only useful if it can propose experiments a lab can
actually run. ``CandidateGenerator`` builds a candidate pool directly from the
campaign's observed design space - recombining categorical choices and
perturbing/resampling numeric conditions - subject to practical constraints:
* numeric bounds per column,
* allowed/forbidden category values,
* fixed columns (e.g. your reactor's fixed pressure),
* mixture normalization: composition fractions rescaled to a total
(e.g. mol% summing to 100).
This turns DENIA from "rank my list" into "suggest what to try next" without
requiring the user to enumerate a pool.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Sequence, Tuple
import numpy as np
import pandas as pd
from .schema import CampaignSchema
[docs]
@dataclass
class Constraints:
bounds: Dict[str, Tuple[float, float]] = field(default_factory=dict) # numeric column -> (lo, hi)
allowed: Dict[str, Sequence] = field(default_factory=dict) # category column -> allowed values
forbidden: Dict[str, Sequence] = field(default_factory=dict) # category column -> excluded values
fixed: Dict[str, object] = field(default_factory=dict) # column -> fixed value
mixture_total: Optional[float] = None # e.g. 100.0 for mol%
def filter_categories(self, col: str, values: List) -> List:
vals = [v for v in values if not (isinstance(v, float) and pd.isna(v))]
if col in self.allowed:
vals = [v for v in vals if v in set(self.allowed[col])]
if col in self.forbidden:
vals = [v for v in vals if v not in set(self.forbidden[col])]
return vals
[docs]
class CandidateGenerator:
"""Generate candidate experiments from the observed design space.
Strategy per candidate: pick a random history row as a template, then
mutate it - resample categorical slots from observed values, jitter
numeric conditions within observed (or constrained) ranges, and
renormalize mixtures. Simple, transparent, and guaranteed to stay in a
region the featurizer understands.
"""
def __init__(
self,
schema: CampaignSchema,
constraints: Optional[Constraints] = None,
random_state: int = 0,
):
self.schema = schema
self.constraints = constraints or Constraints()
self.rng = np.random.default_rng(random_state)
def generate(self, history: pd.DataFrame, n: int = 500, mutation_rate: float = 0.4) -> pd.DataFrame:
c = self.constraints
pairs = self.schema.composition_pairs()
paired_cats = {cat for cat, _ in pairs}
cat_pools = {
col: c.filter_categories(col, list(pd.unique(history[col].dropna())))
for col in self.schema.category_inputs
}
num_ranges = {}
for col in self.schema.numeric_inputs:
s = pd.to_numeric(history[col], errors="coerce")
lo, hi = float(s.min()), float(s.max())
if col in c.bounds:
lo, hi = max(lo, c.bounds[col][0]), min(hi, c.bounds[col][1])
num_ranges[col] = (lo, hi)
rows = []
templates = history.sample(n=n, replace=True, random_state=int(self.rng.integers(1 << 31)))
for _, tpl in templates.iterrows():
row = tpl.to_dict()
# categorical: mutate with some probability, and always enforce constraints
for col in self.schema.category_inputs:
pool = cat_pools.get(col)
if not pool:
continue
invalid = row[col] not in pool and not pd.isna(row[col])
if invalid or self.rng.random() < mutation_rate:
row[col] = pool[int(self.rng.integers(len(pool)))]
# numeric: mutate with some probability, and always clip into range
for col, (lo, hi) in num_ranges.items():
if hi <= lo:
row[col] = lo
continue
val = float(pd.to_numeric(pd.Series([row[col]]), errors="coerce").iloc[0])
if not np.isfinite(val):
val = (lo + hi) / 2
if self.rng.random() < mutation_rate:
val += self.rng.normal(0, 0.15 * (hi - lo))
row[col] = float(np.clip(val, lo, hi))
# fraction mutation + mixture renormalization
frac_cols = [fr for _, fr in pairs]
if frac_cols:
fr_vals = np.array([
float(pd.to_numeric(pd.Series([row[fr]]), errors="coerce").iloc[0] or 0.0)
if not pd.isna(row.get(fr)) else 0.0
for fr in frac_cols
])
mut = self.rng.random(len(fr_vals)) < mutation_rate
fr_vals[mut] = np.abs(fr_vals[mut] + self.rng.normal(0, 0.2 * (fr_vals.max() + 1), mut.sum()))
total = c.mixture_total
if total and fr_vals.sum() > 0:
fr_vals = fr_vals / fr_vals.sum() * total
for fr, v in zip(frac_cols, fr_vals):
row[fr] = float(v)
# fixed columns and target blanking
for col, v in c.fixed.items():
row[col] = v
for t in self.schema.targets:
row[t] = np.nan
rows.append(row)
out = pd.DataFrame(rows, columns=history.columns).reset_index(drop=True)
return out.drop_duplicates(subset=self.schema.inputs).reset_index(drop=True)