Skip to content
2 changes: 1 addition & 1 deletion apps/prep/tests/test_pipeline.py
Original file line number Diff line number Diff line change
Expand Up @@ -309,7 +309,7 @@ async def fake_create(*args, **kwargs):
normalized = ctx.output_dir / "input.normalized.fasta"
assert seen_inputs["embed"] == str(normalized)
assert seen_inputs["annotate"] == str(normalized)
assert ">P12345" in normalized.read_text()
assert normalized.read_text() == ">P12345\nMAAAAAA\n"


async def test_embed_failure_with_connection_refused_is_classified_as_biocentral_unavailable(
Expand Down
12 changes: 6 additions & 6 deletions apps/protspace/docs/annotations.md
Original file line number Diff line number Diff line change
Expand Up @@ -192,12 +192,12 @@ Structure-based domain annotations from [TED (The Encyclopedia of Domains)](http

Per-protein predictions from the [Biocentral API](https://biocentral.rostlab.org/) using pre-trained models. Requires protein sequences (fetched automatically from UniProt).

| Name | Model | Description |
| -------------------------------- | ------------------------------------ | ------------------------------------------ |
| `predicted_subcellular_location` | LightAttention | 10-class subcellular localization |
| `predicted_membrane` | LightAttention | Membrane / Soluble |
| `predicted_signal_peptide` | TMbed | True / False (derived from topology) |
| `predicted_transmembrane` | TMbed | none / alpha-helical / beta-barrel |
| Name | Model | Description |
| -------------------------------- | ------------------------------------ | ----------------------------------------------- |
| `predicted_subcellular_location` | LightAttention | 10-class subcellular localization |
| `predicted_membrane` | LightAttention | Membrane / Soluble |
| `predicted_signal_peptide` | TMbed | True / False (derived from topology) |
| `predicted_transmembrane` | TMbed | non-transmembrane / alpha-helical / beta-barrel |

**Data source**: Batch predictions via Biocentral API (`api.predict()`). TMbed provides per-residue topology labels (`H`=TM helix, `B`=TM beta strand, `S`=signal peptide); signal peptide and transmembrane type are summarized from these labels.

Expand Down
20 changes: 17 additions & 3 deletions apps/protspace/src/protspace/cli/annotate.py
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,9 @@ def annotate(
from protspace.data.loaders.h5 import EMBEDDING_EXTENSIONS

# Extract identifiers from input
if is_fasta_file(input):
input_is_fasta = is_fasta_file(input)
sequences = None
if input_is_fasta:
from protspace.data.loaders.query import extract_identifiers_from_fasta

headers = extract_identifiers_from_fasta(input)
Expand All @@ -82,10 +84,10 @@ def annotate(
logger.info(f"Found {len(headers)} protein identifiers")

# Resolve annotation names
from protspace.data.annotations.configuration import AnnotationConfiguration

annotations_list = None
if annotations:
from protspace.data.annotations.configuration import AnnotationConfiguration

names = []
for item in annotations:
for part in item.split(","):
Expand All @@ -95,11 +97,23 @@ def annotate(
if names:
annotations_list = AnnotationConfiguration(names).user_annotations

if input_is_fasta:
config = AnnotationConfiguration(annotations_list)
if config.interpro_annotations or config.biocentral_annotations:
from protspace.data.io.fasta import parse_fasta
from protspace.data.loaders.h5 import parse_identifier

sequences = {
parse_identifier(header): sequence
for header, sequence in parse_fasta(input).items()
}

# Fetch annotations
df = ProteinAnnotationManager(
headers=headers,
annotations=annotations_list,
output_path=None,
sequences=sequences,
).to_pd()

if not scores:
Expand Down
14 changes: 10 additions & 4 deletions apps/protspace/src/protspace/data/annotations/manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -241,16 +241,20 @@ def _fetch_taxonomy(
return {}

def _build_sequence_map(
self, uniprot_annotations: list[ProteinAnnotations]
self,
uniprot_annotations: list[ProteinAnnotations],
*,
prefer_uniprot: bool = False,
) -> dict[str, str]:
"""Build a mapping from headers to sequences.

Merges local sequences (from FASTA, priority) with UniProt results (fallback).
Local FASTA sequences take priority by default. InterPro can request UniProt
priority because its precomputed matches are keyed by canonical sequence hash.
"""
sequences = dict(self.sequences) if self.sequences else {}
for protein in uniprot_annotations:
seq = protein.annotations.get("sequence", "")
if seq and protein.identifier not in sequences:
if seq and (prefer_uniprot or protein.identifier not in sequences):
sequences[protein.identifier] = seq
return sequences

Expand All @@ -262,7 +266,9 @@ def _fetch_interpro(
return []

try:
sequences = self._build_sequence_map(uniprot_annotations)
sequences = self._build_sequence_map(
uniprot_annotations, prefer_uniprot=True
)

retriever = InterProRetriever(
headers=self.headers,
Expand Down
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
"""Biocentral API prediction retriever for per-protein annotations."""

import logging
import re
import warnings

from protspace.data.annotations.retrievers.base_retriever import BaseAnnotationRetriever
Expand All @@ -24,6 +23,7 @@
}

_BATCH_SIZE = 1000
_TMBED_TOPOLOGY_LABELS = frozenset("BbHhSio.")


class BiocentralPredictionRetriever(BaseAnnotationRetriever):
Expand Down Expand Up @@ -202,28 +202,42 @@ def _extract_signal_peptide(predictions: list) -> str:

TMbed labels: S = signal peptide, H/h = TM helix, B/b = TM beta, i/o = non-TM
"""
for pred in predictions:
if pred.model_name == "TMbed":
topology = str(pred.value) if pred.value else ""
return "True" if "S" in topology else "False"
return ""
topology = BiocentralPredictionRetriever._extract_tmbed_topology(predictions)
if topology is None:
return ""
return "True" if "S" in topology else "False"

@staticmethod
def _extract_transmembrane(predictions: list) -> str:
"""Derive transmembrane type from TMbed per-residue output.

Returns: 'alpha-helical', 'beta-barrel', or 'none'
Returns: 'alpha-helical', 'beta-barrel', or 'non-transmembrane'
"""
topology = BiocentralPredictionRetriever._extract_tmbed_topology(predictions)
if topology is None:
return ""

has_helix = "H" in topology or "h" in topology
has_beta = "B" in topology or "b" in topology
if has_helix and has_beta:
return "alpha-helical;beta-barrel"
elif has_helix:
return "alpha-helical"
elif has_beta:
return "beta-barrel"
return "non-transmembrane"

@staticmethod
def _extract_tmbed_topology(predictions: list) -> str | None:
"""Return a supported TMbed topology, or ``None`` when unavailable."""
for pred in predictions:
if pred.model_name == "TMbed":
topology = str(pred.value) if pred.value else ""
has_helix = bool(re.search(r"[Hh]", topology))
has_beta = bool(re.search(r"[Bb]", topology))
if has_helix and has_beta:
return "alpha-helical;beta-barrel"
elif has_helix:
return "alpha-helical"
elif has_beta:
return "beta-barrel"
return "none"
return ""
value = pred.value
if (
not isinstance(value, str)
or not value
or not set(value) <= _TMBED_TOPOLOGY_LABELS
):
return None
return value
return None
88 changes: 88 additions & 0 deletions apps/protspace/tests/test_annotate_cli.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,88 @@
"""Tests for the standalone ``protspace annotate`` command."""

import pandas as pd
from typer.testing import CliRunner

from protspace.cli.app import app


def test_fasta_sequences_are_passed_to_annotation_manager(tmp_path, monkeypatch):
"""FASTA-only proteins must reach sequence-backed annotation sources."""
import protspace.data.annotations.manager as manager_module

fasta = tmp_path / "input.fasta"
fasta.write_text(
">sp|P12345|KNOWN Known protein\nMKV\n>custom-protein description\nAAAG\n"
)
captured: dict[str, object] = {}

class FakeManager:
def __init__(self, *args, **kwargs):
captured.update(kwargs)

def to_pd(self):
return pd.DataFrame({"identifier": captured["headers"]})

monkeypatch.setattr(manager_module, "ProteinAnnotationManager", FakeManager)

output = tmp_path / "annotations.parquet"
result = CliRunner().invoke(
app,
[
"annotate",
"-i",
str(fasta),
"-a",
"biocentral",
"-o",
str(output),
],
)

assert result.exit_code == 0, result.output
assert captured["headers"] == ["P12345", "custom-protein"]
assert captured["sequences"] == {
"P12345": "MKV",
"custom-protein": "AAAG",
}


def test_fasta_is_not_parsed_for_uniprot_only_annotations(tmp_path, monkeypatch):
"""Sequence-free sources must not materialize an unused FASTA sequence map."""
import protspace.data.annotations.manager as manager_module
import protspace.data.io.fasta as fasta_module

fasta = tmp_path / "input.fasta"
fasta.write_text(">sp|P12345|KNOWN Known protein\nMKV\n")
captured: dict[str, object] = {}

class FakeManager:
def __init__(self, *args, **kwargs):
captured.update(kwargs)

def to_pd(self):
return pd.DataFrame({"identifier": captured["headers"]})

def fail_if_parsed(*args, **kwargs):
raise AssertionError("parse_fasta must not run for UniProt-only annotations")

monkeypatch.setattr(manager_module, "ProteinAnnotationManager", FakeManager)
monkeypatch.setattr(fasta_module, "parse_fasta", fail_if_parsed)

output = tmp_path / "annotations.parquet"
result = CliRunner().invoke(
app,
[
"annotate",
"-i",
str(fasta),
"-a",
"uniprot",
"-o",
str(output),
],
)

assert result.exit_code == 0, result.output
assert captured["headers"] == ["P12345"]
assert captured["sequences"] is None
32 changes: 32 additions & 0 deletions apps/protspace/tests/test_annotation_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -439,6 +439,38 @@ def test_transform_protein_families_with_semicolon(self):
class TestIntegration:
"""Integration tests for complete workflows."""

@patch("src.protspace.data.annotations.manager.BiocentralPredictionRetriever")
@patch("src.protspace.data.annotations.manager.InterProRetriever")
def test_sequence_precedence_is_source_specific(
self, mock_interpro_retriever, mock_biocentral_retriever
):
"""InterPro hashes canonical sequences; Biocentral predicts submitted ones."""
manager = ProteinAnnotationManager(
headers=["P12345", "custom-protein"],
annotations=["pfam", "predicted_transmembrane"],
sequences={"P12345": "LOCAL", "custom-protein": "CUSTOM"},
)
uniprot_annotations = [
ProteinAnnotations(
identifier="P12345", annotations={"sequence": "CANONICAL"}
),
ProteinAnnotations(identifier="custom-protein", annotations={}),
]
mock_interpro_retriever.return_value.fetch_annotations.return_value = []
mock_biocentral_retriever.return_value.fetch_annotations.return_value = []

manager._fetch_interpro(uniprot_annotations, [])
manager._fetch_biocentral(uniprot_annotations, [])

assert mock_interpro_retriever.call_args.kwargs["sequences"] == {
"P12345": "CANONICAL",
"custom-protein": "CUSTOM",
}
assert mock_biocentral_retriever.call_args.kwargs["sequences"] == {
"P12345": "LOCAL",
"custom-protein": "CUSTOM",
}

@patch("src.protspace.data.annotations.manager.TaxonomyRetriever")
@patch("src.protspace.data.annotations.manager.UniProtRetriever")
def test_to_pd_complete_workflow(
Expand Down
Loading
Loading