use plaintext for snakemake [ci skip]

Author: abhi18av

docs/shells.qmd

@@ -0,0 +1,329 @@
+#!/usr/bin/env python3
+"""
+Academic Data Processing Pipeline with Snakemake
+=================================================
+
+This pipeline demonstrates a comprehensive academic data processing workflow
+with built-in resume capabilities and experiment tracking.
+
+Features:
+- Automatic experiment tracking
+- Resume-friendly design with checkpoints
+- Comprehensive logging and reporting  
+- Modular rule design
+- Resource-aware execution
+- Quality control at each step
+
+Author: Research Team
+Version: 1.0.0
+"""
+
+import pandas as pd
+from pathlib import Path
+
+# Configuration loading
+configfile: "config.yaml"
+
+# Load sample information
+samples_df = pd.read_csv(config["samples"], sep="\t")
+SAMPLES = samples_df["sample_id"].tolist()
+
+# Global variables from config
+EXPERIMENT_ID = config.get("experiment_id", "unknown")
+OUTPUT_DIR = config.get("output_dir", "results")
+LOG_DIR = config.get("log_dir", "logs")
+
+# Ensure output directories exist
+Path(OUTPUT_DIR).mkdir(parents=True, exist_ok=True)
+Path(LOG_DIR).mkdir(parents=True, exist_ok=True)
+
+# Define final outputs
+rule all:
+    input:
+        # Quality control reports
+        expand(f"{OUTPUT_DIR}/qc/{{sample}}_fastqc.html", sample=SAMPLES),
+        
+        # Processed data
+        expand(f"{OUTPUT_DIR}/processed/{{sample}}_processed.txt", sample=SAMPLES),
+        
+        # Summary reports
+        f"{OUTPUT_DIR}/reports/experiment_summary.html",
+        f"{OUTPUT_DIR}/reports/quality_summary.txt",
+        
+        # Final consolidated output
+        f"{OUTPUT_DIR}/final/consolidated_results.csv"
+
+# Rule: Quality control analysis
+rule quality_control:
+    """Perform quality control analysis on input data"""
+    input:
+        sample = lambda wildcards: samples_df[samples_df["sample_id"] == wildcards.sample]["fastq_1"].iloc[0]
+    output:
+        html = f"{OUTPUT_DIR}/qc/{{sample}}_fastqc.html",
+        zip = f"{OUTPUT_DIR}/qc/{{sample}}_fastqc.zip"
+    params:
+        outdir = f"{OUTPUT_DIR}/qc"
+    log:
+        f"{LOG_DIR}/qc/{{sample}}_fastqc.log"
+    threads: 2
+    resources:
+        mem_mb = 4000,
+        time_min = 30
+    conda:
+        "envs/qc.yaml"
+    shell:
+        """
+        echo "Starting quality control for {wildcards.sample}" > {log}
+        
+        # Simulate FastQC analysis
+        mkdir -p {params.outdir}
+        
+        # Mock quality analysis (replace with actual FastQC)
+        echo "Quality control analysis for {wildcards.sample}" > {output.html}
+        echo "Sample: {wildcards.sample}" >> {output.html}
+        echo "Input: {input.sample}" >> {output.html}
+        echo "Analysis completed: $(date)" >> {output.html}
+        
+        # Create zip output
+        echo "FastQC results" > {output.zip}
+        
+        echo "Quality control completed for {wildcards.sample}" >> {log}
+        """
+
+# Rule: Data preprocessing
+rule preprocess_data:
+    """Preprocess raw data with quality filtering"""
+    input:
+        fastq = lambda wildcards: samples_df[samples_df["sample_id"] == wildcards.sample]["fastq_1"].iloc[0],
+        qc_html = f"{OUTPUT_DIR}/qc/{{sample}}_fastqc.html"
+    output:
+        processed = f"{OUTPUT_DIR}/processed/{{sample}}_processed.txt",
+        stats = f"{OUTPUT_DIR}/processed/{{sample}}_stats.json"
+    params:
+        quality_threshold = config.get("quality_threshold", 30),
+        min_length = config.get("min_length", 50)
+    log:
+        f"{LOG_DIR}/preprocess/{{sample}}_preprocess.log"
+    threads: 4
+    resources:
+        mem_mb = 8000,
+        time_min = 60
+    conda:
+        "envs/preprocessing.yaml"
+    shell:
+        """
+        echo "Starting preprocessing for {wildcards.sample}" > {log}
+        echo "Quality threshold: {params.quality_threshold}" >> {log}
+        echo "Minimum length: {params.min_length}" >> {log}
+        
+        # Mock preprocessing (replace with actual tools)
+        echo "Processed data for {wildcards.sample}" > {output.processed}
+        echo "Quality filtering applied with threshold {params.quality_threshold}" >> {output.processed}
+        echo "Minimum length filter: {params.min_length}" >> {output.processed}
+        echo "Processing completed: $(date)" >> {output.processed}
+        
+        # Generate processing statistics
+        cat > {output.stats} << EOF
+{{
+    "sample_id": "{wildcards.sample}",
+    "input_reads": 1000000,
+    "processed_reads": 950000,
+    "quality_threshold": {params.quality_threshold},
+    "min_length": {params.min_length},
+    "processing_date": "$(date -I)"
+}}
+EOF
+        
+        echo "Preprocessing completed for {wildcards.sample}" >> {log}
+        """
+
+# Checkpoint: Intermediate results validation
+checkpoint validate_processing:
+    """Validate that all processing steps completed successfully"""
+    input:
+        processed = expand(f"{OUTPUT_DIR}/processed/{{sample}}_processed.txt", sample=SAMPLES),
+        stats = expand(f"{OUTPUT_DIR}/processed/{{sample}}_stats.json", sample=SAMPLES)
+    output:
+        validation = f"{OUTPUT_DIR}/checkpoints/processing_validation.txt"
+    log:
+        f"{LOG_DIR}/validation/processing_validation.log"
+    resources:
+        mem_mb = 2000,
+        time_min = 10
+    shell:
+        """
+        echo "Validating processing results" > {log}
+        
+        # Create checkpoint directory
+        mkdir -p $(dirname {output.validation})
+        
+        # Validate all files exist and are non-empty
+        all_valid=true
+        for file in {input.processed} {input.stats}; do
+            if [ ! -s "$file" ]; then
+                echo "ERROR: $file is missing or empty" >> {log}
+                all_valid=false
+            else
+                echo "VALID: $file" >> {log}
+            fi
+        done
+        
+        if [ "$all_valid" = true ]; then
+            echo "All processing validation checks passed" > {output.validation}
+            echo "Validation completed: $(date)" >> {output.validation}
+        else
+            echo "Validation failed - see log for details" >&2
+            exit 1
+        fi
+        
+        echo "Processing validation completed" >> {log}
+        """
+
+# Rule: Generate quality summary
+rule quality_summary:
+    """Generate comprehensive quality summary across all samples"""
+    input:
+        validation = f"{OUTPUT_DIR}/checkpoints/processing_validation.txt",
+        stats = expand(f"{OUTPUT_DIR}/processed/{{sample}}_stats.json", sample=SAMPLES)
+    output:
+        summary = f"{OUTPUT_DIR}/reports/quality_summary.txt"
+    log:
+        f"{LOG_DIR}/reports/quality_summary.log"
+    resources:
+        mem_mb = 4000,
+        time_min = 15
+    conda:
+        "envs/analysis.yaml"
+    script:
+        "scripts/generate_quality_summary.py"
+
+# Rule: Advanced analysis
+rule advanced_analysis:
+    """Perform advanced analysis on processed data"""
+    input:
+        processed = expand(f"{OUTPUT_DIR}/processed/{{sample}}_processed.txt", sample=SAMPLES),
+        validation = f"{OUTPUT_DIR}/checkpoints/processing_validation.txt"
+    output:
+        analysis = f"{OUTPUT_DIR}/analysis/advanced_results.txt",
+        plots = f"{OUTPUT_DIR}/analysis/analysis_plots.png"
+    params:
+        analysis_type = config.get("analysis_type", "standard")
+    log:
+        f"{LOG_DIR}/analysis/advanced_analysis.log"
+    threads: 8
+    resources:
+        mem_mb = 16000,
+        time_min = 120
+    conda:
+        "envs/analysis.yaml"
+    shell:
+        """
+        echo "Starting advanced analysis" > {log}
+        echo "Analysis type: {params.analysis_type}" >> {log}
+        
+        # Create analysis directory
+        mkdir -p $(dirname {output.analysis})
+        
+        # Mock advanced analysis
+        echo "Advanced Analysis Results" > {output.analysis}
+        echo "Experiment ID: {EXPERIMENT_ID}" >> {output.analysis}
+        echo "Analysis type: {params.analysis_type}" >> {output.analysis}
+        echo "Samples analyzed: {SAMPLES}" >> {output.analysis}
+        echo "Analysis completed: $(date)" >> {output.analysis}
+        
+        # Generate mock plots
+        echo "Analysis plots generated" > {output.plots}
+        
+        echo "Advanced analysis completed" >> {log}
+        """
+
+# Rule: Consolidate results
+rule consolidate_results:
+    """Consolidate all analysis results into final output"""
+    input:
+        analysis = f"{OUTPUT_DIR}/analysis/advanced_results.txt",
+        quality_summary = f"{OUTPUT_DIR}/reports/quality_summary.txt",
+        stats = expand(f"{OUTPUT_DIR}/processed/{{sample}}_stats.json", sample=SAMPLES)
+    output:
+        consolidated = f"{OUTPUT_DIR}/final/consolidated_results.csv"
+    log:
+        f"{LOG_DIR}/final/consolidation.log"
+    resources:
+        mem_mb = 4000,
+        time_min = 30
+    conda:
+        "envs/analysis.yaml"
+    script:
+        "scripts/consolidate_results.py"
+
+# Rule: Generate experiment report
+rule experiment_report:
+    """Generate comprehensive experiment report"""
+    input:
+        consolidated = f"{OUTPUT_DIR}/final/consolidated_results.csv",
+        qc_files = expand(f"{OUTPUT_DIR}/qc/{{sample}}_fastqc.html", sample=SAMPLES),
+        analysis = f"{OUTPUT_DIR}/analysis/advanced_results.txt"
+    output:
+        report = f"{OUTPUT_DIR}/reports/experiment_summary.html"
+    params:
+        experiment_id = EXPERIMENT_ID,
+        researcher = config.get("researcher", "Unknown"),
+        project = config.get("project", "Research Project")
+    log:
+        f"{LOG_DIR}/reports/experiment_report.log"
+    resources:
+        mem_mb = 2000,
+        time_min = 20
+    conda:
+        "envs/reporting.yaml"
+    script:
+        "scripts/generate_experiment_report.py"
+
+# Rule: Cleanup temporary files
+rule cleanup:
+    """Clean up temporary files while preserving important results"""
+    input:
+        report = f"{OUTPUT_DIR}/reports/experiment_summary.html",
+        consolidated = f"{OUTPUT_DIR}/final/consolidated_results.csv"
+    output:
+        cleanup_log = f"{OUTPUT_DIR}/cleanup.log"
+    shell:
+        """
+        echo "Cleaning up temporary files" > {output.cleanup_log}
+        echo "Cleanup completed: $(date)" >> {output.cleanup_log}
+        
+        # Add actual cleanup commands here if needed
+        # find {OUTPUT_DIR}/temp -name "*.tmp" -delete
+        """
+
+# Error handling and resume support
+onerror:
+    print(f"Pipeline failed for experiment {EXPERIMENT_ID}")
+    print("Check logs in {LOG_DIR} for details")
+    print("Resume with: snakemake --rerun-incomplete")
+
+onsuccess:
+    print(f"Pipeline completed successfully for experiment {EXPERIMENT_ID}")
+    print(f"Results available in: {OUTPUT_DIR}")
+    print(f"Report: {OUTPUT_DIR}/reports/experiment_summary.html")
+
+# Resource allocation functions
+def get_mem_mb(wildcards, attempt):
+    """Get memory allocation with increasing retry attempts"""
+    base_mem = 4000
+    return base_mem * (2 ** (attempt - 1))
+
+def get_time_min(wildcards, attempt):
+    """Get time allocation with increasing retry attempts"""
+    base_time = 60
+    return min(base_time * (2 ** (attempt - 1)), 1440)  # Max 24 hours
+
+# Resume-friendly intermediate file handling
+temp_files = [
+    f"{OUTPUT_DIR}/temp/{{sample}}_intermediate.txt"
+]
+
+# Benchmark tracking
+benchmark_files = [
+    f"{OUTPUT_DIR}/benchmarks/{{rule}}_{{sample}}.txt"
+]