mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-10-25 10:24:38 +08:00
Use pysam instead of samtools binary
This will make it possible to remove the requirement of having `samtools` installed when testing tools, e.g. in the tools-iuc TravisCI setup: https://github.com/galaxyproject/tools-iuc/blob/master/.travis.yml#L59
This commit is contained in:
@@ -7,9 +7,13 @@ import logging
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import tempfile
|
||||
|
||||
try:
|
||||
import pysam
|
||||
except ImportError:
|
||||
pysam = None
|
||||
|
||||
from galaxy.util.compression_utils import get_fileobj
|
||||
|
||||
from .asserts import verify_assertions
|
||||
@@ -132,10 +136,14 @@ def _bam_to_sam(local_name, temp_name):
|
||||
temp_local = tempfile.NamedTemporaryFile(suffix='.sam', prefix='local_bam_converted_to_sam_')
|
||||
fd, temp_temp = tempfile.mkstemp(suffix='.sam', prefix='history_bam_converted_to_sam_')
|
||||
os.close(fd)
|
||||
command = 'samtools view -h -o "%s" "%s"' % (temp_local.name, local_name)
|
||||
check_command(command, 'Converting local (test-data) bam to sam')
|
||||
command = 'samtools view -h -o "%s" "%s"' % (temp_temp, temp_name)
|
||||
check_command(command, 'Converting history bam to sam ')
|
||||
try:
|
||||
pysam.view('-h', '-o%s' % temp_local.name, local_name)
|
||||
except Exception as e:
|
||||
raise Exception("Converting local (test-data) BAM to SAM failed: %s" % e)
|
||||
try:
|
||||
pysam.view('-h', '-o%s' % temp_temp, temp_name)
|
||||
except Exception as e:
|
||||
raise Exception("Converting history BAM to SAM failed: %s" % e)
|
||||
os.remove(temp_name)
|
||||
return temp_local, temp_temp
|
||||
|
||||
@@ -153,18 +161,6 @@ def _verify_checksum(data, checksum_type, expected_checksum_value):
|
||||
raise AssertionError(message)
|
||||
|
||||
|
||||
def check_command(command, description):
|
||||
"""Verify a command runs with an exit code of 0."""
|
||||
# TODO: also collect ``which samtools`` and ``samtools --version``
|
||||
p = subprocess.Popen(args=command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
|
||||
(stdout, stderr) = p.communicate()
|
||||
if p.returncode:
|
||||
template = description
|
||||
template += " failed: (cmd=[%s], stdout=[%s], stderr=[%s])"
|
||||
message = template % (command, stdout, stderr)
|
||||
raise AssertionError(message)
|
||||
|
||||
|
||||
def files_diff(file1, file2, attributes=None):
|
||||
"""Check the contents of 2 files for differences."""
|
||||
def get_lines_diff(diff):
|
||||
|
||||
Reference in New Issue
Block a user