LiAuto-DriveAction/croissant-checker

1import mlcroissant._src.operation_graph.operations.download as dl_mod2import requests3import os4 5# Make sure the HF token is loaded6HF_TOKEN = os.environ.get("HF_TOKEN")7 8# Set the environment variables Croissant expects9os.environ["CROISSANT_BASIC_AUTH_USERNAME"] = "hf_user"10os.environ["CROISSANT_BASIC_AUTH_PASSWORD"] = HF_TOKEN or ""11 12print("[DEBUG] HF_TOKEN is", "set" if HF_TOKEN else "missing")13print("[DEBUG] Basic auth env set for Croissant")14 15import mlcroissant as mlc16import func_timeout17import json18import traceback19 20WAIT_TIME = 10 * 60  # seconds21 22def validate_json(file_path):23    """Validate that the file is proper JSON."""24    try:25        with open(file_path, 'r') as f:26            json_data = json.load(f)27        return True, "The file is valid JSON.", json_data28    except json.JSONDecodeError as e:29        error_message = f"Invalid JSON format: {str(e)}"30        return False, error_message, None31    except Exception as e:32        error_message = f"Error reading file: {str(e)}"33        return False, error_message, None34 35def validate_croissant(json_data):36    """Validate that the JSON follows Croissant schema."""37    try:38        dataset = mlc.Dataset(jsonld=json_data)39        return True, "The dataset passes Croissant validation."40    except mlc.ValidationError as e:41        error_details = traceback.format_exc()42        error_message = f"Validation failed: {str(e)}\n\n{error_details}"43        return False, error_message44    except Exception as e:45        error_details = traceback.format_exc()46        error_message = f"Unexpected error during validation: {str(e)}\n\n{error_details}"47        return False, error_message48    49def try_generate_record(record_collection):50    try:51        for i, record in enumerate(record_collection):52            if i == 0:53                break54        return "success"55    except Exception as e:56        return e57 58def validate_records(json_data):59    """Validate that records can be generated within the time limit."""60    try:61        dataset = mlc.Dataset(jsonld=json_data)62        record_sets = dataset.metadata.record_sets63        64        if not record_sets:65            return True, "No record sets found to validate.", "pass"66        67        results = []68        69        for record_set in record_sets:70            try:71                result = func_timeout.func_timeout(72                    WAIT_TIME,73                    lambda: try_generate_record(dataset.records(record_set=record_set.uuid))74                )75 76                if isinstance(result, Exception):77                    raise result  # re-raise actual error outside timeout78 79                results.append(f"Record set '{record_set.uuid}' passed validation.")80 81            except func_timeout.exceptions.FunctionTimedOut:82                error_message = f"Record set '{record_set.uuid}' generation took too long (>10 minutes)."83                return False, error_message, "warning"84 85            except Exception as e:86                error_details = traceback.format_exc()87                error_message = (88                    f"Record set '{record_set.uuid}' failed due to generation error:\n\n"89                    f"```text\n{str(e)}\n\n{error_details}```"90                )91                return False, error_message, "warning"92        93        return True, "\n".join(results), "pass"94    except Exception as e:95        error_details = traceback.format_exc()96        error_message = f"Unexpected error during records validation: {str(e)}\n\n{error_details}"97        return False, error_message, "error"98    99def generate_validation_report(filename, json_data, results):100    """Generate a detailed validation report in markdown format."""101    report = []102    report.append("# CROISSANT VALIDATION REPORT")103    report.append("=" * 80)104    report.append("## VALIDATION RESULTS")105    report.append("-" * 80)106    report.append(f"Starting validation for file: {filename}")107 108    # Add validation results109    for result in results:110        if len(result) == 4:111            test_name, passed, message, status = result112        else:113            test_name, passed, message = result114            status = "pass" if passed else "error"115 116        report.append(f"### {test_name}")117        if status == "pass":118            report.append("✓")119        elif status == "warning":120            report.append("?")  # Question mark for warning121        else:122            report.append("✗")123        report.append(message.strip())  # Remove any trailing newlines124 125    # Add JSON-LD reference126    report.append("## JSON-LD REFERENCE")127    report.append("=" * 80)128    report.append("```json")129    report.append(json.dumps(json_data, indent=2))130    report.append("```")131 132    return "\n".join(report)