Skip to content

fasta_lake.contaminants#

fasta_lake.contaminants #

Contaminant handling (v5.1).

Common proteomics contaminants — keratins, trypsin, BSA, caseins, etc. — must be tagged explicitly so they're identifiable in search results and can be excluded from target/entrapment FDR accounting.

This module provides:

  • A bundled list of common contaminant UniProt accessions (DEFAULT_CONTAMINANTS).
  • tag_contaminants(): rewrite FASTA headers with CON_ prefix where needed.
  • is_contaminant(): O(1) check from protein accession.
  • strip_contaminants_from_fdr(): exclusion helper for FDR tables.

The bundled list is NOT a full cRAP — it's a minimal conservative set (~50 accessions). Users needing the full cRAP should download it from thegpm.org and pass via --contaminant-db.

All contaminant lookups are by accession, not by sequence. This means a contaminant from one source (e.g. SwissProt P00761 trypsin) is correctly tagged even if the same sequence appears in another source.

ContaminantDB dataclass #

In-memory contaminant accession database.

Source code in fasta_lake/contaminants.py 
 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
@dataclass
class ContaminantDB:
    """In-memory contaminant accession database."""

    accessions: set[str] = field(default_factory=set)
    prefix: str = "CON_"

    def __post_init__(self) -> None:
        """Populate the default contaminant accessions when none were supplied."""
        if not self.accessions:
            self.accessions = set(DEFAULT_CONTAMINANTS)

    def is_contaminant(self, protein_id: str) -> bool:
        """True if this protein ID corresponds to a contaminant.

        Matches:
          - exact accession (e.g., ``P00761``)
          - SwissProt-style ``sp|ACC|NAME`` (extracts ACC)
          - already-tagged ``CON_ACC``
        """
        if protein_id.startswith(self.prefix):
            return True
        acc = _extract_accession(protein_id)
        return acc in self.accessions

    def load(self, path: str | Path) -> None:
        """Load additional accessions from a text file (one per line)."""
        p = Path(path)
        if not p.exists():
            return
        for line in p.read_text().splitlines():
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            self.accessions.add(line)

    def load_from_fasta(self, fasta: str | Path) -> None:
        """Extract accessions from a FASTA file."""
        p = Path(fasta)
        if not p.exists():
            return
        with open(p) as f:
            for line in f:
                if line.startswith(">"):
                    acc = _extract_accession(line[1:].split()[0])
                    if acc:
                        self.accessions.add(acc)

__post_init__ #

__post_init__() -> None

Populate the default contaminant accessions when none were supplied.

Source code in fasta_lake/contaminants.py 
100
101
102
103
def __post_init__(self) -> None:
    """Populate the default contaminant accessions when none were supplied."""
    if not self.accessions:
        self.accessions = set(DEFAULT_CONTAMINANTS)

is_contaminant #

is_contaminant(protein_id: str) -> bool

True if this protein ID corresponds to a contaminant.

Matches: - exact accession (e.g., P00761) - SwissProt-style sp|ACC|NAME (extracts ACC) - already-tagged CON_ACC

Source code in fasta_lake/contaminants.py 
105
106
107
108
109
110
111
112
113
114
115
116
def is_contaminant(self, protein_id: str) -> bool:
    """True if this protein ID corresponds to a contaminant.

    Matches:
      - exact accession (e.g., ``P00761``)
      - SwissProt-style ``sp|ACC|NAME`` (extracts ACC)
      - already-tagged ``CON_ACC``
    """
    if protein_id.startswith(self.prefix):
        return True
    acc = _extract_accession(protein_id)
    return acc in self.accessions

load #

load(path: str | Path) -> None

Load additional accessions from a text file (one per line).

Source code in fasta_lake/contaminants.py 
118
119
120
121
122
123
124
125
126
127
def load(self, path: str | Path) -> None:
    """Load additional accessions from a text file (one per line)."""
    p = Path(path)
    if not p.exists():
        return
    for line in p.read_text().splitlines():
        line = line.strip()
        if not line or line.startswith("#"):
            continue
        self.accessions.add(line)

load_from_fasta #

load_from_fasta(fasta: str | Path) -> None

Extract accessions from a FASTA file.

Source code in fasta_lake/contaminants.py 
129
130
131
132
133
134
135
136
137
138
139
def load_from_fasta(self, fasta: str | Path) -> None:
    """Extract accessions from a FASTA file."""
    p = Path(fasta)
    if not p.exists():
        return
    with open(p) as f:
        for line in f:
            if line.startswith(">"):
                acc = _extract_accession(line[1:].split()[0])
                if acc:
                    self.accessions.add(acc)

tag_contaminants #

tag_contaminants(proteins: dict[str, str], db: ContaminantDB | None = None) -> tuple[dict[str, str], list[str]]

Rewrite protein IDs with CON_ prefix where they match contaminants.

Returns (tagged_proteins, tagged_accessions).

Source code in fasta_lake/contaminants.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def tag_contaminants(
    proteins: dict[str, str],
    db: ContaminantDB | None = None,
) -> tuple[dict[str, str], list[str]]:
    """Rewrite protein IDs with ``CON_`` prefix where they match contaminants.

    Returns (tagged_proteins, tagged_accessions).
    """
    if db is None:
        db = ContaminantDB()
    tagged: dict[str, str] = {}
    contam_ids: list[str] = []
    for pid, seq in proteins.items():
        if db.is_contaminant(pid):
            new_id = pid if pid.startswith(db.prefix) else f"{db.prefix}{pid}"
            tagged[new_id] = seq
            contam_ids.append(new_id)
        else:
            tagged[pid] = seq
    return tagged, contam_ids

is_contaminant_id #

is_contaminant_id(protein_id: str, db: ContaminantDB | None = None) -> bool

Check if a protein_id is a contaminant. Cheap wrapper for FDR filters.

Source code in fasta_lake/contaminants.py 
193
194
195
196
197
def is_contaminant_id(protein_id: str, db: ContaminantDB | None = None) -> bool:
    """Check if a protein_id is a contaminant. Cheap wrapper for FDR filters."""
    if db is None:
        db = ContaminantDB()
    return db.is_contaminant(protein_id)

exclude_contaminants #

exclude_contaminants(rows: Iterable[dict], protein_field: str = 'proteins', db: ContaminantDB | None = None) -> list[dict]

Filter out PSM rows whose protein is a contaminant.

For use in FDR calculation — contaminants should not count as targets or entrapment.

Source code in fasta_lake/contaminants.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
def exclude_contaminants(
    rows: Iterable[dict],
    protein_field: str = "proteins",
    db: ContaminantDB | None = None,
) -> list[dict]:
    """Filter out PSM rows whose protein is a contaminant.

    For use in FDR calculation — contaminants should not count as targets
    or entrapment.
    """
    if db is None:
        db = ContaminantDB()
    out = []
    for r in rows:
        prot = r.get(protein_field, "")
        # Header may contain multiple proteins separated by ; or ,
        first = prot.split(";")[0].split(",")[0].strip()
        if db.is_contaminant(first):
            continue
        out.append(r)
    return out