Nextflow Development

scripts/utils/validators.py

"""
Samplesheet validation utilities.

Validates samplesheet rows against pipeline configuration before writing,
catching errors early with helpful messages.
"""

import os
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, List, Optional
import yaml


@dataclass
class ValidationResult:
    """Result of samplesheet validation."""
    valid: bool
    errors: List[str] = field(default_factory=list)
    warnings: List[str] = field(default_factory=list)
    suggestions: List[str] = field(default_factory=list)

    def __bool__(self):
        return self.valid

    def summary(self) -> str:
        """Generate human-readable summary."""
        lines = []
        if self.errors:
            lines.append("Errors:")
            for e in self.errors:
                lines.append(f"  - {e}")
        if self.warnings:
            lines.append("Warnings:")
            for w in self.warnings:
                lines.append(f"  - {w}")
        if self.suggestions:
            lines.append("Suggestions:")
            for s in self.suggestions:
                lines.append(f"  - {s}")
        return "\n".join(lines)


def load_pipeline_config(pipeline: str) -> Optional[Dict]:
    """Load pipeline configuration from YAML file."""
    # Find config directory relative to this file
    script_dir = Path(__file__).parent.parent.parent
    config_path = script_dir / "config" / "pipelines" / f"{pipeline}.yaml"

    if not config_path.exists():
        return None

    with open(config_path) as f:
        return yaml.safe_load(f)


def validate_samplesheet(
    rows: List[Dict],
    pipeline: str,
    config: Optional[Dict] = None
) -> ValidationResult:
    """
    Validate samplesheet rows against pipeline requirements.

    Args:
        rows: List of row dictionaries
        pipeline: Pipeline name (e.g., 'rnaseq', 'sarek')
        config: Optional pre-loaded config dict

    Returns:
        ValidationResult with errors, warnings, and suggestions
    """
    errors = []
    warnings = []
    suggestions = []

    # Load config if not provided
    if config is None:
        config = load_pipeline_config(pipeline)

    if config is None:
        errors.append(f"Unknown pipeline: {pipeline}")
        return ValidationResult(valid=False, errors=errors)

    columns = config.get("samplesheet", {}).get("columns", [])
    required_cols = [c["name"] for c in columns if c.get("required", False)]

    if not rows:
        errors.append("Samplesheet is empty - no samples found")
        return ValidationResult(valid=False, errors=errors)

    # Validate each row
    for i, row in enumerate(rows):
        row_num = i + 2  # Account for header row

        # Check required columns
        for col_name in required_cols:
            col_config = next((c for c in columns if c["name"] == col_name), None)

            # Skip columns with conditions that don't apply
            if col_config and "condition" in col_config:
                # Simple condition check - skip for now
                # Full implementation would evaluate conditions
                pass

            if col_name not in row or row[col_name] is None or row[col_name] == "":
                # Check if there's a default
                if col_config and "default" in col_config:
                    continue
                errors.append(f"Row {row_num}: Missing required column '{col_name}'")

        # Validate path columns exist
        for col_name in ["fastq_1", "fastq_2", "bam", "bai"]:
            if col_name in row and row[col_name]:
                path = row[col_name]
                if not os.path.exists(path):
                    errors.append(f"Row {row_num}: File not found: {path}")
                elif not os.path.isfile(path):
                    errors.append(f"Row {row_num}: Not a file: {path}")

        # Validate enum values
        for col_config in columns:
            col_name = col_config["name"]
            if col_name in row and row[col_name] and "allowed" in col_config:
                value = row[col_name]
                allowed = col_config["allowed"]
                if value not in allowed:
                    errors.append(
                        f"Row {row_num}: Invalid value '{value}' for '{col_name}'. "
                        f"Allowed: {allowed}"
                    )

        # Check R1/R2 pairing consistency
        r1 = row.get("fastq_1", "")
        r2 = row.get("fastq_2", "")
        if r1 and not r2:
            warnings.append(f"Row {row_num}: Single-end data (no R2 file)")
        elif r2 and not r1:
            errors.append(f"Row {row_num}: R2 present but R1 missing")

    # Check for duplicate samples
    sample_col = "sample" if "sample" in rows[0] else "patient"
    if sample_col in rows[0]:
        samples = [r.get(sample_col, "") for r in rows]
        duplicates = [s for s in set(samples) if samples.count(s) > 1]
        if duplicates:
            warnings.append(f"Duplicate sample names: {duplicates}")
            suggestions.append(
                "Duplicates may be intentional (multi-lane sequencing). "
                "Verify sample grouping is correct."
            )

    # Pipeline-specific validation
    if pipeline == "sarek":
        _validate_sarek_specific(rows, errors, warnings, suggestions)
    elif pipeline == "atacseq":
        _validate_atacseq_specific(rows, errors, warnings, suggestions)

    return ValidationResult(
        valid=len(errors) == 0,
        errors=errors,
        warnings=warnings,
        suggestions=suggestions
    )


def _validate_sarek_specific(
    rows: List[Dict],
    errors: List[str],
    warnings: List[str],
    suggestions: List[str]
):
    """Sarek-specific validation for tumor/normal pairing."""
    # Group by patient
    patients = {}
    for row in rows:
        patient = row.get("patient", "")
        status = row.get("status")

        if patient not in patients:
            patients[patient] = {"tumor": 0, "normal": 0, "unknown": 0}

        if status == 1:
            patients[patient]["tumor"] += 1
        elif status == 0:
            patients[patient]["normal"] += 1
        else:
            patients[patient]["unknown"] += 1

    # Check pairing
    for patient, counts in patients.items():
        if counts["tumor"] > 0 and counts["normal"] == 0:
            warnings.append(
                f"Patient '{patient}': Tumor sample(s) without matched normal. "
                "Somatic calling works best with paired tumor-normal."
            )
            suggestions.append(
                f"For patient '{patient}': Add a normal sample or use tumor-only mode."
            )

        if counts["unknown"] > 0:
            warnings.append(
                f"Patient '{patient}': {counts['unknown']} sample(s) with unknown status. "
                "Set status column to 0 (normal) or 1 (tumor)."
            )


def _validate_atacseq_specific(
    rows: List[Dict],
    errors: List[str],
    warnings: List[str],
    suggestions: List[str]
):
    """ATAC-seq specific validation for replicates."""
    # Group by sample (condition)
    samples = {}
    for row in rows:
        sample = row.get("sample", "")
        replicate = row.get("replicate", 1)

        if sample not in samples:
            samples[sample] = []

        samples[sample].append(replicate)

    # Check replicates
    for sample, reps in samples.items():
        if len(reps) < 2:
            warnings.append(
                f"Sample '{sample}': Only {len(reps)} replicate(s). "
                "Consensus peaks require 2+ replicates."
            )

        # Check for duplicate replicate numbers
        if len(reps) != len(set(reps)):
            errors.append(
                f"Sample '{sample}': Duplicate replicate numbers detected. "
                "Each replicate must have a unique number."
            )

    # Check all samples have R2 (ATAC-seq requires paired-end)
    for i, row in enumerate(rows):
        if not row.get("fastq_2"):
            errors.append(
                f"Row {i+2}: ATAC-seq requires paired-end data. R2 file missing."
            )


def validate_file_exists(path: str) -> bool:
    """Check if file exists and is accessible."""
    return os.path.isfile(path) and os.access(path, os.R_OK)


def validate_absolute_path(path: str) -> bool:
    """Check if path is absolute."""
    return os.path.isabs(path)