<commit_before>"""
Date of Creation: 11th Dec 2014.

Description:   Code for reading/writing genome alignments.

Copyright (C) 2010-2014
<NAME>,

Authors: <NAME>

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program.  If not, see <http://www.gnu.org/licenses/>.
"""

# standard python imports
import StringIO
import os
import os.path
import functools

# import for unit testing
import mock
import unittest

# pyokit imports
from pyokit.datastruct.genomeAlignment import GenomeAlignmentBlock
from pyokit.datastruct.genomeAlignment import JustInTimeGenomeAlignmentBlock
from pyokit.datastruct.genomeAlignment import JustInTimeGenomeAlignment
from pyokit.datastruct.genomeAlignment import GenomeAlignment
from pyokit.datastruct.sequence import Sequence
from pyokit.datastruct.sequence import UnknownSequence
from pyokit.io import maf
from pyokit.io.indexedFile import IndexedFile
from pyokit.util.progressIndicator import ProgressIndicator
from pyokit.io.ioError import PyokitIOError


# def genome_alignment_block_hash(b):
#  """Hash a genome alignment block to unique ID: its location in the genome."""
#  return b.chrom + "\t" + str(b.start) + "\t" + str(b.end)

###############################################################################
#                              HELPER FUNCTIONS                               #
###############################################################################

def __trim_extension_dot(ext):
  """trim leading dots from extension."""
  if ext is None:
    return None
  if ext == "":
    return ""
  while ext[0] == ".":
    ext = ext[1:]
  return ext


def __trim_extensions_dot(exts):
  """trim leading dots from extensions and drop any empty strings."""
  if exts is None:
    return None
  res = []
  for i in range(0, len(exts)):
    if exts[i] == "":
      continue
    res.append(__trim_extension_dot(exts[i]))
  return res


def __split_genomic_interval_filename(fn):
  """
  Split a filename of the format chrom:start-end.ext or chrom.ext (full chrom).

  :return: tuple of (chrom, start, end) -- 'start' and 'end' are None if not
           present in the filename.
  """
  if fn is None or fn == "":
    raise ValueError("invalid filename: " + str(fn))
  fn = ".".join(fn.split(".")[:-1])
  parts = fn.split(":")
  if len(parts) == 1:
    return (parts[0].strip(), None, None)
  else:
    r_parts = parts[1].split("-")
    if len(r_parts) != 2:
      raise ValueError("Invalid filename: " + str(fn))
    return (parts[0].strip(), int(r_parts[0]), int(r_parts[1]))


###############################################################################
#                MANAGING AN ON-DISK MULTI-FILE GENOME ALIGNMENT              #
###############################################################################

def load_just_in_time_genome_alignment(path, ref_spec, extensions=None,
                                       index_exts=None, fail_no_index=True,
                                       verbose=False):
    """Load a just-in-time genome alignment from a directory."""
    extensions = __trim_extensions_dot(extensions)
    index_exts = __trim_extensions_dot(index_exts)

    partial_chrom_files = {}
    whole_chrom_files = {}
    for fn in os.listdir(path):
      pth = os.path.join(path, fn)
      if os.path.isfile(pth):
        base, ext = os.path.splitext(pth)
        ext = __trim_extension_dot(ext)
        if extensions is None or ext in extensions:
          idx_path = __find_index(pth, index_exts)
          if idx_path is None and fail_no_index:
            raise PyokitIOError("No index file for " + fn)
          chrom, start, end = __split_genomic_interval_filename(fn)
          assert((start is None and end is None) or
                 (start is not None and end is not None))
          if start is None:
            if chrom in whole_chrom_files:
              raise PyokitIOError("multiple files for chrom " + chrom)
            whole_chrom_files[chrom] = (pth, idx_path)
          else:
            k = (chrom, start, end)
            if k in partial_chrom_files:
              raise PyokitIOError("multiple files for " + str(k))
            partial_chrom_files[k] = (pth, idx_path)

    def factory(k):
      pth, idx = k
      return build_genome_alignment_from_file(pth, ref_spec, idx, verbose)

    return JustInTimeGenomeAlignment(whole_chrom_files, partial_chrom_files,
                                     factory)


###############################################################################
#                   BULK LOADING GENOME ALIGNMENTS FROM DISK                  #
###############################################################################

def __find_index(alig_file_pth, idx_extensions):
  """
  Find an index file for a genome alignment file in the same directory.

  :param alig_file_path: path to the alignment file.
  :param idx_extensions: check for index files with these extensions
  :return: path to first index file that matches the name of the alignment file
           and has one of the specified extensions.
  """
  if idx_extensions is None:
    return None
  base, ext = os.path.splitext(alig_file_pth)
  for idx_ext in idx_extensions:
    candidate = base + os.extsep + idx_ext
    if os.path.isfile(candidate):
      return candidate<commit_msg>improved exception message when multiple files found covering the same region in on-disk genome alignment; added raising of more descriptive exception when trying to load on-disk genome alignment and requesting fail on missing index, but not providing any index extensions to look for.<commit_after>"""
Date of Creation: 11th Dec 2014.

Description:   Code for reading/writing genome alignments.

Copyright (C) 2010-2014
<NAME>,

Authors: <NAME>

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program.  If not, see <http://www.gnu.org/licenses/>.
"""

