#!/usr/bin/env python3 """Check declarations in the Operavue synthetic implant-lineage CSV, locally. Usage: python3 check_implant_lineage.py implant-design-lineage-example.csv Python 3 standard library only. Reads one UTF-8 CSV; writes JSON to stdout. Exit 0: no declaration findings; 1: findings; 2: invalid input. No release pass is implied. No network, artifact reads, file writes, or latest-version selection. Run with --help for the exact schema and supported declarations. """ import csv import io import json import re import sys FIELDS = ( 'case_ref', 'source_series_ref', 'anatomy_ref', 'requirements_ref', 'design_ref', 'design_version', 'predecessor_ref', 'export_kind', 'export_format', 'geometry_units', 'geometry_frame_ref', 'image_frame_ref', 'mapping_status', 'review_record_ref', 'reviewed_design_ref', 'review_decision', 'split_group_ref', 'example_interpretation', ) REQUIRED_REFS = ( 'case_ref', 'source_series_ref', 'anatomy_ref', 'requirements_ref', 'design_ref', 'review_record_ref', 'reviewed_design_ref', 'split_group_ref', ) ENUMS = { 'export_kind': ('implant_mesh', 'viewer_screenshot'), 'export_format': ('STL', 'PNG'), 'geometry_units': ('mm', 'cm', 'm', 'um'), 'mapping_status': ('shared_frame_declared', 'mapping_not_supplied', 'no_geometry_supplied'), 'review_decision': ('approved_for_case',), } REF_PATTERN = re.compile(r'[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}\Z') VERSION_PATTERN = re.compile(r'[1-9][0-9]{0,8}\Z') UNKNOWN = {'unknown', 'unspecified', 'missing', 'none', 'null', 'na', 'n/a', 'n.a.', 'tbd', 'unavailable', 'not_supplied', 'not_known', 'not_available'} MAX_BYTES = 2 * 1024 * 1024 MAX_ROWS = 10000 MAX_FIELD = 4096 SCOPE = 'Manifest declaration consistency only; no artifact validation or release approval.' LIMITATIONS = [ 'References, review decisions, units and frames are declarations, not verified evidence.', 'No DICOM, CAD, mesh, anatomical alignment, clinical, manufacturing or regulatory validation.', 'No privacy, licensing, rights, patient independence or dataset availability determination.', 'Only shared-frame declarations are supported; supplied transforms require a different schema and review.', ] class InputError(Exception): """A fixed error code, never input content or an operating-system message.""" def base_report(status): return {'schema_version': '1.0', 'status': status, 'scope': SCOPE, 'target': 'approved_implant_mesh_at_one_snapshot', 'limitations': LIMITATIONS} def valid_ref(value): return bool(REF_PATTERN.fullmatch(value)) and value.lower() not in UNKNOWN def parse_csv(data): """Parse a bounded UTF-8 document. Header order may vary; fields may not.""" if len(data) > MAX_BYTES: raise InputError('input_too_large') try: text = data.decode('utf-8-sig') except UnicodeError: raise InputError('invalid_utf8') from None csv.field_size_limit(MAX_FIELD) try: reader = csv.reader(io.StringIO(text, newline=''), strict=True) header = next(reader, None) if header is None or len(header) != len(FIELDS) or set(header) != set(FIELDS): raise InputError('header_schema_mismatch') rows = [] seen_designs = set() for values in reader: if len(values) != len(header): raise InputError('wrong_column_count') if len(rows) >= MAX_ROWS: raise InputError('too_many_records') row = {key: value.strip() for key, value in zip(header, values)} design = row['design_ref'] if design and design in seen_designs: raise InputError('duplicate_design_ref') seen_designs.add(design) rows.append(row) except csv.Error: raise InputError('malformed_csv_or_oversized_field') from None if not rows: raise InputError('no_data_records') return rows def check(rows): """Check parsed declarations. Row numbers count CSV records, including header.""" findings = [] seen_findings = set() def add(index, code, *fields): key = (index, code, fields) if key not in seen_findings: findings.append({'row': index + 2, 'code': code, 'fields': list(fields)}) seen_findings.add(key) def require_ref(index, row, field): if not row[field]: add(index, 'missing_required_reference', field) elif not valid_ref(row[field]): add(index, 'invalid_or_unknown_reference', field) designs = {row['design_ref']: i for i, row in enumerate(rows) if valid_ref(row['design_ref'])} versions = {} for i, row in enumerate(rows): for field in REQUIRED_REFS: require_ref(i, row, field) if VERSION_PATTERN.fullmatch(row['design_version']): versions[i] = int(row['design_version']) else: add(i, 'invalid_design_version', 'design_version') for field in ('export_kind', 'export_format', 'mapping_status', 'review_decision'): if row[field] not in ENUMS[field]: add(i, 'missing_or_unsupported_declaration', field) if valid_ref(row['design_ref']) and valid_ref(row['reviewed_design_ref']): if row['design_ref'] != row['reviewed_design_ref']: add(i, 'reviewed_design_mismatch', 'design_ref', 'reviewed_design_ref') reviewed = designs.get(row['reviewed_design_ref']) if reviewed is None: add(i, 'reviewed_design_not_in_manifest', 'reviewed_design_ref') elif valid_ref(row['case_ref']) and valid_ref(rows[reviewed]['case_ref']): if row['case_ref'] != rows[reviewed]['case_ref']: add(i, 'reviewed_design_crosses_case', 'reviewed_design_ref', 'case_ref') if row['export_kind'] == 'implant_mesh': if row['export_format'] != 'STL': add(i, 'mesh_format_mismatch', 'export_kind', 'export_format') if not row['geometry_units']: add(i, 'missing_geometry_units', 'geometry_units') elif row['geometry_units'] not in ENUMS['geometry_units']: add(i, 'unsupported_geometry_units', 'geometry_units') for field in ('geometry_frame_ref', 'image_frame_ref'): require_ref(i, row, field) if row['mapping_status'] == 'shared_frame_declared': if (valid_ref(row['geometry_frame_ref']) and valid_ref(row['image_frame_ref']) and row['geometry_frame_ref'] != row['image_frame_ref']): add(i, 'shared_frame_mismatch', 'geometry_frame_ref', 'image_frame_ref', 'mapping_status') elif row['mapping_status'] == 'mapping_not_supplied': add(i, 'unresolved_mapping', 'mapping_status') elif row['mapping_status'] == 'no_geometry_supplied': add(i, 'mesh_mapping_contradiction', 'export_kind', 'mapping_status') elif row['export_kind'] == 'viewer_screenshot': add(i, 'missing_implant_mesh', 'export_kind') if row['export_format'] != 'PNG': add(i, 'screenshot_format_mismatch', 'export_kind', 'export_format') if row['mapping_status'] != 'no_geometry_supplied': add(i, 'screenshot_mapping_contradiction', 'export_kind', 'mapping_status') if any(row[f] for f in ('geometry_units', 'geometry_frame_ref', 'image_frame_ref')): add(i, 'screenshot_geometry_declarations', 'geometry_units', 'geometry_frame_ref', 'image_frame_ref') # Check references and revision order without choosing a target or rewriting rows. edges = {} case_versions = {} case_groups = {} review_records = {} for i, row in enumerate(rows): predecessor = row['predecessor_ref'] if predecessor: if not valid_ref(predecessor): add(i, 'invalid_or_unknown_reference', 'predecessor_ref') elif predecessor not in designs: add(i, 'missing_predecessor', 'predecessor_ref') else: prior = designs[predecessor] edges[i] = prior if prior == i: add(i, 'self_predecessor', 'predecessor_ref') if valid_ref(row['case_ref']) and valid_ref(rows[prior]['case_ref']): if row['case_ref'] != rows[prior]['case_ref']: add(i, 'predecessor_crosses_case', 'predecessor_ref', 'case_ref') if i in versions and prior in versions and versions[prior] >= versions[i]: add(i, 'predecessor_not_earlier_version', 'predecessor_ref', 'design_version') if valid_ref(row['case_ref']): case_groups.setdefault(row['case_ref'], []).append(i) if i in versions: case_versions.setdefault((row['case_ref'], versions[i]), []).append(i) if valid_ref(row['review_record_ref']): review_records.setdefault(row['review_record_ref'], []).append(i) for indices in case_versions.values(): if len(indices) > 1: for i in indices: add(i, 'duplicate_case_version', 'case_ref', 'design_version') for indices in case_groups.values(): groups = {rows[i]['split_group_ref'] for i in indices if valid_ref(rows[i]['split_group_ref'])} if len(groups) > 1: for i in indices: add(i, 'case_group_mismatch', 'case_ref', 'split_group_ref') for indices in review_records.values(): declarations = {(rows[i]['reviewed_design_ref'], rows[i]['review_decision']) for i in indices} if len(declarations) > 1: for i in indices: add(i, 'review_record_conflict', 'review_record_ref', 'reviewed_design_ref', 'review_decision') # Iterative cycle detection also handles long chains without recursion limits. visited = set() for start in edges: path = [] positions = {} current = start while current in edges and current not in visited: if current in positions: for i in path[positions[current]:]: add(i, 'predecessor_cycle', 'predecessor_ref') break positions[current] = len(path) path.append(current) current = edges[current] visited.update(path) findings.sort(key=lambda f: (f['row'], f['code'], f['fields'])) report = base_report('declaration_findings' if findings else 'no_declaration_findings') report['counts'] = {'records': len(rows), 'records_with_findings': len({f['row'] for f in findings}), 'findings': len(findings)} report['findings'] = findings return report def main(argv): if argv == ['--help']: report = base_report('usage') report.update({'usage': 'python3 check_implant_lineage.py manifest.csv', 'columns': FIELDS, 'enumerations': ENUMS, 'reference_syntax': '1-128 ASCII letters, digits, underscore, period, colon or hyphen; first character alphanumeric; reserved unknown markers rejected', 'design_version': 'Integer 1 through 999999999, without leading zeros', 'row_numbers': 'CSV record number, including header; not physical line number', 'optional_values': ['predecessor_ref may be blank to declare no predecessor', 'example_interpretation is ignored', 'screenshot geometry fields must be blank'], 'limits': {'bytes': MAX_BYTES, 'records': MAX_ROWS, 'field_characters': MAX_FIELD}, 'exit_codes': {'0': 'no declaration findings (or this help)', '1': 'declaration findings', '2': 'invalid input'}}) code = 0 else: try: if len(argv) != 1: raise InputError('expected_one_csv_argument') try: with open(argv[0], 'rb') as source: data = source.read(MAX_BYTES + 1) except (OSError, ValueError): raise InputError('unreadable_input') from None report = check(parse_csv(data)) code = 1 if report['findings'] else 0 except InputError as error: report = base_report('invalid_input') report['error'] = str(error) code = 2 print(json.dumps(report, indent=2, ensure_ascii=True)) return code if __name__ == '__main__': sys.exit(main(sys.argv[1:]))