"""Count reporting fields in the frozen ChemVerify JSON export with Python 3.9+.""" import argparse import csv import hashlib import json import math from collections import Counter from datetime import date from pathlib import Path DIRECTORY = Path(__file__).resolve().parent UNSET = object() STATUS_LABELS = ( 'numeric_eq', 'numeric_lt', 'numeric_gt', 'numeric_other_comparator', 'not_detected_without_number', 'pass_without_number', 'other_verdict_without_number', 'no_numeric_value_or_verdict', ) def is_number(value): return type(value) in (int, float) and math.isfinite(value) def populated(value): if value is UNSET or value is None: return False if isinstance(value, str): return bool(value.strip()) if isinstance(value, (dict, list)): return bool(value) return True def availability(records, level): fields = sorted({field for record in records for field in record}) rows = [] for field in fields: values = [record.get(field, UNSET) for record in records] present = sum(populated(value) for value in values) rows.append({ 'level': level, 'field': field, 'records': len(records), 'populated': present, 'json_null': sum(value is None for value in values), 'missing_key': sum(value is UNSET for value in values), 'empty_string_or_container': sum(value is not UNSET and value is not None and not populated(value) for value in values), 'numeric_zero': sum(is_number(value) and value == 0 for value in values), }) return rows def result_status(result): if is_number(result.get('value')): comparator = result.get('comparator') if comparator in ('eq', 'lt', 'gt'): return f'numeric_{comparator}' return 'numeric_other_comparator' verdict = result.get('verdict') if verdict == 'not_detected': return 'not_detected_without_number' if verdict == 'pass': return 'pass_without_number' if populated(verdict): return 'other_verdict_without_number' return 'no_numeric_value_or_verdict' def counts_by_measurand(result_rows): values = sorted({result.get('measurand') for _, _, result in result_rows}, key=lambda value: (value is None, value or '')) rows = [] for measurand in values: group = [(sample, report, result) for sample, report, result in result_rows if result.get('measurand') == measurand] status_counts = Counter(result_status(result) for _, _, result in group) rows.append({ 'measurand': 'JSON null' if measurand is None else measurand, 'samples_with_category': len({sample['protocol'] for sample, _, _ in group}), 'reports_with_category': len({report['reference'] for _, report, _ in group}), 'results': len(group), 'numeric_value': sum(is_number(result.get('value')) for _, _, result in group), 'json_null_value': sum(result.get('value', UNSET) is None for _, _, result in group), 'unit_populated': sum(populated(result.get('unit', UNSET)) for _, _, result in group), 'uncertainty_populated': sum(populated(result.get('uncertainty', UNSET)) for _, _, result in group), **{label: status_counts[label] for label in STATUS_LABELS}, }) return rows def date_bounds(records, key): dates = [date.fromisoformat(record[key]) for record in records if populated(record.get(key))] return {'first': min(dates).isoformat(), 'last': max(dates).isoformat()} def write_csv(path, rows): with path.open('w', encoding='utf-8', newline='') as file: writer = csv.DictWriter(file, fieldnames=list(rows[0])) writer.writeheader() writer.writerows(rows) def analyze(snapshot, metadata, output): raw = snapshot.read_bytes() source = json.loads(raw) metadata = json.loads(metadata.read_text()) if hashlib.sha256(raw).hexdigest() != metadata['sha256'] or len(raw) != metadata['bytes']: raise ValueError('Snapshot does not match the recorded SHA-256 and byte size.') samples = source['samples'] report_rows = [(sample, report) for sample in samples for report in sample['reports']] result_rows = [(sample, report, result) for sample, report in report_rows for result in report['results']] reports = [report for _, report in report_rows] results = [result for _, _, result in result_rows] if len({sample['protocol'] for sample in samples}) != len(samples): raise ValueError('Duplicate sample protocol URL.') if len({report['reference'] for report in reports}) != len(reports): raise ValueError('Duplicate report reference.') fields = availability(samples, 'sample') + availability(reports, 'report') + availability(results, 'result') measurands = counts_by_measurand(result_rows) status_counts = Counter(result_status(result) for result in results) statuses = [{'status': label, 'records': status_counts[label], 'denominator': len(results)} for label in STATUS_LABELS] numeric_rows = [(sample, report, result) for sample, report, result in result_rows if is_number(result.get('value'))] analysed = date_bounds(samples, 'analysed_at') tested = date_bounds(reports, 'tested_at') source_date_bounds = {'first': source['first_analysed'], 'last': source['last_analysed']} vendor_sizes = Counter(sample['vendor_slug'] for sample in samples) summary = { 'source': {**metadata, 'name': source['name'], 'citation': source['citation'], 'license_url': source['license']}, 'method': 'Descriptive field availability audit of every sample, nested report and nested result in this snapshot. Raw values were not imputed or changed.', 'cohort': { 'samples': len(samples), 'reports': len(reports), 'results': len(results), 'sample_analysed_at': analysed, 'report_tested_at': tested, 'source_top_level_analysed_range': source_date_bounds, 'source_range_matches_sample_analysed_dates': source_date_bounds == analysed, 'source_range_matches_report_tested_dates': source_date_bounds == tested, 'distinct_vendor_slugs': len(vendor_sizes), 'largest_vendor_sample_count': max(vendor_sizes.values()), 'distinct_laboratory_names': len({sample['laboratory'] for sample in samples}), 'reports_per_sample_distribution': dict(sorted(Counter(len(sample['reports']) for sample in samples).items())), 'results_per_report_distribution': dict(sorted(Counter(len(report['results']) for report in reports).items())), 'distinct_report_urls': len({report['report_url'] for report in reports}), }, 'numeric_values': { 'results': len(numeric_rows), 'samples_with_any': len({sample['protocol'] for sample, _, _ in numeric_rows}), 'reports_with_any': len({report['reference'] for _, report, _ in numeric_rows}), 'with_populated_unit': sum(populated(result.get('unit', UNSET)) for _, _, result in numeric_rows), 'explicit_zero': sum(result['value'] == 0 for _, _, result in numeric_rows), 'comparators': dict(sorted(Counter(result['comparator'] for _, _, result in numeric_rows).items())), }, 'comparators_all_results': dict(sorted(Counter(result['comparator'] for result in results).items())), 'result_method_differs_from_report_method_when_populated': sum( populated(result.get('method', UNSET)) and result['method'] != report['method'] for _, report, result in result_rows ), 'reports_tested_on_different_day_from_sample_analysed': sum( report['tested_at'] != sample['analysed_at'] for sample, report in report_rows ), 'field_availability': fields, 'result_statuses': statuses, 'measurand_coverage': measurands, } output.mkdir(parents=True, exist_ok=True) write_csv(output / 'field-availability.csv', fields) write_csv(output / 'result-status.csv', statuses) write_csv(output / 'measurand-coverage.csv', measurands) (output / 'summary.json').write_text(json.dumps(summary, indent=2, ensure_ascii=False) + '\n', encoding='utf-8') def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--snapshot', type=Path, default=DIRECTORY / 'source-snapshot.json') parser.add_argument('--metadata', type=Path, default=DIRECTORY / 'snapshot-metadata.json') parser.add_argument('--output-dir', type=Path, default=DIRECTORY) args = parser.parse_args() analyze(args.snapshot, args.metadata, args.output_dir) if __name__ == '__main__': main()