# standard python imports
import StringIO
import os
import os.path
import functools

# import for unit testing
import mock
import unittest

# pyokit imports
from pyokit.datastruct.genomeAlignment import GenomeAlignmentBlock
from pyokit.datastruct.genomeAlignment import JustInTimeGenomeAlignmentBlock
from pyokit.datastruct.genomeAlignment import JustInTimeGenomeAlignment
from pyokit.datastruct.genomeAlignment import GenomeAlignment
from pyokit.datastruct.sequence import Sequence
from pyokit.datastruct.sequence import UnknownSequence
from pyokit.io import maf
from pyokit.io.indexedFile import IndexedFile
from pyokit.util.progressIndicator import ProgressIndicator
from pyokit.io.ioError import PyokitIOError


# def genome_alignment_block_hash(b):
#  """Hash a genome alignment block to unique ID: its location in the genome."""
#  return b.chrom + "\t" + str(b.start) + "\t" + str(b.end)

###############################################################################
#                              HELPER FUNCTIONS                               #
###############################################################################

def __trim_extension_dot(ext):
  """trim leading dots from extension."""
  if ext is None:
    return None
  if ext == "":
    return ""
  while ext[0] == ".":
    ext = ext[1:]
  return ext


def __trim_extensions_dot(exts):
  """trim leading dots from extensions and drop any empty strings."""
  if exts is None:
    return None
  res = []
  for i in range(0, len(exts)):
    if exts[i] == "":
      continue
    res.append(__trim_extension_dot(exts[i]))
  return res


def __split_genomic_interval_filename(fn):
  """
  Split a filename of the format chrom:start-end.ext or chrom.ext (full chrom).

  :return: tuple of (chrom, start, end) -- 'start' and 'end' are None if not
           present in the filename.
  """
  if fn is None or fn == "":
    raise ValueError("invalid filename: " + str(fn))
  fn = ".".join(fn.split(".")[:-1])
  parts = fn.split(":")
  if len(parts) == 1:
    return (parts[0].strip(), None, None)
  else:
    r_parts = parts[1].split("-")
    if len(r_parts) != 2:
      raise ValueError("Invalid filename: " + str(fn))
    return (parts[0].strip(), int(r_parts[0]), int(r_parts[1]))


###############################################################################
#                MANAGING AN ON-DISK MULTI-FILE GENOME ALIGNMENT              #
###############################################################################

def load_just_in_time_genome_alignment(path, ref_spec, extensions=None,
                                       index_exts=None, fail_no_index=True,
                                       verbose=False):
    """Load a just-in-time genome alignment from a directory."""
    if index_exts is None and fail_no_index:
      raise ValueError("Failure on no index specified for loading genome " +
                       "alignment, but no index extensions specified")

    extensions = __trim_extensions_dot(extensions)
    index_exts = __trim_extensions_dot(index_exts)

    partial_chrom_files = {}
    whole_chrom_files = {}
    for fn in os.listdir(path):
      pth = os.path.join(path, fn)
      if os.path.isfile(pth):
        base, ext = os.path.splitext(pth)
        ext = __trim_extension_dot(ext)
        if extensions is None or ext in extensions:
          idx_path = __find_index(pth, index_exts)
          if idx_path is None and fail_no_index:
            raise PyokitIOError("No index file for " + fn)
          chrom, start, end = __split_genomic_interval_filename(fn)
          assert((start is None and end is None) or
                 (start is not None and end is not None))
          if start is None:
            if chrom in whole_chrom_files:
              raise PyokitIOError("multiple files for chrom " + chrom)
            whole_chrom_files[chrom] = (pth, idx_path)
          else:
            k = (chrom, start, end)
            if k in partial_chrom_files:
              other_pth = partial_chrom_files[k][0]
              raise PyokitIOError("multiple files for " + str(k) + " --> " +
                                  pth + " and " + other_pth)
            partial_chrom_files[k] = (pth, idx_path)

    def factory(k):
      pth, idx = k
      return build_genome_alignment_from_file(pth, ref_spec, idx, verbose)

    return JustInTimeGenomeAlignment(whole_chrom_files, partial_chrom_files,
                                     factory)


###############################################################################
#                   BULK LOADING GENOME ALIGNMENTS FROM DISK                  #
###############################################################################

def __find_index(alig_file_pth, idx_extensions):
  """
  Find an index file for a genome alignment file in the same directory.

  :param alig_file_path: path to the alignment file.
  :param idx_extensions: check for index files with these extensions
  :return: path to first index file that matches the name of the alignment file
           and has one of the specified extensions.
  """
  if idx_extensions is None:
    return None
  base, ext = os.path.splitext(alig_file_pth)
  for idx_ext in idx_extensions:
    candidate = base + os.extsep + idx_ext
    if os.path.isfile(candidate):
      return candidate
{'fraction_non_alphanumeric': 0.15514972030273116, 'fraction_numerical': 0.004359986837775584, 'mean_word_length': 3.1862947658402203, 'pattern_counts': {'":': 10, '<': 9, '<?xml version=': 0, '>': 10, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 7, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2954700', 'n_tokens_mistral': 3489, 'n_tokens_neox': 3299, 'n_words': 1131}