Instrument Data To Allotrope

scripts/export_parser.py

#!/usr/bin/env python3
"""
Export Parser Code

Generates standalone Python scripts that can be handed off to data engineers
or run in Jupyter notebooks. The exported code is self-contained and
production-ready.

Usage:
    python export_parser.py --vendor VI_CELL_BLU --output vicell_parser.py
    python export_parser.py --vendor NANODROP_EIGHT --format notebook --output nanodrop_parser.ipynb
"""

import sys
from pathlib import Path
from datetime import datetime
from typing import Optional


# Template for standalone Python script
SCRIPT_TEMPLATE = '''#!/usr/bin/env python3
"""
{instrument_name} to Allotrope Simple Model (ASM) Parser

Auto-generated by Claude instrument-data-to-allotrope skill
Generated: {timestamp}
Vendor: {vendor}

This script converts {instrument_name} output files to Allotrope Simple Model (ASM)
JSON format for LIMS import, data lakes, or downstream analysis.

Requirements:
    pip install allotropy pandas openpyxl

Usage:
    python {script_name} input_file.csv --output output_asm.json
    python {script_name} input_file.csv --flatten  # Also generate CSV

Input file format:
    {file_format_description}
"""

import json
import argparse
from pathlib import Path
from typing import Dict, Any, Optional

try:
    from allotropy.parser_factory import Vendor
    from allotropy.to_allotrope import allotrope_from_file
    ALLOTROPY_AVAILABLE = True
except ImportError:
    ALLOTROPY_AVAILABLE = False
    print("Warning: allotropy not installed. Install with: pip install allotropy")

try:
    import pandas as pd
    PANDAS_AVAILABLE = True
except ImportError:
    PANDAS_AVAILABLE = False


def convert_to_asm(filepath: str) -> Optional[Dict[str, Any]]:
    """
    Convert {instrument_name} file to ASM format.
    
    Args:
        filepath: Path to input file
        
    Returns:
        ASM dictionary or None if conversion fails
    """
    if not ALLOTROPY_AVAILABLE:
        raise ImportError("allotropy library required. Install with: pip install allotropy")
    
    try:
        asm = allotrope_from_file(filepath, Vendor.{vendor})
        return asm
    except Exception as e:
        print(f"Conversion error: {{e}}")
        return None


def flatten_asm(asm: Dict[str, Any]) -> list:
    """
    Flatten ASM to list of row dictionaries for CSV export.
    
    Args:
        asm: ASM dictionary
        
    Returns:
        List of flattened row dictionaries
    """
    technique = "{technique}"
    rows = []
    
    agg_key = f"{{technique}}-aggregate-document"
    agg_doc = asm.get(agg_key, {{}})
    
    # Extract device info
    device = agg_doc.get("device-system-document", {{}})
    device_info = {{
        "instrument_serial_number": device.get("device-identifier"),
        "instrument_model": device.get("model-number"),
    }}
    
    doc_key = f"{{technique}}-document"
    for doc in agg_doc.get(doc_key, []):
        meas_agg = doc.get("measurement-aggregate-document", {{}})
        
        common = {{
            "analyst": meas_agg.get("analyst"),
            "measurement_time": meas_agg.get("measurement-time"),
            **device_info
        }}
        
        for meas in meas_agg.get("measurement-document", []):
            row = {{**common}}
            for key, value in meas.items():
                clean_key = key.replace("-", "_")
                if isinstance(value, dict) and "value" in value:
                    row[clean_key] = value["value"]
                    if "unit" in value:
                        row[f"{{clean_key}}_unit"] = value["unit"]
                else:
                    row[clean_key] = value
            rows.append(row)
    
    return rows


def main():
    parser = argparse.ArgumentParser(description="Convert {instrument_name} to ASM")
    parser.add_argument("input", help="Input file path")
    parser.add_argument("--output", "-o", help="Output JSON path")
    parser.add_argument("--flatten", action="store_true", help="Also generate CSV")
    
    args = parser.parse_args()
    
    input_path = Path(args.input)
    if not input_path.exists():
        print(f"Error: File not found: {{args.input}}")
        return 1
    
    # Convert to ASM
    print(f"Converting {{args.input}}...")
    asm = convert_to_asm(str(input_path))
    
    if asm is None:
        print("Conversion failed")
        return 1
    
    # Write ASM JSON
    output_path = args.output or str(input_path.with_suffix('.asm.json'))
    with open(output_path, 'w') as f:
        json.dump(asm, f, indent=2, default=str)
    print(f"ASM written to: {{output_path}}")
    
    # Optionally flatten
    if args.flatten and PANDAS_AVAILABLE:
        rows = flatten_asm(asm)
        df = pd.DataFrame(rows)
        flat_path = str(input_path.with_suffix('.flat.csv'))
        df.to_csv(flat_path, index=False)
        print(f"CSV written to: {{flat_path}}")
    
    return 0


if __name__ == "__main__":
    sys.exit(main())
'''


