Instrument Data To Allotrope

Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/

TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION

1. Definitions.

"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.

"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.

"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.

"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.

"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.

"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.

"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).

"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.

"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."

"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.

2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.

3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.

4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:

(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and

(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and

(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and

(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.

You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.

5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.

6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.

7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.

8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.

9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.

END OF TERMS AND CONDITIONS

APPENDIX: How to apply the Apache License to your work.

To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "[]"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.

Copyright [yyyy] [name of copyright owner]

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

# Instrument Data to Allotrope Skill - Pinned Dependencies
#
# These versions are pinned for reproducibility and determinism.
# All scientists using this skill should install these exact versions
# to ensure identical ASM output from the same input files.
#
# Installation:
#   pip install -r requirements.txt --break-system-packages
#
# Note: Versions pinned as of 2025-01-05

# Core parsing library - provides native instrument parsers
allotropy==0.1.55

# Data manipulation and file reading
pandas==2.0.3

# Excel file support (required by pandas for .xlsx files)
openpyxl==3.1.2

# PDF parsing support (for instruments that export PDFs)
pdfplumber==0.9.0

# Scientific computing (optional, but recommended for advanced analysis)
# numpy==1.24.3  # Uncomment if needed
# scipy==1.11.1  # Uncomment if needed

#!/usr/bin/env python3
"""
Instrument Data to ASM Converter

Converts laboratory instrument output files to Allotrope Simple Model (ASM) JSON format.
Supports auto-detection of instrument types and fallback parsing for unsupported formats.

Usage:
    python convert_to_asm.py <input_file> [--vendor VENDOR] [--output OUTPUT]
"""

import json
import sys
import re
import hashlib
import importlib.metadata
from pathlib import Path
from typing import Optional, Tuple, Dict, Any
from datetime import datetime


# Lazy imports to avoid errors if not installed
def get_allotropy():
    try:
        from allotropy.parser_factory import Vendor
        from allotropy.to_allotrope import allotrope_from_file, allotrope_from_io

        return Vendor, allotrope_from_file, allotrope_from_io
    except ImportError:
        return None, None, None


def get_pandas():
    try:
        import pandas as pd

        return pd
    except ImportError:
        return None


# Detection patterns for instrument identification
DETECTION_PATTERNS = {
    "BECKMAN_VI_CELL_BLU": {
        "columns": [
            "Sample ID",
            "Viable cells",
            "Viability",
            "Total cells",
            "Average diameter",
        ],
        "keywords": ["Vi-CELL BLU", "Beckman Coulter"],
        "file_patterns": [r".*\.csv$"],
        "confidence_boost": 20,
    },
    "BECKMAN_VI_CELL_XR": {
        "columns": ["Sample", "Total cells/ml", "Viable cells/ml", "Viability (%)"],
        "keywords": ["Vi-CELL XR", "Cell Viability Analyzer"],
        "file_patterns": [r".*\.(txt|xls|xlsx)$"],
        "confidence_boost": 20,
    },
    "THERMO_FISHER_NANODROP_EIGHT": {
        "columns": ["Sample Name", "Nucleic Acid Conc.", "A260", "A280", "260/280"],
        "keywords": ["NanoDrop Eight", "NanoDrop 8"],
        "file_patterns": [r".*\.(tsv|txt)$"],
        "confidence_boost": 15,
    },
    "THERMO_FISHER_NANODROP_ONE": {
        "columns": ["Sample Name", "Nucleic Acid(ng/uL)", "A260", "A280"],
        "keywords": ["NanoDrop One", "NanoDrop"],
        "file_patterns": [r".*\.(csv|xlsx)$"],
        "confidence_boost": 15,
    },
    "MOLDEV_SOFTMAX_PRO": {
        "columns": ["Well", "Sample", "Values", "Mean", "SD"],
        "keywords": ["SoftMax Pro", "SpectraMax", "Molecular Devices"],
        "file_patterns": [r".*\.txt$"],
        "confidence_boost": 15,
    },
    "BMG_MARS": {
        "columns": ["Well", "Content", "Conc.", "Mean", "SD", "CV"],
        "keywords": ["BMG LABTECH", "MARS", "CLARIOstar", "PHERAstar"],
        "file_patterns": [r".*\.(csv|txt)$"],
        "confidence_boost": 15,
    },
    "AGILENT_GEN5": {
        "columns": ["Well", "Read", "Time", "Temperature"],
        "keywords": ["Gen5", "BioTek", "Synergy"],
        "file_patterns": [r".*\.xlsx$"],
        "confidence_boost": 15,
    },
    "APPBIO_QUANTSTUDIO": {
        "columns": ["Well", "Sample Name", "Target Name", "CT", "Ct Mean"],
        "keywords": ["QuantStudio", "Applied Biosystems", "qPCR"],
        "file_patterns": [r".*\.xlsx$"],
        "confidence_boost": 15,
    },
}


def detect_instrument_type(
    filepath: str, file_content: Optional[str] = None
) -> Tuple[str, float]:
    """
    Auto-detect instrument type from file contents.

    Returns:
        Tuple of (vendor_name, confidence_score)
        confidence_score is 0-100
    """
    path = Path(filepath)
    filename = path.name.lower()
    extension = path.suffix.lower()

    # Read file content if not provided
    if file_content is None:
        try:
            if extension in [".xlsx", ".xls"]:
                pd = get_pandas()
                if pd:
                    df = pd.read_excel(filepath, nrows=50)
                    file_content = df.to_string() + "\n" + "\n".join(df.columns)
                else:
                    file_content = ""
            else:
                with open(filepath, "r", encoding="utf-8", errors="ignore") as f:
                    file_content = f.read(10000)  # First 10KB
        except Exception as e:
            print(f"Warning: Could not read file for detection: {e}")
            file_content = ""

    content_lower = file_content.lower()
    scores = {}

    for vendor, patterns in DETECTION_PATTERNS.items():
        score = 0

        # Check file extension patterns
        for pattern in patterns.get("file_patterns", []):
            if re.match(pattern, filename, re.IGNORECASE):
                score += 10
                break

        # Check column headers
        columns_found = 0
        for col in patterns.get("columns", []):
            if col.lower() in content_lower:
                columns_found += 1
        if columns_found > 0:
            score += min(50, columns_found * 15)

        # Check keywords
        for keyword in patterns.get("keywords", []):
            if keyword.lower() in content_lower:
                score += patterns.get("confidence_boost", 10)

        scores[vendor] = min(100, score)

    # Return best match
    if scores:
        best = max(scores.items(), key=lambda x: x[1])
        return best[0], best[1]

    return "UNKNOWN", 0


def convert_with_allotropy(filepath: str, vendor_name: str) -> Optional[Dict[str, Any]]:
    """
    Convert file using allotropy library.

    Returns:
        ASM dictionary or None if conversion fails
    """
    Vendor, allotrope_from_file, _ = get_allotropy()

    if Vendor is None:
        print(
            "Warning: allotropy not installed. Run: pip install allotropy --break-system-packages"
        )
        return None

    try:
        vendor = getattr(Vendor, vendor_name, None)
        if vendor is None:
            print(f"Warning: Vendor {vendor_name} not found in allotropy")
            return None

        asm = allotrope_from_file(filepath, vendor)
        return asm
    except Exception as e:
        print(f"Allotropy conversion failed: {e}")
        return None


def get_deterministic_timestamp(filepath: str) -> str:
    """
    Get deterministic timestamp for file.
    Uses file modification time for reproducibility.

    Returns:
        ISO format timestamp string
    """
    try:
        path = Path(filepath)
        mtime = path.stat().st_mtime
        return datetime.fromtimestamp(mtime).isoformat()
    except Exception:
        return "TIMESTAMP_NOT_AVAILABLE"


def calculate_file_hash(filepath: str) -> str:
    """Calculate SHA256 hash of file for provenance tracking."""
    try:
        with open(filepath, "rb") as f:
            return hashlib.sha256(f.read()).hexdigest()
    except Exception:
        return "HASH_NOT_AVAILABLE"


def get_library_version(library: str) -> str:
    """Get version of installed library."""
    try:
        return importlib.metadata.version(library)
    except Exception:
        return "VERSION_NOT_AVAILABLE"


def add_provenance_metadata(
    asm: Dict[str, Any],
    filepath: str,
    vendor: str,
    confidence: float,
    used_fallback: bool,
    warnings: list = None,
) -> Dict[str, Any]:
    """
    Add provenance metadata to ASM for reproducibility and audit trail.

    This metadata enables:
    - Reproducing conversions months later
    - Determining which version generated data
    - Auditing data lineage for regulatory compliance
    """
    pd = get_pandas()

    asm["$conversion_metadata"] = {
        "skill_version": "1.0.0",
        "allotropy_version": get_library_version("allotropy"),
        "pandas_version": pd.__version__ if pd else "NOT_INSTALLED",
        "conversion_timestamp_utc": datetime.utcnow().isoformat(),
        "input_file_sha256": calculate_file_hash(filepath),
        "input_file_size_bytes": Path(filepath).stat().st_size,
        "input_file_name": Path(filepath).name,
        "parser_used": "fallback" if used_fallback else "allotropy",
        "detection_confidence": confidence,
        "vendor_detected": vendor,
        "warnings": warnings or [],
    }

    return asm


def flexible_parse(filepath: str, detected_type: str) -> Optional[Dict[str, Any]]:
    """
    Flexible fallback parser when allotropy fails.
    Creates ASM-like structure from parsed data.

    **WARNING:** This parser creates simplified ASM that:
    - Does NOT distinguish raw vs. calculated data
    - LACKS instrument control parameters (temperature, wavelengths, etc.)
    - MAY NOT be compatible with regulatory requirements (GxP)
    - Should be used for exploratory analysis only, not production LIMS import
    """
    pd = get_pandas()
    if pd is None:
        print("Warning: pandas not installed for flexible parsing")
        return None

    path = Path(filepath)
    extension = path.suffix.lower()

    try:
        # Read file based on extension
        if extension in [".xlsx", ".xls"]:
            df = pd.read_excel(filepath, engine="openpyxl")
        elif extension == ".tsv":
            df = pd.read_csv(filepath, sep="\t")
        elif extension == ".csv":
            df = pd.read_csv(filepath)
        else:
            df = pd.read_csv(filepath, sep=None, engine="python")

        # Build ASM-like structure
        asm = build_flexible_asm(df, detected_type, filepath)
        return asm

    except Exception as e:
        print(f"Flexible parsing failed: {e}")
        return None


def build_flexible_asm(df, detected_type: str, filepath: str) -> Dict[str, Any]:
    """
    Build ASM-like JSON structure from parsed DataFrame.
    """
    timestamp = get_deterministic_timestamp(filepath)

    # Determine technique from detected type
    technique = "generic"
    if "VI_CELL" in detected_type:
        technique = "cell-counting"
    elif "NANODROP" in detected_type:
        technique = "spectrophotometry"
    elif detected_type in ["MOLDEV_SOFTMAX_PRO", "BMG_MARS", "AGILENT_GEN5"]:
        technique = "plate-reader"
    elif "QUANTSTUDIO" in detected_type:
        technique = "pcr"

    # Build base structure
    asm = {
        "$asm.manifest": {
            "vocabulary": ["http://purl.allotrope.org/voc/afo/REC/2023/09/"],
            "contexts": [
                "http://purl.allotrope.org/json-ld/afo-context-REC-2023-09.jsonld"
            ],
        },
        f"{technique}-aggregate-document": {
            "device-system-document": {
                "device-identifier": "FLEXIBLE_PARSER",
                "product-manufacturer": (
                    detected_type.split("_")[0] if "_" in detected_type else "Unknown"
                ),
            },
            f"{technique}-document": [
                {
                    "measurement-aggregate-document": {
                        "measurement-time": timestamp,
                        "measurement-document": [],
                    }
                }
            ],
        },
    }

    # Add measurements from DataFrame
    measurements = asm[f"{technique}-aggregate-document"][f"{technique}-document"][0][
        "measurement-aggregate-document"
    ]["measurement-document"]

    for _, row in df.iterrows():
        meas = {}
        for col in df.columns:
            value = row[col]
            if pd.notna(value):
                # Clean column name
                clean_col = str(col).lower().replace(" ", "-").replace("_", "-")
                clean_col = re.sub(r"[^a-z0-9-]", "", clean_col)

                # Handle numeric values
                if isinstance(value, (int, float)):
                    meas[clean_col] = {"value": value, "unit": "(unitless)"}
                else:
                    meas[clean_col] = str(value)

        if meas:
            measurements.append(meas)

    return asm


def main():
    """Main entry point."""
    import argparse

    parser = argparse.ArgumentParser(
        description="Convert instrument data to ASM format"
    )
    parser.add_argument("input", help="Input file path")
    parser.add_argument(
        "--vendor", help="Vendor enum name (auto-detected if not provided)"
    )
    parser.add_argument(
        "--output", "-o", help="Output file path (default: input_asm.json)"
    )
    parser.add_argument(
        "--flatten", action="store_true", help="Also generate flattened CSV"
    )
    parser.add_argument(
        "--allow-fallback",
        action="store_true",
        help="Allow fallback to simplified parser (reduced metadata)",
    )
    parser.add_argument(
        "--skip-validation",
        action="store_true",
        help="Skip automatic validation (not recommended)",
    )
    parser.add_argument(
        "--force",
        action="store_true",
        help="Force conversion even with low confidence detection",
    )

    args = parser.parse_args()

    input_path = Path(args.input)
    if not input_path.exists():
        print(f"Error: File not found: {args.input}")
        sys.exit(1)

    warnings = []

    # Detect or use provided vendor
    if args.vendor:
        vendor = args.vendor.upper()
        confidence = 100
        print(f"Using specified vendor: {vendor}")
    else:
        vendor, confidence = detect_instrument_type(str(input_path))
        print(f"Detected instrument: {vendor} (confidence: {confidence}%)")

        # Enforce confidence thresholds
        if confidence < 30:
            print(
                f"ERROR: Detection confidence too low ({confidence}%). Cannot proceed."
            )
            print("Please specify --vendor explicitly.")
            sys.exit(1)
        elif confidence < 60:
            warning_msg = f"WARNING: Low confidence detection ({confidence}%)."
            print(warning_msg)
            warnings.append(warning_msg)
            if not args.force:
                print("Use --force to proceed anyway (not recommended).")
                sys.exit(1)

    # Try allotropy first
    asm = convert_with_allotropy(str(input_path), vendor)
    used_fallback = False

    # Fall back to flexible parser
    if asm is None:
        print("\n" + "=" * 60)
        print("ALLOTROPY PARSING FAILED - USING REDUCED METADATA PARSER")
        print("=" * 60)
        print("Output will lack:")
        print("  - Calculated data traceability")
        print("  - Device control settings")
        print("  - Data processing metadata")
        print("\nNot suitable for:")
        print("  - Regulatory submissions")
        print("  - LIMS import with validation")
        print("=" * 60 + "\n")

        if not args.allow_fallback:
            print(
                "ERROR: Allotropy parsing failed. Use --allow-fallback to continue with"
            )
            print("simplified parser, but note that output will lack required metadata")
            print("for GxP compliance.")
            sys.exit(1)

        asm = flexible_parse(str(input_path), vendor)
        used_fallback = True
        warnings.append("Used fallback parser - reduced metadata")

    if asm is None:
        print("Error: Could not convert file")
        sys.exit(1)

    # Add provenance metadata
    asm = add_provenance_metadata(
        asm, str(input_path), vendor, confidence, used_fallback, warnings
    )

    # Determine output path
    if args.output:
        output_path = Path(args.output)
    else:
        output_path = input_path.with_suffix(".asm.json")

    # Write to temporary file first
    temp_path = output_path.with_suffix(".tmp")

    try:
        with open(temp_path, "w") as f:
            json.dump(asm, f, indent=2, default=str)

        # Validate unless skipped
        if not args.skip_validation:
            print("Running validation...")
            try:
                from validate_asm import validate_asm

                result = validate_asm(str(temp_path))

                if not result.is_valid():
                    print("\n" + "=" * 60)
                    print("VALIDATION FAILED")
                    print("=" * 60)
                    for error in result.errors:
                        print(f"ERROR: {error}")
                    for warning in result.warnings:
                        print(f"WARNING: {warning}")
                    print("=" * 60)

                    # Remove temp file
                    temp_path.unlink()
                    print("\nValidation failed. Output file not created.")
                    sys.exit(1)
                else:
                    if result.warnings:
                        print("\nValidation warnings:")
                        for warning in result.warnings:
                            print(f"  WARNING: {warning}")
                    print("Validation passed.")
            except ImportError:
                print(
                    "Warning: validate_asm.py not found. Skipping validation. "
                    "Consider adding validation script."
                )

        # Move temp file to final location
        temp_path.replace(output_path)
        print(f"ASM output written to: {output_path}")

    except Exception as e:
        # Clean up temp file on error
        if temp_path.exists():
            temp_path.unlink()
        raise e

    # Optionally flatten
    if args.flatten:
        from flatten_asm import flatten_asm_to_csv

        flat_path = input_path.with_suffix(".flat.csv")
        flatten_asm_to_csv(asm, str(flat_path))
        print(f"Flattened CSV written to: {flat_path}")


if __name__ == "__main__":
    main()

#!/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())

#!/usr/bin/env python3
"""
Flatten ASM JSON to 2D CSV

Converts hierarchical Allotrope Simple Model (ASM) JSON to flat tabular format
suitable for LIMS import, spreadsheet analysis, or database loading.

Usage:
    python flatten_asm.py <input_asm.json> [--output OUTPUT.csv]
"""

import json
import sys
import re
from pathlib import Path
from typing import Dict, Any, List, Optional
from datetime import datetime

try:
    import pandas as pd

    PANDAS_AVAILABLE = True
except ImportError:
    PANDAS_AVAILABLE = False


def detect_technique(asm: Dict[str, Any]) -> str:
    """Detect the ASM technique type from document structure."""
    for key in asm.keys():
        if key.endswith("-aggregate-document"):
            return key.replace("-aggregate-document", "")
    return "generic"


def flatten_value(value: Any, prefix: str = "") -> Dict[str, Any]:
    """
    Flatten a single ASM value, handling value datum patterns.

    Returns dict of {column_name: value}
    """
    result = {}

    if isinstance(value, dict):
        if "value" in value:
            # Value datum pattern
            result[prefix] = value["value"]
            if "unit" in value:
                result[f"{prefix}_unit"] = value["unit"]
        else:
            # Nested dict - recurse
            for k, v in value.items():
                clean_key = k.replace("-", "_")
                nested_prefix = f"{prefix}_{clean_key}" if prefix else clean_key
                result.update(flatten_value(v, nested_prefix))
    elif isinstance(value, list):
        # Array - could be data cube or list of items
        if len(value) > 0 and isinstance(value[0], dict):
            # List of objects - this shouldn't happen at leaf level
            result[prefix] = json.dumps(value)
        else:
            # Simple array - store as JSON string
            result[prefix] = json.dumps(value)
    else:
        # Scalar value
        result[prefix] = value

    return result


def extract_device_info(asm: Dict[str, Any], technique: str) -> Dict[str, Any]:
    """Extract device/instrument information from ASM."""
    agg_key = f"{technique}-aggregate-document"
    agg_doc = asm.get(agg_key, {})

    device = agg_doc.get("device-system-document", {})

    return {
        "instrument_serial_number": device.get("device-identifier"),
        "instrument_model": device.get("model-number"),
        "instrument_manufacturer": device.get("product-manufacturer"),
        "software_name": device.get("software-name"),
        "software_version": device.get("software-version"),
    }


def flatten_asm(asm: Dict[str, Any]) -> List[Dict[str, Any]]:
    """
    Flatten ASM JSON to list of row dictionaries.

    Each measurement becomes one row with metadata repeated.
    """
    technique = detect_technique(asm)
    rows = []

    # Extract device info (shared across all rows)
    device_info = extract_device_info(asm, technique)
    device_info = {k: v for k, v in device_info.items() if v is not None}

    # Navigate to measurements
    agg_key = f"{technique}-aggregate-document"
    agg_doc = asm.get(agg_key, {})

    doc_key = f"{technique}-document"
    technique_docs = agg_doc.get(doc_key, [])

    for doc in technique_docs:
        # Get measurement aggregate
        meas_agg = doc.get("measurement-aggregate-document", {})

        # Extract common measurement metadata
        common_meta = {}
        for key, value in meas_agg.items():
            if key == "measurement-document":
                continue
            clean_key = key.replace("-", "_")
            if isinstance(value, (str, int, float, bool)):
                common_meta[clean_key] = value
            elif isinstance(value, dict) and "value" in value:
                common_meta[clean_key] = value["value"]
                if "unit" in value:
                    common_meta[f"{clean_key}_unit"] = value["unit"]

        # Extract each measurement as a row
        measurements = meas_agg.get("measurement-document", [])
        for meas in measurements:
            row = {**device_info, **common_meta}

            for key, value in meas.items():
                clean_key = key.replace("-", "_")
                flattened = flatten_value(value, clean_key)
                row.update(flattened)

            rows.append(row)

    return rows


def flatten_asm_to_csv(asm: Dict[str, Any], output_path: str) -> None:
    """
    Flatten ASM and write to CSV file.

    Args:
        asm: Parsed ASM JSON dictionary
        output_path: Path for output CSV
    """
    if not PANDAS_AVAILABLE:
        raise ImportError(
            "pandas is required for CSV output. Install with: pip install pandas"
        )

    rows = flatten_asm(asm)

    if not rows:
        print("Warning: No measurements found to flatten")
        # Create empty CSV with header
        with open(output_path, "w") as f:
            f.write("# No measurements found in ASM\n")
        return

    df = pd.DataFrame(rows)

    # Reorder columns for readability
    priority_cols = [
        "sample_identifier",
        "sample_id",
        "well_location",
        "well_position",
        "measurement_time",
        "measurement_datetime",
        "analyst",
    ]

    ordered_cols = []
    for col in priority_cols:
        if col in df.columns:
            ordered_cols.append(col)

    remaining = [c for c in df.columns if c not in ordered_cols]
    df = df[ordered_cols + remaining]

    df.to_csv(output_path, index=False)


def flatten_asm_to_dict(asm: Dict[str, Any]) -> Dict[str, Any]:
    """
    Flatten ASM and return as dictionary with rows and columns.

    Useful for non-CSV outputs or further processing.
    """
    rows = flatten_asm(asm)

    if not rows:
        return {"columns": [], "rows": []}

    columns = list(rows[0].keys())
    return {
        "columns": columns,
        "rows": [[row.get(col) for col in columns] for row in rows],
    }


def main():
    """Main entry point."""
    import argparse

    parser = argparse.ArgumentParser(description="Flatten ASM JSON to CSV")
    parser.add_argument("input", help="Input ASM JSON file")
    parser.add_argument(
        "--output", "-o", help="Output CSV path (default: input_flat.csv)"
    )
    parser.add_argument(
        "--format",
        choices=["csv", "json"],
        default="csv",
        help="Output format (default: csv)",
    )

    args = parser.parse_args()

    input_path = Path(args.input)
    if not input_path.exists():
        print(f"Error: File not found: {args.input}")
        sys.exit(1)

    # Load ASM
    with open(input_path) as f:
        asm = json.load(f)

    # Determine output path
    if args.output:
        output_path = args.output
    else:
        suffix = ".flat.csv" if args.format == "csv" else ".flat.json"
        output_path = str(input_path.with_suffix("")) + suffix

    # Flatten and write
    if args.format == "csv":
        flatten_asm_to_csv(asm, output_path)
    else:
        result = flatten_asm_to_dict(asm)
        with open(output_path, "w") as f:
            json.dump(result, f, indent=2)

    print(f"Flattened output written to: {output_path}")

    # Report stats
    rows = flatten_asm(asm)
    print(f"  Rows: {len(rows)}")
    if rows:
        print(f"  Columns: {len(rows[0])}")


if __name__ == "__main__":
    main()