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
- Number
- 1.5
- Position
- 5 of 10
- Type
- Python
- Size
- 17 KB
- Lines
- 543
scripts/convert_to_asm.py
Python·543 lines·17 KB
16
import
importlib.metadata
17from pathlib import Path
18from typing import Optional, Tuple, Dict, Any
19from datetime import datetime
20
21
22# Lazy imports to avoid errors if not installed
23def get_allotropy():
24 try:
25 from allotropy.parser_factory import Vendor
26 from allotropy.to_allotrope import allotrope_from_file, allotrope_from_io
27
28 return Vendor, allotrope_from_file, allotrope_from_io
29 except ImportError:
30 return None, None, None
31
32
33def get_pandas():
34 try:
35 import pandas as pd
36
37 return pd
38 except ImportError:
39 return None
40
41
42# Detection patterns for instrument identification
43DETECTION_PATTERNS = {
44 "BECKMAN_VI_CELL_BLU": {
45 "columns": [
46 "Sample ID",
47 "Viable cells",
48 "Viability",
49 "Total cells",
50 "Average diameter",
51 ],
52 "keywords": ["Vi-CELL BLU", "Beckman Coulter"],
53 "file_patterns": [r".*\.csv$"],
54 "confidence_boost": 20,
55 },
56 "BECKMAN_VI_CELL_XR": {
57 "columns": ["Sample", "Total cells/ml", "Viable cells/ml", "Viability (%)"],
58 "keywords": ["Vi-CELL XR", "Cell Viability Analyzer"],
59 "file_patterns": [r".*\.(txt|xls|xlsx)$"],
60 "confidence_boost": 20,
61 },
62 "THERMO_FISHER_NANODROP_EIGHT": {
63 "columns": ["Sample Name", "Nucleic Acid Conc.", "A260", "A280", "260/280"],
64 "keywords": ["NanoDrop Eight", "NanoDrop 8"],
65 "file_patterns": [r".*\.(tsv|txt)$"],
66 "confidence_boost": 15,
67 },
68 "THERMO_FISHER_NANODROP_ONE": {
69 "columns": ["Sample Name", "Nucleic Acid(ng/uL)", "A260", "A280"],
70 "keywords": ["NanoDrop One", "NanoDrop"],
71 "file_patterns": [r".*\.(csv|xlsx)$"],
72 "confidence_boost": 15,
73 },
74 "MOLDEV_SOFTMAX_PRO": {
75 "columns": ["Well", "Sample", "Values", "Mean", "SD"],
76 "keywords": ["SoftMax Pro", "SpectraMax", "Molecular Devices"],
77 "file_patterns": [r".*\.txt$"],
78 "confidence_boost": 15,
79 },
80 "BMG_MARS": {
81 "columns": ["Well", "Content", "Conc.", "Mean", "SD", "CV"],
82 "keywords": ["BMG LABTECH", "MARS", "CLARIOstar", "PHERAstar"],
83 "file_patterns": [r".*\.(csv|txt)$"],
84 "confidence_boost": 15,
85 },
86 "AGILENT_GEN5": {
87 "columns": ["Well", "Read", "Time", "Temperature"],
88 "keywords": ["Gen5", "BioTek", "Synergy"],
89 "file_patterns": [r".*\.xlsx$"],
90 "confidence_boost": 15,
91 },
92 "APPBIO_QUANTSTUDIO": {
93 "columns": ["Well", "Sample Name", "Target Name", "CT", "Ct Mean"],
94 "keywords": ["QuantStudio", "Applied Biosystems", "qPCR"],
95 "file_patterns": [r".*\.xlsx$"],
96 "confidence_boost": 15,
97 },
98}
99
100
101def detect_instrument_type(
102 filepath: str, file_content: Optional[str] = None
103) -> Tuple[str, float]:
104 """
105 Auto-detect instrument type from file contents.
106
107 Returns:
108 Tuple of (vendor_name, confidence_score)
109 confidence_score is 0-100
110 """
111 path = Path(filepath)
112 filename = path.name.lower()
113 extension = path.suffix.lower()
114
115 # Read file content if not provided
116 if file_content is None:
117 try:
118 if extension in [".xlsx", ".xls"]:
119 pd = get_pandas()
120 if pd:
121 df = pd.read_excel(filepath, nrows=50)
122 file_content = df.to_string() + "\n" + "\n".join(df.columns)
123 else:
124 file_content = ""
125 else:
126 with open(filepath, "r", encoding="utf-8", errors="ignore") as f:
127 file_content = f.read(10000) # First 10KB
128 except Exception as e:
129 print(f"Warning: Could not read file for detection: {e}")
130 file_content = ""
131
132 content_lower = file_content.lower()
133 scores = {}
134
135 for vendor, patterns in DETECTION_PATTERNS.items():
136 score = 0
137
138 # Check file extension patterns
139 for pattern in patterns.get("file_patterns", []):
140 if re.match(pattern, filename, re.IGNORECASE):
141 score += 10
142 break
143
144 # Check column headers
145 columns_found = 0
146 for col in patterns.get("columns", []):
147 if col.lower() in content_lower:
148 columns_found += 1
149 if columns_found > 0:
150 score += min(50, columns_found * 15)
151
152 # Check keywords
153 for keyword in patterns.get("keywords", []):
154 if keyword.lower() in content_lower:
155 score += patterns.get("confidence_boost", 10)
156
157 scores[vendor] = min(100, score)
158
159 # Return best match
160 if scores:
161 best = max(scores.items(), key=lambda x: x[1])
162 return best[0], best[1]
163
164 return "UNKNOWN", 0
165
166
167def convert_with_allotropy(filepath: str, vendor_name: str) -> Optional[Dict[str, Any]]:
168 """
169 Convert file using allotropy library.
170
171 Returns:
172 ASM dictionary or None if conversion fails
173 """
174 Vendor, allotrope_from_file, _ = get_allotropy()
175
176 if Vendor is None:
177 print(
178 "Warning: allotropy not installed. Run: pip install allotropy --break-system-packages"
179 )
180 return None
181
182 try:
183 vendor = getattr(Vendor, vendor_name, None)
184 if vendor is None:
185 print(f"Warning: Vendor {vendor_name} not found in allotropy")
186 return None
187
188 asm = allotrope_from_file(filepath, vendor)
189 return asm
190 except Exception as e:
191 print(f"Allotropy conversion failed: {e}")
192 return None
193
194
195def get_deterministic_timestamp(filepath: str) -> str:
196 """
197 Get deterministic timestamp for file.
198 Uses file modification time for reproducibility.
199
200 Returns:
201 ISO format timestamp string
202 """
203 try:
204 path = Path(filepath)
205 mtime = path.stat().st_mtime
206 return datetime.fromtimestamp(mtime).isoformat()
207 except Exception:
208 return "TIMESTAMP_NOT_AVAILABLE"
209
210
211def calculate_file_hash(filepath: str) -> str:
212 """Calculate SHA256 hash of file for provenance tracking."""
213 try:
214 with open(filepath, "rb") as f:
215 return hashlib.sha256(f.read()).hexdigest()
216 except Exception:
217 return "HASH_NOT_AVAILABLE"
218
219
220def get_library_version(library: str) -> str:
221 """Get version of installed library."""
222 try:
223 return importlib.metadata.version(library)
224 except Exception:
225 return "VERSION_NOT_AVAILABLE"
226
227
228def add_provenance_metadata(
229 asm: Dict[str, Any],
230 filepath: str,
231 vendor: str,
232 confidence: float,
233 used_fallback: bool,
234 warnings: list = None,
235) -> Dict[str, Any]:
236 """
237 Add provenance metadata to ASM for reproducibility and audit trail.
238
239 This metadata enables:
240 - Reproducing conversions months later
241 - Determining which version generated data
242 - Auditing data lineage for regulatory compliance
243 """
244 pd = get_pandas()
245
246 asm["$conversion_metadata"] = {
247 "skill_version": "1.0.0",
248 "allotropy_version": get_library_version("allotropy"),
249 "pandas_version": pd.__version__ if pd else "NOT_INSTALLED",
250 "conversion_timestamp_utc": datetime.utcnow().isoformat(),
251 "input_file_sha256": calculate_file_hash(filepath),
252 "input_file_size_bytes": Path(filepath).stat().st_size,
253 "input_file_name": Path(filepath).name,
254 "parser_used": "fallback" if used_fallback else "allotropy",
255 "detection_confidence": confidence,
256 "vendor_detected": vendor,
257 "warnings": warnings or [],
258 }
259
260 return asm
261
262
263def flexible_parse(filepath: str, detected_type: str) -> Optional[Dict[str, Any]]:
264 """
265 Flexible fallback parser when allotropy fails.
266 Creates ASM-like structure from parsed data.
267
268 **WARNING:** This parser creates simplified ASM that:
269 - Does NOT distinguish raw vs. calculated data
270 - LACKS instrument control parameters (temperature, wavelengths, etc.)
271 - MAY NOT be compatible with regulatory requirements (GxP)
272 - Should be used for exploratory analysis only, not production LIMS import
273 """
274 pd = get_pandas()
275 if pd is None:
276 print("Warning: pandas not installed for flexible parsing")
277 return None
278
279 path = Path(filepath)
280 extension = path.suffix.lower()
281
282 try:
283 # Read file based on extension
284 if extension in [".xlsx", ".xls"]:
285 df = pd.read_excel(filepath, engine="openpyxl")
286 elif extension == ".tsv":
287 df = pd.read_csv(filepath, sep="\t")
288 elif extension == ".csv":
289 df = pd.read_csv(filepath)
290 else:
291 df = pd.read_csv(filepath, sep=None, engine="python")
292
293 # Build ASM-like structure
294 asm = build_flexible_asm(df, detected_type, filepath)
295 return asm
296
297 except Exception as e:
298 print(f"Flexible parsing failed: {e}")
299 return None
300
301
302def build_flexible_asm(df, detected_type: str, filepath: str) -> Dict[str, Any]:
303 """
304 Build ASM-like JSON structure from parsed DataFrame.
305 """
306 timestamp = get_deterministic_timestamp(filepath)
307
308 # Determine technique from detected type
309 technique = "generic"
310 if "VI_CELL" in detected_type:
311 technique = "cell-counting"
312 elif "NANODROP" in detected_type:
313 technique = "spectrophotometry"
314 elif detected_type in ["MOLDEV_SOFTMAX_PRO", "BMG_MARS", "AGILENT_GEN5"]:
315 technique = "plate-reader"
316 elif "QUANTSTUDIO" in detected_type:
317 technique = "pcr"
318
319 # Build base structure
320 asm = {
321 "$asm.manifest": {
322 "vocabulary": ["http://purl.allotrope.org/voc/afo/REC/2023/09/"],
323 "contexts": [
324 "http://purl.allotrope.org/json-ld/afo-context-REC-2023-09.jsonld"
325 ],
326 },
327 f"{technique}-aggregate-document": {
328 "device-system-document": {
329 "device-identifier": "FLEXIBLE_PARSER",
330 "product-manufacturer": (
331 detected_type.split("_")[0] if "_" in detected_type else "Unknown"
332 ),
333 },
334 f"{technique}-document": [
335 {
336 "measurement-aggregate-document": {
337 "measurement-time": timestamp,
338 "measurement-document": [],
339 }
340 }
341 ],
342 },
343 }
344
345 # Add measurements from DataFrame
346 measurements = asm[f"{technique}-aggregate-document"][f"{technique}-document"][0][
347 "measurement-aggregate-document"
348 ]["measurement-document"]
349
350 for _, row in df.iterrows():
351 meas = {}
352 for col in df.columns:
353 value = row[col]
354 if pd.notna(value):
355 # Clean column name
356 clean_col = str(col).lower().replace(" ", "-").replace("_", "-")
357 clean_col = re.sub(r"[^a-z0-9-]", "", clean_col)
358
359 # Handle numeric values
360 if isinstance(value, (int, float)):
361 meas[clean_col] = {"value": value, "unit": "(unitless)"}
362 else:
363 meas[clean_col] = str(value)
364
365 if meas:
366 measurements.append(meas)
367
368 return asm
369
370
371def main():
372 """Main entry point."""
373 import argparse
374
375 parser = argparse.ArgumentParser(
376 description="Convert instrument data to ASM format"
377 )
378 parser.add_argument("input", help="Input file path")
379 parser.add_argument(
380 "--vendor", help="Vendor enum name (auto-detected if not provided)"
381 )
382 parser.add_argument(
383 "--output", "-o", help="Output file path (default: input_asm.json)"
384 )
385 parser.add_argument(
386 "--flatten", action="store_true", help="Also generate flattened CSV"
387 )
388 parser.add_argument(
389 "--allow-fallback",
390 action="store_true",
391 help="Allow fallback to simplified parser (reduced metadata)",
392 )
393 parser.add_argument(
394 "--skip-validation",
395 action="store_true",
396 help="Skip automatic validation (not recommended)",
397 )
398 parser.add_argument(
399 "--force",
400 action="store_true",
401 help="Force conversion even with low confidence detection",
402 )
403
404 args = parser.parse_args()
405
406 input_path = Path(args.input)
407 if not input_path.exists():
408 print(f"Error: File not found: {args.input}")
409 sys.exit(1)
410
411 warnings = []
412
413 # Detect or use provided vendor
414 if args.vendor:
415 vendor = args.vendor.upper()
416 confidence = 100
417 print(f"Using specified vendor: {vendor}")
418 else:
419 vendor, confidence = detect_instrument_type(str(input_path))
420 print(f"Detected instrument: {vendor} (confidence: {confidence}%)")
421
422 # Enforce confidence thresholds
423 if confidence < 30:
424 print(
425 f"ERROR: Detection confidence too low ({confidence}%). Cannot proceed."
426 )
427 print("Please specify --vendor explicitly.")
428 sys.exit(1)
429 elif confidence < 60:
430 warning_msg = f"WARNING: Low confidence detection ({confidence}%)."
431 print(warning_msg)
432 warnings.append(warning_msg)
433 if not args.force:
434 print("Use --force to proceed anyway (not recommended).")
435 sys.exit(1)
436
437 # Try allotropy first
438 asm = convert_with_allotropy(str(input_path), vendor)
439 used_fallback = False
440
441 # Fall back to flexible parser
442 if asm is None:
443 print("\n" + "=" * 60)
444 print("ALLOTROPY PARSING FAILED - USING REDUCED METADATA PARSER")
445 print("=" * 60)
446 print("Output will lack:")
447 print(" - Calculated data traceability")
448 print(" - Device control settings")
449 print(" - Data processing metadata")
450 print("\nNot suitable for:")
451 print(" - Regulatory submissions")
452 print(" - LIMS import with validation")
453 print("=" * 60 + "\n")
454
455 if not args.allow_fallback:
456 print(
457 "ERROR: Allotropy parsing failed. Use --allow-fallback to continue with"
458 )
459 print("simplified parser, but note that output will lack required metadata")
460 print("for GxP compliance.")
461 sys.exit(1)
462
463 asm = flexible_parse(str(input_path), vendor)
464 used_fallback = True
465 warnings.append("Used fallback parser - reduced metadata")
466
467 if asm is None:
468 print("Error: Could not convert file")
469 sys.exit(1)
470
471 # Add provenance metadata
472 asm = add_provenance_metadata(
473 asm, str(input_path), vendor, confidence, used_fallback, warnings
474 )
475
476 # Determine output path
477 if args.output:
478 output_path = Path(args.output)
479 else:
480 output_path = input_path.with_suffix(".asm.json")
481
482 # Write to temporary file first
483 temp_path = output_path.with_suffix(".tmp")
484
485 try:
486 with open(temp_path, "w") as f:
487 json.dump(asm, f, indent=2, default=str)
488
489 # Validate unless skipped
490 if not args.skip_validation:
491 print("Running validation...")
492 try:
493 from validate_asm import validate_asm
494
495 result = validate_asm(str(temp_path))
496
497 if not result.is_valid():
498 print("\n" + "=" * 60)
499 print("VALIDATION FAILED")
500 print("=" * 60)
501 for error in result.errors:
502 print(f"ERROR: {error}")
503 for warning in result.warnings:
504 print(f"WARNING: {warning}")
505 print("=" * 60)
506
507 # Remove temp file
508 temp_path.unlink()
509 print("\nValidation failed. Output file not created.")
510 sys.exit(1)
511 else:
512 if result.warnings:
513 print("\nValidation warnings:")
514 for warning in result.warnings:
515 print(f" WARNING: {warning}")
516 print("Validation passed.")
517 except ImportError:
518 print(
519 "Warning: validate_asm.py not found. Skipping validation. "
520 "Consider adding validation script."
521 )
522
523 # Move temp file to final location
524 temp_path.replace(output_path)
525 print(f"ASM output written to: {output_path}")
526
527 except Exception as e:
528 # Clean up temp file on error
529 if temp_path.exists():
530 temp_path.unlink()
531 raise e
532
533 # Optionally flatten
534 if args.flatten:
535 from flatten_asm import flatten_asm_to_csv
536
537 flat_path = input_path.with_suffix(".flat.csv")
538 flatten_asm_to_csv(asm, str(flat_path))
539 print(f"Flattened CSV written to: {flat_path}")
540
541
542if __name__ == "__main__":
543 main()