analyze_database.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
Database Analysis Script

Performs a comprehensive row count analysis of all tables in the database
to ensure data has been loaded correctly.
"""

import duckdb
import pandas as pd
from pathlib import Path
import logging
import json

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# Define paths
PROJECT_ROOT = Path("/home/labrys/labrys_platform")
DB_PATH = PROJECT_ROOT / "LabrysPlatform" / "labrys.duckdb"
DATA_DIR = PROJECT_ROOT / "LabrysPlatform" / "data"
INTEGRATED_DIR = DATA_DIR / "integrated"
PREPROCESSED_DIR = DATA_DIR / "preprocessed"
RAW_DIR = DATA_DIR / "raw"

def check_source_files():
    """Analyze the source data files."""
    logger.info("\n=== SOURCE FILES ANALYSIS ===")
    
    # Check raw data files
    raw_files = list(RAW_DIR.glob('**/*.csv'))
    if raw_files:
        logger.info(f"Found {len(raw_files)} raw data files")
        for file in raw_files[:10]:  # Show first 10 files only
            try:
                df = pd.read_csv(file)
                logger.info(f"  {file.relative_to(PROJECT_ROOT)}: {len(df)} rows")
            except Exception as e:
                logger.error(f"  Error reading {file.relative_to(PROJECT_ROOT)}: {e}")
    else:
        logger.warning("No raw data files found")
    
    # Check preprocessed data files
    preprocessed_files = list(PREPROCESSED_DIR.glob('**/*.csv'))
    if preprocessed_files:
        logger.info(f"\nFound {len(preprocessed_files)} preprocessed data files")
        for file in preprocessed_files[:10]:
            try:
                df = pd.read_csv(file)
                logger.info(f"  {file.relative_to(PROJECT_ROOT)}: {len(df)} rows")
            except Exception as e:
                logger.error(f"  Error reading {file.relative_to(PROJECT_ROOT)}: {e}")
    else:
        logger.warning("No preprocessed data files found")
    
    # Check integrated data files
    integrated_files = list(INTEGRATED_DIR.glob('*.csv'))
    if integrated_files:
        logger.info(f"\nFound {len(integrated_files)} integrated data files")
        for file in integrated_files:
            try:
                df = pd.read_csv(file)
                logger.info(f"  {file.relative_to(PROJECT_ROOT)}: {len(df)} rows")
            except Exception as e:
                logger.error(f"  Error reading {file.relative_to(PROJECT_ROOT)}: {e}")
    else:
        logger.warning("No integrated data files found")

def analyze_database_tables():
    """Analyze all tables in the database."""
    logger.info("\n=== DATABASE TABLES ANALYSIS ===")
    
    if not DB_PATH.exists():
        logger.error(f"Database file not found at {DB_PATH}")
        return
    
    conn = duckdb.connect(str(DB_PATH))
    
    # Get all schemas
    schemas = []
    try:
        schemas_df = conn.execute("SELECT schema_name FROM information_schema.schemata").fetchdf()
        schemas = schemas_df['schema_name'].tolist()
        logger.info(f"Found {len(schemas)} schemas: {schemas}")
    except Exception as e:
        # If the information_schema query fails, fall back to a simpler approach
        logger.warning(f"Could not get schemas using information_schema: {e}")
        try:
            # Alternative approach to get all schemas
            all_tables = conn.execute("PRAGMA show_tables").fetchdf()
            if 'schema' in all_tables.columns:
                schemas = all_tables['schema'].unique().tolist()
                logger.info(f"Found {len(schemas)} schemas using alternative method: {schemas}")
        except Exception as e2:
            logger.error(f"Could not get schemas using alternative method: {e2}")
    
    if not schemas:
        # If we still couldn't get schemas, assume the main ones
        schemas = ['main', 'raw', 'stg', 'int']
        logger.warning(f"Falling back to assumed schemas: {schemas}")
    
    # Analyze tables in each schema
    table_counts = {}
    for schema in schemas:
        try:
            # Get all tables in this schema
            tables_df = conn.execute(f"PRAGMA show_tables").fetchdf()
            
            # Filter for this schema
            if 'schema' in tables_df.columns:
                schema_tables = tables_df[tables_df['schema'] == schema]
            else:
                # If there's no schema column, assume all tables are in this schema
                schema_tables = tables_df
            
            if len(schema_tables) == 0:
                logger.info(f"No tables found in schema '{schema}'")
                continue
            
            logger.info(f"\nAnalyzing schema '{schema}' with {len(schema_tables)} tables")
            
            # Get row counts for each table
            for _, row in schema_tables.iterrows():
                table_name = row['name'] if 'name' in row.index else row['table_name']
                try:
                    full_table_name = f"{schema}.{table_name}"
                    count = conn.execute(f"SELECT COUNT(*) FROM {full_table_name}").fetchone()[0]
                    table_counts[full_table_name] = count
                    logger.info(f"  {full_table_name}: {count} rows")
                    
                    # For the main tables, show some sample data
                    if (schema == 'raw' and table_name in ['compounds', 'measurements']) or \
                       count > 0 and count < 10:
                        sample = conn.execute(f"SELECT * FROM {full_table_name} LIMIT 3").fetchdf()
                        print(f"Sample data from {full_table_name}:")
                        print(sample.head(3))
                        print("\n")
                except Exception as e:
                    logger.error(f"  Error getting row count for {schema}.{table_name}: {e}")
        except Exception as e:
            logger.error(f"Error analyzing schema '{schema}': {e}")
    
    # Summary of table counts
    logger.info("\n=== TABLE COUNT SUMMARY ===")
    for table, count in sorted(table_counts.items(), key=lambda x: (-x[1], x[0])):
        logger.info(f"{table}: {count} rows")
    
    conn.close()

def analyze_data_coverage():
    """Analyze data coverage and relationships."""
    logger.info("\n=== DATA COVERAGE ANALYSIS ===")
    
    if not DB_PATH.exists():
        logger.error(f"Database file not found at {DB_PATH}")
        return
    
    conn = duckdb.connect(str(DB_PATH))
    
    # Check compound-measurement relationship
    try:
        compounds_count = conn.execute("SELECT COUNT(*) FROM raw.compounds").fetchone()[0]
        measurements_count = conn.execute("SELECT COUNT(*) FROM raw.measurements").fetchone()[0]
        compounds_with_measurements = conn.execute("""
            SELECT COUNT(DISTINCT compound_id) 
            FROM raw.measurements
        """).fetchone()[0]
        
        logger.info(f"Total compounds: {compounds_count}")
        logger.info(f"Total measurements: {measurements_count}")
        logger.info(f"Compounds with measurements: {compounds_with_measurements} ({compounds_with_measurements/compounds_count*100:.2f}%)")
        logger.info(f"Average measurements per compound: {measurements_count/compounds_with_measurements:.2f}")
        
        # Check ALF calculation coverage
        alf_count = conn.execute("""
            SELECT COUNT(*) 
            FROM raw.measurements 
            WHERE alf IS NOT NULL
        """).fetchone()[0]
        logger.info(f"Measurements with ALF value: {alf_count} ({alf_count/measurements_count*100:.2f}%)")
        
        # Check parameter availability
        epsilon_count = conn.execute("""
            SELECT COUNT(*) 
            FROM raw.measurements 
            WHERE epsilon IS NOT NULL
        """).fetchone()[0]
        qy_count = conn.execute("""
            SELECT COUNT(*) 
            FROM raw.measurements 
            WHERE quantum_yield IS NOT NULL
        """).fetchone()[0]
        lifetime_count = conn.execute("""
            SELECT COUNT(*) 
            FROM raw.measurements 
            WHERE lifetime IS NOT NULL
        """).fetchone()[0]
        
        logger.info(f"Measurements with epsilon: {epsilon_count} ({epsilon_count/measurements_count*100:.2f}%)")
        logger.info(f"Measurements with quantum yield: {qy_count} ({qy_count/measurements_count*100:.2f}%)")
        logger.info(f"Measurements with lifetime: {lifetime_count} ({lifetime_count/measurements_count*100:.2f}%)")
        
        # Check complete parameter sets (all three required for ALF calculation)
        complete_params = conn.execute("""
            SELECT COUNT(*) 
            FROM raw.measurements 
            WHERE epsilon IS NOT NULL 
              AND quantum_yield IS NOT NULL 
              AND lifetime IS NOT NULL
        """).fetchone()[0]
        logger.info(f"Measurements with complete parameter set: {complete_params} ({complete_params/measurements_count*100:.2f}%)")
        
    except Exception as e:
        logger.error(f"Error analyzing data coverage: {e}")
    
    conn.close()

def main():
    """Main function to analyze the database."""
    logger.info(f"Starting database analysis for {DB_PATH}")
    
    # Check source files
    check_source_files()
    
    # Analyze database tables
    analyze_database_tables()
    
    # Analyze data coverage
    analyze_data_coverage()
    
    logger.info("Database analysis complete")

if __name__ == "__main__":
    main()