Setting the file. One moment.
Export Parser · Instrument Data To Allotrope · anthropics/knowledge-work-plugins · Skills Docs
Back to the top of the page 22
Validate Data
Tech Debt
62
Recruiting Pipeline
71
Vendor Check
125
Zoom Meeting SDK Web
88
Vendor Review
181
Create An Asset
Video Sdk/web
Next
Script Flatten Asm
scripts/ export_parser.py
Python · 481 lines · 14 KB
16 from datetime import datetime
17 from typing import Optional
18
19
20 # Template for standalone Python script
21 SCRIPT_TEMPLATE = '''#!/usr/bin/env python3
22 """
23 {instrument_name} to Allotrope Simple Model (ASM) Parser
24
25 Auto-generated by Claude instrument-data-to-allotrope skill
26 Generated: {timestamp}
27 Vendor: {vendor}
28
29 This script converts {instrument_name} output files to Allotrope Simple Model (ASM)
30 JSON format for LIMS import, data lakes, or downstream analysis.
31
32 Requirements:
33 pip install allotropy pandas openpyxl
34
35 Usage:
36 python {script_name} input_file.csv --output output_asm.json
37 python {script_name} input_file.csv --flatten # Also generate CSV
38
39 Input file format:
40 {file_format_description}
41 """
42
43 import json
44 import argparse
45 from pathlib import Path
46 from typing import Dict, Any, Optional
47
48 try:
49 from allotropy.parser_factory import Vendor
50 from allotropy.to_allotrope import allotrope_from_file
51 ALLOTROPY_AVAILABLE = True
52 except ImportError:
53 ALLOTROPY_AVAILABLE = False
54 print("Warning: allotropy not installed. Install with: pip install allotropy")
55
56 try:
57 import pandas as pd
58 PANDAS_AVAILABLE = True
59 except ImportError:
60 PANDAS_AVAILABLE = False
61
62
63 def convert_to_asm(filepath: str) -> Optional[Dict[str, Any]]:
64 """
65 Convert {instrument_name} file to ASM format.
66
67 Args:
68 filepath: Path to input file
69
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")
75
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
82
83
84 def flatten_asm(asm: Dict[str, Any]) -> list:
85 """
86 Flatten ASM to list of row dictionaries for CSV export.
87
88 Args:
89 asm: ASM dictionary
90
91 Returns:
92 List of flattened row dictionaries
93 """
94 technique = " {technique} "
95 rows = []
96
97 agg_key = f" {{ technique }} -aggregate-document"
98 agg_doc = asm.get(agg_key, {{}} )
99
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 }}
106
107 doc_key = f" {{ technique }} -document"
108 for doc in agg_doc.get(doc_key, []):
109 meas_agg = doc.get("measurement-aggregate-document", {{}} )
110
111 common = {{
112 "analyst": meas_agg.get("analyst"),
113 "measurement_time": meas_agg.get("measurement-time"),
114 **device_info
115 }}
116
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)
128
129 return rows
130
131
132 def 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")
137
138 args = parser.parse_args()
139
140 input_path = Path(args.input)
141 if not input_path.exists():
142 print(f"Error: File not found: {{ args.input }} ")
143 return 1
144
145 # Convert to ASM
146 print(f"Converting {{ args.input }} ...")
147 asm = convert_to_asm(str(input_path))
148
149 if asm is None:
150 print("Conversion failed")
151 return 1
152
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 }} ")
158
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 }} ")
166
167 return 0
168
169
170 if __name__ == "__main__":
171 sys.exit(main())
172 '''
173
174
175 # Template for Jupyter notebook
176 NOTEBOOK_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 }} """
347
348
349 # Instrument metadata for templates
350 INSTRUMENT_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 }
392
393
394 def 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 )
404
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 )
413
414 with open (output_path, "w" ) as f:
415 f.write(script)
416
417
418 def 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 )
428
429 notebook = NOTEBOOK_TEMPLATE .format(
430 instrument_name = info[ "name" ],
431 timestamp = datetime.now().isoformat(),
432 vendor = vendor,
433 technique = info[ "technique" ],
434 )
435
436 with open (output_path, "w" ) as f:
437 f.write(notebook)
438
439
440 def main ():
441 import argparse
442
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 )
457
458 args = parser.parse_args()
459
460 if args.list_vendors:
461 print ( "Supported vendors:" )
462 for vendor in INSTRUMENT_INFO .keys():
463 print ( f " { vendor } " )
464 return 0
465
466 if not args.vendor or not args.output:
467 parser.error( "--vendor and --output are required when not using --list-vendors" )
468
469 vendor = args.vendor.upper()
470
471 if args.format == "notebook" :
472 generate_notebook(vendor, args.output)
473 else :
474 generate_script(vendor, args.output)
475
476 print ( f "Parser code exported to: { args.output } " )
477 return 0
478
479
480 if __name__ == "__main__" :
481 sys.exit(main())