flax_model/alphafold3/data/tools/hmmalign.py

"""A Python wrapper for hmmalign from the HMMER Suite."""

from collections.abc import Mapping, Sequence
import os
import tempfile

from flax_model.alphafold3.data import parsers
from flax_model.alphafold3.data.tools import subprocess_utils


def _to_a3m(sequences: Sequence[str], name_prefix: str = 'sequence') -> str:
  a3m = ''
  for i, sequence in enumerate(sequences, 1):
    a3m += f'> {name_prefix} {i}\n{sequence}\n'
  return a3m


class Hmmalign:
  """Python wrapper of the hmmalign binary."""

  def __init__(self, binary_path: str):
    """Initializes the Python hmmalign wrapper.

    Args:
      binary_path: Path to the hmmalign binary.

    Raises:
      RuntimeError: If hmmalign binary not found within the path.
    """
    self._binary_path = binary_path

    subprocess_utils.check_binary_exists(path=self._binary_path, name='hmmalign')

  def align_sequences(
      self,
      sequences: Sequence[str],
      profile: str,
      extra_flags: Mapping[str, str] | None = None,
  ) -> str:
    """Aligns sequence list to the profile and returns the alignment in A3M."""
    return self.align(
        a3m_str=_to_a3m(sequences, name_prefix='query'),
        profile=profile,
        extra_flags=extra_flags,
    )

  def align(
      self,
      a3m_str: str,
      profile: str,
      extra_flags: Mapping[str, str] | None = None,
  ) -> str:
    """Aligns sequences in A3M to the profile and returns the alignment in A3M.

    Args:
      a3m_str: A list of sequence strings.
      profile: A hmm file with the hmm profile to align the sequences to.
      extra_flags: Dictionary with extra flags, flag_name: flag_value, that are
        added to hmmalign.

    Returns:
      An A3M string with the aligned sequences.

    Raises:
      RuntimeError: If hmmalign fails.
    """
    with tempfile.TemporaryDirectory() as query_tmp_dir:
      input_profile = os.path.join(query_tmp_dir, 'profile.hmm')
      input_sequences = os.path.join(query_tmp_dir, 'sequences.a3m')
      output_a3m_path = os.path.join(query_tmp_dir, 'output.a3m')

      with open(input_profile, 'w') as f:
        f.write(profile)

      with open(input_sequences, 'w') as f:
        f.write(a3m_str)

      cmd = [
          self._binary_path,
          *('-o', output_a3m_path),
          *('--outformat', 'A2M'),  # A2M is A3M in the HMMER suite.
      ]
      if extra_flags:
        for flag_name, flag_value in extra_flags.items():
          cmd.extend([flag_name, flag_value])
      cmd.extend([input_profile, input_sequences])

      subprocess_utils.run(
          cmd=cmd,
          cmd_name='hmmalign',
          log_stdout=False,
          log_stderr=True,
          log_on_process_error=True,
      )

      with open(output_a3m_path, encoding='utf-8') as f:
        a3m = f.read()

    return a3m

  def align_sequences_to_profile(self, profile: str, sequences_a3m: str) -> str:
    """Aligns the sequences to profile and returns the alignment in A3M string.

    Uses hmmalign to align the sequences to the profile, then ouputs the
    sequence contatenated at the beginning of the sequences in the A3M format.
    As the sequences are represented by an alignment with possible gaps ('-')
    and insertions (lowercase characters), the method first removes the gaps,
    then uppercases the insertions to prepare the sequences for realignment.
    Sequences with gaps cannot be aligned, as '-'s are not a valid symbol to
    align; lowercase characters must be uppercased to preserve the original
    sequences before realignment.

    Args:
      profile: The Hmmbuild profile to align the sequences to.
      sequences_a3m: Sequences in A3M format to align to the profile.

    Returns:
      An A3M string with the aligned sequences.

    Raises:
      RuntimeError: If hmmalign fails.
    """
    deletion_table = str.maketrans('', '', '-')
    sequences_no_gaps_a3m = []
    for seq, desc in parsers.lazy_parse_fasta_string(sequences_a3m):
      sequences_no_gaps_a3m.append(f'>{desc}')
      sequences_no_gaps_a3m.append(seq.translate(deletion_table))
    sequences_no_gaps_a3m = '\n'.join(sequences_no_gaps_a3m)

    aligned_sequences = self.align(sequences_no_gaps_a3m, profile)

    return aligned_sequences