| """
|
| FASTQ Processing Pipeline
|
| Quality control and preprocessing of sequencing data
|
| """
|
|
|
| from pathlib import Path
|
| from typing import Dict, List, Optional
|
| import yaml
|
| import logging
|
| from Bio import SeqIO
|
| from Bio.SeqIO.QualityIO import FastqGeneralIterator
|
|
|
| logging.basicConfig(level=logging.INFO)
|
| logger = logging.getLogger(__name__)
|
|
|
|
|
| class FASTQProcessor:
|
| """Process FASTQ sequencing files"""
|
|
|
| def __init__(self, config_path: str = "config.yml"):
|
| with open(config_path, 'r') as f:
|
| self.config = yaml.safe_load(f)['pipeline']['fastq']
|
|
|
| self.quality_threshold = self.config['quality_threshold']
|
| self.min_length = self.config['min_length']
|
| self.output_dir = Path(self.config['output_dir'])
|
| self.output_dir.mkdir(parents=True, exist_ok=True)
|
|
|
| def quality_filter(
|
| self,
|
| input_file: Path,
|
| output_file: Optional[Path] = None
|
| ) -> Dict:
|
| """
|
| Filter FASTQ reads by quality score
|
|
|
| Args:
|
| input_file: Input FASTQ file
|
| output_file: Output filtered FASTQ file
|
|
|
| Returns:
|
| Statistics dictionary
|
| """
|
| if output_file is None:
|
| output_file = self.output_dir / f"{input_file.stem}_filtered.fastq"
|
|
|
| stats = {
|
| 'total_reads': 0,
|
| 'passed_reads': 0,
|
| 'failed_reads': 0,
|
| 'total_bases': 0,
|
| 'passed_bases': 0
|
| }
|
|
|
| try:
|
| with open(input_file, 'r') as in_f, open(output_file, 'w') as out_f:
|
| for title, sequence, quality in FastqGeneralIterator(in_f):
|
| stats['total_reads'] += 1
|
| stats['total_bases'] += len(sequence)
|
|
|
|
|
| quality_scores = [ord(q) - 33 for q in quality]
|
| avg_quality = sum(quality_scores) / len(quality_scores)
|
|
|
|
|
| if avg_quality >= self.quality_threshold and len(sequence) >= self.min_length:
|
| out_f.write(f"@{title}\n{sequence}\n+\n{quality}\n")
|
| stats['passed_reads'] += 1
|
| stats['passed_bases'] += len(sequence)
|
| else:
|
| stats['failed_reads'] += 1
|
|
|
| stats['pass_rate'] = stats['passed_reads'] / stats['total_reads'] if stats['total_reads'] > 0 else 0
|
|
|
| logger.info(f"Filtered {input_file.name}: {stats['passed_reads']}/{stats['total_reads']} reads passed")
|
| return stats
|
|
|
| except Exception as e:
|
| logger.error(f"Error filtering FASTQ: {e}")
|
| return stats
|
|
|
| def trim_adapters(
|
| self,
|
| input_file: Path,
|
| adapter_sequence: str,
|
| output_file: Optional[Path] = None
|
| ) -> Path:
|
| """
|
| Trim adapter sequences from reads
|
|
|
| Args:
|
| input_file: Input FASTQ file
|
| adapter_sequence: Adapter sequence to trim
|
| output_file: Output trimmed file
|
| """
|
| if output_file is None:
|
| output_file = self.output_dir / f"{input_file.stem}_trimmed.fastq"
|
|
|
| trimmed_count = 0
|
|
|
| try:
|
| with open(input_file, 'r') as in_f, open(output_file, 'w') as out_f:
|
| for title, sequence, quality in FastqGeneralIterator(in_f):
|
|
|
| adapter_pos = sequence.find(adapter_sequence)
|
|
|
| if adapter_pos != -1:
|
|
|
| sequence = sequence[:adapter_pos]
|
| quality = quality[:adapter_pos]
|
| trimmed_count += 1
|
|
|
| if len(sequence) >= self.min_length:
|
| out_f.write(f"@{title}\n{sequence}\n+\n{quality}\n")
|
|
|
| logger.info(f"Trimmed adapters from {trimmed_count} reads")
|
| return output_file
|
|
|
| except Exception as e:
|
| logger.error(f"Error trimming adapters: {e}")
|
| return input_file
|
|
|
| def calculate_statistics(self, fastq_file: Path) -> Dict:
|
| """
|
| Calculate statistics for FASTQ file
|
|
|
| Returns:
|
| Dictionary with read count, length distribution, quality scores
|
| """
|
| stats = {
|
| 'total_reads': 0,
|
| 'total_bases': 0,
|
| 'min_length': float('inf'),
|
| 'max_length': 0,
|
| 'avg_length': 0,
|
| 'avg_quality': 0,
|
| 'gc_content': 0
|
| }
|
|
|
| lengths = []
|
| qualities = []
|
| gc_count = 0
|
|
|
| try:
|
| with open(fastq_file, 'r') as f:
|
| for title, sequence, quality in FastqGeneralIterator(f):
|
| stats['total_reads'] += 1
|
| seq_len = len(sequence)
|
| stats['total_bases'] += seq_len
|
|
|
| lengths.append(seq_len)
|
| stats['min_length'] = min(stats['min_length'], seq_len)
|
| stats['max_length'] = max(stats['max_length'], seq_len)
|
|
|
|
|
| quality_scores = [ord(q) - 33 for q in quality]
|
| qualities.extend(quality_scores)
|
|
|
|
|
| gc_count += sequence.count('G') + sequence.count('C')
|
|
|
| if stats['total_reads'] > 0:
|
| stats['avg_length'] = sum(lengths) / len(lengths)
|
| stats['avg_quality'] = sum(qualities) / len(qualities)
|
| stats['gc_content'] = (gc_count / stats['total_bases']) * 100
|
|
|
| return stats
|
|
|
| except Exception as e:
|
| logger.error(f"Error calculating statistics: {e}")
|
| return stats
|
|
|
| def convert_to_fasta(
|
| self,
|
| input_file: Path,
|
| output_file: Optional[Path] = None
|
| ) -> Path:
|
| """Convert FASTQ to FASTA format"""
|
| if output_file is None:
|
| output_file = self.output_dir / f"{input_file.stem}.fasta"
|
|
|
| try:
|
| count = SeqIO.convert(str(input_file), "fastq", str(output_file), "fasta")
|
| logger.info(f"Converted {count} sequences to FASTA")
|
| return output_file
|
|
|
| except Exception as e:
|
| logger.error(f"Error converting to FASTA: {e}")
|
| return input_file
|
|
|
|
|
| class FASTQQualityControl:
|
| """Quality control analysis for FASTQ files"""
|
|
|
| def __init__(self):
|
| self.processor = FASTQProcessor()
|
|
|
| def run_qc(self, fastq_file: Path) -> Dict:
|
| """
|
| Run comprehensive QC on FASTQ file
|
|
|
| Returns:
|
| QC report dictionary
|
| """
|
| report = {
|
| 'file': str(fastq_file),
|
| 'statistics': {},
|
| 'quality_check': 'PASS',
|
| 'warnings': []
|
| }
|
|
|
|
|
| stats = self.processor.calculate_statistics(fastq_file)
|
| report['statistics'] = stats
|
|
|
|
|
| if stats['avg_quality'] < 20:
|
| report['warnings'].append('Low average quality score')
|
| report['quality_check'] = 'WARN'
|
|
|
| if stats['avg_length'] < 50:
|
| report['warnings'].append('Short average read length')
|
| report['quality_check'] = 'WARN'
|
|
|
| if stats['gc_content'] < 30 or stats['gc_content'] > 70:
|
| report['warnings'].append(f'Unusual GC content: {stats["gc_content"]:.1f}%')
|
|
|
| return report
|
|
|
| def generate_qc_report(self, fastq_files: List[Path]) -> Dict:
|
| """Generate QC report for multiple FASTQ files"""
|
| reports = {}
|
|
|
| for fastq_file in fastq_files:
|
| report = self.run_qc(fastq_file)
|
| reports[fastq_file.name] = report
|
|
|
|
|
| summary = {
|
| 'total_files': len(fastq_files),
|
| 'passed': sum(1 for r in reports.values() if r['quality_check'] == 'PASS'),
|
| 'warnings': sum(1 for r in reports.values() if r['quality_check'] == 'WARN'),
|
| 'failed': sum(1 for r in reports.values() if r['quality_check'] == 'FAIL')
|
| }
|
|
|
| return {
|
| 'summary': summary,
|
| 'file_reports': reports
|
| }
|
|
|
