<filename>src/pbrdna/_utils.py
import os
import sys
from collections import namedtuple

import numpy as np

from pbrdna.io.FastaIO import FastaReader
from pbrdna.io.FastqIO import FastqRecord

BlasrM1 = namedtuple('BlasrM1', ['qname', 'tname', 'qstrand', 'tstrand',
                                 'score', 'pctsimilarity', 
                                 'tstart', 'tend', 'tlength',
                                 'qstart', 'qend', 'qlength',
                                 'ncells'])

def fasta_count( fasta_file ):
    count = 0
    try:
        for record in FastaReader( fasta_file ):
            if len(record.sequence) > 0:
                count += 1
    except:
        pass
    return count

def meanP( record ):
    try:
        assert isinstance(record, FastqRecord)
    except:
        raise TypeError("Record is not a FastqRecord!")
    pValues = [10**-(i/10) for i in np.float32(record.quality)]
    return sum(pValues) / len(pValues)

def meanPQv( record ):
    try:
        assert isinstance(record, FastqRecord)
    except:
        raise TypeError("Record is not a FastqRecord!")
    return pValueToQv( meanP(record) )

def pValueToQv( pValue ):
    return -10 * np.log10( pValue )

def fileExists( filename ):
    return os.path.exists( filename ) and os.path.getsize( filename ) > 0

def allFilesExist( filenames ):
    return all( map(fileExists, filenames) )

def isExe( filePath ):
    if filePath is None:
        return False
    return os.path.isfile(filePath) and os.access(filePath, os.X_OK)

def getZmw( read ):
    parts = read.split('/')
    return '/'.join(parts[0:2])

def splitRootFromExt( inputFile ):
    root, ext = os.path.splitext( inputFile )
    if ext == '.h5':
        root, ext = os.path.splitext( root )
        return (root, ext+'.h5')
    return (root, ext)

def predictOutputFile( inputFile, outputType ):
    root, ext = os.path.splitext( inputFile )
    return '{0}.{1}'.format(root, outputType)

def returnEmpty():
    return []

def createDirectory( dirName ):
    try:
        os.mkdir( dirName )
    except OSError:
        pass
    if not os.path.isdir( dirName ):
        raise OSError('Could not create directory "%s"!' % dirName)
    return dirName

def which(program):
    """
    Find and return path to local executables  
    """
    fpath, fname = os.path.split(program)
    if fpath:
        if isExe(program):
            return program
    else:
        for path in os.environ["PATH"].split(os.pathsep):
            path = path.strip('"')
            exeFile = os.path.join(path, program)
            if isExe(exeFile):
                return exeFile
    return None

def validateInputFile( fileName, allowedSuffixes ):
    allowedSuffixes = [allowedSuffixes] if isinstance(str, type(allowedSuffixes)) \
                                        else allowedSuffixes
    # First we check whether the input file has a valid suffix
    try:            
        assert any( [fileName.endswith(suffix) for suffix in allowedSuffixes] )
    except AssertionError:
        raise ValueError('File does not have an allowed suffix! %s' % \
                                                        allowedSuffixes)
    # Next we check whether the input file exists where specified
    try: 
        assert fileExists( fileName )
    except AssertionError:
        raise OSError('Input file does not exist!')
    # Finally we return the absolute path to the file
    return os.path.abspath( fileName )

def validateOutputFile( fileName ):
    if fileName in [sys.stdout, sys.stderr]:
        return fileName
    return os.path.abspath( fileName )

def validateExecutable( executable ):
    exePath = which( executable )
    try: 
        assert exePath is not None
    except AssertionError:
        raise ValueError('"%s" is not a valid executable!' % executable)
    return exePath

def validateInt( integer, minValue=None, maxValue=None ):
    try: # First we check whether the supplied parameter is an Int
        assert isinstance(integer, int)
    except AssertionError:
        raise TypeError('Parameter is not an Integer!')
    # If a minimum value is supplied, compare it to the Integer
    if minValue is not None:
        try:
            assert integer >= minValue
        except AssertionError:
            raise ValueError("Integer is less than Minimum Value!")
    # If a maximum value is supplied, compare it to the Integer
    if maxValue is not None:
        try:
            assert integer <= maxValue
        except AssertionError:
            raise ValueError("Integer is greater than Maximum Value!")
    return integer

def validateFloat( floating_point, minValue=None, maxValue=None ):
    try: # First we check whether the supplied parameter is an Int
        assert isinstance(floating_point, float)
    except AssertionError:
        raise TypeError('Parameter is not a Floating Point!')
    # If a minimum value is supplied, compare it to the Integer
    if minValue is not None:
        try:
            assert floating_point >= minValue
        except AssertionError:
            raise ValueError("Float is less than Minimum Value!")
    # If a maximum value is supplied, compare it to the Integer
    if maxValue is not None:
        try:
            assert floating_point <= maxValue
        except AssertionError:
            raise ValueError("Float is greater than Maximum Value!")
    return floating_point

<filename>src/pbrdna/_utils.py
import os
import sys
from collections import namedtuple

import numpy as np

from pbrdna.io.FastaIO import FastaReader
from pbrdna.io.FastqIO import FastqRecord

