🔄 Reproducible Workflows¶

# FIGURES/fig_analysis.py
"""
Automated figure generation with complete reproducibility.
"""
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from pathlib import Path
import hashlib

# Data versioning with checksums
def verify_data_integrity(filepath, expected_hash=None):
    """Verify data hasn't changed unexpectedly."""
    with open(filepath, 'rb') as f:
        file_hash = hashlib.sha256(f.read()).hexdigest()

    if expected_hash and file_hash != expected_hash:
        raise ValueError(f"Data integrity check failed for {filepath}")

    return file_hash

# Load and verify data
data_file = Path("data/experiment_results.csv")
data_hash = verify_data_integrity(data_file)
print(f"Data verified: {data_hash[:8]}...")

# Create reproducible figure
df = pd.read_csv(data_file)

plt.figure(figsize=(10, 6))
sns.boxplot(data=df, x='condition', y='response_time')
plt.title(f'Response Times by Condition (n={len(df)})')
plt.ylabel('Response Time (ms)')
plt.xlabel('Experimental Condition')

# Add statistical annotations
from scipy import stats
conditions = df['condition'].unique()
if len(conditions) == 2:
    group1 = df[df['condition'] == conditions[0]]['response_time']
    group2 = df[df['condition'] == conditions[1]]['response_time']
    stat, p_value = stats.ttest_ind(group1, group2)
    plt.text(0.5, 0.95, f'p = {p_value:.3f}',
             transform=plt.gca().transAxes, ha='center')

plt.tight_layout()
plt.savefig('fig_analysis.png', dpi=300, bbox_inches='tight')
plt.close()

# Log analysis metadata
metadata = {
    'data_hash': data_hash,
    'n_subjects': len(df),
    'conditions': list(conditions),
    'analysis_date': pd.Timestamp.now().isoformat()
}

import json
with open('fig_analysis_metadata.json', 'w') as f:
    json.dump(metadata, f, indent=2)

# FIGURES/data_pipeline.py
"""
Reproducible data processing pipeline.
"""
import pandas as pd
from pathlib import Path
import logging

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class DataPipeline:
    def __init__(self, config_file="pipeline_config.yml"):
        self.config = self.load_config(config_file)
        self.steps = []

    def load_raw_data(self, filepath):
        """Load and validate raw data."""
        logger.info(f"Loading raw data from {filepath}")
        df = pd.read_csv(filepath)

        # Data validation
        required_columns = self.config.get('required_columns', [])
        missing_cols = set(required_columns) - set(df.columns)
        if missing_cols:
            raise ValueError(f"Missing required columns: {missing_cols}")

        logger.info(f"Loaded {len(df)} rows, {len(df.columns)} columns")
        self.steps.append(f"load_raw: {len(df)} rows")
        return df

    def clean_data(self, df):
        """Apply data cleaning procedures."""
        logger.info("Cleaning data...")
        initial_rows = len(df)

        # Remove duplicates
        df = df.drop_duplicates()

        # Handle missing values
        df = df.dropna(subset=self.config.get('required_fields', []))

        final_rows = len(df)
        logger.info(f"Cleaning: {initial_rows} → {final_rows} rows")
        self.steps.append(f"clean: {initial_rows} → {final_rows} rows")
        return df

    def generate_report(self):
        """Generate processing report."""
        report = {
            'pipeline_steps': self.steps,
            'config': self.config,
            'timestamp': pd.Timestamp.now().isoformat()
        }
        return report

# Execute pipeline
if __name__ == "__main__":
    pipeline = DataPipeline()

    # Process data
    raw_data = pipeline.load_raw_data("data/raw_experiment.csv")
    clean_data = pipeline.clean_data(raw_data)

    # Save processed data
    clean_data.to_csv("data/processed_experiment.csv", index=False)

    # Generate report
    report = pipeline.generate_report()
    with open("data/processing_report.json", "w") as f:
        json.dump(report, f, indent=2)

# tests/test_figures.py
"""
Test suite for figure generation.
"""
import pytest
import pandas as pd
from pathlib import Path
import matplotlib
matplotlib.use('Agg')  # Non-interactive backend

class TestFigures:
    def test_data_loading(self):
        """Test that required data files exist and are valid."""
        data_path = Path("data/experiment_results.csv")
        assert data_path.exists(), "Missing experiment data file"

        df = pd.read_csv(data_path)
        assert len(df) > 0, "Empty data file"
        assert 'condition' in df.columns, "Missing condition column"
        assert 'response_time' in df.columns, "Missing response_time column"

    def test_figure_generation(self):
        """Test that figures generate without errors."""
        import subprocess
        result = subprocess.run(['python', 'FIGURES/fig_analysis.py'],
                              capture_output=True, text=True)

        assert result.returncode == 0, f"Figure generation failed: {result.stderr}"

        # Check output files
        assert Path("FIGURES/fig_analysis.png").exists(), "Figure not generated"
        assert Path("FIGURES/fig_analysis_metadata.json").exists(), "Metadata not generated"

    def test_statistical_validity(self):
        """Test statistical analysis components."""
        df = pd.read_csv("data/experiment_results.csv")

        # Basic statistical checks
        assert df['response_time'].std() > 0, "No variance in response times"
        assert df['condition'].nunique() > 1, "Only one condition found"

        # Check for reasonable values
        rt = df['response_time']
        assert rt.min() > 0, "Negative response times found"
        assert rt.max() < 10000, "Unreasonably large response times"

if __name__ == "__main__":
    pytest.main([__file__, "-v"])

# requirements.txt
rxiv-maker==2.5.0
pandas==2.0.3
matplotlib==3.7.2
seaborn==0.12.2
scipy==1.11.1
numpy==1.24.3
pytest==7.4.0

# Use caching for expensive computations
from functools import lru_cache
import pickle

@lru_cache(maxsize=None)
def expensive_analysis(data_hash):
    """Cache expensive analysis results."""
    # Load data, perform analysis
    return results

# Or use file-based caching
def cached_analysis(data_file, cache_file):
    if Path(cache_file).exists() and Path(cache_file).stat().st_mtime > Path(data_file).stat().st_mtime:
        with open(cache_file, 'rb') as f:
            return pickle.load(f)

    # Perform analysis
    results = perform_analysis(data_file)

    with open(cache_file, 'wb') as f:
        pickle.dump(results, f)

    return results