Setting the file. One moment. 22
Validate Data
Tech Debt
62
Recruiting Pipeline
71
Vendor Check
125
Zoom Meeting SDK Web
88
Vendor Review
181
Create An Asset
Video Sdk/web
Script Export Parser
scripts/flatten_asm.py
Python·254 lines·7 KB
Path
16from typing import Dict, Any, List, Optional
17from datetime import datetime
18
19try:
20 import pandas as pd
21
22 PANDAS_AVAILABLE = True
23except ImportError:
24 PANDAS_AVAILABLE = False
25
26
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"
33
34
35def flatten_value(value: Any, prefix: str = "") -> Dict[str, Any]:
36 """
37 Flatten a single ASM value, handling value datum patterns.
38
39 Returns dict of {column_name: value}
40 """
41 result = {}
42
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
66
67 return result
68
69
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, {})
74
75 device = agg_doc.get("device-system-document", {})
76
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 }
84
85
86def flatten_asm(asm: Dict[str, Any]) -> List[Dict[str, Any]]:
87 """
88 Flatten ASM JSON to list of row dictionaries.
89
90 Each measurement becomes one row with metadata repeated.
91 """
92 technique = detect_technique(asm)
93 rows = []
94
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}
98
99 # Navigate to measurements
100 agg_key = f"{technique}-aggregate-document"
101 agg_doc = asm.get(agg_key, {})
102
103 doc_key = f"{technique}-document"
104 technique_docs = agg_doc.get(doc_key, [])
105
106 for doc in technique_docs:
107 # Get measurement aggregate
108 meas_agg = doc.get("measurement-aggregate-document", {})
109
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"]
122
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}
127
128 for key, value in meas.items():
129 clean_key = key.replace("-", "_")
130 flattened = flatten_value(value, clean_key)
131 row.update(flattened)
132
133 rows.append(row)
134
135 return rows
136
137
138def flatten_asm_to_csv(asm: Dict[str, Any], output_path: str) -> None:
139 """
140 Flatten ASM and write to CSV file.
141
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 )
150
151 rows = flatten_asm(asm)
152
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
159
160 df = pd.DataFrame(rows)
161
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 ]
172
173 ordered_cols = []
174 for col in priority_cols:
175 if col in df.columns:
176 ordered_cols.append(col)
177
178 remaining = [c for c in df.columns if c not in ordered_cols]
179 df = df[ordered_cols + remaining]
180
181 df.to_csv(output_path, index=False)
182
183
184def flatten_asm_to_dict(asm: Dict[str, Any]) -> Dict[str, Any]:
185 """
186 Flatten ASM and return as dictionary with rows and columns.
187
188 Useful for non-CSV outputs or further processing.
189 """
190 rows = flatten_asm(asm)
191
192 if not rows:
193 return {"columns": [], "rows": []}
194
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 }
200
201
202def main():
203 """Main entry point."""
204 import argparse
205
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 )
217
218 args = parser.parse_args()
219
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)
224
225 # Load ASM
226 with open(input_path) as f:
227 asm = json.load(f)
228
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
235
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)
243
244 print(f"Flattened output written to: {output_path}")
245
246 # Report stats
247 rows = flatten_asm(asm)
248 print(f" Rows: {len(rows)}")
249 if rows:
250 print(f" Columns: {len(rows[0])}")
251
252
253if __name__ == "__main__":
254 main()