# Template for Jupyter notebook
NOTEBOOK_TEMPLATE = """{{
 "cells": [
  {{
   "cell_type": "markdown",
   "metadata": {{}},
   "source": [
    "# {instrument_name} to Allotrope Simple Model (ASM) Parser\\n",
    "\\n",
    "Auto-generated by Claude instrument-data-to-allotrope skill\\n",
    "Generated: {timestamp}\\n",
    "Vendor: {vendor}\\n",
    "\\n",
    "This notebook converts {instrument_name} output files to Allotrope Simple Model (ASM) JSON format."
   ]
  }},
  {{
   "cell_type": "code",
   "execution_count": null,
   "metadata": {{}},
   "source": [
    "# Install requirements (uncomment if needed)\\n",
    "# !pip install allotropy pandas openpyxl"
   ]
  }},
  {{
   "cell_type": "code",
   "execution_count": null,
   "metadata": {{}},
   "source": [
    "import json\\n",
    "from pathlib import Path\\n",
    "import pandas as pd\\n",
    "\\n",
    "from allotropy.parser_factory import Vendor\\n",
    "from allotropy.to_allotrope import allotrope_from_file"
   ]
  }},
  {{
   "cell_type": "markdown",
   "metadata": {{}},
   "source": [
    "## Configuration\\n",
    "\\n",
    "Set your input file path here:"
   ]
  }},
  {{
   "cell_type": "code",
   "execution_count": null,
   "metadata": {{}},
   "source": [
    "# Configure input/output paths\\n",
    "INPUT_FILE = \\"your_data_file.csv\\"  # <-- Change this\\n",
    "OUTPUT_ASM = \\"output_asm.json\\"\\n",
    "OUTPUT_CSV = \\"output_flat.csv\\""
   ]
  }},
  {{
   "cell_type": "markdown",
   "metadata": {{}},
   "source": [
    "## Convert to ASM"
   ]
  }},
  {{
   "cell_type": "code",
   "execution_count": null,
   "metadata": {{}},
   "source": [
    "# Convert file to ASM\\n",
    "asm = allotrope_from_file(INPUT_FILE, Vendor.{vendor})\\n",
    "\\n",
    "# Save ASM JSON\\n",
    "with open(OUTPUT_ASM, 'w') as f:\\n",
    "    json.dump(asm, f, indent=2, default=str)\\n",
    "\\n",
    "print(f\\"ASM saved to: {{OUTPUT_ASM}}\\")"
   ]
  }},
  {{
   "cell_type": "markdown",
   "metadata": {{}},
   "source": [
    "## Preview ASM Structure"
   ]
  }},
  {{
   "cell_type": "code",
   "execution_count": null,
   "metadata": {{}},
   "source": [
    "# Show ASM structure\\n",
    "print(json.dumps(asm, indent=2, default=str)[:2000])"
   ]
  }},
  {{
   "cell_type": "markdown",
   "metadata": {{}},
   "source": [
    "## Flatten to CSV"
   ]
  }},
  {{
   "cell_type": "code",
   "execution_count": null,
   "metadata": {{}},
   "source": [
    "def flatten_asm(asm, technique=\\"{technique}\\"):\\n",
    "    rows = []\\n",
    "    agg_key = f\\"{{technique}}-aggregate-document\\"\\n",
    "    agg_doc = asm.get(agg_key, {{}})\\n",
    "    \\n",
    "    device = agg_doc.get(\\"device-system-document\\", {{}})\\n",
    "    device_info = {{\\n",
    "        \\"instrument_serial_number\\": device.get(\\"device-identifier\\"),\\n",
    "        \\"instrument_model\\": device.get(\\"model-number\\"),\\n",
    "    }}\\n",
    "    \\n",
    "    doc_key = f\\"{{technique}}-document\\"\\n",
    "    for doc in agg_doc.get(doc_key, []):\\n",
    "        meas_agg = doc.get(\\"measurement-aggregate-document\\", {{}})\\n",
    "        common = {{\\n",
    "            \\"analyst\\": meas_agg.get(\\"analyst\\"),\\n",
    "            \\"measurement_time\\": meas_agg.get(\\"measurement-time\\"),\\n",
    "            **device_info\\n",
    "        }}\\n",
    "        \\n",
    "        for meas in meas_agg.get(\\"measurement-document\\", []):\\n",
    "            row = {{**common}}\\n",
    "            for key, value in meas.items():\\n",
    "                clean_key = key.replace(\\"-\\", \\"_\\")\\n",
    "                if isinstance(value, dict) and \\"value\\" in value:\\n",
    "                    row[clean_key] = value[\\"value\\"]\\n",
    "                    if \\"unit\\" in value:\\n",
    "                        row[f\\"{{clean_key}}_unit\\"] = value[\\"unit\\"]\\n",
    "                else:\\n",
    "                    row[clean_key] = value\\n",
    "            rows.append(row)\\n",
    "    return rows\\n",
    "\\n",
    "# Flatten and save\\n",
    "rows = flatten_asm(asm)\\n",
    "df = pd.DataFrame(rows)\\n",
    "df.to_csv(OUTPUT_CSV, index=False)\\n",
    "print(f\\"CSV saved to: {{OUTPUT_CSV}}\\")"
   ]
  }},
  {{
   "cell_type": "code",
   "execution_count": null,
   "metadata": {{}},
   "source": [
    "# Preview flattened data\\n",
    "df.head()"
   ]
  }}
 ],
 "metadata": {{
  "kernelspec": {{
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  }},
  "language_info": {{
   "name": "python",
   "version": "3.10.0"
  }}
 }},
 "nbformat": 4,
 "nbformat_minor": 4
}}"""


