Skills › Knowledge Work Plugins › instrument-data-to-allotrope › export_parser.py
export_parser.py
481 lines · bio-research/skills/instrument-data-to-allotrope/scripts/export_parser.py at 2574e29
1#!/usr/bin/env python3
2"""
3Export Parser Code
5Generates standalone Python scripts that can be handed off to data engineers
6or run in Jupyter notebooks. The exported code is self-contained and
7production-ready.
9Usage:
10 python export_parser.py --vendor VI_CELL_BLU --output vicell_parser.py
11 python export_parser.py --vendor NANODROP_EIGHT --format notebook --output nanodrop_parser.ipynb
12"""
14import sys
15from pathlib import Path
16from datetime import datetime
17from typing import Optional
20# Template for standalone Python script
21SCRIPT_TEMPLATE = '''#!/usr/bin/env python3
22"""
23{instrument_name} to Allotrope Simple Model (ASM) Parser
25Auto-generated by Claude instrument-data-to-allotrope skill
26Generated: {timestamp}
27Vendor: {vendor}
29This script converts {instrument_name} output files to Allotrope Simple Model (ASM)
30JSON format for LIMS import, data lakes, or downstream analysis.
32Requirements:
33 pip install allotropy pandas openpyxl
35Usage:
36 python {script_name} input_file.csv --output output_asm.json
37 python {script_name} input_file.csv --flatten # Also generate CSV
39Input file format:
40 {file_format_description}
41"""
43import json
44import argparse
45from pathlib import Path
46from typing import Dict, Any, Optional
48try:
49 from allotropy.parser_factory import Vendor
50 from allotropy.to_allotrope import allotrope_from_file
51 ALLOTROPY_AVAILABLE = True
52except ImportError:
53 ALLOTROPY_AVAILABLE = False
54 print("Warning: allotropy not installed. Install with: pip install allotropy")
56try:
57 import pandas as pd
58 PANDAS_AVAILABLE = True
59except ImportError:
60 PANDAS_AVAILABLE = False
63def convert_to_asm(filepath: str) -> Optional[Dict[str, Any]]:
64 """
65 Convert {instrument_name} file to ASM format.
67 Args:
68 filepath: Path to input file
70 Returns:
71 ASM dictionary or None if conversion fails
72 """
73 if not ALLOTROPY_AVAILABLE:
74 raise ImportError("allotropy library required. Install with: pip install allotropy")
76 try:
77 asm = allotrope_from_file(filepath, Vendor.{vendor})
78 return asm
79 except Exception as e:
80 print(f"Conversion error: {{e}}")
81 return None
84def flatten_asm(asm: Dict[str, Any]) -> list:
85 """
86 Flatten ASM to list of row dictionaries for CSV export.
88 Args:
89 asm: ASM dictionary
91 Returns:
92 List of flattened row dictionaries
93 """
94 technique = "{technique}"
95 rows = []
97 agg_key = f"{{technique}}-aggregate-document"
98 agg_doc = asm.get(agg_key, {{}})
100 # Extract device info
101 device = agg_doc.get("device-system-document", {{}})
102 device_info = {{
103 "instrument_serial_number": device.get("device-identifier"),
104 "instrument_model": device.get("model-number"),
105 }}
107 doc_key = f"{{technique}}-document"
108 for doc in agg_doc.get(doc_key, []):
109 meas_agg = doc.get("measurement-aggregate-document", {{}})
111 common = {{
112 "analyst": meas_agg.get("analyst"),
113 "measurement_time": meas_agg.get("measurement-time"),
114 **device_info
115 }}
117 for meas in meas_agg.get("measurement-document", []):
118 row = {{**common}}
119 for key, value in meas.items():
120 clean_key = key.replace("-", "_")
121 if isinstance(value, dict) and "value" in value:
122 row[clean_key] = value["value"]
123 if "unit" in value:
124 row[f"{{clean_key}}_unit"] = value["unit"]
125 else:
126 row[clean_key] = value
127 rows.append(row)
129 return rows
132def main():
133 parser = argparse.ArgumentParser(description="Convert {instrument_name} to ASM")
134 parser.add_argument("input", help="Input file path")
135 parser.add_argument("--output", "-o", help="Output JSON path")
136 parser.add_argument("--flatten", action="store_true", help="Also generate CSV")
138 args = parser.parse_args()
140 input_path = Path(args.input)
141 if not input_path.exists():
142 print(f"Error: File not found: {{args.input}}")
143 return 1
145 # Convert to ASM
146 print(f"Converting {{args.input}}...")
147 asm = convert_to_asm(str(input_path))
149 if asm is None:
150 print("Conversion failed")
151 return 1
153 # Write ASM JSON
154 output_path = args.output or str(input_path.with_suffix('.asm.json'))
155 with open(output_path, 'w') as f:
156 json.dump(asm, f, indent=2, default=str)
157 print(f"ASM written to: {{output_path}}")
159 # Optionally flatten
160 if args.flatten and PANDAS_AVAILABLE:
161 rows = flatten_asm(asm)
162 df = pd.DataFrame(rows)
163 flat_path = str(input_path.with_suffix('.flat.csv'))
164 df.to_csv(flat_path, index=False)
165 print(f"CSV written to: {{flat_path}}")
167 return 0
170if __name__ == "__main__":
171 sys.exit(main())
172'''
175# Template for Jupyter notebook
176NOTEBOOK_TEMPLATE = """{{
177 "cells": [
178 {{
179 "cell_type": "markdown",
180 "metadata": {{}},
181 "source": [
182 "# {instrument_name} to Allotrope Simple Model (ASM) Parser\\n",
183 "\\n",
184 "Auto-generated by Claude instrument-data-to-allotrope skill\\n",
185 "Generated: {timestamp}\\n",
186 "Vendor: {vendor}\\n",
187 "\\n",
188 "This notebook converts {instrument_name} output files to Allotrope Simple Model (ASM) JSON format."
189 ]
190 }},
191 {{
192 "cell_type": "code",
193 "execution_count": null,
194 "metadata": {{}},
195 "source": [
196 "# Install requirements (uncomment if needed)\\n",
197 "# !pip install allotropy pandas openpyxl"
198 ]
199 }},
200 {{
201 "cell_type": "code",
202 "execution_count": null,
203 "metadata": {{}},
204 "source": [
205 "import json\\n",
206 "from pathlib import Path\\n",
207 "import pandas as pd\\n",
208 "\\n",
209 "from allotropy.parser_factory import Vendor\\n",
210 "from allotropy.to_allotrope import allotrope_from_file"
211 ]
212 }},
213 {{
214 "cell_type": "markdown",
215 "metadata": {{}},
216 "source": [
217 "## Configuration\\n",
218 "\\n",
219 "Set your input file path here:"
220 ]
221 }},
222 {{
223 "cell_type": "code",
224 "execution_count": null,
225 "metadata": {{}},
226 "source": [
227 "# Configure input/output paths\\n",
228 "INPUT_FILE = \\"your_data_file.csv\\" # <-- Change this\\n",
229 "OUTPUT_ASM = \\"output_asm.json\\"\\n",
230 "OUTPUT_CSV = \\"output_flat.csv\\""
231 ]
232 }},
233 {{
234 "cell_type": "markdown",
235 "metadata": {{}},
236 "source": [
237 "## Convert to ASM"
238 ]
239 }},
240 {{
241 "cell_type": "code",
242 "execution_count": null,
243 "metadata": {{}},
244 "source": [
245 "# Convert file to ASM\\n",
246 "asm = allotrope_from_file(INPUT_FILE, Vendor.{vendor})\\n",
247 "\\n",
248 "# Save ASM JSON\\n",
249 "with open(OUTPUT_ASM, 'w') as f:\\n",
250 " json.dump(asm, f, indent=2, default=str)\\n",
251 "\\n",
252 "print(f\\"ASM saved to: {{OUTPUT_ASM}}\\")"
253 ]
254 }},
255 {{
256 "cell_type": "markdown",
257 "metadata": {{}},
258 "source": [
259 "## Preview ASM Structure"
260 ]
261 }},
262 {{
263 "cell_type": "code",
264 "execution_count": null,
265 "metadata": {{}},
266 "source": [
267 "# Show ASM structure\\n",
268 "print(json.dumps(asm, indent=2, default=str)[:2000])"
269 ]
270 }},
271 {{
272 "cell_type": "markdown",
273 "metadata": {{}},
274 "source": [
275 "## Flatten to CSV"
276 ]
277 }},
278 {{
279 "cell_type": "code",
280 "execution_count": null,
281 "metadata": {{}},
282 "source": [
283 "def flatten_asm(asm, technique=\\"{technique}\\"):\\n",
284 " rows = []\\n",
285 " agg_key = f\\"{{technique}}-aggregate-document\\"\\n",
286 " agg_doc = asm.get(agg_key, {{}})\\n",
287 " \\n",
288 " device = agg_doc.get(\\"device-system-document\\", {{}})\\n",
289 " device_info = {{\\n",
290 " \\"instrument_serial_number\\": device.get(\\"device-identifier\\"),\\n",
291 " \\"instrument_model\\": device.get(\\"model-number\\"),\\n",
292 " }}\\n",
293 " \\n",
294 " doc_key = f\\"{{technique}}-document\\"\\n",
295 " for doc in agg_doc.get(doc_key, []):\\n",
296 " meas_agg = doc.get(\\"measurement-aggregate-document\\", {{}})\\n",
297 " common = {{\\n",
298 " \\"analyst\\": meas_agg.get(\\"analyst\\"),\\n",
299 " \\"measurement_time\\": meas_agg.get(\\"measurement-time\\"),\\n",
300 " **device_info\\n",
301 " }}\\n",
302 " \\n",
303 " for meas in meas_agg.get(\\"measurement-document\\", []):\\n",
304 " row = {{**common}}\\n",
305 " for key, value in meas.items():\\n",
306 " clean_key = key.replace(\\"-\\", \\"_\\")\\n",
307 " if isinstance(value, dict) and \\"value\\" in value:\\n",
308 " row[clean_key] = value[\\"value\\"]\\n",
309 " if \\"unit\\" in value:\\n",
310 " row[f\\"{{clean_key}}_unit\\"] = value[\\"unit\\"]\\n",
311 " else:\\n",
312 " row[clean_key] = value\\n",
313 " rows.append(row)\\n",
314 " return rows\\n",
315 "\\n",
316 "# Flatten and save\\n",
317 "rows = flatten_asm(asm)\\n",
318 "df = pd.DataFrame(rows)\\n",
319 "df.to_csv(OUTPUT_CSV, index=False)\\n",
320 "print(f\\"CSV saved to: {{OUTPUT_CSV}}\\")"
321 ]
322 }},
323 {{
324 "cell_type": "code",
325 "execution_count": null,
326 "metadata": {{}},
327 "source": [
328 "# Preview flattened data\\n",
329 "df.head()"
330 ]
331 }}
332 ],
333 "metadata": {{
334 "kernelspec": {{
335 "display_name": "Python 3",
336 "language": "python",
337 "name": "python3"
338 }},
339 "language_info": {{
340 "name": "python",
341 "version": "3.10.0"
342 }}
343 }},
344 "nbformat": 4,
345 "nbformat_minor": 4
346}}"""
349# Instrument metadata for templates
350INSTRUMENT_INFO = {
351 "BECKMAN_VI_CELL_BLU": {
352 "name": "Beckman Coulter Vi-CELL BLU",
353 "technique": "cell-counting",
354 "file_format": "CSV export from Vi-CELL BLU software with columns: Sample ID, Viable cells, Viability, Total cells, etc.",
355 },
356 "BECKMAN_VI_CELL_XR": {
357 "name": "Beckman Coulter Vi-CELL XR",
358 "technique": "cell-counting",
359 "file_format": "TXT or XLS/XLSX export from Vi-CELL XR with sample and measurement data",
360 },
361 "THERMO_FISHER_NANODROP_EIGHT": {
362 "name": "Thermo Fisher NanoDrop Eight",
363 "technique": "spectrophotometry",
364 "file_format": "TSV or TXT export with Sample Name, Nucleic Acid Conc., A260, A280, 260/280 ratio",
365 },
366 "THERMO_FISHER_NANODROP_ONE": {
367 "name": "Thermo Fisher NanoDrop One",
368 "technique": "spectrophotometry",
369 "file_format": "CSV or XLSX export with spectrophotometry measurements",
370 },
371 "MOLDEV_SOFTMAX_PRO": {
372 "name": "Molecular Devices SoftMax Pro",
373 "technique": "plate-reader",
374 "file_format": "TXT export from SoftMax Pro with plate reader data",
375 },
376 "BMG_MARS": {
377 "name": "BMG MARS (CLARIOstar)",
378 "technique": "plate-reader",
379 "file_format": "CSV or TXT export from BMG MARS with Well, Content, Conc., Mean, SD, CV columns",
380 },
381 "AGILENT_GEN5": {
382 "name": "Agilent Gen5 (BioTek)",
383 "technique": "plate-reader",
384 "file_format": "XLSX export from Gen5 software",
385 },
386 "APPBIO_QUANTSTUDIO": {
387 "name": "Applied Biosystems QuantStudio",
388 "technique": "pcr",
389 "file_format": "XLSX export with qPCR data including Well, Sample Name, Target Name, CT values",
390 },
391}
394def generate_script(vendor: str, output_path: str) -> None:
395 """Generate standalone Python script for given vendor."""
396 info = INSTRUMENT_INFO.get(
397 vendor,
398 {
399 "name": vendor.replace("_", " ").title(),
400 "technique": "generic",
401 "file_format": "Instrument output file",
402 },
403 )
405 script = SCRIPT_TEMPLATE.format(
406 instrument_name=info["name"],
407 timestamp=datetime.now().isoformat(),
408 vendor=vendor,
409 script_name=Path(output_path).name,
410 file_format_description=info["file_format"],
411 technique=info["technique"],
412 )
414 with open(output_path, "w") as f:
415 f.write(script)
418def generate_notebook(vendor: str, output_path: str) -> None:
419 """Generate Jupyter notebook for given vendor."""
420 info = INSTRUMENT_INFO.get(
421 vendor,
422 {
423 "name": vendor.replace("_", " ").title(),
424 "technique": "generic",
425 "file_format": "Instrument output file",
426 },
427 )
429 notebook = NOTEBOOK_TEMPLATE.format(
430 instrument_name=info["name"],
431 timestamp=datetime.now().isoformat(),
432 vendor=vendor,
433 technique=info["technique"],
434 )
436 with open(output_path, "w") as f:
437 f.write(notebook)
440def main():
441 import argparse
443 parser = argparse.ArgumentParser(
444 description="Export parser code for data engineers"
445 )
446 parser.add_argument("--vendor", help="Vendor enum name (e.g., VI_CELL_BLU)")
447 parser.add_argument("--output", "-o", help="Output file path")
448 parser.add_argument(
449 "--format",
450 choices=["script", "notebook"],
451 default="script",
452 help="Output format (default: script)",
453 )
454 parser.add_argument(
455 "--list-vendors", action="store_true", help="List supported vendors"
456 )
458 args = parser.parse_args()
460 if args.list_vendors:
461 print("Supported vendors:")
462 for vendor in INSTRUMENT_INFO.keys():
463 print(f" {vendor}")
464 return 0
466 if not args.vendor or not args.output:
467 parser.error("--vendor and --output are required when not using --list-vendors")
469 vendor = args.vendor.upper()
471 if args.format == "notebook":
472 generate_notebook(vendor, args.output)
473 else:
474 generate_script(vendor, args.output)
476 print(f"Parser code exported to: {args.output}")
477 return 0
480if __name__ == "__main__":
481 sys.exit(main())