BlasrM1 = namedtuple('BlasrM1', ['qname', 'tname', 'qstrand', 'tstrand',
                                 'score', 'pctsimilarity', 
                                 'tstart', 'tend', 'tlength',
                                 'qstart', 'qend', 'qlength',
                                 'ncells'])

def fasta_count( fasta_file ):
    count = 0
    try:
        for record in FastaReader( fasta_file ):
            if len(record.sequence) > 0:
                count += 1
    except:
        pass
    return count

def meanP( record ):
    try:
        assert isinstance(record, FastqRecord)
    except:
        raise TypeError("Record is not a FastqRecord!")
    pValues = [10**-(i/10) for i in np.float32(record.quality)]
    return sum(pValues) / len(pValues)

def meanPQv( record ):
    try:
        assert isinstance(record, FastqRecord)
    except:
        raise TypeError("Record is not a FastqRecord!")
    return pValueToQv( meanP(record) )

def pValueToQv( pValue ):
    return -10 * np.log10( pValue )

def fileExists( filename ):
    return os.path.exists( filename ) and os.path.getsize( filename ) > 0

def allFilesExist( filenames ):
    return all( map(fileExists, filenames) )

def isExe( filePath ):
    if filePath is None:
        return False
    return os.path.isfile(filePath) and os.access(filePath, os.X_OK)

def getZmw( read ):
    parts = read.split('/')
    return '/'.join(parts[0:2])

def splitRootFromExt( inputFile ):
    root, ext = os.path.splitext( inputFile )
    if ext == '.h5':
        root, ext = os.path.splitext( root )
        return (root, ext+'.h5')
    return (root, ext)

def predictOutputFile( inputFile, outputType ):
    root, ext = os.path.splitext( inputFile )
    return '{0}.{1}'.format(root, outputType)

def returnEmpty():
    return []

def createDirectory( dirName ):
    try:
        os.mkdir( dirName )
    except OSError:
        pass
    if not os.path.isdir( dirName ):
        raise OSError('Could not create directory "%s"!' % dirName)
    return dirName

def which(program):
    """
    Find and return path to local executables  
    """
    fpath, fname = os.path.split(program)
    if fpath:
        if isExe(program):
            return program
    else:
        for path in os.environ["PATH"].split(os.pathsep):
            path = path.strip('"')
            exeFile = os.path.join(path, program)
            if isExe(exeFile):
                return exeFile
    return None

def validateInputFile( fileName, allowedSuffixes ):
    allowedSuffixes = [allowedSuffixes] if isinstance(str, type(allowedSuffixes)) \
                                        else allowedSuffixes
    # First we check whether the input file has a valid suffix
    try:            
        assert any( [fileName.endswith(suffix) for suffix in allowedSuffixes] )
    except AssertionError:
        raise ValueError('File does not have an allowed suffix! %s' % \
                                                        allowedSuffixes)
    # Next we check whether the input file exists where specified
    try: 
        assert fileExists( fileName )
    except AssertionError:
        raise OSError('Input file does not exist!')
    # Finally we return the absolute path to the file
    return os.path.abspath( fileName )

def validateOutputFile( fileName ):
    if fileName in [sys.stdout, sys.stderr]:
        return fileName
    return os.path.abspath( fileName )

def validateExecutable( executable ):
    exePath = which( executable )
    try: 
        assert exePath is not None
    except AssertionError:
        raise ValueError('"%s" is not a valid executable!' % executable)
    return exePath

def validateInt( integer, minValue=None, maxValue=None ):
    try: # First we check whether the supplied parameter is an Int
        assert isinstance(integer, int)
    except AssertionError:
        raise TypeError('Parameter is not an Integer!')
    # If a minimum value is supplied, compare it to the Integer
    if minValue is not None:
        try:
            assert integer >= minValue
        except AssertionError:
            raise ValueError("Integer is less than Minimum Value!")
    # If a maximum value is supplied, compare it to the Integer
    if maxValue is not None:
        try:
            assert integer <= maxValue
        except AssertionError:
            raise ValueError("Integer is greater than Maximum Value!")
    return integer

def validateFloat( floating_point, minValue=None, maxValue=None ):
    try: # First we check whether the supplied parameter is an Int
        assert isinstance(floating_point, float)
    except AssertionError:
        raise TypeError('Parameter is not a Floating Point!')
    # If a minimum value is supplied, compare it to the Integer
    if minValue is not None:
        try:
            assert floating_point >= minValue
        except AssertionError:
            raise ValueError("Float is less than Minimum Value!")
    # If a maximum value is supplied, compare it to the Integer
    if maxValue is not None:
        try:
            assert floating_point <= maxValue
        except AssertionError:
            raise ValueError("Float is greater than Maximum Value!")
    return floating_point

Find and return path to local executables # First we check whether the input file has a valid suffix # Next we check whether the input file exists where specified # Finally we return the absolute path to the file # First we check whether the supplied parameter is an Int # If a minimum value is supplied, compare it to the Integer # If a maximum value is supplied, compare it to the Integer # First we check whether the supplied parameter is an Int # If a minimum value is supplied, compare it to the Integer # If a maximum value is supplied, compare it to the Integer