Coverage for src/instawell/cli.py: 0%
200 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-07 15:47 -0600
1"""
2Command-line interface for InstaWell.
4Provides an intuitive CLI for running DSF data analysis pipelines.
5"""
7import json
8import sys
9from pathlib import Path
11import click
13from instawell import (
14 StepFiles,
15 average_accross_replicates,
16 calculate_derivative,
17 filter_wells,
18 find_min_temperature,
19 ingest_data,
20 load_experiment_context,
21 min_max_scale,
22 setup_experiment,
23 subtract_background,
24)
27@click.group()
28@click.version_option(package_name="instawell")
29def cli():
30 """
31 InstaWell - Differential Scanning Fluorimetry (DSF) Data Analysis.
33 A powerful toolkit for processing thermal shift assay experiments.
35 \b
36 Quick Start:
37 1. Create experiment: instawell init my_exp --raw data.csv --layout layout.csv
38 2. Run full pipeline: instawell run my_exp --all
39 3. View results: instawell info my_exp
41 \b
42 Common Workflows:
43 # Run specific steps
44 instawell run my_exp --steps ingest,filter,average
46 # Filter wells interactively
47 instawell filter my_exp --wells A1,B2,C3
49 # List all experiments
50 instawell list
51 """
52 pass
55@cli.command()
56@click.argument("experiment_name")
57@click.option("--raw", "-r", "raw_data_path", required=True, type=click.Path(exists=True),
58 help="Path to raw fluorescence data CSV file")
59@click.option("--layout", "-l", "layout_path", required=True, type=click.Path(exists=True),
60 help="Path to plate layout CSV file")
61@click.option("--root", default="experiments", type=click.Path(),
62 help="Root directory for experiments (default: ./experiments)")
63@click.option("--separator", default="_", type=str,
64 help="Condition separator character (default: _)")
65@click.option("--fields", default="concentration,ligand,protein,buffer", type=str,
66 help="Comma-separated field order (default: concentration,ligand,protein,buffer)")
67@click.option("--temp-col", default="Temperature", type=str,
68 help="Temperature column name (default: Temperature)")
69def init(experiment_name, raw_data_path, layout_path, root, separator, fields, temp_col):
70 """
71 Initialize a new experiment.
73 Creates experiment directory and copies data files.
75 \b
76 Example:
77 instawell init TSA_001 --raw raw.csv --layout layout.csv
78 instawell init TSA_002 -r data.csv -l plate.csv --separator '|'
79 """
80 click.echo(f"🔬 Initializing experiment: {click.style(experiment_name, fg='cyan', bold=True)}")
82 try:
83 # Parse fields
84 fields_tuple = tuple(f.strip() for f in fields.split(","))
86 # Validate separator
87 if len(separator) != 1:
88 raise click.BadParameter("Separator must be exactly one character")
90 # Setup experiment
91 ctx = setup_experiment(
92 experiment_name=experiment_name,
93 raw_data_path=raw_data_path,
94 layout_data_path=layout_path,
95 experiments_root=root,
96 condition_separator=separator,
97 fields=fields_tuple,
98 temperature_column=temp_col,
99 )
101 click.echo(f"✅ Experiment created at: {click.style(str(ctx.experiment_dir), fg='green')}")
102 click.echo(f" 📁 Raw data: {ctx.raw_data_path.name}")
103 click.echo(f" 📁 Layout: {ctx.layout_data_path.name}")
104 click.echo(f" ⚙️ Separator: '{separator}'")
105 click.echo(f" ⚙️ Fields: {', '.join(fields_tuple)}")
106 click.echo()
107 click.echo(f"💡 Next step: Run pipeline with 'instawell run {experiment_name} --all'")
109 except Exception as e:
110 click.echo(f"❌ Error: {e}", err=True)
111 sys.exit(1)
114@cli.command()
115@click.argument("experiment_name")
116@click.option("--all", "-a", "run_all", is_flag=True,
117 help="Run all pipeline steps")
118@click.option("--steps", "-s", type=str,
119 help="Comma-separated list of steps: ingest,filter,average,background,scale,derivative,min_temp")
120@click.option("--filter-wells", type=str,
121 help="Comma-separated list of wells to filter (e.g., A1,B2)")
122@click.option("--root", default="experiments", type=click.Path(),
123 help="Root directory for experiments")
124def run(experiment_name, run_all, steps, filter_wells, root):
125 """
126 Run processing pipeline steps.
128 \b
129 Steps:
130 ingest - Parse layout and organize raw data
131 filter - Remove problematic wells
132 average - Average technical replicates
133 background - Subtract background signal
134 scale - Normalize to 0-1 range
135 derivative - Calculate -dY/dT
136 min_temp - Find melting temperatures
138 \b
139 Examples:
140 # Run entire pipeline
141 instawell run TSA_001 --all
143 # Run specific steps
144 instawell run TSA_001 --steps ingest,filter,average
146 # Run with well filtering
147 instawell run TSA_001 --all --filter-wells A1,B2,G20
148 """
149 try:
150 # Load experiment
151 ctx = load_experiment_context(experiment_name, experiments_root=root)
152 click.echo(f"🔬 Running pipeline for: {click.style(experiment_name, fg='cyan', bold=True)}")
154 # Parse filter wells
155 wells_to_filter = []
156 if filter_wells:
157 wells_to_filter = [w.strip() for w in filter_wells.split(",")]
159 # Determine which steps to run
160 if run_all:
161 step_names = ["ingest", "filter", "average", "background", "scale", "derivative", "min_temp"]
162 elif steps:
163 step_names = [s.strip() for s in steps.split(",")]
164 else:
165 click.echo("❌ Error: Must specify --all or --steps", err=True)
166 sys.exit(1)
168 # Step mapping
169 step_map = {
170 "ingest": ("Ingesting data", lambda: ingest_data(ctx)),
171 "filter": ("Filtering wells", lambda: filter_wells(ctx, wells_to_filter=wells_to_filter)),
172 "average": ("Averaging replicates", lambda: average_accross_replicates(ctx)),
173 "background": ("Subtracting background", lambda: subtract_background(ctx)),
174 "scale": ("Min-max scaling", lambda: min_max_scale(ctx)),
175 "derivative": ("Calculating derivative", lambda: calculate_derivative(ctx)),
176 "min_temp": ("Finding min temperatures", lambda: find_min_temperature(ctx)),
177 }
179 # Run steps
180 click.echo()
181 for step_name in step_names:
182 if step_name not in step_map:
183 click.echo(f"⚠️ Warning: Unknown step '{step_name}', skipping", err=True)
184 continue
186 description, func = step_map[step_name]
187 click.echo(f" ▶️ {description}...", nl=False)
189 try:
190 func()
191 click.echo(click.style(" ✓", fg="green"))
192 except Exception as e:
193 click.echo(click.style(f" ✗ ({e})", fg="red"))
194 raise
196 click.echo()
197 click.echo(click.style("✅ Pipeline completed successfully!", fg="green", bold=True))
198 click.echo(f" 📁 Results: {ctx.experiment_dir}")
199 click.echo()
200 click.echo(f"💡 View results with: instawell info {experiment_name}")
202 except FileNotFoundError:
203 click.echo(f"❌ Error: Experiment '{experiment_name}' not found in {root}", err=True)
204 click.echo(f" Run 'instawell list' to see available experiments", err=True)
205 sys.exit(1)
206 except Exception as e:
207 click.echo(f"❌ Error: {e}", err=True)
208 sys.exit(1)
211@cli.command()
212@click.option("--root", default="experiments", type=click.Path(),
213 help="Root directory for experiments")
214def list(root):
215 """
216 List all experiments.
218 Shows all initialized experiments in the experiments directory.
220 \b
221 Example:
222 instawell list
223 instawell list --root /path/to/experiments
224 """
225 experiments_root = Path(root)
227 if not experiments_root.exists():
228 click.echo(f"📁 No experiments directory found at: {experiments_root}")
229 click.echo(f" Create your first experiment with 'instawell init'")
230 return
232 # Find all experiment directories (ones with experiment.json)
233 experiments = []
234 for item in experiments_root.iterdir():
235 if item.is_dir() and (item / "experiment.json").exists():
236 experiments.append(item)
238 if not experiments:
239 click.echo(f"📁 No experiments found in: {experiments_root}")
240 click.echo(f" Create your first experiment with 'instawell init'")
241 return
243 # Display experiments
244 click.echo(f"📊 Experiments in {click.style(str(experiments_root), fg='cyan')}:")
245 click.echo()
247 for exp_dir in sorted(experiments):
248 # Load metadata
249 try:
250 with open(exp_dir / "experiment.json") as f:
251 metadata = json.load(f)
253 name = metadata.get("experiment_name", exp_dir.name)
254 created = metadata.get("created_at_iso", "Unknown")
256 # Check which steps have been completed
257 completed_steps = []
258 if (exp_dir / StepFiles.INGESTED_DATA).exists():
259 completed_steps.append("ingest")
260 if (exp_dir / StepFiles.FILTERED_DATA).exists():
261 completed_steps.append("filter")
262 if (exp_dir / StepFiles.AVERAGED_DATA).exists():
263 completed_steps.append("average")
264 if (exp_dir / StepFiles.MIN_TEMPERATURES_DATA).exists():
265 completed_steps.append("complete")
267 status = " → ".join(completed_steps) if completed_steps else "initialized"
269 click.echo(f" • {click.style(name, fg='cyan', bold=True)}")
270 click.echo(f" Created: {created}")
271 click.echo(f" Status: {status}")
272 click.echo()
274 except Exception as e:
275 click.echo(f" • {exp_dir.name} (error reading metadata: {e})")
276 click.echo()
279@cli.command()
280@click.argument("experiment_name")
281@click.option("--root", default="experiments", type=click.Path(),
282 help="Root directory for experiments")
283@click.option("--verbose", "-v", is_flag=True,
284 help="Show detailed information")
285def info(experiment_name, root, verbose):
286 """
287 Show experiment information.
289 Displays configuration, status, and output files.
291 \b
292 Example:
293 instawell info TSA_001
294 instawell info TSA_001 --verbose
295 """
296 try:
297 ctx = load_experiment_context(experiment_name, experiments_root=root)
299 click.echo(f"🔬 Experiment: {click.style(experiment_name, fg='cyan', bold=True)}")
300 click.echo(f" 📁 Location: {ctx.experiment_dir}")
301 click.echo()
303 # Configuration
304 click.echo(click.style("⚙️ Configuration:", fg="yellow", bold=True))
305 click.echo(f" Separator: '{ctx.condition_separator}'")
306 click.echo(f" Fields: {', '.join(ctx.fields)}")
307 click.echo(f" Temperature column: {ctx.temperature_column}")
308 click.echo(f" NPC marker: {ctx.non_protein_control_marker}")
309 click.echo()
311 # Files
312 click.echo(click.style("📄 Output Files:", fg="yellow", bold=True))
313 step_files = [
314 (StepFiles.INGESTED_DATA, "Raw organized data"),
315 (StepFiles.FILTERED_DATA, "Filtered data"),
316 (StepFiles.AVERAGED_DATA, "Averaged replicates"),
317 (StepFiles.BG_SUB_DATA, "Background subtracted"),
318 (StepFiles.MIN_MAX_SCALED_DATA, "Min-max scaled"),
319 (StepFiles.DERIVATIVE_DATA, "Derivative"),
320 (StepFiles.MIN_TEMPERATURES_DATA, "Min temperatures"),
321 ]
323 for file_name, description in step_files:
324 file_path = ctx.experiment_dir / file_name
325 if file_path.exists():
326 size = file_path.stat().st_size / 1024 # KB
327 status = click.style("✓", fg="green")
328 size_str = f"({size:.1f} KB)"
329 else:
330 status = click.style("✗", fg="red")
331 size_str = ""
333 click.echo(f" {status} {description:<25} {size_str}")
335 # Filtered wells
336 filtered_wells_path = ctx.experiment_dir / StepFiles.FILTERED_WELLS
337 if filtered_wells_path.exists():
338 with open(filtered_wells_path) as f:
339 wells = [w.strip() for w in f.readlines() if w.strip()]
340 if wells:
341 click.echo()
342 click.echo(click.style("🚫 Filtered Wells:", fg="yellow", bold=True))
343 click.echo(f" {', '.join(wells)}")
345 # Verbose: show min temperatures summary
346 if verbose:
347 min_temp_path = ctx.experiment_dir / StepFiles.MIN_TEMPERATURES_DATA
348 if min_temp_path.exists():
349 import pandas as pd
351 df = pd.read_csv(min_temp_path)
352 click.echo()
353 click.echo(click.style("🌡️ Min Temperatures Summary:", fg="yellow", bold=True))
354 click.echo(f" Total conditions: {len(df)}")
355 click.echo(f" Temp range: {df['min_temperature'].min():.1f}°C - {df['min_temperature'].max():.1f}°C")
356 click.echo(f" Mean Tm: {df['min_temperature'].mean():.1f}°C")
358 except FileNotFoundError:
359 click.echo(f"❌ Error: Experiment '{experiment_name}' not found", err=True)
360 sys.exit(1)
361 except Exception as e:
362 click.echo(f"❌ Error: {e}", err=True)
363 sys.exit(1)
366@cli.command()
367@click.argument("experiment_name")
368@click.argument("wells", nargs=-1)
369@click.option("--root", default="experiments", type=click.Path(),
370 help="Root directory for experiments")
371def filter(experiment_name, wells, root):
372 """
373 Filter wells from an experiment.
375 Add wells to the filter list and re-filter the data.
377 \b
378 Examples:
379 instawell filter TSA_001 A1 B2 G20
380 instawell filter TSA_001 C3
381 """
382 try:
383 ctx = load_experiment_context(experiment_name, experiments_root=root)
385 # Check if data has been ingested
386 if not (ctx.experiment_dir / StepFiles.INGESTED_DATA).exists():
387 click.echo("❌ Error: Data not ingested yet. Run 'instawell run ... --steps ingest' first", err=True)
388 sys.exit(1)
390 if not wells:
391 click.echo("❌ Error: No wells specified", err=True)
392 sys.exit(1)
394 # Convert wells to list
395 wells_to_filter = list(wells)
397 click.echo(f"🚫 Filtering wells from {click.style(experiment_name, fg='cyan', bold=True)}")
398 click.echo(f" Wells: {', '.join(wells_to_filter)}")
400 # Run filter
401 filter_wells(ctx, wells_to_filter=wells_to_filter)
403 click.echo(click.style("✅ Wells filtered successfully!", fg="green"))
404 click.echo(f" 📁 Filtered data: {ctx.experiment_dir / StepFiles.FILTERED_DATA}")
406 except FileNotFoundError:
407 click.echo(f"❌ Error: Experiment '{experiment_name}' not found", err=True)
408 sys.exit(1)
409 except Exception as e:
410 click.echo(f"❌ Error: {e}", err=True)
411 sys.exit(1)
414def main():
415 """Entry point for the CLI."""
416 cli()
419if __name__ == "__main__":
420 main()