| | """
|
| | 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
|
| | }
|
| |
|
