File size: 4,281 Bytes
35cdf53 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 |
"""A Python wrapper for hmmbuild - construct HMM profiles from MSA."""
import os
import re
import tempfile
from typing import Literal
from flax_model.alphafold3.data import parsers
from flax_model.alphafold3.data.tools import subprocess_utils
class Hmmbuild(object):
"""Python wrapper of the hmmbuild binary."""
def __init__(
self,
*,
binary_path: str,
singlemx: bool = False,
alphabet: str | None = None,
):
"""Initializes the Python hmmbuild wrapper.
Args:
binary_path: The path to the hmmbuild executable.
singlemx: Whether to use --singlemx flag. If True, it forces HMMBuild to
just use a common substitution score matrix.
alphabet: The alphabet to assert when building a profile. Useful when
hmmbuild cannot guess the alphabet. If None, no alphabet is asserted.
Raises:
RuntimeError: If hmmbuild binary not found within the path.
"""
self._binary_path = binary_path
self._singlemx = singlemx
self._alphabet = alphabet
subprocess_utils.check_binary_exists(path=self._binary_path, name='hmmbuild')
def build_profile_from_sto(self, sto: str, model_construction='fast') -> str:
"""Builds a HHM for the aligned sequences given as an A3M string.
Args:
sto: A string with the aligned sequences in the Stockholm format.
model_construction: Whether to use reference annotation in the msa to
determine consensus columns ('hand') or default ('fast').
Returns:
A string with the profile in the HMM format.
Raises:
RuntimeError: If hmmbuild fails.
"""
return self._build_profile(
sto, informat='stockholm', model_construction=model_construction
)
def build_profile_from_a3m(self, a3m: str) -> str:
"""Builds a HHM for the aligned sequences given as an A3M string.
Args:
a3m: A string with the aligned sequences in the A3M format.
Returns:
A string with the profile in the HMM format.
Raises:
RuntimeError: If hmmbuild fails.
"""
lines = []
for sequence, description in parsers.lazy_parse_fasta_string(a3m):
sequence = re.sub('[a-z]+', '', sequence) # Remove inserted residues.
lines.append(f'>{description}\n{sequence}\n')
msa = ''.join(lines)
return self._build_profile(msa, informat='afa')
def _build_profile(
self,
msa: str,
informat: Literal['afa', 'stockholm'],
model_construction: str = 'fast',
) -> str:
"""Builds a HMM for the aligned sequences given as an MSA string.
Args:
msa: A string with the aligned sequences, in A3M or STO format.
informat: One of 'afa' (aligned FASTA) or 'sto' (Stockholm).
model_construction: Whether to use reference annotation in the msa to
determine consensus columns ('hand') or default ('fast').
Returns:
A string with the profile in the HMM format.
Raises:
RuntimeError: If hmmbuild fails.
ValueError: If unspecified arguments are provided.
"""
if model_construction not in {'hand', 'fast'}:
raise ValueError(f'Bad {model_construction=}. Only hand or fast allowed.')
with tempfile.TemporaryDirectory() as query_tmp_dir:
input_msa_path = os.path.join(query_tmp_dir, 'query.msa')
output_hmm_path = os.path.join(query_tmp_dir, 'output.hmm')
with open(input_msa_path, 'w') as f:
f.write(msa)
# Specify the format as we don't specify the input file extension. See
# https://github.com/EddyRivasLab/hmmer/issues/321 for more details.
cmd_flags = ['--informat', informat]
# If adding flags, we have to do so before the output and input:
if model_construction == 'hand':
cmd_flags.append(f'--{model_construction}')
if self._singlemx:
cmd_flags.append('--singlemx')
if self._alphabet:
cmd_flags.append(f'--{self._alphabet}')
cmd_flags.extend([output_hmm_path, input_msa_path])
cmd = [self._binary_path, *cmd_flags]
subprocess_utils.run(
cmd=cmd,
cmd_name='Hmmbuild',
log_stdout=False,
log_stderr=True,
log_on_process_error=True,
)
with open(output_hmm_path) as f:
hmm = f.read()
return hmm
|