validate_conversion.py

#!/usr/bin/env python3
"""
Validation Script for CSV to JSON Conversion
This script validates that all requested modifications have been properly implemented.
"""

import json
import pandas as pd
from collections import Counter

def validate_json_structure(json_file: str) -> dict:
    """Validate the JSON structure and metadata extraction."""
    
    print("🔍 Validating JSON Structure and Metadata...")
    
    with open(json_file, 'r', encoding='utf-8') as f:
        data = json.load(f)
    
    df = pd.DataFrame(data)
    
    validation_results = {
        'structure_validation': {},
        'metadata_validation': {},
        'data_integrity': {},
        'sample_records': {}
    }
    
    # 1. Structure validation
    required_fields = ['source_scenario', 'source_rcp', 'source_model', 'year', 'source_file']
    missing_fields = [field for field in required_fields if field not in df.columns]
    
    validation_results['structure_validation'] = {
        'total_records': len(df),
        'total_columns': len(df.columns),
        'has_all_required_metadata_fields': len(missing_fields) == 0,
        'missing_fields': missing_fields,
        'required_fields_present': required_fields
    }
    
    # 2. Metadata validation
    scenarios = df['source_scenario'].value_counts().to_dict()
    rcps = df['source_rcp'].value_counts().to_dict()
    models = df['source_model'].value_counts().to_dict()
    years = df['year'].value_counts().to_dict()
    
    validation_results['metadata_validation'] = {
        'scenario_distribution': scenarios,
        'rcp_distribution': rcps,
        'model_distribution': models,
        'year_distribution': years,
        'unique_scenarios': list(scenarios.keys()),
        'unique_rcps': list(rcps.keys()),
        'unique_models': list(models.keys()),
        'unique_years': [int(y) for y in years.keys()]
    }
    
    # 3. Data integrity checks
    baseline_records = df[df['source_scenario'] == 'baseline']
    ssp_records = df[df['source_scenario'].str.startswith('ssp')]
    
    validation_results['data_integrity'] = {
        'baseline_has_correct_year': (baseline_records['year'] == 2010).all() if len(baseline_records) > 0 else False,
        'baseline_has_correct_rcp': (baseline_records['source_rcp'] == 'N/A').all() if len(baseline_records) > 0 else False,
        'baseline_has_correct_model': (baseline_records['source_model'] == 'baseline').all() if len(baseline_records) > 0 else False,
        'ssp_has_correct_year': (ssp_records['year'] == 2050).all() if len(ssp_records) > 0 else False,
        'ssp_has_rcp_scenarios': ssp_records['source_rcp'].str.startswith('rcp').all() if len(ssp_records) > 0 else False,
        'baseline_record_count': len(baseline_records),
        'ssp_record_count': len(ssp_records)
    }
    
    # 4. Sample records for verification
    validation_results['sample_records'] = {
        'baseline_sample': baseline_records.iloc[0].to_dict() if len(baseline_records) > 0 else None,
        'ssp1_sample': df[df['source_scenario'] == 'ssp1'].iloc[0].to_dict() if len(df[df['source_scenario'] == 'ssp1']) > 0 else None,
        'ssp2_sample': df[df['source_scenario'] == 'ssp2'].iloc[0].to_dict() if len(df[df['source_scenario'] == 'ssp2']) > 0 else None
    }
    
    return validation_results

