| """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 |
|
|
|
|
