diff --git a/.claude/.ruff.toml b/.claude/.ruff.toml new file mode 100644 index 000000000..754a91af2 --- /dev/null +++ b/.claude/.ruff.toml @@ -0,0 +1,7 @@ +# Ruff configuration for Claude skills +# These are CLI scripts, so print statements are allowed + +extend-ignore = ["T201", "T203", "E501"] # Allow print, pprint, and long lines in skills + +[per-file-ignores] +"skills/*/scripts/*.py" = ["T201", "T203", "FBT001", "FBT002"] # Allow print and boolean args in skill scripts \ No newline at end of file diff --git a/.claude/skills/parser-generator/README.md b/.claude/skills/parser-generator/README.md new file mode 100644 index 000000000..2114da949 --- /dev/null +++ b/.claude/skills/parser-generator/README.md @@ -0,0 +1,120 @@ +# Allotropy Parser Generator Skill + +This Claude skill helps you generate complete Allotropy instrument parsers from example input files. It analyzes file structure, auto-detects appropriate Allotrope schemas, and generates fully functional parser code. + +## Installation + +This skill is already installed in the `.claude/skills/parser-generator/` directory. + +## Usage with Claude + +When working with Claude in this repository, you can invoke the parser generator skill by saying: + +- "Use the parser-generator skill to create a new parser" +- "Analyze this file and suggest a schema" (with a file path) +- "Generate a parser for [instrument name]" +- "/parser-generator" (if configured as a slash command) + +## Manual Usage + +You can also run the scripts directly: + +### 1. Analyze an Input File + +```bash +python .claude/skills/parser-generator/scripts/analyze_file.py +``` + +This will analyze the file and suggest an appropriate schema based on the content. + +### 2. List Available Schemas + +```bash +# List all schemas +python .claude/skills/parser-generator/scripts/list_schemas.py + +# Filter schemas +python .claude/skills/parser-generator/scripts/list_schemas.py plate + +# Verbose output with paths +python .claude/skills/parser-generator/scripts/list_schemas.py --verbose +``` + +### 3. Generate a Parser + +```bash +python .claude/skills/parser-generator/scripts/create_parser.py \ + \ + "" \ + --schema \ + --example +``` + +Example: +```bash +python .claude/skills/parser-generator/scripts/create_parser.py \ + beckman_pharmspec \ + "Beckman Coulter PharmSpec" \ + --schema adm/plate_reader/BENCHLING/2024/06/plate_reader \ + --example ~/Downloads/pharmspec_data.xlsx +``` + +## Generated Files + +The skill creates a complete parser structure: + +``` +src/allotropy/parsers/{parser_name}/ +├── __init__.py # Module exports +├── {parser_name}_parser.py # Main parser class +├── {parser_name}_reader.py # File reading logic +└── {parser_name}_structure.py # Data structures + +tests/parsers/{parser_name}/ +├── __init__.py +├── test_{parser_name}_parser.py # Test file +└── testdata/ + └── example.xlsx # Test data +``` + +## Workflow + +1. **Analyze** your example file to detect the schema +2. **List** available schemas if you need to override +3. **Generate** the parser with the create script +4. **Implement** the TODO sections in the generated code +5. **Test** the parser with your example data +6. **Register** the parser in `parser_factory.py` + +## Supported File Formats + +- Excel files (.xlsx, .xls) +- CSV files (.csv) +- Tab-delimited files (.txt, .tsv) +- Text files with sections + +## Schema Detection + +The skill detects schemas based on keywords in the file: + +- **plate-reader**: absorbance, fluorescence, luminescence, well, plate +- **pcr**: ct, cq, cycle, amplification, target +- **cell-counting**: viability, cell count, density +- **spectrophotometry**: wavelength, spectrum, nm +- **solution-analyzer**: pH, osmolality, particle size +- **binding-affinity**: ka, kd, kon, koff, resonance + +## Tips + +- Always start with `analyze_file.py` to understand your input format +- Look at similar existing parsers for patterns +- Use the existing utility functions in `allotropy.parsers.utils` +- Test with real data files early and often +- Start with `WORKING_DRAFT` release state + +## Troubleshooting + +- **Schema not detected**: Manually specify with `list_schemas.py` +- **Parser already exists**: Choose a different name or delete existing +- **Import errors**: Make sure you're in the allotropy repository +- **Test failures**: Check that your Data structure matches the schema \ No newline at end of file diff --git a/.claude/skills/parser-generator/scripts/analyze_file.py b/.claude/skills/parser-generator/scripts/analyze_file.py new file mode 100755 index 000000000..eabf74ba7 --- /dev/null +++ b/.claude/skills/parser-generator/scripts/analyze_file.py @@ -0,0 +1,315 @@ +#!/usr/bin/env python3 +# type: ignore +""" +Analyzes an input file to infer structure and suggest appropriate Allotrope schema. +Usage: python analyze_file.py +""" +from pathlib import Path +import sys +from typing import Any + +import pandas as pd + + +def analyze_excel_file(file_path: Path) -> dict[str, Any]: + """Analyze Excel file structure.""" + try: + # Try to read Excel file (try calamine first, fall back to openpyxl) + try: + df = pd.read_excel(file_path, header=None, engine="calamine") + except Exception: + df = pd.read_excel(file_path, header=None, engine="openpyxl") + + analysis: dict[str, Any] = { + "format": "excel", + "shape": df.shape, + "columns": df.shape[1], + "rows": df.shape[0], + "has_headers": False, + "potential_sections": [], + "measurement_indicators": [], + "suggested_schema": None, + } + + # Check for headers (first row contains mostly strings) + if df.iloc[0].dtype == "object": + analysis["has_headers"] = True + + # Look for measurement type indicators + content_str = df.to_string().lower() + + # Schema detection based on keywords + indicators = { + "plate-reader": [ + "absorbance", + "fluorescence", + "luminescence", + "well", + "plate", + "od", + "optical density", + ], + "pcr": [ + "ct", + "cq", + "cycle", + "amplification", + "melt", + "target", + "threshold", + ], + "solution-analyzer": [ + "ph", + "osmolality", + "particle", + "size", + "distribution", + "conductivity", + ], + "cell-counting": [ + "viability", + "cell density", + "cell count", + "live cells", + "dead cells", + "total cells", + ], + "spectrophotometry": [ + "wavelength", + "spectrum", + "absorbance", + "transmittance", + "nm", + ], + "electrophoresis": ["lane", "band", "migration", "gel", "ladder"], + "liquid-chromatography": [ + "retention time", + "peak", + "chromatogram", + "area", + "height", + ], + "binding-affinity": [ + "ka", + "kd", + "kon", + "koff", + "resonance", + "binding", + "affinity", + ], + "flow-cytometry": ["fsc", "ssc", "fluorescence", "population", "gating"], + } + + detected_schemas = [] + for schema, keywords in indicators.items(): + matches = sum(1 for kw in keywords if kw in content_str) + if matches > 0: + detected_schemas.append((schema, matches)) + analysis["measurement_indicators"].extend( + [kw for kw in keywords if kw in content_str] + ) + + # Sort by number of matches + if detected_schemas: + detected_schemas.sort(key=lambda x: x[1], reverse=True) + analysis["suggested_schema"] = detected_schemas[0][0] + analysis["all_matches"] = detected_schemas + + # Check for sheet names if Excel + try: + try: + xl = pd.ExcelFile(file_path, engine="calamine") + except Exception: + xl = pd.ExcelFile(file_path, engine="openpyxl") + analysis["sheet_names"] = xl.sheet_names + except Exception: + pass # noqa: S110 + + return analysis + + except Exception as e: + return {"format": "excel", "error": str(e)} + + +def analyze_text_file(file_path: Path) -> dict[str, Any]: + """Analyze text file structure.""" + try: + with open(file_path, encoding="utf-8") as f: + content = f.read() + + lines = content.split("\n") + + analysis: dict[str, Any] = { + "format": "text", + "lines": len(lines), + "has_sections": False, + "section_markers": [], + "delimiter": None, + "measurement_indicators": [], + "suggested_schema": None, + } + + # Check for section markers + for line in lines[:50]: # Check first 50 lines + if line.strip().startswith("[") and line.strip().endswith("]"): + analysis["has_sections"] = True + analysis["section_markers"].append(line.strip()) + + # Detect delimiter + if "\t" in content: + analysis["delimiter"] = "tab" + elif "," in content: + analysis["delimiter"] = "comma" + + # Schema detection (same as Excel) + content_lower = content.lower() + indicators = { + "plate-reader": [ + "absorbance", + "fluorescence", + "luminescence", + "well", + "plate", + "od", + "optical density", + ], + "pcr": [ + "ct", + "cq", + "cycle", + "amplification", + "melt", + "target", + "threshold", + ], + "solution-analyzer": [ + "ph", + "osmolality", + "particle", + "size", + "distribution", + "conductivity", + ], + "cell-counting": [ + "viability", + "cell density", + "cell count", + "live cells", + "dead cells", + ], + "spectrophotometry": [ + "wavelength", + "spectrum", + "absorbance", + "transmittance", + "nm", + ], + "binding-affinity": [ + "ka", + "kd", + "kon", + "koff", + "resonance", + "binding", + "affinity", + ], + } + + detected_schemas = [] + for schema, keywords in indicators.items(): + matches = sum(1 for kw in keywords if kw in content_lower) + if matches > 0: + detected_schemas.append((schema, matches)) + analysis["measurement_indicators"].extend( + [kw for kw in keywords if kw in content_lower] + ) + + if detected_schemas: + detected_schemas.sort(key=lambda x: x[1], reverse=True) + analysis["suggested_schema"] = detected_schemas[0][0] + analysis["all_matches"] = detected_schemas + + return analysis + + except Exception as e: + return {"format": "text", "error": str(e)} + + +def main(): + if len(sys.argv) < 2: + print("Usage: python analyze_file.py ") + print( + "\nThis script analyzes an input file to suggest the appropriate " + "Allotrope schema." + ) + sys.exit(1) + + file_path = Path(sys.argv[1]) + + if not file_path.exists(): + print(f"Error: File not found: {file_path}") + sys.exit(1) + + # Determine file type and analyze + if file_path.suffix.lower() in [".xlsx", ".xls"]: + analysis = analyze_excel_file(file_path) + elif file_path.suffix.lower() in [".txt", ".csv", ".tsv"]: + analysis = analyze_text_file(file_path) + else: + print(f"Unsupported file format: {file_path.suffix}") + print("Supported formats: .xlsx, .xls, .txt, .csv, .tsv") + sys.exit(1) + + # Print results + print("\n" + "=" * 60) + print(f"FILE ANALYSIS: {file_path.name}") + print("=" * 60) + + for key, value in analysis.items(): + if key == "all_matches": + print(f"\n{key.upper()}:") + for schema, count in value: + print(f" - {schema}: {count} indicators") + elif key == "sheet_names" and value: + print("\nSHEET NAMES:") + for sheet in value: + print(f" - {sheet}") + elif key == "measurement_indicators" and value: + print("\nMEASUREMENT INDICATORS FOUND:") + unique_indicators = list(set(value))[:10] # Limit to first 10 unique + for item in unique_indicators: + print(f" - {item}") + elif key == "section_markers" and value: + print("\nSECTION MARKERS:") + for item in value[:10]: # Limit to first 10 + print(f" - {item}") + elif key not in [ + "all_matches", + "measurement_indicators", + "section_markers", + "sheet_names", + ]: + print(f"{key}: {value}") + + print("\n" + "=" * 60) + if analysis.get("suggested_schema"): + print(f"✅ RECOMMENDED SCHEMA: {analysis['suggested_schema']}") + print("\nNext steps:") + print( + f"1. Run 'python .claude/skills/parser-generator/scripts/list_schemas.py {analysis['suggested_schema']}' for more details" + ) + print( + "2. Review existing parsers in src/allotropy/parsers/ for similar instruments" + ) + print("3. Run 'python scripts/create_parser.py' to generate the parser") + else: + print("⚠️ Could not determine schema - manual selection required") + print( + "\nRun 'python .claude/skills/parser-generator/scripts/list_schemas.py' to see all available schemas" + ) + print("=" * 60 + "\n") + + +if __name__ == "__main__": + main() diff --git a/.claude/skills/parser-generator/scripts/create_parser.py b/.claude/skills/parser-generator/scripts/create_parser.py new file mode 100755 index 000000000..1280c9bfd --- /dev/null +++ b/.claude/skills/parser-generator/scripts/create_parser.py @@ -0,0 +1,377 @@ +#!/usr/bin/env python3 +# type: ignore +""" +Creates a new Allotropy parser with complete file structure. +Usage: python create_parser.py [--schema SCHEMA_PATH] [--example EXAMPLE_FILE] +""" +from pathlib import Path +import sys + + +def create_parser_init(parser_name: str) -> str: + """Generate __init__.py content.""" + class_name = ( + "".join(word.capitalize() for word in parser_name.split("_")) + "Parser" + ) + return f'''"""Allotropy parser for {parser_name.replace('_', ' ').title()}.""" + +from allotropy.parsers.{parser_name}.{parser_name}_parser import {class_name} + +__all__ = ["{class_name}"] +''' + + +def create_parser_file( + parser_name: str, display_name: str, schema_path: str, extension: str +) -> str: + """Generate parser.py content.""" + class_name = ( + "".join(word.capitalize() for word in parser_name.split("_")) + "Parser" + ) + + # Extract schema details from path + # Expected format: adm/technique/ORG/YEAR/MONTH/technique + path_parts = schema_path.split("/") + path_parts[1] if len(path_parts) > 1 else "plate_reader" + + return f'''"""Parser for {display_name}.""" + +from allotropy.allotrope.models.{schema_path.replace('/', '.')} import Model +from allotropy.allotrope.schema_mappers.{schema_path.replace('/', '.')} import ( + Data, + Mapper, +) +from allotropy.named_file_contents import NamedFileContents +from allotropy.parsers.release_state import ReleaseState +from allotropy.parsers.vendor_parser import VendorParser + +from allotropy.parsers.{parser_name}.{parser_name}_reader import {class_name}Reader +from allotropy.parsers.{parser_name}.{parser_name}_structure import ( + create_metadata, + create_measurement_groups, +) + + +class {class_name}(VendorParser[Data, Model]): + """Parser for {display_name} files.""" + + DISPLAY_NAME = "{display_name}" + RELEASE_STATE = ReleaseState.WORKING_DRAFT + SUPPORTED_EXTENSIONS = "{extension}" + SCHEMA_MAPPER = Mapper + + def create_data(self, named_file_contents: NamedFileContents) -> Data: + """Create Data object from input file.""" + reader = {class_name}Reader(named_file_contents) + + return Data( + create_metadata(reader.header, named_file_contents.original_file_path), + create_measurement_groups(reader.measurements), + ) +''' + + +def create_reader_file(parser_name: str, extension: str) -> str: + """Generate reader.py content.""" + class_name = ( + "".join(word.capitalize() for word in parser_name.split("_")) + "Reader" + ) + + if extension == "xlsx": + reader_content = '''"""Reader for {display_name} Excel files.""" + +from typing import Any, Dict, List +import pandas as pd + +from allotropy.named_file_contents import NamedFileContents +from allotropy.parsers.utils.pandas import read_excel + + +class {class_name}: + """Reads and parses {display_name} Excel files.""" + + def __init__(self, named_file_contents: NamedFileContents) -> None: + """Initialize reader with file contents.""" + # Read the Excel file + df = read_excel( + named_file_contents.contents, + header=None, + sheet_name=0, # Or specify sheet name + ) + + # Parse header section + self.header = self._parse_header(df) + + # Parse measurements section + self.measurements = self._parse_measurements(df) + + def _parse_header(self, df: pd.DataFrame) -> Dict[str, Any]: + """Parse header/metadata section.""" + header = {{}} + + # TODO: Extract metadata from appropriate rows + # Example: + # header["instrument_model"] = df.iloc[0, 1] + # header["run_date"] = df.iloc[1, 1] + + return header + + def _parse_measurements(self, df: pd.DataFrame) -> List[Dict[str, Any]]: + """Parse measurement data.""" + measurements = [] + + # TODO: Find data section and parse + # Example: + # data_start_row = 10 # Adjust based on file format + # data = df.iloc[data_start_row:].reset_index(drop=True) + # data.columns = df.iloc[data_start_row - 1] # Use row above as headers + + return measurements +''' + else: + reader_content = '''"""Reader for {display_name} text files.""" + +from typing import Any, Dict, List + +from allotropy.named_file_contents import NamedFileContents +from allotropy.parsers.lines_reader import LinesReader + + +class {class_name}: + """Reads and parses {display_name} text files.""" + + def __init__(self, named_file_contents: NamedFileContents) -> None: + """Initialize reader with file contents.""" + reader = LinesReader(named_file_contents.contents) + + # Parse header section + self.header = self._parse_header(reader) + + # Parse measurements section + self.measurements = self._parse_measurements(reader) + + def _parse_header(self, reader: LinesReader) -> Dict[str, Any]: + """Parse header/metadata section.""" + header = {{}} + + # TODO: Extract metadata from lines + # Example: + # for line in reader.pop_until_empty(): + # if ":" in line: + # key, value = line.split(":", 1) + # header[key.strip()] = value.strip() + + return header + + def _parse_measurements(self, reader: LinesReader) -> List[Dict[str, Any]]: + """Parse measurement data.""" + measurements = [] + + # TODO: Parse data section + # Example: + # headers = reader.pop().split("\\t") + # for line in reader.lines: + # values = line.split("\\t") + # measurement = dict(zip(headers, values)) + # measurements.append(measurement) + + return measurements +''' + + return reader_content.format( + display_name=parser_name.replace("_", " ").title(), class_name=class_name + ) + + +def create_structure_file(parser_name: str) -> str: + """Generate structure.py content.""" + return f'''"""Data structures for {parser_name.replace('_', ' ').title()} parser.""" + +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Dict, List, Optional + +from allotropy.allotrope.schema_mappers.adm.plate_reader.rec._2024._06.plate_reader import ( + Measurement, + MeasurementGroup, + MeasurementType, + Metadata, +) +from allotropy.parsers.constants import NOT_APPLICABLE +from allotropy.parsers.utils.pandas import SeriesData +from allotropy.parsers.utils.uuids import random_uuid_str + + +def create_metadata(header: Dict[str, Any], file_path: Optional[Path]) -> Metadata: + """Create Metadata object from header data.""" + return Metadata( + device_identifier=header.get("instrument_id", NOT_APPLICABLE), + model_number=header.get("model", NOT_APPLICABLE), + software_name=header.get("software", NOT_APPLICABLE), + unc_path=str(file_path) if file_path else NOT_APPLICABLE, + data_system_instance_id=NOT_APPLICABLE, + ) + + +def create_measurement_groups( + measurements_data: List[Dict[str, Any]] +) -> List[MeasurementGroup]: + """Create MeasurementGroup objects from measurement data.""" + measurements = [] + + for data in measurements_data: + measurement = Measurement( + identifier=random_uuid_str(), + type_=MeasurementType.UNKNOWN, # TODO: Set appropriate type + sample_identifier=data.get("sample_id", NOT_APPLICABLE), + # TODO: Add more measurement fields + ) + measurements.append(measurement) + + # Group measurements as needed + return [ + MeasurementGroup( + measurements=measurements, + ) + ] +''' + + +def create_test_file(parser_name: str) -> str: + """Generate test file content.""" + return f'''"""Tests for {parser_name.replace('_', ' ').title()} parser.""" + +from allotropy.testing.utils import run_allotropy + + +def test_to_allotrope_{parser_name}() -> None: + """Test parsing of example {parser_name.replace('_', ' ')} file.""" + test_file = "tests/parsers/{parser_name}/testdata/example.xlsx" + expected_file = "tests/parsers/{parser_name}/testdata/example.json" + run_allotropy(test_file, expected_file) +''' + + +def main(): + """Main function to create parser structure.""" + if len(sys.argv) < 3: + print( + "Usage: python create_parser.py [--schema SCHEMA_PATH] [--example EXAMPLE_FILE]" + ) + print("\nExample:") + print( + ' python create_parser.py beckman_pharmspec "Beckman Coulter PharmSpec" --schema adm/plate_reader/BENCHLING/2024/06/plate_reader' + ) + sys.exit(1) + + parser_name = sys.argv[1].lower().replace(" ", "_").replace("-", "_") + display_name = sys.argv[2] + + # Parse optional arguments + schema_path = "adm/plate_reader/BENCHLING/2024/06/plate_reader" # Default + example_file = None + extension = "xlsx" # Default + + i = 3 + while i < len(sys.argv): + if sys.argv[i] == "--schema" and i + 1 < len(sys.argv): + schema_path = sys.argv[i + 1] + i += 2 + elif sys.argv[i] == "--example" and i + 1 < len(sys.argv): + example_file = Path(sys.argv[i + 1]) + extension = example_file.suffix[1:] if example_file.suffix else "xlsx" + i += 2 + else: + i += 1 + + # Find repository root + current = Path.cwd() + repo_root = None + for parent in [current, *list(current.parents)]: + if (parent / "src" / "allotropy").exists(): + repo_root = parent + break + + if not repo_root: + print("Error: Must run from within allotropy repository") + sys.exit(1) + + # Create parser directory structure + parser_dir = repo_root / "src" / "allotropy" / "parsers" / parser_name + test_dir = repo_root / "tests" / "parsers" / parser_name + testdata_dir = test_dir / "testdata" + + # Check if parser already exists + if parser_dir.exists(): + print(f"Error: Parser '{parser_name}' already exists at {parser_dir}") + sys.exit(1) + + # Create directories + parser_dir.mkdir(parents=True) + test_dir.mkdir(parents=True) + testdata_dir.mkdir(parents=True) + + # Create parser files + files_created = [] + + # __init__.py + init_file = parser_dir / "__init__.py" + init_file.write_text(create_parser_init(parser_name)) + files_created.append(init_file) + + # parser.py + parser_file = parser_dir / f"{parser_name}_parser.py" + parser_file.write_text( + create_parser_file(parser_name, display_name, schema_path, extension) + ) + files_created.append(parser_file) + + # reader.py + reader_file = parser_dir / f"{parser_name}_reader.py" + reader_file.write_text(create_reader_file(parser_name, extension)) + files_created.append(reader_file) + + # structure.py + structure_file = parser_dir / f"{parser_name}_structure.py" + structure_file.write_text(create_structure_file(parser_name)) + files_created.append(structure_file) + + # Test file + test_file = test_dir / f"test_{parser_name}_parser.py" + test_file.write_text(create_test_file(parser_name)) + files_created.append(test_file) + + # Test __init__.py + test_init = test_dir / "__init__.py" + test_init.write_text("") + files_created.append(test_init) + + # Copy example file if provided + if example_file and example_file.exists(): + import shutil + + dest_file = testdata_dir / f"example.{extension}" + shutil.copy(example_file, dest_file) + files_created.append(dest_file) + + # Print summary + print(f"\n✅ Successfully created parser: {parser_name}") + print(f" Display name: {display_name}") + print(f" Schema: {schema_path}") + print(f" Extension: {extension}") + + print("\n📁 Files created:") + for file in files_created: + print(f" - {file.relative_to(repo_root)}") + + print("\n📝 Next steps:") + print(f"1. Edit {parser_name}_reader.py to implement file parsing") + print(f"2. Edit {parser_name}_structure.py to map data to schema") + print(f"3. Add test data to {testdata_dir}") + print(f"4. Run tests: hatch run test:pytest tests/parsers/{parser_name}/") + print("5. Register parser in src/allotropy/parser_factory.py") + print("6. Update README.md with parser information") + + +if __name__ == "__main__": + main() diff --git a/.claude/skills/parser-generator/scripts/list_schemas.py b/.claude/skills/parser-generator/scripts/list_schemas.py new file mode 100755 index 000000000..7e049f153 --- /dev/null +++ b/.claude/skills/parser-generator/scripts/list_schemas.py @@ -0,0 +1,208 @@ +#!/usr/bin/env python3 +# type: ignore +""" +Lists available Allotrope schemas by scanning the local allotropy repository. +Usage: python list_schemas.py [filter] [--verbose] +""" +from collections import defaultdict +import os +from pathlib import Path +import sys + +# Fallback descriptions for common schema types +SCHEMA_DESCRIPTIONS = { + "plate-reader": "Microplate readers measuring absorbance, fluorescence, luminescence in wells", + "plate_reader": "Microplate readers measuring absorbance, fluorescence, luminescence in wells", + "pcr": "PCR instruments (qPCR and dPCR) measuring amplification cycles and targets", + "qpcr": "Quantitative PCR (qPCR) measuring real-time amplification", + "dpcr": "Digital PCR (dPCR) measuring absolute quantification", + "solution-analyzer": "Instruments analyzing solution properties (pH, osmolality, particle size)", + "solution_analyzer": "Instruments analyzing solution properties (pH, osmolality, particle size)", + "cell-counting": "Cell counters measuring viability, density, and cell populations", + "cell_counting": "Cell counters measuring viability, density, and cell populations", + "spectrophotometry": "Spectrophotometers measuring absorbance/transmittance across wavelengths", + "electrophoresis": "Electrophoresis systems analyzing DNA/RNA/protein separation", + "liquid-chromatography": "Liquid chromatography systems measuring compound separation and peaks", + "liquid_chromatography": "Liquid chromatography systems measuring compound separation and peaks", + "multi-analyte-profiling": "Multi-analyte detection platforms (e.g., Luminex)", + "multi_analyte_profiling": "Multi-analyte detection platforms (e.g., Luminex)", + "binding-affinity": "Surface plasmon resonance and binding affinity analyzers", + "binding_affinity": "Surface plasmon resonance and binding affinity analyzers", + "binding-affinity-analyzer": "Surface plasmon resonance and binding affinity analyzers", + "binding_affinity_analyzer": "Surface plasmon resonance and binding affinity analyzers", + "flow-cytometry": "Flow cytometers analyzing cell populations and markers", + "flow_cytometry": "Flow cytometers analyzing cell populations and markers", + "mass_spectrometry": "Mass spectrometers analyzing molecular mass and structure", +} + + +def find_allotropy_repo() -> Path | None: + """Find the allotropy repository in the environment.""" + # Check if we're in the allotropy repo + current = Path.cwd() + for parent in [current, *list(current.parents)]: + if (parent / "src" / "allotropy" / "allotrope" / "schemas").exists(): + return parent + + # Check environment variable + if allotropy_path := os.getenv("ALLOTROPY_PATH"): + path = Path(allotropy_path) + if path.exists(): + return path + + return None + + +def scan_schemas(allotropy_path: Path) -> dict[str, list[dict[str, str]]]: + """Scan the allotropy repository for available schemas.""" + schemas_dir = allotropy_path / "src" / "allotropy" / "allotrope" / "schemas" / "adm" + + if not schemas_dir.exists(): + print(f"Error: Schema directory not found: {schemas_dir}") + sys.exit(1) + + # Group schemas by technique + schemas: dict[str, list[dict[str, str]]] = defaultdict(list) + + for technique_dir in sorted(schemas_dir.iterdir()): + if not technique_dir.is_dir(): + continue + + technique_name = technique_dir.name + + # Scan for schema files + for schema_file in technique_dir.rglob("*.schema.json"): + # Build relative path from schemas/adm/ + rel_path = schema_file.relative_to(schemas_dir) + # Remove .schema.json extension + schema_path = str(rel_path).replace(".schema.json", "") + + # Try to get the version info from path + path_parts = schema_path.split("/") + version_info = "" + if len(path_parts) >= 4: + # Format: technique/ORG/YEAR/MONTH + version_info = f"{path_parts[1]}/{path_parts[2]}/{path_parts[3]}" + + schemas[technique_name].append( + { + "path": f"adm/{schema_path}", + "version": version_info, + "file": str(schema_file), + } + ) + + return dict(schemas) + + +def find_parsers_using_schema(allotropy_path: Path, technique_name: str) -> list[str]: + """Find parsers that might use a specific schema technique.""" + parsers_dir = allotropy_path / "src" / "allotropy" / "parsers" + matching_parsers = [] + + if not parsers_dir.exists(): + return matching_parsers + + # Look for parsers that import from this schema technique + for parser_dir in parsers_dir.iterdir(): + if not parser_dir.is_dir() or parser_dir.name.startswith("_"): + continue + + # Check if parser file exists + parser_file = parser_dir / f"{parser_dir.name}_parser.py" + if parser_file.exists(): + try: + with open(parser_file) as f: + content = f.read() + # Check for imports from the schema + if ( + f"adm.{technique_name}" in content + or f"adm/{technique_name}" in content + ): + matching_parsers.append(parser_dir.name) + except Exception: + pass # noqa: S110 + + return matching_parsers + + +def list_schemas( + allotropy_path: Path, filter_term: str | None = None, verbose: bool = False +) -> None: + """List all available schemas from the local repository.""" + print("\n" + "=" * 80) + print("AVAILABLE ALLOTROPE SCHEMAS") + print(f"Repository: {allotropy_path}") + print("=" * 80 + "\n") + + schemas = scan_schemas(allotropy_path) + + if not schemas: + print("No schemas found in repository.") + return + + total_count = 0 + for technique_name in sorted(schemas.keys()): + if filter_term and filter_term.lower() not in technique_name.lower(): + continue + + schema_list = sorted(schemas[technique_name], key=lambda x: x["path"]) + total_count += len(schema_list) + + print(f"📋 {technique_name.upper()}") + + # Show description if available + description = SCHEMA_DESCRIPTIONS.get(technique_name) + if description: + print(f" {description}") + + print(f"\n Available schema paths ({len(schema_list)} version(s)):") + for schema_info in schema_list: + if verbose: + print(f" - {schema_info['path']}") + print(f" Version: {schema_info['version']}") + else: + print(f" - {schema_info['path']}") + + # Find example parsers + matching_parsers = find_parsers_using_schema(allotropy_path, technique_name) + if matching_parsers: + print("\n 📁 Example parsers using this schema:") + for parser in matching_parsers[:5]: # Limit to 5 + print(f" - {parser}") + + print("\n" + "-" * 80 + "\n") + + if not filter_term: + print(f"Total schemas found: {total_count}") + print("\nUse a filter to see specific schemas: python list_schemas.py ") + print("Use --verbose for more details: python list_schemas.py --verbose") + + +def main() -> None: + # Parse arguments + args = sys.argv[1:] + filter_term = None + verbose = False + + for arg in args: + if arg in ["--verbose", "-v"]: + verbose = True + elif not arg.startswith("-"): + filter_term = arg + + # Find allotropy repository + allotropy_path = find_allotropy_repo() + + if not allotropy_path: + print("Error: Could not find allotropy repository.") + print("\nPlease run this script from within the allotropy repository,") + print("or set ALLOTROPY_PATH environment variable:") + print(" export ALLOTROPY_PATH=/path/to/allotropy") + sys.exit(1) + + list_schemas(allotropy_path, filter_term, verbose) + + +if __name__ == "__main__": + main() diff --git a/.claude/skills/parser-generator/skill.md b/.claude/skills/parser-generator/skill.md new file mode 100644 index 000000000..8c7052199 --- /dev/null +++ b/.claude/skills/parser-generator/skill.md @@ -0,0 +1,233 @@ +--- +name: parser-generator +description: | + Generate complete Allotropy instrument parsers from example input files. This skill analyzes file structure, auto-detects appropriate Allotrope schemas, and generates fully functional parser code including reader, structure, schema mapper, and tests. Use when creating new parsers for scientific instruments that output data files (Excel, TXT, CSV, binary formats) that need conversion to Allotrope ASM format. +--- + +# Allotropy Parser Generator + +Generate complete, production-ready parsers for scientific instrument data files. + +## Quick Start Workflow + +1. **Analyze input file** - Run analysis script to detect structure and schema +2. **Review suggestions** - Confirm or override detected schema +3. **Generate parser** - Create complete parser with all files +4. **Test and validate** - Run tests to ensure correctness +5. **Register parser** - Add to parser factory + +## Step 1: Analyze Input File + +Always start by analyzing the example input file: + +```bash +python scripts/analyze_file.py +``` + +This script: +- Detects file format (Excel, TXT, CSV, binary) +- Identifies measurement types (absorbance, pH, CT values, etc.) +- Suggests appropriate Allotrope schema +- Shows file structure overview + +Example output: +``` +============================================================ +FILE ANALYSIS: example_data.xlsx +============================================================ +format: excel +shape: (150, 12) +suggested_schema: plate-reader +measurement_indicators: ['absorbance', 'well', 'plate'] + +============================================================ +✅ RECOMMENDED SCHEMA: plate-reader +============================================================ +``` + +## Step 2: List Available Schemas + +Review available schemas if auto-detection needs override: + +```bash +# From within allotropy repository +python scripts/list_schemas.py [optional_filter] + +# Use --verbose for detailed output with schema paths +python scripts/list_schemas.py --verbose +``` + +The script dynamically scans your local allotropy repository for all available schemas and shows: +- Schema technique names (plate-reader, pcr, etc.) +- All available schema versions and paths +- Description and use cases +- Example parsers that use each schema + +## Step 3: Generate Parser Code + +Use the `create_parser.py` script to generate the complete parser: + +```bash +python scripts/create_parser.py --example +``` + +### Parser Structure Created + +``` +src/allotropy/parsers/{parser_name}/ +├── __init__.py # Exports parser class +├── {parser_name}_parser.py # VendorParser subclass +├── {parser_name}_reader.py # File format parser +├── {parser_name}_structure.py # Dataclasses and factories +├── constants.py # Constants (if needed) +└── README.md # Documentation + +tests/parsers/{parser_name}/ +├── __init__.py +├── test_{parser_name}_parser.py # Test file +└── testdata/ + └── example.xlsx # Example test file +``` + +## Code Generation Approach + +### Reader Generation + +Based on file format detection: + +**Excel files**: +- Use `read_excel` with calamine engine +- Detect header/data sections using markers +- Handle well plate layouts if present +- Extract metadata and measurements + +**Text files**: +- Detect delimiter (tab, comma) +- Identify section markers (`[Section]` patterns) +- Parse using `SectionLinesReader` or `read_csv` + +### Structure Generation + +Create dataclasses for: +- `Header` - Metadata from file header +- `Measurement` - Individual measurement data +- Helper functions: + - `create_metadata()` - Build Metadata object + - `create_measurement_groups()` - Build MeasurementGroup list + - `create_calculated_data()` - Build calculated data (if applicable) + +### Parser Generation + +Generate `VendorParser` subclass with: +- `DISPLAY_NAME` - User-friendly instrument name +- `RELEASE_STATE` - Start with `ReleaseState.WORKING_DRAFT` +- `SUPPORTED_EXTENSIONS` - File extensions (from analysis) +- `SCHEMA_MAPPER` - Reference to schema mapper +- `create_data()` - Orchestrate reader + structure → Data + +## Step 4: Schema Mapping + +The schema mapper defines the intermediate `Data` structure and maps it to Allotrope models. + +### If Schema Mapper Exists + +Reuse existing mapper: +```python +from allotropy.allotrope.schema_mappers.adm.{technique}.{org}.{year}.{month}.{technique} import ( + Data, + Mapper, +) +``` + +Conform your `create_data()` to return the expected `Data` structure. + +### If Schema Mapper Needs Creation + +This is rare - most techniques have existing mappers. If needed: + +1. Define Data structure matching schema requirements +2. Implement `Mapper.map_model()` method +3. Handle quantity conversions and units + +## Step 5: Testing + +Generate test file: + +```python +# tests/parsers/{parser_name}/test_{parser_name}_parser.py + +def test_to_allotrope_{parser_name}() -> None: + test_file = "testdata/example.xlsx" + expected_file = "testdata/example.json" + run_allotropy(test_file, expected_file) +``` + +Run tests: +```bash +hatch run test:pytest tests/parsers/{parser_name}/ +``` + +## Step 6: Register Parser + +Add to `src/allotropy/parser_factory.py`: + +1. Import your parser: +```python +from allotropy.parsers.{parser_name}.{parser_name}_parser import {ParserName}Parser +``` + +2. Add to `Vendor` enum: +```python +class Vendor(Enum): + YOUR_INSTRUMENT = "YOUR_INSTRUMENT" +``` + +3. Add to `_VENDOR_TO_PARSER` mapping: +```python +_VENDOR_TO_PARSER: dict[Vendor, type[VendorParser]] = { + Vendor.YOUR_INSTRUMENT: YourInstrumentParser, + # ... existing parsers +} +``` + +## Implementation Checklist + +- [ ] Analyze example input file with `analyze_file.py` +- [ ] Confirm or select schema with `list_schemas.py` +- [ ] Generate parser using `create_parser.py` +- [ ] Review and adjust generated code +- [ ] Add example test data to `testdata/` +- [ ] Run tests and validate output +- [ ] Register parser in `parser_factory.py` +- [ ] Update README with parser info +- [ ] Update `RELEASE_STATE` when stable + +## Key Design Principles + +1. **Follow existing patterns** - Look at similar parsers for guidance +2. **Reuse utilities** - Use `read_excel`, `SeriesData`, `quantity_or_none`, etc. +3. **Type safety** - Use proper quantity types for all measurements +4. **Error handling** - Capture errors in Error objects, don't fail silently +5. **Validation** - Test against real files and validate ASM output + +## Common Measurement Types → Schema Mapping + +- **Absorbance, fluorescence, luminescence in wells** → `plate-reader` +- **CT/Cq values, amplification curves** → `pcr` (qpcr or dpcr) +- **pH, osmolality, particle size, pO2/pCO2** → `solution-analyzer` +- **Cell density, viability, cell counts** → `cell-counting` +- **Wavelength scans, UV-Vis spectra** → `spectrophotometry` +- **DNA/RNA/protein bands, lanes** → `electrophoresis` +- **Retention time, chromatogram peaks** → `liquid-chromatography` +- **Binding kinetics, SPR responses** → `binding-affinity` +- **Flow cytometry markers, populations** → `flow-cytometry` + +## Troubleshooting + +**Schema detection fails**: Manually specify schema after reviewing `list_schemas.py` output + +**File format unclear**: Look at similar parsers in the repository + +**Mapping errors**: Check schema mapper Data structure requirements + +**Tests fail**: Validate ASM output structure matches expected schema \ No newline at end of file