def print_validation_summary(results: dict):
    """Print a human-readable validation summary."""
    
    print("\n" + "="*60)
    print("📋 VALIDATION SUMMARY REPORT")
    print("="*60)
    
    # Structure validation
    print("\n🏗️  STRUCTURE VALIDATION:")
    struct = results['structure_validation']
    print(f"   ✅ Total records processed: {struct['total_records']:,}")
    print(f"   ✅ Total columns standardized: {struct['total_columns']}")
    
    if struct['has_all_required_metadata_fields']:
        print(f"   ✅ All required metadata fields present")
    else:
        print(f"   ❌ Missing metadata fields: {struct['missing_fields']}")
    
    # Metadata validation
    print("\n📊 METADATA VALIDATION:")
    meta = results['metadata_validation']
    
    print(f"   📈 Scenarios found: {', '.join(meta['unique_scenarios'])}")
    for scenario, count in meta['scenario_distribution'].items():
        print(f"      - {scenario}: {count:,} records")
    
    print(f"   🌡️  RCP scenarios: {', '.join(meta['unique_rcps'])}")
    for rcp, count in meta['rcp_distribution'].items():
        print(f"      - {rcp}: {count:,} records")
    
    print(f"   🔬 Models: {', '.join(meta['unique_models'])}")
    for model, count in meta['model_distribution'].items():
        print(f"      - {model}: {count:,} records")
    
    print(f"   📅 Years: {', '.join(map(str, sorted(meta['unique_years'])))}")
    for year, count in meta['year_distribution'].items():
        print(f"      - {int(year)}: {count:,} records")
    
    # Data integrity
    print("\n🔍 DATA INTEGRITY CHECKS:")
    integrity = results['data_integrity']
    
    checks = [
        ("Baseline records have year 2010", integrity['baseline_has_correct_year']),
        ("Baseline records have RCP 'N/A'", integrity['baseline_has_correct_rcp']),
        ("Baseline records have model 'baseline'", integrity['baseline_has_correct_model']),
        ("SSP records have year 2050", integrity['ssp_has_correct_year']),
        ("SSP records have RCP scenarios", integrity['ssp_has_rcp_scenarios'])
    ]
    
    for check_name, passed in checks:
        status = "✅" if passed else "❌"
        print(f"   {status} {check_name}")
    
    print(f"\n   📊 Record distribution:")
    print(f"      - Baseline records: {integrity['baseline_record_count']:,}")
    print(f"      - SSP records: {integrity['ssp_record_count']:,}")
    
    # Sample verification
    print(f"\n🔍 SAMPLE RECORD VERIFICATION:")
    samples = results['sample_records']
    
    if samples['baseline_sample']:
        b = samples['baseline_sample']
        print(f"   📋 Baseline sample (Country: {b['countryname']}):")
        print(f"      - Scenario: {b['source_scenario']}, Year: {b['year']}")
        print(f"      - RCP: {b['source_rcp']}, Model: {b['source_model']}")
    
    if samples['ssp1_sample']:
        s1 = samples['ssp1_sample']
        print(f"   📋 SSP1 sample (Country: {s1['countryname']}):")
        print(f"      - Scenario: {s1['source_scenario']}, Year: {s1['year']}")
        print(f"      - RCP: {s1['source_rcp']}, Model: {s1['source_model']}")
    
    if samples['ssp2_sample']:
        s2 = samples['ssp2_sample']
        print(f"   📋 SSP2 sample (Country: {s2['countryname']}):")
        print(f"      - Scenario: {s2['source_scenario']}, Year: {s2['year']}")
        print(f"      - RCP: {s2['source_rcp']}, Model: {s2['source_model']}")

def main():
    """Main validation function."""
    
    json_file = r"c:\Users\user\source\scenarios_out_ug\consolidated_data.json"
    
    print("🚀 Starting Validation of CSV to JSON Conversion...")
    
    try:
        validation_results = validate_json_structure(json_file)
        print_validation_summary(validation_results)
        
        # Save detailed validation report
        validation_report_file = r"c:\Users\user\source\scenarios_out_ug\validation_report.json"
        with open(validation_report_file, 'w', encoding='utf-8') as f:
            json.dump(validation_results, f, indent=2, ensure_ascii=False)
        
        print(f"\n📄 Detailed validation report saved to: {validation_report_file}")
        
        # Overall status
        struct_ok = validation_results['structure_validation']['has_all_required_metadata_fields']
        integrity_ok = all([
            validation_results['data_integrity']['baseline_has_correct_year'],
            validation_results['data_integrity']['baseline_has_correct_rcp'],
            validation_results['data_integrity']['baseline_has_correct_model'],
            validation_results['data_integrity']['ssp_has_correct_year'],
            validation_results['data_integrity']['ssp_has_rcp_scenarios']
        ])
        
        if struct_ok and integrity_ok:
            print(f"\n🎉 VALIDATION PASSED! All requirements have been successfully implemented.")
        else:
            print(f"\n⚠️  VALIDATION ISSUES FOUND. Please review the report above.")
            
    except Exception as e:
        print(f"❌ Validation failed with error: {e}")

if __name__ == "__main__":
    main()