Skills › Knowledge Work Plugins › instrument-data-to-allotrope › flatten_asm.py
flatten_asm.py
254 lines · bio-research/skills/instrument-data-to-allotrope/scripts/flatten_asm.py at 2574e29
1#!/usr/bin/env python3
2"""
3Flatten ASM JSON to 2D CSV
5Converts hierarchical Allotrope Simple Model (ASM) JSON to flat tabular format
6suitable for LIMS import, spreadsheet analysis, or database loading.
8Usage:
9 python flatten_asm.py <input_asm.json> [--output OUTPUT.csv]
10"""
12import json
13import sys
14import re
15from pathlib import Path
16from typing import Dict, Any, List, Optional
17from datetime import datetime
19try:
20 import pandas as pd
22 PANDAS_AVAILABLE = True
23except ImportError:
24 PANDAS_AVAILABLE = False
27def detect_technique(asm: Dict[str, Any]) -> str:
28 """Detect the ASM technique type from document structure."""
29 for key in asm.keys():
30 if key.endswith("-aggregate-document"):
31 return key.replace("-aggregate-document", "")
32 return "generic"
35def flatten_value(value: Any, prefix: str = "") -> Dict[str, Any]:
36 """
37 Flatten a single ASM value, handling value datum patterns.
39 Returns dict of {column_name: value}
40 """
41 result = {}
43 if isinstance(value, dict):
44 if "value" in value:
45 # Value datum pattern
46 result[prefix] = value["value"]
47 if "unit" in value:
48 result[f"{prefix}_unit"] = value["unit"]
49 else:
50 # Nested dict - recurse
51 for k, v in value.items():
52 clean_key = k.replace("-", "_")
53 nested_prefix = f"{prefix}_{clean_key}" if prefix else clean_key
54 result.update(flatten_value(v, nested_prefix))
55 elif isinstance(value, list):
56 # Array - could be data cube or list of items
57 if len(value) > 0 and isinstance(value[0], dict):
58 # List of objects - this shouldn't happen at leaf level
59 result[prefix] = json.dumps(value)
60 else:
61 # Simple array - store as JSON string
62 result[prefix] = json.dumps(value)
63 else:
64 # Scalar value
65 result[prefix] = value
67 return result
70def extract_device_info(asm: Dict[str, Any], technique: str) -> Dict[str, Any]:
71 """Extract device/instrument information from ASM."""
72 agg_key = f"{technique}-aggregate-document"
73 agg_doc = asm.get(agg_key, {})
75 device = agg_doc.get("device-system-document", {})
77 return {
78 "instrument_serial_number": device.get("device-identifier"),
79 "instrument_model": device.get("model-number"),
80 "instrument_manufacturer": device.get("product-manufacturer"),
81 "software_name": device.get("software-name"),
82 "software_version": device.get("software-version"),
83 }
86def flatten_asm(asm: Dict[str, Any]) -> List[Dict[str, Any]]:
87 """
88 Flatten ASM JSON to list of row dictionaries.
90 Each measurement becomes one row with metadata repeated.
91 """
92 technique = detect_technique(asm)
93 rows = []
95 # Extract device info (shared across all rows)
96 device_info = extract_device_info(asm, technique)
97 device_info = {k: v for k, v in device_info.items() if v is not None}
99 # Navigate to measurements
100 agg_key = f"{technique}-aggregate-document"
101 agg_doc = asm.get(agg_key, {})
103 doc_key = f"{technique}-document"
104 technique_docs = agg_doc.get(doc_key, [])
106 for doc in technique_docs:
107 # Get measurement aggregate
108 meas_agg = doc.get("measurement-aggregate-document", {})
110 # Extract common measurement metadata
111 common_meta = {}
112 for key, value in meas_agg.items():
113 if key == "measurement-document":
114 continue
115 clean_key = key.replace("-", "_")
116 if isinstance(value, (str, int, float, bool)):
117 common_meta[clean_key] = value
118 elif isinstance(value, dict) and "value" in value:
119 common_meta[clean_key] = value["value"]
120 if "unit" in value:
121 common_meta[f"{clean_key}_unit"] = value["unit"]
123 # Extract each measurement as a row
124 measurements = meas_agg.get("measurement-document", [])
125 for meas in measurements:
126 row = {**device_info, **common_meta}
128 for key, value in meas.items():
129 clean_key = key.replace("-", "_")
130 flattened = flatten_value(value, clean_key)
131 row.update(flattened)
133 rows.append(row)
135 return rows
138def flatten_asm_to_csv(asm: Dict[str, Any], output_path: str) -> None:
139 """
140 Flatten ASM and write to CSV file.
142 Args:
143 asm: Parsed ASM JSON dictionary
144 output_path: Path for output CSV
145 """
146 if not PANDAS_AVAILABLE:
147 raise ImportError(
148 "pandas is required for CSV output. Install with: pip install pandas"
149 )
151 rows = flatten_asm(asm)
153 if not rows:
154 print("Warning: No measurements found to flatten")
155 # Create empty CSV with header
156 with open(output_path, "w") as f:
157 f.write("# No measurements found in ASM\n")
158 return
160 df = pd.DataFrame(rows)
162 # Reorder columns for readability
163 priority_cols = [
164 "sample_identifier",
165 "sample_id",
166 "well_location",
167 "well_position",
168 "measurement_time",
169 "measurement_datetime",
170 "analyst",
171 ]
173 ordered_cols = []
174 for col in priority_cols:
175 if col in df.columns:
176 ordered_cols.append(col)
178 remaining = [c for c in df.columns if c not in ordered_cols]
179 df = df[ordered_cols + remaining]
181 df.to_csv(output_path, index=False)
184def flatten_asm_to_dict(asm: Dict[str, Any]) -> Dict[str, Any]:
185 """
186 Flatten ASM and return as dictionary with rows and columns.
188 Useful for non-CSV outputs or further processing.
189 """
190 rows = flatten_asm(asm)
192 if not rows:
193 return {"columns": [], "rows": []}
195 columns = list(rows[0].keys())
196 return {
197 "columns": columns,
198 "rows": [[row.get(col) for col in columns] for row in rows],
199 }
202def main():
203 """Main entry point."""
204 import argparse
206 parser = argparse.ArgumentParser(description="Flatten ASM JSON to CSV")
207 parser.add_argument("input", help="Input ASM JSON file")
208 parser.add_argument(
209 "--output", "-o", help="Output CSV path (default: input_flat.csv)"
210 )
211 parser.add_argument(
212 "--format",
213 choices=["csv", "json"],
214 default="csv",
215 help="Output format (default: csv)",
216 )
218 args = parser.parse_args()
220 input_path = Path(args.input)
221 if not input_path.exists():
222 print(f"Error: File not found: {args.input}")
223 sys.exit(1)
225 # Load ASM
226 with open(input_path) as f:
227 asm = json.load(f)
229 # Determine output path
230 if args.output:
231 output_path = args.output
232 else:
233 suffix = ".flat.csv" if args.format == "csv" else ".flat.json"
234 output_path = str(input_path.with_suffix("")) + suffix
236 # Flatten and write
237 if args.format == "csv":
238 flatten_asm_to_csv(asm, output_path)
239 else:
240 result = flatten_asm_to_dict(asm)
241 with open(output_path, "w") as f:
242 json.dump(result, f, indent=2)
244 print(f"Flattened output written to: {output_path}")
246 # Report stats
247 rows = flatten_asm(asm)
248 print(f" Rows: {len(rows)}")
249 if rows:
250 print(f" Columns: {len(rows[0])}")
253if __name__ == "__main__":
254 main()