# Instrument metadata for templates
INSTRUMENT_INFO = {
    "BECKMAN_VI_CELL_BLU": {
        "name": "Beckman Coulter Vi-CELL BLU",
        "technique": "cell-counting",
        "file_format": "CSV export from Vi-CELL BLU software with columns: Sample ID, Viable cells, Viability, Total cells, etc.",
    },
    "BECKMAN_VI_CELL_XR": {
        "name": "Beckman Coulter Vi-CELL XR",
        "technique": "cell-counting",
        "file_format": "TXT or XLS/XLSX export from Vi-CELL XR with sample and measurement data",
    },
    "THERMO_FISHER_NANODROP_EIGHT": {
        "name": "Thermo Fisher NanoDrop Eight",
        "technique": "spectrophotometry",
        "file_format": "TSV or TXT export with Sample Name, Nucleic Acid Conc., A260, A280, 260/280 ratio",
    },
    "THERMO_FISHER_NANODROP_ONE": {
        "name": "Thermo Fisher NanoDrop One",
        "technique": "spectrophotometry",
        "file_format": "CSV or XLSX export with spectrophotometry measurements",
    },
    "MOLDEV_SOFTMAX_PRO": {
        "name": "Molecular Devices SoftMax Pro",
        "technique": "plate-reader",
        "file_format": "TXT export from SoftMax Pro with plate reader data",
    },
    "BMG_MARS": {
        "name": "BMG MARS (CLARIOstar)",
        "technique": "plate-reader",
        "file_format": "CSV or TXT export from BMG MARS with Well, Content, Conc., Mean, SD, CV columns",
    },
    "AGILENT_GEN5": {
        "name": "Agilent Gen5 (BioTek)",
        "technique": "plate-reader",
        "file_format": "XLSX export from Gen5 software",
    },
    "APPBIO_QUANTSTUDIO": {
        "name": "Applied Biosystems QuantStudio",
        "technique": "pcr",
        "file_format": "XLSX export with qPCR data including Well, Sample Name, Target Name, CT values",
    },
}


def generate_script(vendor: str, output_path: str) -> None:
    """Generate standalone Python script for given vendor."""
    info = INSTRUMENT_INFO.get(
        vendor,
        {
            "name": vendor.replace("_", " ").title(),
            "technique": "generic",
            "file_format": "Instrument output file",
        },
    )

    script = SCRIPT_TEMPLATE.format(
        instrument_name=info["name"],
        timestamp=datetime.now().isoformat(),
        vendor=vendor,
        script_name=Path(output_path).name,
        file_format_description=info["file_format"],
        technique=info["technique"],
    )

    with open(output_path, "w") as f:
        f.write(script)


def generate_notebook(vendor: str, output_path: str) -> None:
    """Generate Jupyter notebook for given vendor."""
    info = INSTRUMENT_INFO.get(
        vendor,
        {
            "name": vendor.replace("_", " ").title(),
            "technique": "generic",
            "file_format": "Instrument output file",
        },
    )

    notebook = NOTEBOOK_TEMPLATE.format(
        instrument_name=info["name"],
        timestamp=datetime.now().isoformat(),
        vendor=vendor,
        technique=info["technique"],
    )

    with open(output_path, "w") as f:
        f.write(notebook)


def main():
    import argparse

    parser = argparse.ArgumentParser(
        description="Export parser code for data engineers"
    )
    parser.add_argument("--vendor", help="Vendor enum name (e.g., VI_CELL_BLU)")
    parser.add_argument("--output", "-o", help="Output file path")
    parser.add_argument(
        "--format",
        choices=["script", "notebook"],
        default="script",
        help="Output format (default: script)",
    )
    parser.add_argument(
        "--list-vendors", action="store_true", help="List supported vendors"
    )

    args = parser.parse_args()

    if args.list_vendors:
        print("Supported vendors:")
        for vendor in INSTRUMENT_INFO.keys():
            print(f"  {vendor}")
        return 0

    if not args.vendor or not args.output:
        parser.error("--vendor and --output are required when not using --list-vendors")

    vendor = args.vendor.upper()

    if args.format == "notebook":
        generate_notebook(vendor, args.output)
    else:
        generate_script(vendor, args.output)

    print(f"Parser code exported to: {args.output}")
    return 0


if __name__ == "__main__":
    sys.exit(main())