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
509
def count_measurements
— line 509
This file
- Number
- 1.8
- Position
- 8 of 10
- Type
- Python
- Size
- 36 KB
- Lines
- 1,102
scripts/validate_asm.py
Python·1,102 lines·36 KB
17 Source: https://gitlab.com/allotrope-public/asm/-/tree/main/json-schemas/adm
18
19Note: Unknown techniques/units generate WARNINGS (not errors) to allow for new
20additions to the Allotrope specification. This prevents blocking valid data
21when the Allotrope foundation adds new techniques or units.
22
23Usage:
24 python validate_asm.py output.json
25 python validate_asm.py output.json --reference reference.json
26 python validate_asm.py output.json --strict
27"""
28
29import json
30import re
31import sys
32import argparse
33from typing import Dict, List, Tuple, Any, Optional
34
35# Validation metadata
36ASM_SPEC_VERSION = "2024-12"
37VALIDATION_RULES_DATE = "2026-01-07"
38SCHEMA_SOURCE = "https://gitlab.com/allotrope-public/asm"
39
40
41# All valid ASM techniques from https://gitlab.com/allotrope-public/asm/-/tree/main/json-schemas/adm
42VALID_TECHNIQUES = [
43 "absorbance",
44 "automated-reactors",
45 "balance",
46 "bga",
47 "binding-affinity",
48 "bulk-density",
49 "cell-counting",
50 "cell-culture-analyzer",
51 "chromatography",
52 "code-reader",
53 "conductance",
54 "conductivity",
55 "disintegration",
56 "dsc",
57 "dvs",
58 "electronic-lab-notebook",
59 "electronic-spectrometry",
60 "electrophoresis",
61 "flow-cytometry",
62 "fluorescence",
63 "foam-height",
64 "foam-qualification",
65 "fplc",
66 "ftir",
67 "gas-chromatography",
68 "gc-ms",
69 "gloss",
70 "hot-tack",
71 "impedance",
72 "lc-ms",
73 "light-obscuration",
74 "liquid-chromatography",
75 "liquid-handler", # Added for liquid handler support
76 "loss-on-drying",
77 "luminescence",
78 "mass-spectrometry",
79 "metabolite-analyzer",
80 "multi-analyte-profiling",
81 "nephelometry",
82 "nmr",
83 "optical-imaging",
84 "optical-microscopy",
85 "osmolality",
86 "oven-kf",
87 "pcr",
88 "ph",
89 "plate-reader",
90 "pressure-monitoring",
91 "psd",
92 "pumping",
93 "raman",
94 "rheometry",
95 "sem",
96 "solution-analyzer",
97 "specific-rotation",
98 "spectrophotometry",
99 "stirring",
100 "surface-area-analysis",
101 "tablet-hardness",
102 "temperature-monitoring",
103 "tensile-test",
104 "thermogravimetric-analysis",
105 "titration",
106 "ultraviolet-absorbance",
107 "x-ray-powder-diffraction",
108]
109
110# Instrument keywords that indicate specific techniques
111TECHNIQUE_INDICATORS = {
112 "multi-analyte-profiling": [
113 "bead",
114 "luminex",
115 "bio-plex",
116 "bioplex",
117 "multiplex",
118 "plex",
119 "msd",
120 "region",
121 ],
122 "electrophoresis": [
123 "tapestation",
124 "bioanalyzer",
125 "labchip",
126 "fragment",
127 "din",
128 "rin",
129 "gel",
130 "capillary",
131 ],
132 "spectrophotometry": ["nanodrop", "lunatic", "a260", "a280", "wavelength"],
133 "cell-counting": [
134 "viability",
135 "viable cell",
136 "cell count",
137 "vi-cell",
138 "vicell",
139 "nucleocounter",
140 "cell density",
141 ],
142 "pcr": [
143 "ct",
144 "quantstudio",
145 "cfx",
146 "amplification",
147 "melt curve",
148 "qpcr",
149 "cycle threshold",
150 ],
151 "plate-reader": [
152 "microplate",
153 "96-well",
154 "384-well",
155 "plate reader",
156 "envision",
157 "spectramax",
158 ],
159 "liquid-chromatography": [
160 "hplc",
161 "uplc",
162 "retention time",
163 "chromatogram",
164 "empower",
165 "chromeleon",
166 ],
167 "flow-cytometry": ["facs", "flow cytometry", "scatter", "gating", "cytometer"],
168 "mass-spectrometry": ["m/z", "mass spec", "ms/ms", "lcms", "maldi"],
169 "fluorescence": ["fluorescence", "excitation", "emission", "fluorimeter"],
170 "luminescence": ["luminescence", "bioluminescence", "chemiluminescence"],
171 "absorbance": ["absorbance", "optical density", "od600"],
172 "ph": ["ph meter", "ph measurement"],
173 "osmolality": ["osmolality", "osmometer"],
174 "conductivity": ["conductivity", "conductance"],
175 "balance": ["balance", "weight", "mass measurement"],
176 "nmr": ["nmr", "nuclear magnetic resonance"],
177 "ftir": ["ftir", "infrared", "ir spectrum"],
178 "raman": ["raman", "raman spectroscopy"],
179 "liquid-handler": [
180 "biomek",
181 "liquid handler",
182 "aspirate",
183 "dispense",
184 "transfer volume",
185 "liquid handling",
186 ],
187}
188
189# Fields that should typically be in calculated-data-document, not measurement-document
190SHOULD_BE_CALCULATED = [
191 "dna integrity number",
192 "rna integrity number",
193 "din",
194 "rin",
195 "viability",
196 "260/280",
197 "a260/a280",
198 "concentration", # When derived from standard curve
199 "percent of total",
200 "average size",
201 "molarity", # When calculated from concentration
202 "relative quantity",
203 "fold change",
204 "coefficient of variation",
205]
206
207# =============================================================================
208# NESTED DOCUMENT STRUCTURE DEFINITIONS
209# =============================================================================
210
211# Fields that MUST be inside 'sample document' (space or hyphen separated)
212SAMPLE_DOCUMENT_FIELDS = {
213 # Core sample identification
214 "sample identifier",
215 "sample-identifier",
216 "written name",
217 "written-name",
218 "batch identifier",
219 "batch-identifier",
220 "sample role type",
221 "sample-role-type",
222 "description",
223 # Location fields (should be in sample document for most techniques)
224 "location identifier",
225 "location-identifier",
226 "well location identifier",
227 "well-location-identifier",
228 "well plate identifier",
229 "well-plate-identifier",
230 # Liquid handler specific - source/destination pairs
231 "source location identifier",
232 "source-location-identifier",
233 "destination location identifier",
234 "destination-location-identifier",
235 "source well plate identifier",
236 "source-well-plate-identifier",
237 "destination well plate identifier",
238 "destination-well-plate-identifier",
239 "source well location identifier",
240 "source-well-location-identifier",
241 "destination well location identifier",
242 "destination-well-location-identifier",
243}
244
245# Fields that MUST be inside 'device control aggregate document' -> 'device control document'
246DEVICE_CONTROL_FIELDS = {
247 # General device control
248 "device type",
249 "device-type",
250 "detector wavelength setting",
251 "detector-wavelength-setting",
252 "compartment temperature",
253 "compartment-temperature",
254 "sample volume setting",
255 "sample-volume-setting",
256 "flow rate",
257 "flow-rate",
258 "exposure duration setting",
259 "exposure-duration-setting",
260 "detector gain setting",
261 "detector-gain-setting",
262 "illumination setting",
263 "illumination-setting",
264 # Liquid handler specific
265 "liquid handling technique",
266 "liquid-handling-technique",
267 "source liquid handling technique",
268 "source-liquid-handling-technique",
269 "destination liquid handling technique",
270 "destination-liquid-handling-technique",
271}
272
273# Fields that should be in 'custom information document' (vendor-specific)
274CUSTOM_INFO_FIELDS = {
275 # Liquid handler specific
276 "probe",
277 "pod",
278 "source labware name",
279 "source-labware-name",
280 "destination labware name",
281 "destination-labware-name",
282 "deck position",
283 "deck-position",
284}
285
286# Fields that commonly get incorrectly flattened (superset for general checking)
287COMMONLY_FLATTENED_FIELDS = {
288 # Sample-related (often incorrectly put directly on measurement)
289 "sample identifier",
290 "sample-identifier",
291 "sample barcode",
292 "sample-barcode",
293 "well index",
294 "well-index",
295 "location identifier",
296 "location-identifier",
297 # Device control related (often incorrectly put directly on measurement)
298 "probe identifier",
299 "probe-identifier",
300 "device identifier", # When it should be in device control doc, not measurement
301 "device-identifier",
302 "technique", # Should be "liquid handling technique" in device control
303 "transfer type", # Should be structured differently
304 "transfer-type",
305}
306
307# Standard ASM units
308VALID_UNITS = {
309 "fluorescence": ["RFU", "MFI", "(unitless)"],
310 "counts": ["#"],
311 "volume": ["μL", "mL", "L", "µL"],
312 "concentration": [
313 "ng/μL",
314 "ng/mL",
315 "pg/mL",
316 "mg/mL",
317 "μg/mL",
318 "M",
319 "mM",
320 "μM",
321 "nM",
322 ],
323 "temperature": ["degC"],
324 "unitless": ["(unitless)", "%"],
325 "molecular_weight": ["bp", "Da", "kDa"],
326 "time": ["s", "min", "h"],
327}
328
329# Standard sample role types
330VALID_SAMPLE_ROLES = [
331 "standard sample role",
332 "blank role",
333 "control sample role",
334 "unknown sample role",
335 "reference sample role",
336 "calibration sample role",
337]
338
339# Standard statistic datum roles
340VALID_STATISTIC_ROLES = [
341 "median role",
342 "arithmetic mean role",
343 "coefficient of variation role",
344 "standard deviation role",
345 "standard error role",
346 "trimmed arithmetic mean role",
347 "trimmed standard deviation role",
348 "minimum value role",
349 "maximum value role",
350]
351
352
353class ValidationResult:
354 """Container for validation results."""
355
356 def __init__(self):
357 self.errors: List[str] = []
358 self.warnings: List[str] = []
359 self.info: List[str] = []
360 self.metrics: Dict[str, Any] = {}
361
362 def add_error(self, msg: str):
363 self.errors.append(f"ERROR: {msg}")
364
365 def add_warning(self, msg: str):
366 self.warnings.append(f"WARNING: {msg}")
367
368 def add_info(self, msg: str):
369 self.info.append(f"INFO: {msg}")
370
371 def is_valid(self) -> bool:
372 return len(self.errors) == 0
373
374 def print_report(self):
375 print("\n" + "=" * 60)
376 print("ASM VALIDATION REPORT")
377 print("=" * 60)
378
379 # Print metrics
380 if self.metrics:
381 print("\nMetrics:")
382 for key, value in self.metrics.items():
383 print(f" {key}: {value}")
384
385 # Print info
386 if self.info:
387 print("\n" + "\n".join(self.info))
388
389 # Print warnings
390 if self.warnings:
391 print("\n" + "\n".join(self.warnings))
392
393 # Print errors
394 if self.errors:
395 print("\n" + "\n".join(self.errors))
396
397 # Summary
398 print("\n" + "-" * 60)
399 if self.is_valid():
400 if self.warnings:
401 print(f"PASSED with {len(self.warnings)} warning(s)")
402 else:
403 print("PASSED - No issues found")
404 else:
405 print(
406 f"FAILED - {len(self.errors)} error(s), {len(self.warnings)} warning(s)"
407 )
408 print("=" * 60 + "\n")
409
410
411def validate_manifest(asm: Dict, result: ValidationResult):
412 """Check for valid manifest."""
413 if "$asm.manifest" not in asm:
414 result.add_error("Missing $asm.manifest")
415 return
416
417 manifest = asm["$asm.manifest"]
418 if isinstance(manifest, str):
419 if "allotrope.org" in manifest:
420 result.add_info(f"Manifest: {manifest}")
421 else:
422 result.add_warning(f"Non-standard manifest URL: {manifest}")
423 elif isinstance(manifest, dict):
424 if "vocabulary" in manifest or "contexts" in manifest:
425 result.add_info("Manifest: Object format with vocabulary/contexts")
426 else:
427 result.add_warning("Manifest object missing vocabulary or contexts")
428
429
430def detect_technique(asm: Dict) -> Tuple[str, float]:
431 """Detect technique from ASM structure."""
432 # Check for technique in top-level keys
433 for key in asm.keys():
434 if key == "$asm.manifest":
435 continue
436 # Extract technique name from aggregate document key
437 # Handle both "liquid handler aggregate document" and "liquid-handler-aggregate-document"
438 key_normalized = key.lower().replace("-", " ")
439 if "aggregate document" in key_normalized:
440 technique = key_normalized.replace(" aggregate document", "").strip()
441 return technique, 100.0
442
443 return "unknown", 0.0
444
445
446def validate_technique(asm: Dict, result: ValidationResult, content_str: str):
447 """Validate technique selection."""
448 technique, confidence = detect_technique(asm)
449 result.metrics["technique"] = technique
450 result.metrics["technique_confidence"] = confidence
451
452 if technique == "unknown":
453 result.add_warning("Could not detect technique from ASM structure")
454 return
455
456 result.add_info(f"Detected technique: {technique}")
457
458 # Check if technique is in known list (soft validation)
459 technique_normalized = technique.replace(" ", "-")
460 if technique_normalized not in VALID_TECHNIQUES:
461 result.add_warning(
462 f"Unknown technique '{technique}' not in known list (as of {VALIDATION_RULES_DATE}). "
463 f"This may be a new Allotrope addition. Verify at: {SCHEMA_SOURCE}"
464 )
465
466 # Check if technique seems appropriate for content
467 content_lower = content_str.lower()
468 suggested_technique = None
469
470 for tech, keywords in TECHNIQUE_INDICATORS.items():
471 matches = sum(1 for kw in keywords if kw in content_lower)
472 if matches >= 2: # Multiple keyword matches
473 if tech != technique.replace(" ", "-"):
474 suggested_technique = tech
475 break
476
477 if suggested_technique:
478 result.add_warning(
479 f"Content suggests '{suggested_technique}' but ASM uses '{technique}' - "
480 "verify correct technique selection"
481 )
482
483
484def validate_naming_conventions(content_str: str, result: ValidationResult):
485 """Check for proper space-separated naming (not hyphens)."""
486 # Find all keys that look like ASM field names
487 # Hyphenated keys in ASM are typically wrong (should be space-separated)
488 hyphenated_keys = re.findall(r'"([a-z]+-[a-z]+-?[a-z]*-?[a-z]*)":', content_str)
489
490 # Filter to likely ASM fields (not URLs, not manifest)
491 asm_hyphenated = []
492 for key in hyphenated_keys:
493 if "http" in key or "manifest" in key:
494 continue
495 # Known hyphenated keys that are OK
496 if key in ["data-source-identifier", "data-source-feature"]:
497 continue
498 asm_hyphenated.append(key)
499
500 if asm_hyphenated:
501 unique = list(set(asm_hyphenated))[:10]
502 result.add_warning(
503 f"Found hyphenated field names (ASM uses spaces): {unique}"
504 + (" ... and more" if len(set(asm_hyphenated)) > 10 else "")
505 )
506 result.add_info("Tip: Use 'sample identifier' not 'sample-identifier'")
507
508
509def count_measurements(content_str: str) -> int:
510 """Count measurement documents in ASM."""
511 # Count occurrences of measurement document patterns
512 count = len(re.findall(r'"measurement identifier":', content_str))
513 if count == 0:
514 count = len(re.findall(r'"measurement-identifier":', content_str))
515 return count
516
517
518def validate_measurements(content_str: str, result: ValidationResult):
519 """Validate measurement documents."""
520 count = count_measurements(content_str)
521 result.metrics["measurement_count"] = count
522
523 if count == 0:
524 result.add_warning("No measurement documents found")
525 else:
526 result.add_info(f"Measurement count: {count}")
527
528
529def validate_sample_roles(content_str: str, result: ValidationResult):
530 """Check for valid sample roles."""
531 roles = re.findall(r'"sample.role.type":\s*"([^"]+)"', content_str)
532 if not roles:
533 roles = re.findall(r'"sample role type":\s*"([^"]+)"', content_str)
534
535 if roles:
536 unknown_roles = [r for r in set(roles) if r not in VALID_SAMPLE_ROLES]
537 if unknown_roles:
538 result.add_warning(
539 f"Unknown sample roles not in known list (as of {VALIDATION_RULES_DATE}): {unknown_roles}. "
540 f"These may be valid Allotrope roles added after spec version {ASM_SPEC_VERSION}. "
541 f"Verify at: {SCHEMA_SOURCE}"
542 )
543
544
545def validate_statistics(asm: Dict, content_str: str, result: ValidationResult):
546 """Check for statistics documents where expected."""
547 technique, _ = detect_technique(asm)
548
549 has_stats = (
550 "statistics aggregate document" in content_str.lower()
551 or "statistics-aggregate-document" in content_str
552 )
553
554 result.metrics["has_statistics"] = has_stats
555
556 # Statistics are required for multi-analyte profiling
557 if "multi analyte" in technique or "multiplex" in content_str.lower():
558 if not has_stats:
559 result.add_warning(
560 "No statistics aggregate document found - bead-based assays should include "
561 "median, mean, CV, std dev per analyte"
562 )
563 else:
564 result.add_info("Statistics document: Present")
565
566
567def validate_units(content_str: str, result: ValidationResult):
568 """Check for valid units."""
569 # Find all unit values
570 units = re.findall(r'"unit":\s*"([^"]+)"', content_str)
571
572 # Check for common case-sensitivity issues
573 case_issues = []
574 for unit in set(units):
575 if unit.lower() in ["rfu", "mfi"] and unit not in ["RFU", "MFI"]:
576 case_issues.append(f"{unit} (should be uppercase)")
577 elif unit in ["ul", "uL", "µl"] and unit != "μL":
578 case_issues.append(f"{unit} (should be μL)")
579
580 if case_issues:
581 result.add_warning(f"Non-standard unit capitalization: {case_issues}")
582
583 # Soft validation: check against known units list
584 all_known_units = set()
585 for unit_list in VALID_UNITS.values():
586 all_known_units.update(unit_list)
587
588 unknown_units = []
589 for unit in set(units):
590 # Skip units that have case issues (already reported above)
591 if unit not in all_known_units and unit not in [u.lower() for u in case_issues]:
592 unknown_units.append(unit)
593
594 if unknown_units:
595 result.add_warning(
596 f"Unknown units not in known list (as of {VALIDATION_RULES_DATE}): {unknown_units}. "
597 f"These may be valid Allotrope units added after spec version {ASM_SPEC_VERSION}. "
598 f"Verify at: {SCHEMA_SOURCE}"
599 )
600
601
602def validate_metadata(content_str: str, result: ValidationResult):
603 """Check for required metadata fields."""
604 required_fields = [
605 ("device system document", "equipment serial number"),
606 ("data system document", "software name"),
607 ("data system document", "software version"),
608 ]
609
610 missing = []
611 for _, field in required_fields:
612 if (
613 field not in content_str.lower()
614 and field.replace(" ", "-") not in content_str
615 ):
616 missing.append(field)
617
618 if missing:
619 result.add_warning(f"Missing recommended metadata: {missing}")
620
621
622def validate_calculated_data(content_str: str, result: ValidationResult):
623 """Check calculated data has proper traceability."""
624 content_lower = content_str.lower()
625
626 has_calculated = (
627 "calculated data document" in content_lower
628 or "calculated-data-document" in content_str
629 )
630 has_data_source = (
631 "data source aggregate document" in content_lower
632 or "data-source-aggregate-document" in content_str
633 )
634
635 result.metrics["has_calculated_data"] = has_calculated
636 result.metrics["has_data_source_traceability"] = has_data_source
637
638 if has_calculated:
639 result.add_info("Calculated data document: Present")
640 if not has_data_source:
641 result.add_error(
642 "Calculated data found without data-source-aggregate-document - "
643 "traceability is required for audit/regulatory compliance"
644 )
645 else:
646 result.add_info("Data source traceability: Present")
647
648 # Check for calculated fields that might be incorrectly placed in measurement-document
649 misplaced = []
650 for field in SHOULD_BE_CALCULATED:
651 # Check if field appears in measurement document context but not in calculated data
652 field_pattern = field.replace("/", ".")
653 if field_pattern in content_lower:
654 # If we have the field but no calculated-data-document, it's misplaced
655 if not has_calculated:
656 misplaced.append(field)
657
658 if misplaced:
659 result.add_warning(
660 f"Fields that should likely be in calculated-data-document: {misplaced[:5]}"
661 + (f" ... and {len(misplaced)-5} more" if len(misplaced) > 5 else "")
662 )
663
664
665def validate_unique_identifiers(content_str: str, result: ValidationResult):
666 """Validate that entities have unique identifiers for traceability."""
667 # Count different identifier types
668 measurement_ids = len(
669 re.findall(r'"measurement identifier":\s*"[^"]+"', content_str)
670 )
671 if measurement_ids == 0:
672 measurement_ids = len(
673 re.findall(r'"measurement-identifier":\s*"[^"]+"', content_str)
674 )
675
676 calculated_ids = len(
677 re.findall(r'"calculated data identifier":\s*"[^"]+"', content_str)
678 )
679 if calculated_ids == 0:
680 calculated_ids = len(
681 re.findall(r'"calculated-data-identifier":\s*"[^"]+"', content_str)
682 )
683
684 data_source_ids = len(
685 re.findall(r'"data source identifier":\s*"[^"]+"', content_str)
686 )
687 if data_source_ids == 0:
688 data_source_ids = len(
689 re.findall(r'"data-source-identifier":\s*"[^"]+"', content_str)
690 )
691
692 result.metrics["measurement_identifiers"] = measurement_ids
693 result.metrics["calculated_data_identifiers"] = calculated_ids
694 result.metrics["data_source_identifiers"] = data_source_ids
695
696 if measurement_ids == 0:
697 result.add_warning(
698 "No measurement identifiers found - required for traceability"
699 )
700
701 # If we have calculated data but no data source identifiers, that's a problem
702 if calculated_ids > 0 and data_source_ids == 0:
703 result.add_error(
704 f"Found {calculated_ids} calculated data entries but no data source identifiers - "
705 "each calculated value should reference its source"
706 )
707
708
709# =============================================================================
710# NEW: NESTED DOCUMENT STRUCTURE VALIDATION
711# =============================================================================
712
713
714def validate_nested_document_structure(
715 asm: Dict, content_str: str, result: ValidationResult
716):
717 """
718 Validate that fields are properly nested in their correct documents.
719
720 This checks for common mistakes like:
721 - Sample fields flattened directly onto measurement instead of in 'sample document'
722 - Device control fields flattened instead of in 'device control aggregate document'
723 - Custom/vendor fields not wrapped in 'custom information document'
724 """
725 content_lower = content_str.lower()
726
727 # Check if proper nested documents exist
728 has_sample_doc = (
729 '"sample document"' in content_lower or '"sample-document"' in content_str
730 )
731 has_device_control_doc = (
732 '"device control aggregate document"' in content_lower
733 or '"device-control-aggregate-document"' in content_str
734 )
735 has_custom_info_doc = (
736 '"custom information document"' in content_lower
737 or '"custom-information-document"' in content_str
738 )
739
740 result.metrics["has_sample_document"] = has_sample_doc
741 result.metrics["has_device_control_document"] = has_device_control_doc
742 result.metrics["has_custom_information_document"] = has_custom_info_doc
743
744 # Parse ASM to check field locations
745 def find_flattened_fields_in_measurements(obj, path=""):
746 """Recursively find fields that appear directly on measurement documents."""
747 issues = {"sample": [], "device_control": [], "custom": []}
748
749 if isinstance(obj, dict):
750 # Check if we're inside a measurement document
751 in_measurement = (
752 "measurement document" in path.lower() or "measurement-document" in path
753 )
754 in_sample_doc = (
755 "sample document" in path.lower() or "sample-document" in path
756 )
757 in_device_control = (
758 "device control" in path.lower() or "device-control" in path
759 )
760 in_custom_info = (
761 "custom information" in path.lower() or "custom-information" in path
762 )
763
764 for key, value in obj.items():
765 key_normalized = key.lower().replace("-", " ")
766 new_path = f"{path}.{key}"
767
768 # Check if this key should be nested but isn't
769 if in_measurement and not in_sample_doc:
770 if key_normalized in [
771 f.lower().replace("-", " ") for f in SAMPLE_DOCUMENT_FIELDS
772 ]:
773 issues["sample"].append(key)
774
775 if in_measurement and not in_device_control:
776 if key_normalized in [
777 f.lower().replace("-", " ") for f in DEVICE_CONTROL_FIELDS
778 ]:
779 issues["device_control"].append(key)
780
781 if in_measurement and not in_custom_info:
782 if key_normalized in [
783 f.lower().replace("-", " ") for f in CUSTOM_INFO_FIELDS
784 ]:
785 issues["custom"].append(key)
786
787 # Recurse
788 child_issues = find_flattened_fields_in_measurements(value, new_path)
789 for k in issues:
790 issues[k].extend(child_issues[k])
791
792 elif isinstance(obj, list):
793 for i, item in enumerate(obj):
794 child_issues = find_flattened_fields_in_measurements(
795 item, f"{path}[{i}]"
796 )
797 for k in issues:
798 issues[k].extend(child_issues[k])
799
800 return issues
801
802 issues = find_flattened_fields_in_measurements(asm)
803 flattened_sample_fields = list(set(issues["sample"]))
804 flattened_device_control_fields = list(set(issues["device_control"]))
805 flattened_custom_fields = list(set(issues["custom"]))
806
807 # Report issues
808 if flattened_sample_fields:
809 result.add_error(
810 f"Fields that should be nested in 'sample document' are flattened on measurement: "
811 f"{flattened_sample_fields[:5]}"
812 + (
813 f" ... and {len(flattened_sample_fields)-5} more"
814 if len(flattened_sample_fields) > 5
815 else ""
816 )
817 )
818 result.add_info(
819 "Tip: Wrap sample fields in a 'sample document' object inside each measurement"
820 )
821
822 if flattened_device_control_fields:
823 result.add_error(
824 f"Fields that should be nested in 'device control aggregate document' are flattened: "
825 f"{flattened_device_control_fields[:5]}"
826 + (
827 f" ... and {len(flattened_device_control_fields)-5} more"
828 if len(flattened_device_control_fields) > 5
829 else ""
830 )
831 )
832 result.add_info(
833 "Tip: Wrap device control fields in 'device control aggregate document' → 'device control document'"
834 )
835
836 if flattened_custom_fields:
837 result.add_warning(
838 f"Vendor-specific fields that should be in 'custom information document': "
839 f"{flattened_custom_fields[:5]}"
840 + (
841 f" ... and {len(flattened_custom_fields)-5} more"
842 if len(flattened_custom_fields) > 5
843 else ""
844 )
845 )
846
847
848def validate_liquid_handler_structure(
849 asm: Dict, content_str: str, result: ValidationResult
850):
851 """
852 Specific validation for liquid handler ASM documents.
853
854 Checks for:
855 - Proper transfer pairing (aspirate + dispense = 1 measurement)
856 - Source/destination field pairs
857 - Aspiration volume + transfer volume instead of single volume
858 """
859 technique, _ = detect_technique(asm)
860
861 # Only run for liquid handler techniques
862 if "liquid" not in technique.lower() and "handler" not in technique.lower():
863 # Also check content for liquid handler indicators
864 content_lower = content_str.lower()
865 if not any(
866 kw in content_lower
867 for kw in ["aspirate", "dispense", "liquid handler", "biomek"]
868 ):
869 return
870
871 result.add_info("Liquid handler specific validation...")
872
873 content_lower = content_str.lower()
874
875 # Check for proper volume field structure
876 has_aspiration_volume = (
877 "aspiration volume" in content_lower or "aspiration-volume" in content_str
878 )
879 has_transfer_volume = (
880 "transfer volume" in content_lower or "transfer-volume" in content_str
881 )
882 has_single_volume = (
883 '"volume"' in content_str
884 and not has_aspiration_volume
885 and not has_transfer_volume
886 )
887
888 if has_single_volume and not has_aspiration_volume:
889 result.add_warning(
890 "Liquid handler ASM uses single 'volume' field - "
891 "consider using 'aspiration volume' and 'transfer volume' for full transfer semantics"
892 )
893
894 if has_aspiration_volume and has_transfer_volume:
895 result.add_info("Volume fields: Proper aspiration/transfer volume structure")
896
897 # Check for source/destination pairing
898 has_source_dest = (
899 "source location" in content_lower or "source-location" in content_str
900 ) and (
901 "destination location" in content_lower or "destination-location" in content_str
902 )
903
904 has_separate_transfer_type = (
905 "transfer type" in content_lower or "transfer-type" in content_str
906 )
907
908 if has_separate_transfer_type and not has_source_dest:
909 result.add_warning(
910 "Found 'transfer type' field (Aspirate/Dispense as separate records) - "
911 "proper ASM pairs source→destination in single measurement with 'source location identifier' "
912 "and 'destination location identifier'"
913 )
914 result.add_info(
915 "Tip: Pair aspirate+dispense operations by probe number into single transfer measurements"
916 )
917
918 if has_source_dest:
919 result.add_info("Source/destination: Proper paired transfer structure")
920
921 # Check for labware name fields in custom information document
922 has_labware_names = (
923 "source labware name" in content_lower
924 or "destination labware name" in content_lower
925 )
926
927 if has_labware_names:
928 result.add_info(
929 "Labware names: Present (should be in custom information document)"
930 )
931
932
933def compare_to_reference(
934 asm: Dict,
935 reference: Dict,
936 content_str: str,
937 ref_content: str,
938 result: ValidationResult,
939):
940 """Compare generated ASM to reference ASM."""
941 result.add_info("Comparing to reference ASM...")
942
943 # Compare techniques
944 gen_tech, _ = detect_technique(asm)
945 ref_tech, _ = detect_technique(reference)
946
947 if gen_tech.replace("-", " ") != ref_tech.replace("-", " "):
948 result.add_error(
949 f"Technique mismatch: generated '{gen_tech}' vs reference '{ref_tech}'"
950 )
951
952 # Compare measurement counts
953 gen_count = count_measurements(content_str)
954 ref_count = count_measurements(ref_content)
955
956 result.metrics["reference_measurement_count"] = ref_count
957
958 if gen_count != ref_count:
959 diff = ref_count - gen_count
960 if diff > 0:
961 result.add_error(
962 f"Missing {diff} measurements: generated {gen_count} vs reference {ref_count}"
963 )
964 else:
965 result.add_warning(
966 f"Extra {-diff} measurements: generated {gen_count} vs reference {ref_count}"
967 )
968
969 # Compare sample roles
970 gen_roles = set(re.findall(r'"sample.role.type":\s*"([^"]+)"', content_str))
971 ref_roles = set(re.findall(r'"sample role type":\s*"([^"]+)"', ref_content))
972
973 missing_roles = ref_roles - gen_roles
974 if missing_roles:
975 result.add_warning(f"Missing sample roles from reference: {missing_roles}")
976
977 # Compare nested document presence
978 ref_has_sample_doc = '"sample document"' in ref_content.lower()
979 gen_has_sample_doc = (
980 '"sample document"' in content_str.lower() or '"sample-document"' in content_str
981 )
982
983 if ref_has_sample_doc and not gen_has_sample_doc:
984 result.add_error(
985 "Reference has 'sample document' but generated ASM does not - fields may be incorrectly flattened"
986 )
987
988 ref_has_device_control = (
989 '"device control aggregate document"' in ref_content.lower()
990 )
991 gen_has_device_control = (
992 '"device control aggregate document"' in content_str.lower()
993 or '"device-control-aggregate-document"' in content_str
994 )
995
996 if ref_has_device_control and not gen_has_device_control:
997 result.add_error(
998 "Reference has 'device control aggregate document' but generated ASM does not"
999 )
1000
1001 ref_has_custom_info = '"custom information document"' in ref_content.lower()
1002 gen_has_custom_info = (
1003 '"custom information document"' in content_str.lower()
1004 or '"custom-information-document"' in content_str
1005 )
1006
1007 if ref_has_custom_info and not gen_has_custom_info:
1008 result.add_warning(
1009 "Reference has 'custom information document' for vendor fields but generated ASM does not"
1010 )
1011
1012
1013def validate_asm(
1014 filepath: str, reference_path: Optional[str] = None, strict: bool = False
1015) -> ValidationResult:
1016 """
1017 Validate ASM JSON file.
1018
1019 Args:
1020 filepath: Path to ASM JSON file
1021 reference_path: Optional path to reference ASM for comparison
1022 strict: If True, treat warnings as errors
1023
1024 Returns:
1025 ValidationResult with errors, warnings, and metrics
1026 """
1027 result = ValidationResult()
1028
1029 # Load ASM file
1030 try:
1031 with open(filepath, "r", encoding="utf-8") as f:
1032 content_str = f.read()
1033 asm = json.loads(content_str)
1034 except json.JSONDecodeError as e:
1035 result.add_error(f"Invalid JSON: {e}")
1036 return result
1037 except FileNotFoundError:
1038 result.add_error(f"File not found: {filepath}")
1039 return result
1040
1041 result.add_info(f"Validating: {filepath}")
1042
1043 # Run validations
1044 validate_manifest(asm, result)
1045 validate_technique(asm, result, content_str)
1046 validate_naming_conventions(content_str, result)
1047 validate_measurements(content_str, result)
1048 validate_sample_roles(content_str, result)
1049 validate_statistics(asm, content_str, result)
1050 validate_units(content_str, result)
1051 validate_metadata(content_str, result)
1052 validate_calculated_data(content_str, result)
1053 validate_unique_identifiers(content_str, result)
1054
1055 # NEW: Nested document structure validation
1056 validate_nested_document_structure(asm, content_str, result)
1057 validate_liquid_handler_structure(asm, content_str, result)
1058
1059 # Compare to reference if provided
1060 if reference_path:
1061 try:
1062 with open(reference_path, "r", encoding="utf-8") as f:
1063 ref_content = f.read()
1064 reference = json.loads(ref_content)
1065 compare_to_reference(asm, reference, content_str, ref_content, result)
1066 except Exception as e:
1067 result.add_warning(f"Could not load reference file: {e}")
1068
1069 # In strict mode, convert warnings to errors
1070 if strict:
1071 result.errors.extend([w.replace("WARNING", "ERROR") for w in result.warnings])
1072 result.warnings = []
1073
1074 return result
1075
1076
1077def main():
1078 parser = argparse.ArgumentParser(description="Validate ASM JSON output")
1079 parser.add_argument("input", help="ASM JSON file to validate")
1080 parser.add_argument("--reference", "-r", help="Reference ASM file for comparison")
1081 parser.add_argument(
1082 "--strict", "-s", action="store_true", help="Treat warnings as errors"
1083 )
1084 parser.add_argument("--quiet", "-q", action="store_true", help="Only show errors")
1085
1086 args = parser.parse_args()
1087
1088 result = validate_asm(args.input, args.reference, args.strict)
1089
1090 if args.quiet:
1091 if result.errors:
1092 for error in result.errors:
1093 print(error)
1094 sys.exit(1)
1095 sys.exit(0)
1096
1097 result.print_report()
1098 sys.exit(0 if result.is_valid() else 1)
1099
1100
1101if __name__ == "__main__":
1102 main()