From f4dcbfef9f44602e0aca538aa4f21628d647526c Mon Sep 17 00:00:00 2001 From: Thomas Shin Date: Tue, 22 Sep 2026 09:18:05 -0400 Subject: [PATCH 1/2] Add local and online PfamScan support --- src/lverage/pfam.py | 240 +++++++++++++++++++++++++++++++++++ tests/v2/test_local_pfam.py | 77 +++++++++++ tests/v2/test_online_pfam.py | 101 +++++++++++++++ 3 files changed, 418 insertions(+) create mode 100644 src/lverage/pfam.py create mode 100644 tests/v2/test_local_pfam.py create mode 100644 tests/v2/test_online_pfam.py diff --git a/src/lverage/pfam.py b/src/lverage/pfam.py new file mode 100644 index 0000000..427de19 --- /dev/null +++ b/src/lverage/pfam.py @@ -0,0 +1,240 @@ +"""Pfam-based domain scanners.""" + +import logging +import shutil +import subprocess +import tempfile +import time + +import requests + +from .domain_scanner import DomainRecord, DomainScannerTemplate + + +LOGGER = logging.getLogger(__name__) +PFAMSCAN_URL = "https://www.ebi.ac.uk/Tools/services/rest/pfamscan/" +PFAMSCAN_RUN_URL = PFAMSCAN_URL + "run/" +PFAMSCAN_STATUS_URL = PFAMSCAN_URL + "status/" +PFAMSCAN_RESULT_TYPES_URL = PFAMSCAN_URL + "resulttypes/" +PFAMSCAN_RESULT_URL = PFAMSCAN_URL + "result/" +FAILED_STATUSES = {"ERROR", "FAILURE", "NOT_FOUND"} + + +def _resolve_executable(executable : str) -> str: + resolved_executable = shutil.which(executable) + if resolved_executable is None: + raise FileNotFoundError(f"PfamScan executable not found: {executable}") + return resolved_executable + + +def _validate_sequence(sequence : str) -> None: + if not isinstance(sequence, str) or not sequence.strip(): + raise ValueError("sequence must be a nonempty string") + + +def _make_domain(result) -> DomainRecord: + try: + name = result["name"] + accession = result["acc"] + start = int(result["env"]["from"]) - 1 + end = int(result["env"]["to"]) + except (KeyError, TypeError, ValueError) as error: + raise ValueError("invalid PfamScan domain result") from error + if not isinstance(name, str) or not name: + raise ValueError("invalid PfamScan domain name") + if not isinstance(accession, str) or not accession: + raise ValueError("invalid PfamScan domain accession") + if start < 0 or end <= start: + raise ValueError("invalid PfamScan domain coordinates") + return DomainRecord(name, accession, start, end) + + +def _parse_online_results(results) -> list[DomainRecord]: + if not isinstance(results, list): + raise ValueError("PfamScan result must be a list") + domains = [] + for result in results: + try: + domains.append(_make_domain(result)) + except ValueError: + LOGGER.warning("Skipping malformed PfamScan domain", exc_info=True) + return domains + + +def _parse_local_results(output) -> list[DomainRecord]: + domains = [] + for line in output.splitlines(): + if not line.strip() or line.lstrip().startswith("#"): + continue + fields = line.split() + if len(fields) < 7: + LOGGER.warning("Skipping malformed local PfamScan result: %s", line) + continue + try: + domains.append(_make_domain({ + "name": fields[6], + "acc": fields[5], + "env": {"from": fields[3], "to": fields[4]}, + })) + except ValueError: + LOGGER.warning("Skipping malformed local PfamScan result: %s", line) + return domains + + +class LocalPfamScanner(DomainScannerTemplate): + """ + Scan protein sequences with a local PfamScan executable. + + Parameters + ---------- + database_path : str + PfamScan database directory + pfamscan_path : str, optional + Executable name or path for ``pfam_scan.pl`` + """ + + def __init__(self, database_path, pfamscan_path="pfam_scan.pl"): + if not isinstance(database_path, str) or not database_path.strip(): + raise ValueError("database_path must be a nonempty string") + + self.database_path = database_path + self.pfamscan_path = _resolve_executable(pfamscan_path) + + def get_domains(self, sequence : str) -> list[DomainRecord]: + """ + Scan a protein sequence with the local PfamScan database. + + Parameters + ---------- + sequence : str + Protein sequence to scan + + Returns + ------- + list + Domain records parsed from PfamScan output + """ + + _validate_sequence(sequence) + with tempfile.NamedTemporaryFile(mode="w", suffix=".fasta") as query_file: + query_file.write(f">query\n{sequence}\n") + query_file.flush() + result = subprocess.run( + [ + self.pfamscan_path, + "-fasta", query_file.name, + "-dir", self.database_path, + ], + capture_output=True, + check=True, + text=True, + ) + return _parse_local_results(result.stdout) + + +class OnlinePfamScanner(DomainScannerTemplate): + """ + Scan protein sequences with the EMBL-EBI online PfamScan service. + + Parameters + ---------- + email : str + Contact email supplied to the service + try_count : int, optional + Maximum number of status requests + time_interval : float, optional + Seconds to wait between status requests + request_timeout_seconds : float, optional + Maximum duration of each HTTP request + """ + + def __init__(self, + email, + try_count=100, + time_interval=5, + request_timeout_seconds=30): + if not isinstance(email, str) or not email.strip() or "@" not in email: + raise ValueError("email must be a nonempty email address") + if isinstance(try_count, bool) or not isinstance(try_count, int) or try_count <= 0: + raise ValueError("try_count must be a positive integer") + if not isinstance(time_interval, (int, float)) or time_interval < 0: + raise ValueError("time_interval must be nonnegative") + if not isinstance(request_timeout_seconds, (int, float)) or request_timeout_seconds <= 0: + raise ValueError("request_timeout_seconds must be positive") + + self.email = email + self.try_count = try_count + self.time_interval = time_interval + self.request_timeout_seconds = request_timeout_seconds + + def get_domains(self, sequence : str) -> list[DomainRecord]: + """ + Submit a protein sequence to the online PfamScan service. + + Parameters + ---------- + sequence : str + Protein sequence to scan + + Returns + ------- + list + Domain records returned by PfamScan + """ + + _validate_sequence(sequence) + response = requests.post( + PFAMSCAN_RUN_URL, + data={ + "email": self.email, + "sequence": sequence, + "format": "json", + }, + timeout=self.request_timeout_seconds, + ) + response.raise_for_status() + job_id = response.text.strip() + if not job_id: + raise ValueError("PfamScan response did not contain a job identifier") + + for attempt in range(self.try_count): + status_response = requests.get( + PFAMSCAN_STATUS_URL + job_id, + timeout=self.request_timeout_seconds, + ) + status_response.raise_for_status() + status = status_response.text.strip().upper() + if status == "QUEUED" or status == "RUNNING": + if attempt + 1 < self.try_count: + time.sleep(self.time_interval) + continue + if status == "FINISHED": + break + if status in FAILED_STATUSES: + raise RuntimeError(f"PfamScan job failed with status: {status}") + raise RuntimeError(f"Unknown PfamScan job status: {status}") + else: + raise TimeoutError("PfamScan job did not finish within try_count") + + result_types_response = requests.get( + PFAMSCAN_RESULT_TYPES_URL + job_id, + timeout=self.request_timeout_seconds, + ) + result_types_response.raise_for_status() + result_types = result_types_response.json() + json_result_type = next( + ( + result_type for result_type in result_types + if result_type.get("mediaType", "").casefold() == "application/json" + ), + None, + ) + if json_result_type is None: + raise RuntimeError("PfamScan did not provide a JSON result type") + + result_response = requests.get( + PFAMSCAN_RESULT_URL + job_id + "/" + json_result_type["identifier"], + timeout=self.request_timeout_seconds, + ) + result_response.raise_for_status() + return _parse_online_results(result_response.json()) \ No newline at end of file diff --git a/tests/v2/test_local_pfam.py b/tests/v2/test_local_pfam.py new file mode 100644 index 0000000..efd5a33 --- /dev/null +++ b/tests/v2/test_local_pfam.py @@ -0,0 +1,77 @@ +import subprocess +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + +from lverage.domain_scanner import DomainRecord +from lverage.pfam import LocalPfamScanner + + +class LocalPfamScannerTests(unittest.TestCase): + + def setUp(self): + self.which_patcher = patch( + "lverage.pfam.shutil.which", + side_effect=lambda path: f"/resolved/{path}", + ) + self.run_patcher = patch("lverage.pfam.subprocess.run") + self.which_patcher.start() + self.mock_run = self.run_patcher.start() + self.addCleanup(self.which_patcher.stop) + self.addCleanup(self.run_patcher.stop) + + def test_constructor_resolves_executable_and_copies_database(self): + scanner = LocalPfamScanner("pfamdb", "custom-pfam_scan.pl") + + self.assertEqual(scanner.database_path, "pfamdb") + self.assertEqual(scanner.pfamscan_path, "/resolved/custom-pfam_scan.pl") + self.assertIsInstance(scanner, LocalPfamScanner) + + def test_scan_builds_command_and_writes_temporary_fasta(self): + scanner = LocalPfamScanner("pfamdb") + captured = {} + + def run_scan(command, **kwargs): + captured["path"] = command[2] + captured["contents"] = Path(command[2]).read_text() + return SimpleNamespace(stdout=( + "# comment\n" + "query 1 8 2 7 PF00046.1 Homeobox Domain 1 6 60\n" + )) + + self.mock_run.side_effect = run_scan + + domains = scanner.get_domains("MPEPTIDE") + + self.assertEqual(domains, [DomainRecord("Homeobox", "PF00046.1", 1, 7)]) + self.assertEqual(captured["contents"], ">query\nMPEPTIDE\n") + self.assertFalse(Path(captured["path"]).exists()) + command = self.mock_run.call_args.args[0] + self.assertEqual(command[0], "/resolved/pfam_scan.pl") + self.assertEqual(command[1], "-fasta") + self.assertTrue(command[2].endswith(".fasta")) + self.assertEqual(command[3:], ["-dir", "pfamdb"]) + + def test_empty_scan_returns_no_domains(self): + scanner = LocalPfamScanner("pfamdb") + self.mock_run.return_value = SimpleNamespace(stdout="# no hits\n") + + self.assertEqual(scanner.get_domains("QUERY"), []) + + def test_scan_process_failure_propagates(self): + scanner = LocalPfamScanner("pfamdb") + self.mock_run.side_effect = subprocess.CalledProcessError(1, "pfam_scan.pl") + + with self.assertRaises(subprocess.CalledProcessError): + scanner.get_domains("QUERY") + + def test_empty_sequence_is_rejected(self): + scanner = LocalPfamScanner("pfamdb") + + with self.assertRaises(ValueError): + scanner.get_domains(" ") + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/v2/test_online_pfam.py b/tests/v2/test_online_pfam.py new file mode 100644 index 0000000..368dcdc --- /dev/null +++ b/tests/v2/test_online_pfam.py @@ -0,0 +1,101 @@ +import unittest +from types import SimpleNamespace +from unittest.mock import patch + +import requests + +from lverage.domain_scanner import DomainRecord +from lverage.pfam import OnlinePfamScanner + + +class OnlinePfamScannerTests(unittest.TestCase): + + def setUp(self): + self.post_patcher = patch("lverage.pfam.requests.post") + self.get_patcher = patch("lverage.pfam.requests.get") + self.sleep_patcher = patch("lverage.pfam.time.sleep") + self.mock_post = self.post_patcher.start() + self.mock_get = self.get_patcher.start() + self.sleep_patcher.start() + self.addCleanup(self.post_patcher.stop) + self.addCleanup(self.get_patcher.stop) + self.addCleanup(self.sleep_patcher.stop) + + def make_response(self, text="", json_data=None): + response = SimpleNamespace(text=text, json=lambda: json_data) + response.raise_for_status = unittest.mock.Mock() + return response + + def make_scanner(self, **kwargs): + return OnlinePfamScanner("researcher@example.org", **kwargs) + + def test_scan_submits_polls_and_parses_domains(self): + self.mock_post.return_value = self.make_response("job-123\n") + self.mock_get.side_effect = [ + self.make_response("RUNNING"), + self.make_response("FINISHED"), + self.make_response(json_data=[ + {"identifier": "txt", "mediaType": "text/plain"}, + {"identifier": "json", "mediaType": "application/json"}, + ]), + self.make_response(json_data=[{ + "name": "Homeobox", + "acc": "PF00046.1", + "env": {"from": 2, "to": 7}, + }]), + ] + + domains = self.make_scanner(time_interval=2).get_domains("MPEPTIDE") + + self.assertEqual(domains, [DomainRecord("Homeobox", "PF00046.1", 1, 7)]) + self.mock_post.assert_called_once_with( + "https://www.ebi.ac.uk/Tools/services/rest/pfamscan/run/", + data={ + "email": "researcher@example.org", + "sequence": "MPEPTIDE", + "format": "json", + }, + timeout=30, + ) + self.assertEqual(self.mock_get.call_count, 4) + self.assertEqual(self.mock_get.call_args_list[0].args[0].rsplit("/", 1)[-1], "job-123") + self.assertEqual(self.mock_get.call_args_list[2].args[0].rsplit("/", 1)[-1], "job-123") + self.assertEqual(self.mock_get.call_args_list[3].args[0].rsplit("/", 1)[-1], "json") + + def test_failed_job_raises(self): + for status in ("ERROR", "FAILURE", "NOT_FOUND"): + with self.subTest(status=status): + self.mock_post.return_value = self.make_response("job-123") + self.mock_get.return_value = self.make_response(status) + + with self.assertRaises(RuntimeError): + self.make_scanner().get_domains("QUERY") + + def test_job_timeout_raises(self): + self.mock_post.return_value = self.make_response("job-123") + self.mock_get.return_value = self.make_response("RUNNING") + + with self.assertRaises(TimeoutError): + self.make_scanner(try_count=2).get_domains("QUERY") + + def test_unknown_job_status_raises(self): + self.mock_post.return_value = self.make_response("job-123") + self.mock_get.return_value = self.make_response("PAUSED") + + with self.assertRaises(RuntimeError): + self.make_scanner().get_domains("QUERY") + + def test_http_error_propagates(self): + self.mock_post.return_value = self.make_response("job-123") + self.mock_post.return_value.raise_for_status.side_effect = requests.HTTPError("unavailable") + + with self.assertRaises(requests.HTTPError): + self.make_scanner().get_domains("QUERY") + + def test_empty_sequence_is_rejected(self): + with self.assertRaises(ValueError): + self.make_scanner().get_domains(" ") + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file From 5c343031df969b8dc76bb7030ca57ca8c9e0bf0c Mon Sep 17 00:00:00 2001 From: Anthony Garza Date: Mon, 28 Sep 2026 13:22:56 -0400 Subject: [PATCH 2/2] Make local Pfam temp FASTA handling Windows-safe --- src/lverage/pfam.py | 19 ++++++++++++++----- 1 file changed, 14 insertions(+), 5 deletions(-) diff --git a/src/lverage/pfam.py b/src/lverage/pfam.py index 427de19..874b089 100644 --- a/src/lverage/pfam.py +++ b/src/lverage/pfam.py @@ -1,6 +1,7 @@ """Pfam-based domain scanners.""" import logging +import os import shutil import subprocess import tempfile @@ -116,19 +117,27 @@ def get_domains(self, sequence : str) -> list[DomainRecord]: """ _validate_sequence(sequence) - with tempfile.NamedTemporaryFile(mode="w", suffix=".fasta") as query_file: - query_file.write(f">query\n{sequence}\n") - query_file.flush() + fd, query_path = tempfile.mkstemp(prefix="lverage_pfam_", suffix=".fasta") + try: + with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as query_file: + query_file.write(f">query\n{sequence}\n") + result = subprocess.run( [ self.pfamscan_path, - "-fasta", query_file.name, + "-fasta", query_path, "-dir", self.database_path, ], capture_output=True, check=True, text=True, ) + finally: + try: + os.unlink(query_path) + except FileNotFoundError: + pass + return _parse_local_results(result.stdout) @@ -237,4 +246,4 @@ def get_domains(self, sequence : str) -> list[DomainRecord]: timeout=self.request_timeout_seconds, ) result_response.raise_for_status() - return _parse_online_results(result_response.json()) \ No newline at end of file + return _parse_online_results(result_response.json())