Spaces:
Sleeping
Sleeping
File size: 6,771 Bytes
16e4ad5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 | """Data parsing and validation for the Pigeon Pea Pangenome Atlas."""
import re
import pandas as pd
import numpy as np
from pathlib import Path
from collections import Counter
from src.utils import logger, timer
@timer
def load_pav(path: str) -> pd.DataFrame:
"""
Load 89_line_PAV.txt.
Returns DataFrame: index=gene_id (str), columns=line_ids (str), values=int {0,1}.
"""
df = pd.read_csv(path, sep="\t", index_col=0)
df.index.name = "gene"
df.index = df.index.astype(str)
df.columns = df.columns.astype(str)
# Validate all values are 0 or 1
unique_vals = set(df.values.flatten())
assert unique_vals.issubset({0, 1}), f"PAV contains values other than 0/1: {unique_vals - {0, 1}}"
logger.info(f"PAV matrix loaded: {df.shape[0]} genes x {df.shape[1]} lines")
return df
@timer
def parse_gff_genes(path: str) -> pd.DataFrame:
"""
Parse GFF3; keep only feature == 'gene' rows.
Returns DataFrame: gene_id, contig_id, start, end, strand.
"""
records = []
with open(path, "r") as f:
for line in f:
if line.startswith("#"):
continue
parts = line.strip().split("\t")
if len(parts) < 9:
continue
if parts[2] != "gene":
continue
contig_id = parts[0]
start = int(parts[3])
end = int(parts[4])
strand = parts[6]
attrs = parts[8]
# Extract gene_id from attributes: ID=<value>
gene_id = None
for attr in attrs.split(";"):
attr = attr.strip()
if attr.startswith("ID="):
gene_id = attr[3:]
break
if gene_id:
records.append({
"gene_id": gene_id,
"contig_id": contig_id,
"start": start,
"end": end,
"strand": strand,
})
df = pd.DataFrame(records)
logger.info(f"GFF parsed: {len(df)} genes on {df['contig_id'].nunique()} contigs")
return df
@timer
def parse_protein_fasta(path: str) -> pd.DataFrame:
"""
Returns DataFrame: gene_id, protein_length, aa_composition (dict as string).
gene_id = header token after '>' up to first whitespace.
"""
records = []
current_id = None
current_seq = []
def flush():
if current_id and current_seq:
seq = "".join(current_seq).replace("*", "")
length = len(seq)
counts = Counter(seq)
total = max(length, 1)
top_aas = sorted(counts.items(), key=lambda x: -x[1])[:5]
comp_str = ", ".join(f"{aa}:{count/total*100:.1f}%" for aa, count in top_aas)
records.append({
"gene_id": current_id,
"protein_length": length,
"composition_summary": comp_str,
})
with open(path, "r") as f:
for line in f:
line = line.strip()
if line.startswith(">"):
flush()
current_id = line[1:].split()[0]
current_seq = []
else:
current_seq.append(line)
flush()
df = pd.DataFrame(records)
logger.info(f"Protein FASTA parsed: {len(df)} proteins")
return df
@timer
def build_contig_index(path: str) -> dict:
"""
Returns dict: {contig_id: length}.
Sequential scan of FASTA headers and sequences.
"""
contig_index = {}
current_contig = None
current_len = 0
with open(path, "r") as f:
for line in f:
if line.startswith(">"):
if current_contig is not None:
contig_index[current_contig] = current_len
current_contig = line[1:].strip().split()[0]
current_len = 0
else:
current_len += len(line.strip())
if current_contig is not None:
contig_index[current_contig] = current_len
logger.info(f"Contig index built: {len(contig_index)} contigs")
return contig_index
def build_contig_name_mapping(gff_genes: pd.DataFrame, contig_index: dict) -> dict:
"""
Build mapping from GFF contig IDs to FASTA contig IDs.
Strategy: exact match first, then substring match on accession tokens.
Returns dict: {gff_contig_id: fasta_contig_id}
"""
gff_contigs = set(gff_genes["contig_id"].unique())
fasta_contigs = set(contig_index.keys())
mapping = {}
# Exact match
for gc in gff_contigs:
if gc in fasta_contigs:
mapping[gc] = gc
# For unmatched, try accession-based matching
unmatched = gff_contigs - set(mapping.keys())
if unmatched:
# Extract accession-like tokens from FASTA headers (e.g. NC_033813.1)
fasta_accession_map = {}
for fc in fasta_contigs:
# Try to extract RefSeq accession
match = re.search(r'(N[CWZ]_\d+\.\d+)', fc)
if match:
fasta_accession_map[match.group(1)] = fc
for gc in unmatched:
match = re.search(r'(N[CWZ]_\d+\.\d+)', gc)
if match and match.group(1) in fasta_accession_map:
mapping[gc] = fasta_accession_map[match.group(1)]
logger.info(f"Contig mapping: {len(mapping)}/{len(gff_contigs)} GFF contigs matched to FASTA")
return mapping
def validate_joins(pav: pd.DataFrame, gff_genes: pd.DataFrame,
protein_index: pd.DataFrame, contig_index: dict) -> dict:
"""
Returns validation report with coverage percentages and orphan genes.
"""
pav_genes = set(pav.index)
gff_gene_set = set(gff_genes["gene_id"])
protein_gene_set = set(protein_index["gene_id"])
contig_set = set(contig_index.keys())
gff_contig_set = set(gff_genes["contig_id"])
pav_in_gff = pav_genes & gff_gene_set
pav_in_protein = pav_genes & protein_gene_set
gff_contigs_in_fasta = gff_contig_set & contig_set
orphans = pav_genes - (gff_gene_set | protein_gene_set)
report = {
"pav_gene_count": len(pav_genes),
"gff_gene_count": len(gff_gene_set),
"protein_gene_count": len(protein_gene_set),
"pav_genes_in_gff_pct": len(pav_in_gff) / max(len(pav_genes), 1) * 100,
"pav_genes_in_protein_pct": len(pav_in_protein) / max(len(pav_genes), 1) * 100,
"gff_contigs_in_fasta_pct": len(gff_contigs_in_fasta) / max(len(gff_contig_set), 1) * 100,
"orphan_genes_count": len(orphans),
}
if orphans:
logger.warning(f"{len(orphans)} orphan genes (in PAV but missing from both GFF and protein)")
for key, val in report.items():
logger.info(f" {key}: {val}")
return report
|