htsworkflow/util/ucsc.py

   1 """Wrap ucsc command line utilities
   2 """
   3
   4 import logging
   5 import os
   6 import sys
   7 from subprocess import Popen, PIPE
   8
   9 LOGGER = logging.getLogger(__name__)
  10
  11 def parseNumber(number):
  12     buffer = []
  13     isFloat = False
  14     for n in number:
  15         if n == ',':
  16             continue
  17         if n == '.':
  18             isFloat = True
  19             buffer.append(n)
  20         else:
  21             buffer.append(n)
  22     if isFloat:
  23         return float(''.join(buffer))
  24     else:
  25         return int(''.join(buffer))
  26
  27 def parseBoolean(value):
  28     if value.lower() in ('yes', '1', 'true'):
  29         return True
  30     elif value.lower() in ('no', '0', 'false'):
  31         return False
  32
  33 class bigWigInfo:
  34     def __init__(self, filename=None):
  35         self.version = None
  36         self.isCompressed = None
  37         self.isSwapped = None
  38         self.primaryDataSize = None
  39         self.primaryIndexSize = None
  40         self.zoomLevels = None
  41         self.chromCount = None
  42         self.basesCovered = None
  43         self.mean = None
  44         self.min = None
  45         self.max = None
  46         self.std = None
  47         self.filename = None
  48         if filename:
  49             self.scan_file(filename)
  50             self.filename = filename
  51
  52     def scan_file(self, filename):
  53         cmd = ['bigWigInfo',
  54                filename]
  55         try:
  56             p = Popen(cmd, stdout=PIPE)
  57             stdout, _ = p.communicate()
  58             for line in stdout.split(os.linesep):
  59                 if len(line) > 0:
  60                     term, value = line.split(': ')
  61                     if term in ('isCompressed', 'isSwapped'):
  62                         value = parseBoolean(value)
  63                     else:
  64                         value = parseNumber(value)
  65                     LOGGER.debug('%s: %s', term, str(value))
  66                     setattr(self, term, value)
  67         except OSError as e:
  68             LOGGER.error("Exception %s trying to run: %s", str(e), ' '.join(cmd))
  69             sys.exit(-1)
  70
  71
  72
  73