` such as `v1.2.3`
+
+The workflow resolves the Docker repository namespace in this order:
+
+1. `DOCKERHUB_NAMESPACE` repository variable
+2. `DOCKERHUB_USERNAME` repository secret
+3. the lowercased GitHub repository owner as a fallback for build-only runs
+
+Required repository secrets:
+
+- `DOCKERHUB_USERNAME`
+- `DOCKERHUB_TOKEN`
+
+Optional repository variable:
+
+- `DOCKERHUB_NAMESPACE`
+
+If `DOCKERHUB_NAMESPACE` is not set, the workflow publishes to the same namespace as `DOCKERHUB_USERNAME`.
+
## About Scop3P[^1]
**Scop3P: A Comprehensive Resource of Human Phosphosites within Their Full Context**
diff --git a/apps/common/__init__.py b/apps/common/__init__.py
index c77e741..1f40e9f 100644
--- a/apps/common/__init__.py
+++ b/apps/common/__init__.py
@@ -1,5 +1,6 @@
"""Shared services for Shiny app conversions."""
+from .logging_utils import get_logger
from .models import PeptideRow, PeptideSelectionMode
from .services import AlphaFoldService, Scop3PClient
from .peptide_mapper import (
@@ -13,6 +14,7 @@
__all__ = [
"PeptideRow",
"PeptideSelectionMode",
+ "get_logger",
"AlphaFoldService",
"Scop3PClient",
"PeptideMapperService",
diff --git a/apps/common/logging_utils.py b/apps/common/logging_utils.py
new file mode 100644
index 0000000..3a2fa07
--- /dev/null
+++ b/apps/common/logging_utils.py
@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+import logging
+import os
+import sys
+from typing import Any
+
+
+_CONFIGURED = False
+
+
+class _SafeExtraFormatter(logging.Formatter):
+ def format(self, record: logging.LogRecord) -> str:
+ if not hasattr(record, "event"):
+ record.event = "-"
+ return super().format(record)
+
+
+def configure_logging() -> None:
+ global _CONFIGURED
+ if _CONFIGURED:
+ return
+
+ level_name = os.getenv("SCOP3P_LOG_LEVEL", "INFO").upper()
+ level = getattr(logging, level_name, logging.INFO)
+
+ handler = logging.StreamHandler(sys.stdout)
+ handler.setFormatter(
+ _SafeExtraFormatter("%(asctime)s %(levelname)s %(name)s event=%(event)s %(message)s")
+ )
+ logging.basicConfig(level=level, handlers=[handler])
+ _CONFIGURED = True
+
+
+class _EventAdapter(logging.LoggerAdapter):
+ def process(self, msg: str, kwargs: dict[str, Any]) -> tuple[str, dict[str, Any]]:
+ extra = dict(kwargs.get("extra", {}))
+ extra.setdefault("event", "-")
+ kwargs["extra"] = extra
+ return msg, kwargs
+
+
+def get_logger(name: str) -> logging.LoggerAdapter:
+ configure_logging()
+ return _EventAdapter(logging.getLogger(name), {})
diff --git a/apps/common/mutation_effect.py b/apps/common/mutation_effect.py
index 1ec33a2..e9cca75 100644
--- a/apps/common/mutation_effect.py
+++ b/apps/common/mutation_effect.py
@@ -60,34 +60,11 @@ def fetch_uniprot_sequence(self, accession: str) -> str:
return sequence
def fetch_scop3p_modifications(self, accession: str) -> pd.DataFrame:
- response = requests.get(
- f"{self.scop3p_client.base_url}/modifications",
- params={"accession": accession},
- headers={"accept": "application/json"},
- timeout=self.timeout,
- )
- response.raise_for_status()
- payload = response.json()
- if isinstance(payload, list):
- payload = payload[0] if payload else {}
-
- dataframe = pd.DataFrame(payload.get("modifications", []))
+ dataframe = self.scop3p_client.fetch_modifications(accession)
if dataframe.empty:
return dataframe
- keep = [
- column
- for column in [
- "position",
- "residue",
- "name",
- "source",
- "evidence",
- "reference",
- "functionalScore",
- ]
- if column in dataframe.columns
- ]
+ keep = [column for column in dataframe.columns if column != "specificSinglyPhosphorylated"]
dataframe = dataframe[keep].copy()
dataframe["position"] = pd.to_numeric(dataframe["position"], errors="coerce")
dataframe = dataframe.dropna(subset=["position"])
@@ -102,19 +79,12 @@ def predict_biophysical(self, accession: str, sequence: str) -> dict:
handle.write(f">{accession}\n{sequence}\n")
handle.flush()
predictor = SingleSeq(handle.name)
- if constants is not None:
- tool_candidates = []
- for name in (
- "TOOL_BACKBONE_DYNAMICS",
- "TOOL_DYNAMINE",
- "TOOL_DISOMINE",
- "TOOL_EFOLDMINE",
- ):
- if hasattr(constants, name):
- tool_candidates.append(getattr(constants, name))
- if tool_candidates:
- return predictor.predict(tools=tool_candidates).get_all_predictions()
- return predictor.predict().get_all_predictions()
+
+ predictor.predict(
+ tools=[constants.TOOL_DYNAMINE, constants.TOOL_DISOMINE, constants.TOOL_EFOLDMINE]
+ )
+
+ return predictor.get_all_predictions()
@staticmethod
def prediction_to_df(prediction: dict, accession: str) -> pd.DataFrame:
@@ -135,14 +105,21 @@ def prediction_to_df(prediction: dict, accession: str) -> pd.DataFrame:
)
dataframe = pd.DataFrame(
{
+ "seq": protein.get("seq", [None] * size),
"seqpos": list(range(1, size + 1)),
"backbone": protein.get("backbone", [None] * size),
+ "sidechain": protein.get("sidechain", [None] * size),
+ "helix": protein.get("helix", [None] * size),
+ "sheet": protein.get("sheet", [None] * size),
+ "coil": protein.get("coil", [None] * size),
+ "ppII": protein.get("ppII", [None] * size),
"disoMine": protein.get("disoMine", [None] * size),
"earlyFolding": protein.get("earlyFolding", [None] * size),
}
)
- for column in ["backbone", "disoMine", "earlyFolding"]:
+ for column in ["backbone", "sidechain", "helix", "sheet", "coil", "ppII", "disoMine", "earlyFolding"]:
dataframe[column] = pd.to_numeric(dataframe[column], errors="coerce")
+
return dataframe
@staticmethod
@@ -519,6 +496,7 @@ def scrollable_table_html(
for index in range(1, sticky_cols + 1):
sticky_css.append(
f"""
+#{table_id} table th {{ text-align:center;}},
#{table_id} table th:nth-child({index}),
#{table_id} table td:nth-child({index}) {{
position: sticky;
diff --git a/apps/common/services.py b/apps/common/services.py
index 6cdbcbd..0257bfa 100644
--- a/apps/common/services.py
+++ b/apps/common/services.py
@@ -5,7 +5,7 @@
import urllib.request
import pandas as pd
-import requests
+from scop3p_api_client.api import Scop3pRestApi
class Scop3PClient:
@@ -14,13 +14,10 @@ class Scop3PClient:
def __init__(self, base_url: str = "https://iomics.ugent.be/scop3p/api", timeout: int = 30) -> None:
self.base_url = base_url.rstrip("/")
self.timeout = timeout
+ self.api = Scop3pRestApi(default_timeout=timeout)
def fetch_peptides_modifications(self, accession: str) -> pd.DataFrame:
- url = f"{self.base_url}/get-peptides-modifications"
- response = requests.get(url, params={"accession": accession}, timeout=self.timeout)
- response.raise_for_status()
- payload = response.json()
-
+ payload = self.api.fetch_peptides(accession)
df = pd.DataFrame(payload.get("peptides", []))
if df.empty:
return df
@@ -36,6 +33,34 @@ def fetch_peptides_modifications(self, accession: str) -> pd.DataFrame:
df["label"] = df.apply(self._format_label, axis=1)
return df
+ def fetch_modifications(self, accession: str) -> pd.DataFrame:
+ payload = self.api.fetch_modifications(accession)
+ if isinstance(payload, list):
+ payload = payload[0] if payload else {}
+
+ dataframe = pd.DataFrame(payload.get("modifications", []))
+ if dataframe.empty:
+ return dataframe
+
+ keep = [
+ column
+ for column in [
+ "position",
+ "residue",
+ "name",
+ "source",
+ "evidence",
+ "reference",
+ "functionalScore",
+ "specificSinglyPhosphorylated",
+ ]
+ if column in dataframe.columns
+ ]
+ dataframe = dataframe[keep].copy()
+ if "position" in dataframe.columns:
+ dataframe["position"] = pd.to_numeric(dataframe["position"], errors="coerce").astype("Int64")
+ return dataframe
+
@staticmethod
def _format_label(row: pd.Series) -> str:
peptide_sequence = str(row.get("peptideSequence", ""))
diff --git a/apps/common/structure_viz.py b/apps/common/structure_viz.py
index 649111a..9d0febe 100644
--- a/apps/common/structure_viz.py
+++ b/apps/common/structure_viz.py
@@ -16,37 +16,33 @@
from b2bTools import SingleSeq, constants
from scipy.spatial import KDTree
+from .services import Scop3PClient
+
+
+B2B_METRIC_COLUMNS = (
+ "backbone",
+ "sidechain",
+ "ppII",
+ "coil",
+ "sheet",
+ "helix",
+ "earlyFolding",
+ "disoMine",
+)
+B2B_NORMALIZED_SUFFIX = "_normalized"
+
class StructureVizService:
def __init__(self, workdir: Path, timeout: int = 60) -> None:
self.workdir = workdir
self.timeout = timeout
+ self.scop3p_client = Scop3PClient(timeout=timeout)
self.workdir.mkdir(parents=True, exist_ok=True)
def fetch_ptms(self, accession: str) -> pd.DataFrame:
- url = "https://iomics.ugent.be/scop3p/api/modifications"
- response = requests.get(url, params={"accession": accession}, headers={"accept": "application/json"}, timeout=self.timeout)
- response.raise_for_status()
- payload = response.json()
- if isinstance(payload, list):
- payload = payload[0] if payload else {}
- dataframe = pd.DataFrame(payload.get("modifications", []))
+ dataframe = self.scop3p_client.fetch_modifications(accession)
if dataframe.empty:
return dataframe
- columns = [
- "residue",
- "name",
- "evidence",
- "position",
- "source",
- "reference",
- "functionalScore",
- "specificSinglyPhosphorylated",
- ]
- keep = [column for column in columns if column in dataframe.columns]
- dataframe = dataframe[keep].copy()
- if "position" in dataframe.columns:
- dataframe["position"] = pd.to_numeric(dataframe["position"], errors="coerce").astype("Int64")
return dataframe
def fetch_variants(self, accession: str) -> pd.DataFrame:
@@ -90,14 +86,94 @@ def predict_b2b(self, accession: str, sequence: str) -> pd.DataFrame:
fasta_file.flush()
predictor = SingleSeq(fasta_file.name)
tools = []
- for name in ["TOOL_BACKBONE_DYNAMICS", "TOOL_DYNAMINE", "TOOL_DISOMINE", "TOOL_EFOLDMINE"]:
+ for name in ["TOOL_DYNAMINE", "TOOL_DISOMINE", "TOOL_EFOLDMINE"]:
if hasattr(constants, name):
tools.append(getattr(constants, name))
prediction = predictor.predict(tools=tools).get_all_predictions() if tools else predictor.predict().get_all_predictions()
+
protein = prediction.get("proteins", {}).get(accession, {})
- dataframe = pd.DataFrame(protein)
+ return self._normalize_b2b_prediction(protein)
+
+ @staticmethod
+ def _normalize_b2b_prediction(protein: dict[str, object]) -> pd.DataFrame:
+ import pprint;
+ print("_normalize_b2b_prediction")
+ pprint.pprint(protein, indent=4, sort_dicts=True)
+
+ sequence = "".join(protein.get("seq", ""))
+ print("SEQUENCE=", sequence, "length=", len(sequence))
+
+ size = (
+ len(sequence)
+ or len(protein.get("backbone", []))
+ or len(protein.get("sidechain", []))
+ or len(protein.get("ppII", []))
+ or len(protein.get("coil", []))
+ or len(protein.get("sheet", []))
+ or len(protein.get("helix", []))
+ or len(protein.get("earlyFolding", []))
+ or len(protein.get("disoMine", []))
+ )
+ print("SIZE=", size)
+
+ def _coerce_series(value: object) -> list[object]:
+ if isinstance(value, (list, tuple)):
+ values = list(value)
+ elif hasattr(value, "tolist"):
+ values = list(value.tolist()) # type: ignore[call-arg]
+ else:
+ values = []
+
+ if len(values) < size:
+ values.extend([None] * (size - len(values)))
+ elif len(values) > size:
+ values = values[:size]
+ return values
+
+ dataframe = pd.DataFrame(
+ {
+ "Position": list(range(1, size + 1)),
+ "Amino acid": protein.get("seq"),
+ "backbone": _coerce_series(protein.get("backbone")),
+ "sidechain": _coerce_series(protein.get("sidechain")),
+ "ppII": _coerce_series(protein.get("ppII")),
+ "coil": _coerce_series(protein.get("coil")),
+ "sheet": _coerce_series(protein.get("sheet")),
+ "helix": _coerce_series(protein.get("helix")),
+ "earlyFolding": _coerce_series(protein.get("earlyFolding")),
+ "disoMine": _coerce_series(protein.get("disoMine")),
+ }
+ )
+ for column in B2B_METRIC_COLUMNS:
+ dataframe[column] = pd.to_numeric(dataframe[column], errors="coerce")
+ dataframe[StructureVizService.b2b_metric_column(column, normalized=True)] = (
+ StructureVizService._min_max_normalize_series(dataframe[column])
+ )
return dataframe
+ @staticmethod
+ def b2b_metric_column(metric: str, *, normalized: bool = False) -> str:
+ return f"{metric}{B2B_NORMALIZED_SUFFIX}" if normalized else metric
+
+ @staticmethod
+ def _min_max_normalize_series(series: pd.Series) -> pd.Series:
+ numeric = pd.to_numeric(series, errors="coerce")
+ non_null = numeric.dropna()
+ if non_null.empty:
+ return pd.Series([pd.NA] * len(numeric), index=numeric.index, dtype="Float64")
+
+ minimum = float(non_null.min())
+ maximum = float(non_null.max())
+ if minimum == maximum:
+ return pd.Series(
+ [0.0 if pd.notna(value) else pd.NA for value in numeric],
+ index=numeric.index,
+ dtype="Float64",
+ )
+
+ normalized = (numeric - minimum) / (maximum - minimum)
+ return normalized.astype("Float64")
+
def download_alphafold_pdb(self, accession: str) -> Path:
out_path = self.workdir / f"AF-{accession}-F1-model_v6.pdb"
url = f"https://alphafold.ebi.ac.uk/files/AF-{accession}-F1-model_v6.pdb"
@@ -161,6 +237,15 @@ def accept_residue(self, residue): # noqa: ANN001
class StructureOps:
+ @staticmethod
+ def validate_pdb_id(pdb_id: str) -> str:
+ pdb_key = pdb_id.strip().upper()
+ if len(pdb_key) != 4 or not pdb_key.isalnum():
+ raise ValueError(
+ f"Invalid PDB ID '{pdb_id}'. Expected a 4-character RCSB identifier such as 2IVT."
+ )
+ return pdb_key
+
@staticmethod
def bfactor_pdb(pdb_path: Path, dataframe: pd.DataFrame, value_col: str, out_path: Path, chain: str | None = None) -> Path:
values = dataframe[value_col].tolist()
@@ -195,20 +280,71 @@ def chain_range_from_pdb(pdb_path: Path, chain_id: str) -> tuple[int, int]:
positions = [res.id[1] for res in chain.get_residues() if res.id[0] == " "]
return min(positions), max(positions)
+ @staticmethod
+ def chain_ranges_from_pdb(pdb_path: Path) -> dict[str, tuple[int, int]]:
+ parser = PDBParser(QUIET=True)
+ structure = parser.get_structure("chains", str(pdb_path))
+ model = next(structure.get_models())
+ chain_ranges: dict[str, tuple[int, int]] = {}
+ for chain in model.get_chains():
+ positions = [res.id[1] for res in chain.get_residues() if res.id[0] == " "]
+ if positions:
+ chain_ranges[chain.id] = (min(positions), max(positions))
+ return chain_ranges
+
@staticmethod
def save_chain_segment(source_pdb: Path, target_pdb: Path, chain_id: str, start: int | None, end: int | None) -> Path:
parser = PDBParser(QUIET=True)
structure = parser.get_structure("segment", str(source_pdb))
+ model = next(structure.get_models())
+ if chain_id not in model:
+ available = ", ".join(chain.id for chain in model.get_chains()) or "(none)"
+ raise ValueError(
+ f"Chain '{chain_id}' not found in {source_pdb.name}. Available chains: {available}."
+ )
+
+ chain = model[chain_id]
+ positions = [res.id[1] for res in chain.get_residues() if res.id[0] == " "]
+ if not positions:
+ raise ValueError(f"Chain '{chain_id}' in {source_pdb.name} does not contain standard residues.")
+
+ chain_start = min(positions)
+ chain_end = max(positions)
+ if start is not None and end is not None and start > end:
+ raise ValueError(f"Invalid range for chain '{chain_id}': start {start} is greater than end {end}.")
+ if start is not None and start > chain_end:
+ raise ValueError(f"Start {start} is outside chain '{chain_id}' range {chain_start}-{chain_end}.")
+ if end is not None and end < chain_start:
+ raise ValueError(f"End {end} is outside chain '{chain_id}' range {chain_start}-{chain_end}.")
+
io = PDBIO()
io.set_structure(structure)
io.save(str(target_pdb), select=ChainRangeSelect(chain_id, start, end))
+ segment_positions = [
+ int(line[22:26].strip())
+ for line in target_pdb.read_text(encoding="utf-8", errors="ignore").splitlines()
+ if line.startswith(("ATOM", "HETATM")) and line[21].strip() == chain_id and line[22:26].strip()
+ ]
+ if not segment_positions:
+ raise ValueError(
+ f"Chain/range selection produced an empty segment for chain '{chain_id}' "
+ f"with range {start or chain_start}-{end or chain_end} in {source_pdb.name}."
+ )
return target_pdb
@staticmethod
def run_tmalign(pdb1: Path, pdb2: Path, out_dir: Path, out_name: str = "aligned") -> tuple[Path, str]:
out_dir.mkdir(parents=True, exist_ok=True)
cmd = ["TM-align", str(pdb1.resolve()), str(pdb2.resolve()), "-o", out_name]
- process = subprocess.run(cmd, cwd=out_dir, capture_output=True, text=True, check=True)
+ try:
+ process = subprocess.run(cmd, cwd=out_dir, capture_output=True, text=True, check=True)
+ except subprocess.CalledProcessError as error:
+ stderr = (error.stderr or "").strip()
+ stdout = (error.stdout or "").strip()
+ details = stderr or stdout or f"exit code {error.returncode}"
+ raise RuntimeError(
+ f"TM-align failed for {pdb1.name} vs {pdb2.name}: {details}"
+ ) from error
candidates = [
out_dir / out_name,
out_dir / f"{out_name}.pdb",
diff --git a/apps/mutation_effect/app.py b/apps/mutation_effect/app.py
index bd6f820..b170fbc 100644
--- a/apps/mutation_effect/app.py
+++ b/apps/mutation_effect/app.py
@@ -14,10 +14,12 @@
MutationEffectService,
MutationEffectViews,
)
+from common.logging_utils import get_logger # noqa: E402
from common.ui_shell import scop3p_card, scop3p_shell, scop3p_footer # noqa: E402
service = MutationEffectService()
+LOGGER = get_logger("scop3p.mutation_effect")
def _bokeh_iframe(html_doc: str) -> ui.Tag:
@@ -117,6 +119,7 @@ def status() -> str:
def _run_wt() -> None:
try:
accession_value = input.accession().strip()
+ LOGGER.info("run_wt requested accession=%s", accession_value or "-", extra={"event": "run_wt"})
status_text.set("Fetching UniProt sequence and Scop3P PTMs...")
sequence_value = service.fetch_uniprot_sequence(accession_value)
mods = service.fetch_scop3p_modifications(accession_value)
@@ -150,12 +153,23 @@ def _run_wt() -> None:
)
status_text.set(f"WT prediction ready. PTMs: {0 if mods is None else len(mods)}.")
except Exception as error:
+ LOGGER.exception("run_wt failed accession=%s", input.accession().strip() or "-", extra={"event": "run_wt"})
status_text.set(f"WT error: {error}")
+ return
+ LOGGER.info(
+ "run_wt completed accession=%s seq_len=%s ptms=%s rows=%s",
+ accession_value,
+ len(sequence_value),
+ 0 if mods is None else len(mods),
+ len(dataframe),
+ extra={"event": "run_wt"},
+ )
@reactive.effect
@reactive.event(input.run_mut)
def _run_mut() -> None:
try:
+ LOGGER.info("run_mut requested positions=%s aas=%s", input.positions(), input.mut_aas(), extra={"event": "run_mut"})
if wt_df.get() is None or not sequence.get():
raise ValueError("Run WT prediction first.")
@@ -189,12 +203,21 @@ def _run_mut() -> None:
inf_sections.set([])
status_text.set("Mutant prediction ready.")
except Exception as error:
+ LOGGER.exception("run_mut failed", extra={"event": "run_mut"})
status_text.set(f"Mutant error: {error}")
+ return
+ LOGGER.info(
+ "run_mut completed mutations=%s rows=%s",
+ len(parsed_mutations),
+ len(dataframe),
+ extra={"event": "run_mut"},
+ )
@reactive.effect
@reactive.event(input.run_inf)
def _run_inf() -> None:
try:
+ LOGGER.info("run_inf requested mutations=%s", len(mutations.get()), extra={"event": "run_inf"})
if wt_df.get() is None or mut_df.get() is None:
raise ValueError("Run WT prediction and Mutant prediction first.")
@@ -223,7 +246,10 @@ def _run_inf() -> None:
inf_sections.set(sections)
status_text.set("Inference ready.")
except Exception as error:
+ LOGGER.exception("run_inf failed", extra={"event": "run_inf"})
status_text.set(f"Inference error: {error}")
+ return
+ LOGGER.info("run_inf completed sections=%s", len(sections), extra={"event": "run_inf"})
@output
@render.ui
diff --git a/apps/peptide_mapper/app.py b/apps/peptide_mapper/app.py
index 9c4e5cd..93d7306 100644
--- a/apps/peptide_mapper/app.py
+++ b/apps/peptide_mapper/app.py
@@ -10,6 +10,7 @@
import pandas as pd
from shiny import App, reactive, render, ui
+from common.logging_utils import get_logger
from common.models import PeptideSelectionMode
from common.peptide_mapper import PeptideMapperService, map_selection
from common.services import AlphaFoldService, Scop3PClient
@@ -17,6 +18,9 @@
from common.viewer import NGLViewerBuilder
+LOGGER = get_logger("scop3p.peptide_mapper")
+
+
class PeptideMapperController:
"""Stateful coordinator for Peptide Mapper app behavior."""
@@ -180,6 +184,7 @@ def server(input, output, session):
@reactive.event(input.load_btn)
def _load_data() -> None:
accession = input.accession().strip()
+ LOGGER.info("load requested accession=%s", accession or "-", extra={"event": "load_btn"})
if not accession:
controller.status_text.set("Enter an accession (e.g., O00571), then click Load.")
return
@@ -187,6 +192,7 @@ def _load_data() -> None:
try:
dataframe = controller.client.fetch_peptides_modifications(accession)
except Exception as error:
+ LOGGER.exception("load failed accession=%s", accession, extra={"event": "load_btn"})
controller.status_text.set(f"Scop3P API error: {error}")
controller.dataframe.set(pd.DataFrame())
controller.filtered_dataframe.set(pd.DataFrame())
@@ -207,6 +213,13 @@ def _load_data() -> None:
options = PeptideMapperService.build_options(dataframe, mode)
ui.update_selectize("peptides", choices=_as_selectize_choices(options), selected=[])
controller.status_text.set(f"Loaded {len(dataframe)} peptide-mod rows for {accession}.")
+ LOGGER.info(
+ "load completed accession=%s rows=%s mode=%s",
+ accession,
+ len(dataframe),
+ mode.value,
+ extra={"event": "load_btn"},
+ )
@reactive.effect
def _update_filter_and_choices() -> None:
@@ -225,11 +238,20 @@ def _update_filter_and_choices() -> None:
restored = [value for value in selected if value in valid_values]
ui.update_selectize("peptides", choices=_as_selectize_choices(options), selected=restored)
+ LOGGER.info(
+ "filter updated query=%r filtered_rows=%s options=%s restored=%s",
+ input.search(),
+ len(filtered),
+ len(options),
+ len(restored),
+ extra={"event": "filter_update"},
+ )
@reactive.effect
@reactive.event(input.map_all)
def _map_all() -> None:
filtered = controller.filtered_dataframe.get()
+ LOGGER.info("map_all requested", extra={"event": "map_all"})
if filtered is None or filtered.empty:
controller.status_text.set("No filtered rows available. Load data first.")
return
@@ -238,6 +260,7 @@ def _map_all() -> None:
options = PeptideMapperService.build_options(filtered, mode)
values = [value for _, value in options]
ui.update_selectize("peptides", selected=values)
+ LOGGER.info("map_all selected_count=%s", len(values), extra={"event": "map_all"})
@reactive.effect
@reactive.event(input.peptides, input.show_mods, input.mods_scope)
@@ -247,6 +270,14 @@ def _render_selection() -> None:
return
accession = input.accession().strip()
+ LOGGER.info(
+ "render requested accession=%s selected=%s show_mods=%s scope=%s",
+ accession or "-",
+ len(selected_values),
+ input.show_mods(),
+ input.mods_scope(),
+ extra={"event": "render_selection"},
+ )
if not accession:
controller.status_text.set("Enter an accession and click Load first.")
return
@@ -260,6 +291,7 @@ def _render_selection() -> None:
try:
pdb_path = controller.af_service.download_pdb(accession)
except Exception as error:
+ LOGGER.exception("alphafold download failed accession=%s", accession, extra={"event": "render_selection"})
controller.status_text.set(f"AlphaFold download error: {error}")
return
@@ -272,6 +304,7 @@ def _render_selection() -> None:
mods_scope=input.mods_scope(),
)
except Exception as error:
+ LOGGER.exception("selection mapping failed accession=%s", accession, extra={"event": "render_selection"})
controller.status_text.set(f"Selection mapping error: {error}")
return
@@ -306,11 +339,21 @@ def _render_selection() -> None:
]
)
)
+ LOGGER.info(
+ "render completed accession=%s pdb=%s ranges=%s intersection=%s mods=%s",
+ accession,
+ pdb_path,
+ len(union_ranges),
+ len(intersection_positions),
+ len(set(modification_positions)),
+ extra={"event": "render_selection"},
+ )
@reactive.effect
@reactive.event(input.export_html)
def _export_html() -> None:
accession = input.accession().strip()
+ LOGGER.info("export requested accession=%s", accession or "-", extra={"event": "export_html"})
if not accession:
controller.status_text.set("Enter an accession first.")
return
@@ -322,6 +365,7 @@ def _export_html() -> None:
export_path = Path("exports") / f"{accession}_styled_session.html"
NGLViewerBuilder.export_html(export_path, controller.viewer_html.get())
controller.status_text.set(f"Exported styled HTML to: {export_path.resolve()}")
+ LOGGER.info("export completed path=%s", export_path.resolve(), extra={"event": "export_html"})
@render.text
def status() -> str:
diff --git a/apps/structure_viz/app.py b/apps/structure_viz/app.py
index 6105034..bc70196 100644
--- a/apps/structure_viz/app.py
+++ b/apps/structure_viz/app.py
@@ -11,10 +11,19 @@
if str(ROOT) not in sys.path:
sys.path.append(str(ROOT))
-from common.structure_viz import StructureOps, StructureViewerBuilder, StructureVizService # noqa: E402
+from common.structure_viz import ( # noqa: E402
+ B2B_METRIC_COLUMNS,
+ StructureOps,
+ StructureViewerBuilder,
+ StructureVizService,
+)
+from common.logging_utils import get_logger # noqa: E402
from common.ui_shell import scop3p_card, scop3p_shell, scop3p_footer # noqa: E402
+LOGGER = get_logger("scop3p.structure_viz")
+
+
class StructureVizController:
def __init__(self) -> None:
self.workdir = Path(tempfile.mkdtemp(prefix="scop3p_structure_viz_"))
@@ -36,11 +45,31 @@ def __init__(self) -> None:
self.tm_report = reactive.value("")
self.b2b_html = reactive.value("")
self.tm_html = reactive.value("")
+ self.tm_input_1 = reactive.value(None)
+ self.tm_input_2 = reactive.value(None)
+ self.tm_chain_ranges_1 = reactive.value({})
+ self.tm_chain_ranges_2 = reactive.value({})
+ self.tm_structures_loaded = reactive.value(False)
+ self.tm_loaded_signature_1 = reactive.value(None)
+ self.tm_loaded_signature_2 = reactive.value(None)
controller = StructureVizController()
+def _tm_source_signature(upload, pdb_id: str) -> tuple[str, str] | None: # noqa: ANN001
+ if upload:
+ row = upload[0]
+ datapath = str(row.get("datapath", ""))
+ name = str(row.get("name", ""))
+ return ("upload", f"{datapath}|{name}")
+
+ pdb_key = pdb_id.strip().upper()
+ if pdb_key:
+ return ("pdb", pdb_key)
+ return None
+
+
def _scroll_df(dataframe: pd.DataFrame) -> ui.Tag:
if dataframe is None or dataframe.empty:
return ui.p("No rows.")
@@ -48,13 +77,48 @@ def _scroll_df(dataframe: pd.DataFrame) -> ui.Tag:
"""
return ui.HTML(css + f"{dataframe.to_html(index=False, escape=False)}
")
+def _reset_b2b_state() -> None:
+ controller.sequence.set("")
+ controller.b2b_df.set(pd.DataFrame())
+ controller.b2b_html.set("")
+ ui.update_select("b2b_metric", choices={}, selected=None)
+
+
+def _b2b_metric_names(dataframe: pd.DataFrame) -> list[str]:
+ if dataframe is None or dataframe.empty:
+ return []
+ return [
+ metric
+ for metric in B2B_METRIC_COLUMNS
+ if StructureVizService.b2b_metric_column(metric) in dataframe.columns
+ ]
+
+
+def _selected_b2b_metric_column(metric: str | None, *, normalized: bool) -> str | None:
+ if not metric:
+ return None
+ return StructureVizService.b2b_metric_column(metric, normalized=normalized)
+
+
+def _b2b_table_dataframe(dataframe: pd.DataFrame, *, normalized: bool) -> pd.DataFrame:
+ if dataframe is None or dataframe.empty:
+ return pd.DataFrame()
+ column_names = ["Position", "Amino acid"] + [
+ StructureVizService.b2b_metric_column(metric, normalized=normalized)
+ for metric in B2B_METRIC_COLUMNS
+ ]
+ table = dataframe.loc[:, column_names].copy()
+ table.columns = ["Position", "Amino acid", *B2B_METRIC_COLUMNS]
+ return table
+
+
app_ui = scop3p_shell(
"Structure Visualisation",
"Inspect PTMs, disease variants, 3D structures, Bio2Byte overlays, residue interaction networks, and TM-align comparisons within one structure-centric workspace.",
@@ -117,8 +181,10 @@ def _scroll_df(dataframe: pd.DataFrame) -> ui.Tag:
ui.input_action_button("fetch_seq", "Fetch sequence", class_="btn-warning"),
ui.input_action_button("run_b2b", "Run predictions", class_="btn-danger"),
ui.input_action_button("render_b2b_3d", "Show 3D", class_="btn-success"),
- col_widths=[4, 4, 4],
+ ui.input_action_button("reset_b2b", "Reset results", class_="btn-secondary"),
+ col_widths=[3, 3, 3, 3],
),
+ ui.input_checkbox("b2b_normalized", "Show normalized values", value=False),
ui.input_select("b2b_metric", "Color by", choices=[]),
ui.output_ui("b2b_table"),
ui.output_ui("b2b_view"),
@@ -159,15 +225,19 @@ def _scroll_df(dataframe: pd.DataFrame) -> ui.Tag:
col_widths=[6, 6],
),
ui.layout_columns(
- ui.input_text("tm_chain1", "Chain 1", value="A"),
+ ui.input_select("tm_chain1", "Chain 1", choices={}),
ui.input_numeric("tm_start1", "Start 1", value=None),
ui.input_numeric("tm_end1", "End 1", value=None),
- ui.input_text("tm_chain2", "Chain 2", value="A"),
+ ui.input_select("tm_chain2", "Chain 2", choices={}),
ui.input_numeric("tm_start2", "Start 2", value=None),
ui.input_numeric("tm_end2", "End 2", value=None),
col_widths=[2, 2, 2, 2, 2, 2],
),
- ui.input_action_button("run_tmalign", "Align + Visualize", class_="btn-primary"),
+ ui.layout_columns(
+ ui.input_action_button("load_tmalign_structures", "Load structures", class_="btn-warning"),
+ ui.output_ui("tm_actions"),
+ col_widths=[6, 6],
+ ),
ui.output_text_verbatim("tm_output"),
ui.output_ui("tm_view"),
),
@@ -188,48 +258,57 @@ def require_accession() -> str | None:
@reactive.event(input.set_accession)
def _set_accession() -> None:
accession = require_accession()
+ LOGGER.info("set_accession requested accession=%s", accession or "-", extra={"event": "set_accession"})
if not accession:
return
controller.accession.set(accession)
+ _reset_b2b_state()
controller.status.set(f"Protein set: {accession} | session: {controller.workdir}")
@reactive.effect
@reactive.event(input.fetch_ptm)
def _fetch_ptm() -> None:
accession = controller.accession.get()
+ LOGGER.info("fetch_ptm requested accession=%s", accession or "-", extra={"event": "fetch_ptm"})
if not accession:
controller.status.set("Set a UniProt accession first.")
return
dataframe = controller.service.fetch_ptms(accession)
controller.ptm_df.set(dataframe)
controller.status.set(f"PTMs fetched: {len(dataframe)} rows.")
+ LOGGER.info("fetch_ptm completed rows=%s", len(dataframe), extra={"event": "fetch_ptm"})
@reactive.effect
@reactive.event(input.fetch_variants)
def _fetch_variants() -> None:
accession = controller.accession.get()
+ LOGGER.info("fetch_variants requested accession=%s", accession or "-", extra={"event": "fetch_variants"})
if not accession:
controller.status.set("Set a UniProt accession first.")
return
dataframe = controller.service.fetch_variants(accession)
controller.var_df.set(dataframe)
controller.status.set(f"Variants fetched: {len(dataframe)} rows.")
+ LOGGER.info("fetch_variants completed rows=%s", len(dataframe), extra={"event": "fetch_variants"})
@reactive.effect
@reactive.event(input.fetch_af)
def _fetch_af() -> None:
accession = controller.accession.get()
+ LOGGER.info("fetch_af requested accession=%s", accession or "-", extra={"event": "fetch_af"})
if not accession:
controller.status.set("Set a UniProt accession first.")
return
af_path = controller.service.download_alphafold_pdb(accession)
controller.af_path.set(af_path)
controller.status.set(f"AlphaFold downloaded: {af_path}")
+ LOGGER.info("fetch_af completed path=%s", af_path, extra={"event": "fetch_af"})
@reactive.effect
@reactive.event(input.render_structure)
def _render_structure() -> None:
accession = controller.accession.get()
+ LOGGER.info("render_structure requested accession=%s source=%s", accession or "-", input.structure_source(), extra={"event": "render_structure"})
if not accession:
controller.status.set("Set a UniProt accession first.")
return
@@ -259,31 +338,40 @@ def _render_structure() -> None:
)
controller.viewer_html.set(html_payload)
controller.status.set(f"Rendered 3D structure from: {Path(pdb_path).name}")
+ LOGGER.info("render_structure completed pdb=%s chain=%s", pdb_path, chain or "-", extra={"event": "render_structure"})
@reactive.effect
@reactive.event(input.fetch_seq)
def _fetch_seq() -> None:
accession = controller.accession.get()
+ LOGGER.info("fetch_seq requested accession=%s", accession or "-", extra={"event": "fetch_seq"})
if not accession:
controller.status.set("Set a UniProt accession first.")
return
+ controller.b2b_html.set("")
sequence = controller.service.fetch_sequence(accession)
controller.sequence.set(sequence)
controller.status.set(f"Sequence fetched: {len(sequence)} aa")
+ LOGGER.info("fetch_seq completed length=%s", len(sequence), extra={"event": "fetch_seq"})
@reactive.effect
@reactive.event(input.run_b2b)
def _run_b2b() -> None:
accession = controller.accession.get()
sequence = controller.sequence.get()
+ LOGGER.info("run_b2b requested accession=%s sequence_length=%s", accession or "-", len(sequence), extra={"event": "run_b2b"})
if not accession or not sequence:
controller.status.set("Fetch sequence first.")
return
+
+ controller.status.set("Predicting biophysical features, please wait...")
dataframe = controller.service.predict_b2b(accession, sequence)
controller.b2b_df.set(dataframe)
- numeric = [column for column in dataframe.columns if pd.api.types.is_numeric_dtype(dataframe[column])]
- ui.update_select("b2b_metric", choices={column: column for column in numeric}, selected=numeric[0] if numeric else None)
+ controller.b2b_html.set("")
+ metrics = _b2b_metric_names(dataframe)
+ ui.update_select("b2b_metric", choices={metric: metric for metric in metrics}, selected=metrics[0] if metrics else None)
controller.status.set(f"Bio2Byte prediction completed ({len(dataframe)} rows).")
+ LOGGER.info("run_b2b completed rows=%s metrics=%s", len(dataframe), len(metrics), extra={"event": "run_b2b"})
@reactive.effect
@reactive.event(input.render_b2b_3d)
@@ -291,37 +379,61 @@ def _render_b2b() -> None:
dataframe = controller.b2b_df.get()
accession = controller.accession.get()
metric = input.b2b_metric()
+ normalized = bool(input.b2b_normalized())
+ metric_column = _selected_b2b_metric_column(metric, normalized=normalized)
af_path = controller.af_path.get()
- if dataframe is None or dataframe.empty or not metric:
+ LOGGER.info(
+ "render_b2b requested accession=%s metric=%s normalized=%s",
+ accession or "-",
+ metric or "-",
+ normalized,
+ extra={"event": "render_b2b"},
+ )
+ if dataframe is None or dataframe.empty or not metric or metric_column is None:
controller.status.set("Run predictions and choose a metric first.")
return
if af_path is None:
+ controller.b2b_html.set("")
controller.status.set("Fetch AlphaFold first (tab 3).")
return
- out_pdb = controller.workdir / f"b2b_{metric}.pdb"
- bfactor_pdb = StructureOps.bfactor_pdb(Path(af_path), dataframe, metric, out_pdb)
+ out_pdb = controller.workdir / f"b2b_{metric_column}.pdb"
+ bfactor_pdb = StructureOps.bfactor_pdb(Path(af_path), dataframe, metric_column, out_pdb)
html_payload = StructureViewerBuilder.b2b_html(
pdb_text=bfactor_pdb.read_text(encoding="utf-8", errors="ignore"),
accession=accession,
- metric=metric,
+ metric=metric_column,
)
controller.b2b_html.set(html_payload)
- controller.status.set(f"Rendered Bio2Byte 3D metric: {metric}")
+ controller.status.set(
+ f"Rendered Bio2Byte 3D metric: {metric}"
+ f"{' (normalized)' if normalized else ''}"
+ )
+ LOGGER.info("render_b2b completed metric=%s normalized=%s", metric, normalized, extra={"event": "render_b2b"})
+
+ @reactive.effect
+ @reactive.event(input.reset_b2b)
+ def _reset_b2b() -> None:
+ LOGGER.info("reset_b2b requested", extra={"event": "reset_b2b"})
+ _reset_b2b_state()
+ controller.status.set("Bio2Byte results cleared.")
@reactive.effect
@reactive.event(input.rin_dl_af)
def _rin_dl_af() -> None:
accession = controller.accession.get()
+ LOGGER.info("rin_dl_af requested accession=%s", accession or "-", extra={"event": "rin_dl_af"})
if not accession:
controller.status.set("Set a UniProt accession first.")
return
path = controller.service.download_alphafold_pdb(accession)
controller.rin_path.set(path)
controller.status.set(f"RIN input set to AlphaFold PDB: {path}")
+ LOGGER.info("rin_dl_af completed path=%s", path, extra={"event": "rin_dl_af"})
@reactive.effect
@reactive.event(input.build_rin)
def _build_rin() -> None:
+ LOGGER.info("build_rin requested", extra={"event": "build_rin"})
pdb_path = controller.service.resolve_uploaded_or_remote_pdb(
input.rin_upload(),
input.rin_pdb_id(),
@@ -349,28 +461,38 @@ def _build_rin() -> None:
StructureOps.rin_to_pyvis_html(graph, html_path, ptm_pos, variant_pos)
controller.rin_html.set(html_path.read_text(encoding="utf-8", errors="ignore"))
controller.status.set(f"RIN built with {graph.number_of_nodes()} nodes / {graph.number_of_edges()} edges.")
+ LOGGER.info(
+ "build_rin completed pdb=%s nodes=%s edges=%s",
+ pdb_path,
+ graph.number_of_nodes(),
+ graph.number_of_edges(),
+ extra={"event": "build_rin"},
+ )
@reactive.effect
@reactive.event(input.run_tmalign)
def _run_tmalign() -> None:
+ LOGGER.info("run_tmalign requested", extra={"event": "run_tmalign"})
try:
- f1 = controller.service.resolve_uploaded_or_remote_pdb(
- input.tm_pdb1(),
- input.tm_pdb1_id(),
- target_name="tm_input_1.pdb",
- )
- f2 = controller.service.resolve_uploaded_or_remote_pdb(
- input.tm_pdb2(),
- input.tm_pdb2_id(),
- target_name="tm_input_2.pdb",
- )
+ current_signature_1 = _tm_source_signature(input.tm_pdb1(), input.tm_pdb1_id().strip())
+ current_signature_2 = _tm_source_signature(input.tm_pdb2(), input.tm_pdb2_id().strip())
+ f1 = controller.tm_input_1.get()
+ f2 = controller.tm_input_2.get()
if f1 is None or f2 is None:
- controller.tm_report.set("Provide both structures via local upload or RCSB PDB ID.")
+ controller.tm_report.set("Load both structures first.")
+ controller.tm_html.set("")
+ return
+ if (
+ current_signature_1 != controller.tm_loaded_signature_1.get()
+ or current_signature_2 != controller.tm_loaded_signature_2.get()
+ ):
+ controller.tm_structures_loaded.set(False)
+ controller.tm_report.set("TM-align inputs changed. Reload both structures first.")
controller.tm_html.set("")
return
- chain1 = (input.tm_chain1() or "A").strip() or "A"
- chain2 = (input.tm_chain2() or "A").strip() or "A"
+ chain1 = input.tm_chain1() or "A"
+ chain2 = input.tm_chain2() or "A"
start1 = int(input.tm_start1()) if input.tm_start1() is not None else None
end1 = int(input.tm_end1()) if input.tm_end1() is not None else None
start2 = int(input.tm_start2()) if input.tm_start2() is not None else None
@@ -391,14 +513,152 @@ def _run_tmalign() -> None:
)
)
controller.status.set("TM-align completed.")
+ LOGGER.info("run_tmalign completed aligned=%s", aligned_path, extra={"event": "run_tmalign"})
except Exception as error:
+ LOGGER.exception("run_tmalign failed", extra={"event": "run_tmalign"})
controller.tm_html.set("")
controller.tm_report.set(f"TM-align error: {error}")
controller.status.set("TM-align failed.")
+ @reactive.effect
+ @reactive.event(input.load_tmalign_structures)
+ def _load_tmalign_structures() -> None:
+ LOGGER.info("load_tmalign_structures requested", extra={"event": "load_tmalign_structures"})
+ controller.tm_html.set("")
+ controller.tm_structures_loaded.set(False)
+ try:
+ tm_pdb1_id = input.tm_pdb1_id().strip()
+ tm_pdb2_id = input.tm_pdb2_id().strip()
+ if tm_pdb1_id:
+ StructureOps.validate_pdb_id(tm_pdb1_id)
+ if tm_pdb2_id:
+ StructureOps.validate_pdb_id(tm_pdb2_id)
+
+ f1 = controller.service.resolve_uploaded_or_remote_pdb(
+ input.tm_pdb1(),
+ tm_pdb1_id,
+ target_name="tm_input_1.pdb",
+ )
+ f2 = controller.service.resolve_uploaded_or_remote_pdb(
+ input.tm_pdb2(),
+ tm_pdb2_id,
+ target_name="tm_input_2.pdb",
+ )
+ if f1 is None or f2 is None:
+ controller.tm_input_1.set(None)
+ controller.tm_input_2.set(None)
+ controller.tm_chain_ranges_1.set({})
+ controller.tm_chain_ranges_2.set({})
+ controller.tm_report.set("Provide both structures via local upload or RCSB PDB ID.")
+ return
+
+ ranges_1 = StructureOps.chain_ranges_from_pdb(f1)
+ ranges_2 = StructureOps.chain_ranges_from_pdb(f2)
+ if not ranges_1 or not ranges_2:
+ raise ValueError("Could not find any standard-residue chains in one of the loaded structures.")
+
+ controller.tm_input_1.set(f1)
+ controller.tm_input_2.set(f2)
+ controller.tm_chain_ranges_1.set(ranges_1)
+ controller.tm_chain_ranges_2.set(ranges_2)
+ controller.tm_loaded_signature_1.set(_tm_source_signature(input.tm_pdb1(), tm_pdb1_id))
+ controller.tm_loaded_signature_2.set(_tm_source_signature(input.tm_pdb2(), tm_pdb2_id))
+ controller.tm_structures_loaded.set(True)
+
+ first_chain_1 = next(iter(ranges_1))
+ first_chain_2 = next(iter(ranges_2))
+ ui.update_select("tm_chain1", choices={chain: chain for chain in ranges_1}, selected=first_chain_1)
+ ui.update_select("tm_chain2", choices={chain: chain for chain in ranges_2}, selected=first_chain_2)
+ start_1, end_1 = ranges_1[first_chain_1]
+ start_2, end_2 = ranges_2[first_chain_2]
+ ui.update_numeric("tm_start1", value=start_1, min=start_1, max=end_1)
+ ui.update_numeric("tm_end1", value=end_1, min=start_1, max=end_1)
+ ui.update_numeric("tm_start2", value=start_2, min=start_2, max=end_2)
+ ui.update_numeric("tm_end2", value=end_2, min=start_2, max=end_2)
+ controller.tm_report.set(
+ "Loaded TM-align structures:\n"
+ f"1) {f1.name}: chains {', '.join(ranges_1)}\n"
+ f"2) {f2.name}: chains {', '.join(ranges_2)}"
+ )
+ controller.status.set("TM-align structures loaded.")
+ LOGGER.info(
+ "load_tmalign_structures completed structure1=%s chains1=%s structure2=%s chains2=%s",
+ f1,
+ list(ranges_1),
+ f2,
+ list(ranges_2),
+ extra={"event": "load_tmalign_structures"},
+ )
+ except Exception as error:
+ LOGGER.exception("load_tmalign_structures failed", extra={"event": "load_tmalign_structures"})
+ controller.tm_input_1.set(None)
+ controller.tm_input_2.set(None)
+ controller.tm_chain_ranges_1.set({})
+ controller.tm_chain_ranges_2.set({})
+ controller.tm_loaded_signature_1.set(None)
+ controller.tm_loaded_signature_2.set(None)
+ controller.tm_report.set(f"TM-align load error: {error}")
+ controller.status.set("TM-align structure load failed.")
+
+ @reactive.effect
+ def _invalidate_loaded_tmalign_inputs() -> None:
+ current_signature_1 = _tm_source_signature(input.tm_pdb1(), input.tm_pdb1_id().strip())
+ current_signature_2 = _tm_source_signature(input.tm_pdb2(), input.tm_pdb2_id().strip())
+ loaded_signature_1 = controller.tm_loaded_signature_1.get()
+ loaded_signature_2 = controller.tm_loaded_signature_2.get()
+
+ if loaded_signature_1 is None and loaded_signature_2 is None:
+ return
+ if current_signature_1 == loaded_signature_1 and current_signature_2 == loaded_signature_2:
+ return
+
+ controller.tm_structures_loaded.set(False)
+ controller.tm_input_1.set(None)
+ controller.tm_input_2.set(None)
+ controller.tm_chain_ranges_1.set({})
+ controller.tm_chain_ranges_2.set({})
+ controller.tm_loaded_signature_1.set(None)
+ controller.tm_loaded_signature_2.set(None)
+ controller.tm_html.set("")
+ controller.tm_report.set("TM-align inputs changed. Reload both structures first.")
+ LOGGER.info("tmalign inputs invalidated after source change", extra={"event": "load_tmalign_structures"})
+
+ @reactive.effect
+ def _sync_tm_chain1_range() -> None:
+ chain_ranges = controller.tm_chain_ranges_1.get()
+ selected_chain = input.tm_chain1()
+ if not chain_ranges or selected_chain not in chain_ranges:
+ return
+ start, end = chain_ranges[selected_chain]
+ current_start = input.tm_start1()
+ current_end = input.tm_end1()
+ next_start = start if current_start is None or current_start < start or current_start > end else current_start
+ next_end = end if current_end is None or current_end < start or current_end > end else current_end
+ if next_start > next_end:
+ next_start, next_end = start, end
+ ui.update_numeric("tm_start1", value=next_start, min=start, max=end)
+ ui.update_numeric("tm_end1", value=next_end, min=start, max=end)
+
+ @reactive.effect
+ def _sync_tm_chain2_range() -> None:
+ chain_ranges = controller.tm_chain_ranges_2.get()
+ selected_chain = input.tm_chain2()
+ if not chain_ranges or selected_chain not in chain_ranges:
+ return
+ start, end = chain_ranges[selected_chain]
+ current_start = input.tm_start2()
+ current_end = input.tm_end2()
+ next_start = start if current_start is None or current_start < start or current_start > end else current_start
+ next_end = end if current_end is None or current_end < start or current_end > end else current_end
+ if next_start > next_end:
+ next_start, next_end = start, end
+ ui.update_numeric("tm_start2", value=next_start, min=start, max=end)
+ ui.update_numeric("tm_end2", value=next_end, min=start, max=end)
+
@reactive.effect
@reactive.event(input.show_rin)
def _show_rin() -> None:
+ LOGGER.info("show_rin requested has_html=%s", bool(controller.rin_html.get()), extra={"event": "show_rin"})
if not controller.rin_html.get():
controller.status.set("No RIN HTML yet. Build RIN first.")
@@ -423,7 +683,12 @@ def structure_view():
@render.ui
def b2b_table():
- return _scroll_df(controller.b2b_df.get())
+ return _scroll_df(
+ _b2b_table_dataframe(
+ controller.b2b_df.get(),
+ normalized=bool(input.b2b_normalized()),
+ )
+ )
@render.ui
def b2b_view():
@@ -445,6 +710,15 @@ def rin_view():
def tm_output() -> str:
return controller.tm_report.get()
+ @render.ui
+ def tm_actions():
+ return ui.input_action_button(
+ "run_tmalign",
+ "Align + Visualize",
+ class_="btn-primary",
+ disabled=not controller.tm_structures_loaded.get(),
+ )
+
@render.ui
def tm_view():
payload = controller.tm_html.get()
diff --git a/docker/Dockerfile b/docker/Dockerfile
index 244ce22..1cd4acf 100644
--- a/docker/Dockerfile
+++ b/docker/Dockerfile
@@ -1,43 +1,67 @@
# syntax=docker/dockerfile:1.7
-FROM python:3.12-slim AS runtime
+FROM python:3.12-slim AS builder
ARG BUILD_DATE
ARG VCS_REF
ARG VERSION=dev
+ENV VIRTUAL_ENV=/opt/venv \
+ PATH=/opt/venv/bin:$PATH \
+ PYTHONDONTWRITEBYTECODE=1 \
+ PYTHONUNBUFFERED=1
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+ build-essential gcc g++ hmmer t-coffee && \
+ python -m venv "$VIRTUAL_ENV" && \
+ "$VIRTUAL_ENV/bin/pip" install -U pip && \
+ rm -rf /var/lib/apt/lists/*
+
+COPY requirements-biophysics.txt /tmp/requirements-biophysics.txt
+RUN --mount=type=cache,target=/root/.cache/pip \
+ "$VIRTUAL_ENV/bin/pip" install -r /tmp/requirements-biophysics.txt
+
+COPY requirements-shiny.txt /tmp/requirements-shiny.txt
+RUN --mount=type=cache,target=/root/.cache/pip \
+ "$VIRTUAL_ENV/bin/pip" install -r /tmp/requirements-shiny.txt
+
+FROM python:3.12-slim AS runtime
+
LABEL org.opencontainers.image.title="Scop3P-Toolkit" \
org.opencontainers.image.description="Scop3P-Toolkit: Tools for exploring and extending Scop3P" \
org.opencontainers.image.url="https://github.com/Bio2Byte/Scop3P-notebooks" \
org.opencontainers.image.source="https://github.com/Bio2Byte/Scop3P-notebooks" \
- org.opencontainers.image.licenses="Apache 2.0" \
+ org.opencontainers.image.licenses="Apache-2.0" \
org.opencontainers.image.version="${VERSION}" \
org.opencontainers.image.revision="${VCS_REF}" \
org.opencontainers.image.created="${BUILD_DATE}"
-ENV PYTHONDONTWRITEBYTECODE=1 \
+ENV VIRTUAL_ENV=/opt/venv \
+ PATH=/opt/venv/bin:$PATH \
+ PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
PYTHONPATH=/apps
-WORKDIR /apps
-
RUN apt-get update && apt-get install -y --no-install-recommends \
- build-essential gcc g++ && \
- rm -rf /var/lib/apt/lists/* && \
- pip install -U pip
+ 't-coffee' hmmer && \
+ rm -rf /var/lib/apt/lists/*
-COPY requirements-biophysics.txt /app/requirements-biophysics.txt
-RUN --mount=type=cache,target=/root/.cache/pip \
- python -m pip install -r /app/requirements-biophysics.txt
+RUN groupadd --system scop3p && \
+ useradd --system --gid scop3p --create-home --home-dir /home/scop3p scop3p && \
+ mkdir -p /apps /tmp/scop3p && \
+ chown -R scop3p:scop3p /apps /tmp/scop3p /home/scop3p
-COPY requirements-shiny.txt /app/requirements-shiny.txt
-RUN --mount=type=cache,target=/root/.cache/pip \
- python -m pip install -r /app/requirements-shiny.txt
+RUN printf '%s\n' 'export VIRTUAL_ENV=/opt/venv' 'export PATH=/opt/venv/bin:$PATH' > /etc/profile.d/scop3p-venv.sh
+
+WORKDIR /apps
+COPY --from=builder /opt/venv /opt/venv
COPY --chmod=755 TM-align /usr/local/bin/TM-align
-COPY ./apps .
+COPY --chown=scop3p:scop3p ./apps .
EXPOSE 8000
+USER scop3p
+
FROM runtime AS peptide-mapper
CMD ["shiny", "run", "--host", "0.0.0.0", "--port", "8000", "/apps/peptide_mapper/app.py"]
diff --git a/pytest.ini b/pytest.ini
index 4584de7..910015f 100644
--- a/pytest.ini
+++ b/pytest.ini
@@ -1,3 +1,3 @@
[pytest]
testpaths = tests
-pythonpath = .
+pythonpath = . apps
diff --git a/requirements-biophysics.txt b/requirements-biophysics.txt
index 0bf9df6..a3908bc 100644
--- a/requirements-biophysics.txt
+++ b/requirements-biophysics.txt
@@ -1 +1,2 @@
-b2bTools==3.0.8b3
+b2bTools~=3.0.8
+scop3p~=1.1.0
diff --git a/requirements-shiny.txt b/requirements-shiny.txt
index 14fd029..51142ec 100644
--- a/requirements-shiny.txt
+++ b/requirements-shiny.txt
@@ -10,9 +10,6 @@ py3Dmol>=2.5.3
pyvis>=0.3.2,<0.4
bokeh>=3.3,<4
-# Extra bio
-hmmer
-
# Testing
pytest>=8.0,<9
httpx>=0.27,<1
diff --git a/tests/integration/test_app_smoke.py b/tests/integration/test_app_smoke.py
index 6eb17ef..f1587fa 100644
--- a/tests/integration/test_app_smoke.py
+++ b/tests/integration/test_app_smoke.py
@@ -30,7 +30,7 @@ def test_portal_root_selector_and_cookie() -> None:
default_response = client.get("/")
assert default_response.status_code == 200
- assert "App selector" in default_response.text
+ assert "Scop3P-Toolkit" in default_response.text
assert "Peptide Mapper" in default_response.text
assert "scop3p_app=peptide-mapper" in default_response.headers["set-cookie"]
diff --git a/tests/unit/test_logging_utils.py b/tests/unit/test_logging_utils.py
new file mode 100644
index 0000000..7c30a55
--- /dev/null
+++ b/tests/unit/test_logging_utils.py
@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+import io
+import logging
+
+from common.logging_utils import _SafeExtraFormatter
+
+
+def test_safe_extra_formatter_defaults_missing_event() -> None:
+ stream = io.StringIO()
+ handler = logging.StreamHandler(stream)
+ handler.setFormatter(_SafeExtraFormatter("%(levelname)s event=%(event)s %(message)s"))
+
+ logger = logging.getLogger("test.logging_utils")
+ logger.handlers = [handler]
+ logger.propagate = False
+ logger.setLevel(logging.INFO)
+
+ logger.info("hello")
+
+ assert "INFO event=- hello" in stream.getvalue()
diff --git a/tests/unit/test_mutation_effect_service.py b/tests/unit/test_mutation_effect_service.py
index cf93d70..bb93e5d 100644
--- a/tests/unit/test_mutation_effect_service.py
+++ b/tests/unit/test_mutation_effect_service.py
@@ -1,6 +1,7 @@
from __future__ import annotations
import pandas as pd
+import pytest
from common.mutation_effect import (
Mutation,
@@ -88,3 +89,70 @@ def test_make_wt_mut_merged_table_adds_ptm_flag() -> None:
merged = MutationEffectViews.make_wt_mut_merged_table(wt_df, mut_df, mods_df)
assert list(merged.columns[:3]) == ["seqpos", "WT_AA", "Mut_AA"]
assert merged.loc[merged["seqpos"] == 2, "PTMs"].iloc[0] == "yes"
+
+
+class _TextResponse:
+ def __init__(self, text: str) -> None:
+ self.text = text
+
+ def raise_for_status(self) -> None:
+ return None
+
+
+def test_fetch_uniprot_sequence_parses_fasta(monkeypatch) -> None:
+ monkeypatch.setattr(
+ "apps.common.mutation_effect.requests.get",
+ lambda *args, **kwargs: _TextResponse(">sp|P12345|\nACD\nEFG\n"),
+ )
+
+ sequence = MutationEffectService().fetch_uniprot_sequence("P12345")
+ assert sequence == "ACDEFG"
+
+
+def test_fetch_uniprot_sequence_rejects_empty_sequence(monkeypatch) -> None:
+ monkeypatch.setattr(
+ "apps.common.mutation_effect.requests.get",
+ lambda *args, **kwargs: _TextResponse(">sp|P12345|\n"),
+ )
+
+ with pytest.raises(ValueError, match="No sequence returned"):
+ MutationEffectService().fetch_uniprot_sequence("P12345")
+
+
+def test_fetch_scop3p_modifications_drops_invalid_positions(monkeypatch) -> None:
+ class _Client:
+ def fetch_modifications(self, accession: str) -> pd.DataFrame:
+ return pd.DataFrame(
+ [
+ {"position": 10, "residue": "S", "name": "Phosphorylation"},
+ {"position": pd.NA, "residue": "T", "name": "Phosphorylation"},
+ {"position": 12, "residue": "Y", "name": "Phosphorylation", "specificSinglyPhosphorylated": 1},
+ ]
+ )
+
+ service = MutationEffectService(
+ scop3p_client=_Client(), # type: ignore[arg-type]
+ )
+
+ dataframe = service.fetch_scop3p_modifications("P12345")
+ assert dataframe["position"].tolist() == [10, 12]
+ assert list(dataframe.columns) == ["position", "residue", "name"]
+
+
+def test_prediction_to_df_coerces_values_and_seqpos() -> None:
+ prediction = {
+ "proteins": {
+ "P12345": {
+ "seq": "AC",
+ "backbone": ["0.7", "bad"],
+ "disoMine": ["0.1", "0.2"],
+ "earlyFolding": ["0.3", None],
+ }
+ }
+ }
+
+ dataframe = MutationEffectService.prediction_to_df(prediction, "P12345")
+ assert dataframe["seqpos"].tolist() == [1, 2]
+ assert dataframe.loc[0, "backbone"] == 0.7
+ assert pd.isna(dataframe.loc[1, "backbone"])
+ assert dataframe.loc[1, "disoMine"] == 0.2
diff --git a/tests/unit/test_scop3p_client.py b/tests/unit/test_scop3p_client.py
index 42f13c7..07fd102 100644
--- a/tests/unit/test_scop3p_client.py
+++ b/tests/unit/test_scop3p_client.py
@@ -5,17 +5,6 @@
from common.services import Scop3PClient
-class _DummyResponse:
- def __init__(self, payload: dict) -> None:
- self._payload = payload
-
- def raise_for_status(self) -> None:
- return None
-
- def json(self) -> dict:
- return self._payload
-
-
def test_fetch_peptides_modifications_normalizes_payload(monkeypatch) -> None:
payload = {
"peptides": [
@@ -31,10 +20,10 @@ def test_fetch_peptides_modifications_normalizes_payload(monkeypatch) -> None:
]
}
- def _mock_get(*args, **kwargs):
- return _DummyResponse(payload)
-
- monkeypatch.setattr("apps.common.services.requests.get", _mock_get)
+ monkeypatch.setattr(
+ "apps.common.services.Scop3pRestApi.fetch_peptides",
+ lambda self, accession: payload,
+ )
client = Scop3PClient()
dataframe = client.fetch_peptides_modifications("O00571")
@@ -43,3 +32,46 @@ def _mock_get(*args, **kwargs):
assert dataframe.iloc[0]["peptideStart"] == 10
assert dataframe.iloc[0]["uniprotPosition"] == 12
assert "label" in dataframe.columns
+
+
+def test_fetch_peptides_modifications_returns_empty_dataframe(monkeypatch) -> None:
+ monkeypatch.setattr(
+ "apps.common.services.Scop3pRestApi.fetch_peptides",
+ lambda self, accession: {"peptides": []},
+ )
+
+ dataframe = Scop3PClient().fetch_peptides_modifications("O00571")
+ assert isinstance(dataframe, pd.DataFrame)
+ assert dataframe.empty
+
+
+def test_fetch_modifications_normalizes_position_column(monkeypatch) -> None:
+ monkeypatch.setattr(
+ "apps.common.services.Scop3pRestApi.fetch_modifications",
+ lambda self, accession: {
+ "modifications": [
+ {"position": "10", "residue": "S", "name": "Phosphorylation"},
+ {"position": "bad", "residue": "T", "name": "Phosphorylation"},
+ ]
+ },
+ )
+
+ dataframe = Scop3PClient().fetch_modifications("O00571")
+ assert dataframe["position"].tolist() == [10, pd.NA]
+ assert str(dataframe["position"].dtype) == "Int64"
+
+
+def test_format_label_uses_placeholders_for_missing_numeric_values() -> None:
+ row = pd.Series(
+ {
+ "peptideSequence": "AAAA",
+ "peptideStart": pd.NA,
+ "peptideEnd": pd.NA,
+ "modifiedResidue": "S",
+ "uniprotPosition": pd.NA,
+ "score": "",
+ }
+ )
+
+ label = Scop3PClient._format_label(row)
+ assert label == "AAAA (?-?) @S? score="
diff --git a/tests/unit/test_structure_ops.py b/tests/unit/test_structure_ops.py
index 6354fca..2d3a6e6 100644
--- a/tests/unit/test_structure_ops.py
+++ b/tests/unit/test_structure_ops.py
@@ -1,9 +1,11 @@
from __future__ import annotations
from pathlib import Path
+import pytest
from common.structure_viz import StructureVizService
from common.structure_viz import StructureOps
+from structure_viz.app import _tm_source_signature
PDB_MINI = """ATOM 1 N ALA A 1 11.104 13.207 2.100 1.00 10.00 N
@@ -14,6 +16,19 @@
END
"""
+PDB_MULTI_CHAIN = """ATOM 1 N ALA A 5 11.104 13.207 2.100 1.00 10.00 N
+ATOM 2 CA ALA A 5 12.200 12.300 2.300 1.00 10.00 C
+ATOM 3 N GLY A 10 13.104 14.207 3.100 1.00 10.00 N
+ATOM 4 CA GLY A 10 14.200 14.300 3.300 1.00 10.00 C
+TER
+ATOM 5 N SER B 200 21.104 23.207 2.100 1.00 10.00 N
+ATOM 6 CA SER B 200 22.200 22.300 2.300 1.00 10.00 C
+ATOM 7 N TYR B 210 23.104 24.207 3.100 1.00 10.00 N
+ATOM 8 CA TYR B 210 24.200 24.300 3.300 1.00 10.00 C
+TER
+END
+"""
+
def test_bfactor_pdb_rewrites_selected_metric(tmp_path: Path) -> None:
source = tmp_path / "in.pdb"
@@ -37,6 +52,14 @@ def test_chain_range_from_pdb(tmp_path: Path) -> None:
assert end == 2
+def test_chain_ranges_from_pdb_returns_all_available_ranges(tmp_path: Path) -> None:
+ source = tmp_path / "multi_chain.pdb"
+ source.write_text(PDB_MULTI_CHAIN)
+
+ ranges = StructureOps.chain_ranges_from_pdb(source)
+ assert ranges == {"A": (5, 10), "B": (200, 210)}
+
+
def test_run_tmalign_finds_existing_candidate(monkeypatch, tmp_path: Path) -> None:
pdb1 = tmp_path / "a.pdb"
pdb2 = tmp_path / "b.pdb"
@@ -59,6 +82,23 @@ def _mock_run(cmd, cwd, capture_output, text, check): # noqa: ANN001
assert report == "Aligned\n"
+def test_run_tmalign_raises_runtime_error_with_subprocess_output(monkeypatch, tmp_path: Path) -> None:
+ pdb1 = tmp_path / "a.pdb"
+ pdb2 = tmp_path / "b.pdb"
+ pdb1.write_text(PDB_MINI)
+ pdb2.write_text(PDB_MINI)
+
+ def _mock_run(cmd, cwd, capture_output, text, check): # noqa: ANN001
+ error = __import__("subprocess").CalledProcessError(139, cmd, output="")
+ error.stderr = "segmentation fault"
+ raise error
+
+ monkeypatch.setattr("apps.common.structure_viz.subprocess.run", _mock_run)
+
+ with pytest.raises(RuntimeError, match="segmentation fault"):
+ StructureOps.run_tmalign(pdb1, pdb2, tmp_path, out_name="aligned")
+
+
def test_resolve_uploaded_or_remote_pdb_prefers_upload(tmp_path: Path) -> None:
service = StructureVizService(tmp_path)
source = tmp_path / "upload_source.pdb"
@@ -72,3 +112,52 @@ def test_resolve_uploaded_or_remote_pdb_prefers_upload(tmp_path: Path) -> None:
assert resolved == tmp_path / "copied.pdb"
assert resolved.read_text() == PDB_MINI
+
+
+def test_validate_pdb_id_accepts_2ivt_and_rejects_bad_id() -> None:
+ assert StructureOps.validate_pdb_id("2IVT") == "2IVT"
+ with pytest.raises(ValueError, match="Expected a 4-character RCSB identifier"):
+ StructureOps.validate_pdb_id("21VTX")
+
+
+def test_tm_source_signature_prefers_upload_then_normalizes_pdb_id() -> None:
+ upload = [{"datapath": "/tmp/file.pdb", "name": "2IVT.pdb"}]
+ assert _tm_source_signature(upload, "1CRN") == ("upload", "/tmp/file.pdb|2IVT.pdb")
+ assert _tm_source_signature(None, "2ivt") == ("pdb", "2IVT")
+ assert _tm_source_signature(None, "") is None
+
+
+def test_save_chain_segment_rejects_missing_chain(tmp_path: Path) -> None:
+ source = tmp_path / "source.pdb"
+ source.write_text(PDB_MINI)
+
+ with pytest.raises(ValueError, match="Chain 'B' not found"):
+ StructureOps.save_chain_segment(source, tmp_path / "seg.pdb", "B", 1, 2)
+
+
+def test_save_chain_segment_rejects_empty_range(tmp_path: Path) -> None:
+ source = tmp_path / "source.pdb"
+ source.write_text(PDB_MINI)
+
+ with pytest.raises(ValueError, match="outside chain 'A' range 1-2"):
+ StructureOps.save_chain_segment(source, tmp_path / "seg.pdb", "A", 5, 10)
+
+
+def test_download_pdb_uses_uppercase_2ivt_url(monkeypatch, tmp_path: Path) -> None:
+ service = StructureVizService(tmp_path)
+
+ class _Response:
+ content = PDB_MINI.encode()
+
+ def raise_for_status(self) -> None:
+ return None
+
+ def _mock_get(url, timeout): # noqa: ANN001
+ assert url == "https://files.rcsb.org/download/2IVT.pdb"
+ return _Response()
+
+ monkeypatch.setattr("apps.common.structure_viz.requests.get", _mock_get)
+
+ downloaded = service.download_pdb("2ivt")
+ assert downloaded.name == "2IVT.pdb"
+ assert downloaded.read_text() == PDB_MINI
diff --git a/tests/unit/test_structure_viz_app.py b/tests/unit/test_structure_viz_app.py
new file mode 100644
index 0000000..7099051
--- /dev/null
+++ b/tests/unit/test_structure_viz_app.py
@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+import pandas as pd
+
+from structure_viz.app import _b2b_table_dataframe, _selected_b2b_metric_column
+
+
+def test_selected_b2b_metric_column_uses_toggle_state() -> None:
+ assert _selected_b2b_metric_column("backbone", normalized=False) == "backbone"
+ assert _selected_b2b_metric_column("backbone", normalized=True) == "backbone_normalized"
+ assert _selected_b2b_metric_column(None, normalized=True) is None
+
+
+def test_b2b_table_dataframe_switches_between_raw_and_normalized_values() -> None:
+ dataframe = pd.DataFrame(
+ {
+ "Position": [1, 2],
+ "Amino acid": ["A", "C"],
+ "backbone": [0.1, 0.2],
+ "sidechain": [0.3, 0.4],
+ "ppII": [0.5, 0.6],
+ "coil": [0.7, 0.8],
+ "sheet": [0.9, 1.0],
+ "helix": [1.1, 1.2],
+ "earlyFolding": [1.3, 1.4],
+ "disoMine": [1.5, 1.6],
+ "backbone_normalized": [0.0, 1.0],
+ "sidechain_normalized": [0.0, 1.0],
+ "ppII_normalized": [0.0, 1.0],
+ "coil_normalized": [0.0, 1.0],
+ "sheet_normalized": [0.0, 1.0],
+ "helix_normalized": [0.0, 1.0],
+ "earlyFolding_normalized": [0.0, 1.0],
+ "disoMine_normalized": [0.0, 1.0],
+ }
+ )
+
+ raw_table = _b2b_table_dataframe(dataframe, normalized=False)
+ normalized_table = _b2b_table_dataframe(dataframe, normalized=True)
+
+ assert list(raw_table.columns) == [
+ "Position",
+ "Amino acid",
+ "backbone",
+ "sidechain",
+ "ppII",
+ "coil",
+ "sheet",
+ "helix",
+ "earlyFolding",
+ "disoMine",
+ ]
+ assert list(normalized_table.columns) == list(raw_table.columns)
+ assert raw_table.loc[0, "backbone"] == 0.1
+ assert normalized_table.loc[0, "backbone"] == 0.0
+ assert raw_table.loc[1, "disoMine"] == 1.6
+ assert normalized_table.loc[1, "disoMine"] == 1.0
diff --git a/tests/unit/test_structure_viz_service.py b/tests/unit/test_structure_viz_service.py
new file mode 100644
index 0000000..9d9be2b
--- /dev/null
+++ b/tests/unit/test_structure_viz_service.py
@@ -0,0 +1,121 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pandas as pd
+
+from common.structure_viz import B2B_METRIC_COLUMNS, StructureVizService
+
+
+class _TextResponse:
+ def __init__(self, text: str) -> None:
+ self.text = text
+
+ def raise_for_status(self) -> None:
+ return None
+
+
+def test_fetch_ptms_normalizes_position_column(monkeypatch, tmp_path: Path) -> None:
+ monkeypatch.setattr(
+ "apps.common.services.Scop3pRestApi.fetch_modifications",
+ lambda self, accession: {
+ "modifications": [
+ {"residue": "S", "position": "10", "name": "Phosphorylation"},
+ {"residue": "T", "position": "bad", "name": "Phosphorylation"},
+ ]
+ },
+ )
+
+ dataframe = StructureVizService(tmp_path).fetch_ptms("P12345")
+ assert dataframe["position"].tolist() == [10, pd.NA]
+ assert str(dataframe["position"].dtype) == "Int64"
+
+
+def test_fetch_ptms_returns_empty_dataframe(monkeypatch, tmp_path: Path) -> None:
+ monkeypatch.setattr(
+ "apps.common.services.Scop3pRestApi.fetch_modifications",
+ lambda self, accession: {"modifications": []},
+ )
+
+ dataframe = StructureVizService(tmp_path).fetch_ptms("P12345")
+ assert dataframe.empty
+
+
+def test_fetch_sequence_parses_fasta(monkeypatch, tmp_path: Path) -> None:
+ monkeypatch.setattr(
+ "apps.common.structure_viz.requests.get",
+ lambda *args, **kwargs: _TextResponse(">sp|P12345|\nACD\nEFG\n"),
+ )
+
+ sequence = StructureVizService(tmp_path).fetch_sequence("P12345")
+ assert sequence == "ACDEFG"
+
+
+def test_normalize_b2b_prediction_uses_expected_columns_and_types(tmp_path: Path) -> None:
+ service = StructureVizService(tmp_path)
+ dataframe = service._normalize_b2b_prediction(
+ {
+ "seq": "AC",
+ "backbone": ["0.1", "0.2"],
+ "sidechain": ["0.3", "0.4"],
+ "ppII": ["0.5", "0.6"],
+ "coil": ["0.7", "0.8"],
+ "sheet": ["0.9", "1.0"],
+ "helix": ["1.1", "1.2"],
+ "earlyFolding": ["1.3", "1.4"],
+ "disoMine": ["1.5", "1.6"],
+ }
+ )
+
+ assert list(dataframe.columns) == [
+ "Position",
+ "Amino acid",
+ *B2B_METRIC_COLUMNS,
+ *(f"{metric}_normalized" for metric in B2B_METRIC_COLUMNS),
+ ]
+ assert dataframe["Position"].tolist() == [1, 2]
+ assert dataframe["Amino acid"].tolist() == ["AC", "AC"]
+ assert dataframe.loc[0, "backbone"] == 0.1
+ assert dataframe.loc[1, "disoMine"] == 1.6
+ assert dataframe.loc[0, "backbone_normalized"] == 0.0
+ assert dataframe.loc[1, "backbone_normalized"] == 1.0
+ assert dataframe.loc[0, "disoMine_normalized"] == 0.0
+ assert dataframe.loc[1, "disoMine_normalized"] == 1.0
+
+
+def test_normalize_b2b_prediction_tolerates_missing_and_uneven_fields(tmp_path: Path) -> None:
+ service = StructureVizService(tmp_path)
+ dataframe = service._normalize_b2b_prediction(
+ {
+ "seq": "ACD",
+ "backbone": [0.1, 0.2, 0.3],
+ "sidechain": [0.4],
+ "ppII": None,
+ "coil": [0.5, 0.6, 0.7, 0.8],
+ "earlyFolding": [0.9, 1.0],
+ }
+ )
+
+ assert dataframe["Position"].tolist() == [1, 2, 3]
+ assert dataframe["Amino acid"].tolist() == ["ACD", "ACD", "ACD"]
+ assert dataframe.loc[0, "sidechain"] == 0.4
+ assert pd.isna(dataframe.loc[1, "sidechain"])
+ assert pd.isna(dataframe.loc[2, "sidechain"])
+ assert dataframe["coil"].tolist() == [0.5, 0.6, 0.7]
+ assert dataframe["ppII"].isna().all()
+ assert dataframe["ppII_normalized"].isna().all()
+ assert dataframe.loc[0, "coil_normalized"] == 0.0
+ assert dataframe.loc[2, "coil_normalized"] == 1.0
+
+
+def test_normalize_b2b_prediction_uses_zero_for_constant_metric_series(tmp_path: Path) -> None:
+ service = StructureVizService(tmp_path)
+ dataframe = service._normalize_b2b_prediction(
+ {
+ "seq": "AC",
+ "backbone": [0.7, 0.7],
+ }
+ )
+
+ assert dataframe["backbone"].tolist() == [0.7, 0.7]
+ assert dataframe["backbone_normalized"].tolist() == [0.0, 0.0]