"""A synthetic engineering fixture. These data are not cellular measurements.""" from pathlib import Path import numpy as np import pandas as pd from .chemistry import canonical_smiles, molecule_id, fingerprints from .splits import make_splits from .io import ResponseData, write_json from .models import train from .pipeline import calibrate, evaluate from .plotting import plot_summary def synthetic_data(seed=0, molecules=48, dimensions=8): rng = np.random.default_rng(seed) smiles = [] for chain in range(1, 9): for ring in ["c1ccccc1", "c1ccncc1", "c1ccoc1", "c1ccsc1", "C1CCCCC1", "N1CCCCC1"]: smiles.append(canonical_smiles("C"*chain+ring)) smiles = sorted(set(smiles))[:molecules] splits = make_splits(pd.DataFrame({"smiles": smiles}), seed, (.55,.1,.2,.15)) fp = fingerprints(smiles) projection = rng.normal(size=(fp.shape[1], dimensions))/np.sqrt(fp.shape[1]) effects = fp@projection rows, response, context, tv, cv = [], [], [], [], [] for c in ["context_A", "context_B"]: baseline = rng.normal(size=dimensions)*.2 for block in ["rep1", "rep2"]: ctrl = baseline+rng.normal(size=dimensions)*.015 for m, s in enumerate(smiles): for dose in [.05,.5,5.]: effect = effects[m]*(1+.25*np.log(dose))+.05*baseline observed = effect+rng.normal(size=dimensions)*.02-(ctrl-baseline) rows.append({"molecule_id": molecule_id(s), "smiles": s, "drug": f"synthetic_{m}", "dose_um": dose, "context": c, "block": block, "control_id": c+"|"+block, "split": splits.set_index("smiles").loc[s,"split"], "n_cells": 64, "n_control_cells": 128}) response.append(observed); context.append(ctrl) tv.append(np.full(dimensions,.02**2)); cv.append(np.full(dimensions,.015**2)) return ResponseData(pd.DataFrame(rows), np.array(response), np.array(context), np.array(tv), np.array(cv), {"source": "synthetic engineering fixture", "feature_space_id": "synthetic-v1", "cell_features": {"representation": "synthetic", "dimensions": dimensions}, "biological_data": False}).validate() def run_demo(output, seed=0, max_queries=8): output = Path(output); output.mkdir(parents=True, exist_ok=True) data = synthetic_data(seed) data.save(output/"data") model = train(data, output/"model", ensemble=3, folds=3, seed=seed) cal = calibrate(data, model, output/"calibration", panel_molecules=5, scenarios=16, seed=seed) frame = evaluate(data, model, cal, output/"evaluation", scenarios=16, seed=seed, max_queries=max_queries, methods=["mean","gibbs","marginal","finite-minimax","remedi"]) plot_summary(output/"evaluation/summary.csv", output/"plots", title="Synthetic engineering check") write_json(output/"run.json", {"source": "synthetic", "biological_validation": False, "queries": int(frame.query_id.nunique()), "seed": seed, "purpose": "Verify serialization, grouped training, calibration, optimization, evaluation, and plots"}) return output