Mercurial > repos > yufei-luo > s_mart
view smart_toolShed/SMART/Java/Python/structure/Interval.py @ 1:30c5431d8ce4
Uploaded
author | yufei-luo |
---|---|
date | Thu, 17 Jan 2013 11:01:51 -0500 |
parents | e0f8dcca02ed |
children |
line wrap: on
line source
# # Copyright INRA-URGI 2009-2010 # # This software is governed by the CeCILL license under French law and # abiding by the rules of distribution of free software. You can use, # modify and/ or redistribute the software under the terms of the CeCILL # license as circulated by CEA, CNRS and INRIA at the following URL # "http://www.cecill.info". # # As a counterpart to the access to the source code and rights to copy, # modify and redistribute granted by the license, users are provided only # with a limited warranty and the software's author, the holder of the # economic rights, and the successive licensors have only limited # liability. # # In this respect, the user's attention is drawn to the risks associated # with loading, using, modifying and/or developing or reproducing the # software by the user in light of its specific status of free software, # that may mean that it is complicated to manipulate, and that also # therefore means that it is reserved for developers and experienced # professionals having in-depth computer knowledge. Users are therefore # encouraged to load and test the software's suitability as regards their # requirements in conditions enabling the security of their systems and/or # data to be ensured and, more generally, to use and operate it in the # same conditions as regards security. # # The fact that you are presently reading this means that you have had # knowledge of the CeCILL license and that you accept its terms. # from SMART.Java.Python.structure.Bins import * from commons.core.coord.Range import Range class Interval(Range): """ Store a genomic interval @ivar name: name of the interval [optional] @type name: string @ivar id: id of the interval [optional] @type id: int @ivar bin: bin in which the interval should be if stored in a database [computed] @type bin: int @ival tags: information about the transcript [optional] @type tags: dict @ivar verbosity: verbosity @type verbosity: int [default: 0] """ def __init__(self, interval = None, verbosity = 0): """ Constructor @param interval: interval to be copied @type interval: class L{Interval<Interval>} @param verbosity: verbosity @type verbosity: int """ Range.__init__(self) self.name = None self.id = None self.bin = None self.verbosity = verbosity self.tags = {} if interval != None: self.copy(interval) #!!!! Warning: two methods getStart() and getEnd() give the information maximum and minimum in interval.!!!!# #In case strand = "+", start < end; strand = "-", start > end def getStart(self): if self.start == -1: return -1 if self.end == -1: return self.start return self.getMin() def getEnd(self): if self.end == -1: return -1 if self.start == -1: return self.end return self.getMax() def getChromosome(self): return self.getSeqname() def getDirection(self): return 1 if self.getStrand() == "+" else -1 def getName(self): return self.name def isSet(self): """ Check if the interval is set """ return self.getStart() == None and self.getEnd() == None def copy(self, interval): """ Copy method @param interval: interval to be copied @type interval: class L{Interval<Interval>} """ self.setStart(interval.getStart()) self.setEnd(interval.getEnd()) self.setChromosome(interval.getChromosome()) self.setDirection(interval.getDirection()) self.name = interval.name self.id = interval.id self.bin = interval.bin self.tags = {} for tag in interval.tags: self.tags[tag] = interval.tags[tag] self.verbosity = interval.verbosity def setName(self, name): """ Set the name @param name: name of the interval @type name: string """ if len(name) > 100: name = name[:100] self.name = name def setChromosome(self, chromosome=""): """ Set the chromosome @param chromosome: chromosome on which the interval is @type chromosome: string """ if not chromosome: self.seqname = None else: self.seqname = chromosome.replace(".", "_").replace("|", "_") def setStart(self, start): """ Set the start point Possibly reset bin @param start: start point of the interval @type start: int """ self.bin = None direction = self.getDirection() if self.start == -1: self.start = start elif self.end == -1: self.end = start else: if direction == 1: self.start = start else: self.end = start if direction == 1: self.start, self.end = min(self.start, self.end), max(self.start, self.end) else: self.start, self.end = max(self.start, self.end), min(self.start, self.end) def setEnd(self, end): """ Set the end point Possibly reset bin @param end: end point of the interval of the interval @type end: int """ self.bin = None direction = self.getDirection() if self.end == -1: self.end = end elif self.start == -1: self.start = end else: if direction == 1: self.end = end else: self.start = end if direction == 1: self.start, self.end = min(self.start, self.end), max(self.start, self.end) else: self.start, self.end = max(self.start, self.end), min(self.start, self.end) def setSize(self, size): """ Possibly modify the end point @param size: size of the transcript @type size: int """ if self.end == None and self.start != None: self.setEnd(self.start + self.getSize() - 1) elif self.start == None and self.end != None: self.setStart(self.end - self.getSize() + 1) def getSize(self): """ Get the size """ return self.getEnd() - self.getStart() + 1 def _setDirection(self, direction): """ Set the direction of the interval (connection to Range) @param direction: direction of the transcript (+ / -) @type direction: int (1 or -1) """ if direction * self.getDirection() < 0: self.reverse() def setDirection(self, direction): """ Set the direction of the interval Possibly parse different formats @param direction: direction of the transcript (+ / -) @type direction: int or string """ if type(direction).__name__ == 'int': self._setDirection(direction / abs(direction)) elif type(direction).__name__ == 'str': if direction == "+": self._setDirection(1) elif direction == "-": self._setDirection(-1) elif direction == "1" or direction == "-1": self._setDirection(int(direction)) elif direction.lower() == "plus": self._setDirection(1) elif direction.lower() == "minus": self._setDirection(-1) else: raise Exception("Cannot understand direction %s" % (direction)) else: raise Exception("Cannot understand direction %s" % (direction)) def extendStart(self, size): """ Extend the interval by the 5' end @param size: the size to be exended @type size: int """ if self.getDirection() == 1: self.setStart(max(0, self.getStart() - size)) else: self.setEnd(self.getEnd() + size) self.bin = None def extendEnd(self, size): """ Extend the interval by the 3' end @param size: the size to be exended @type size: int """ if self.getDirection() == 1: self.setEnd(self.getEnd() + size) else: self.setStart(max(0, self.getStart() - size)) self.bin = None def restrictStart(self, size = 1): """ Restrict the interval by some nucleotides, start from its start position Remove the exons @param size: the size to be restricted to @type size: int """ if self.getDirection() == 1: self.setEnd(min(self.getEnd(), self.getStart() + size - 1)) else: self.setStart(max(self.getStart(), self.getEnd() - size + 1)) self.bin = None def restrictEnd(self, size = 1): """ Restrict the interval by some nucleotides, end from its end position Remove the exons @param size: the size to be restricted to @type size: int """ if self.getDirection() == 1: self.setStart(max(self.getStart(), self.getEnd() - size + 1)) else: self.setEnd(min(self.getEnd(), self.getStart() + size - 1)) self.bin = None def setTagValue(self, name, value): """ Set a tag @param name: name of the tag @type name: string @param value: value of the tag @type value: int or string """ self.tags[name] = value def getTagNames(self): """ Get all the names of the tags """ return self.tags.keys() def getTagValue(self, tag): """ Get the value of a tag @param tag: name of a tag @type tag: string """ if tag not in self.tags: return None return self.tags[tag] def getTagValues(self, tagSep = "; ", fieldSep = " ", surrounder = ""): """ Get the formatted tag values @param tagSep: separator between tags @type tagSep: string @param fieldSep: separator between tag name and tag value @type fieldSep: string @param surrounder: string which optionally surround values @type surrounder: string """ tags = [] for name in self.tags: value = self.tags[name] if value == None: continue if isinstance(value, basestring): tags.append("%s%s%s%s%s" % (name, fieldSep, surrounder, self.tags[name], surrounder)) elif type(value) is int: tags.append("%s%s%s%i%s" % (name, fieldSep, surrounder, self.tags[name], surrounder)) elif type(value) is float: tags.append("%s%s%s%f%s" % (name, fieldSep, surrounder, self.tags[name], surrounder)) else: raise Exception("Do not know how to print '" + value + "'.") if self.getName() != None: tags.append("%s%s%s%s%s" % ("Name", fieldSep, surrounder, self.getName(), surrounder)) return tagSep.join(tags) def setTagValues(self, tags, tagSep = "; ", fieldSep = " "): """ Set the tag values using given string @param tags: the tags, concatenated @type tags: string @param tagSep: separator between tags @type tagSep: string @param fieldSep: separator between tag name and tag value @type fieldSep: string """ if tags == "": self.tags = {} return for splittedTag in tags.split(tagSep): if fieldSep not in splittedTag: raise Exception("Weird field '%s' in tags '%s'" % (splittedTag, tags)) tag, value = splittedTag.split(fieldSep, 1) if tag == "Name": self.setName(value) continue try: intValue = int(value) self.tags[tag] = intValue except ValueError: try: floatValue = float(value) self.tags[tag] = floatValue except ValueError: self.tags[tag] = value def deleteTag(self, tag): """ Remove a tag @param tag: the tag to be removed @type tag: string """ if tag in self.tags: del self.tags[tag] def setNbOccurrences(self, nbOccurrences): """ Set the number of occurrences of the interval @param nbOccurrences: number of occurrences of the interval @type nbOccurrences: int """ self.setTagValue("nbOccurrences", nbOccurrences) def setOccurrence(self, occurrence): """ Set the occurrence of this interval @param occurrence: an occurrence for this transcript @type occurrence: int """ self.setTagValue("occurrence", occurrence) def __eq__(self, interval): """ Whether two intervals are equal (start and end at same position) @param interval: object to be compared to @type interval: class L{Interval<Interval>} """ if not interval: return False return self.getChromosome() == interval.getChromosome() and self.getStart() == interval.getStart() and self.getEnd() == interval.getEnd() and self.getDirection() == interval.getDirection() def overlapWith(self, interval, nbNucleotides = 1): """ Whether two intervals overlap @param interval: object to be compared to @type interval: class L{Interval<Interval>} @param nbNucleotides: minimum number of nucleotides to declare and overlap @type nbNucleotides: int """ if self.getChromosome() != interval.getChromosome(): return False return (min(self.getEnd(), interval.getEnd()) - max(self.getStart(), interval.getStart()) + 1 >= nbNucleotides) def isIncludeIn(self, interval): return interval.include(self) def include(self, interval): """ Whether this interval includes the other one @param interval: object to be compared to @type interval: class L{Interval<Interval>} """ if self.getChromosome() != interval.getChromosome(): return False return ((self.getStart() <= interval.getStart()) and (self.getEnd() >= interval.getEnd())) def getDifference(self, interval, sameStrand = False): """ Get the difference between this cluster and another one @param interval: object to be compared to @type interval: class L{Interval<Interval>} @param sameStrand: do the comparison iff the intervals are on the same strand @type sameStrand: boolean @return: a (possibly empty) list of intervals """ newInterval = Interval() newInterval.copy(self) if self.getChromosome() != interval.getChromosome(): return [newInterval] if not self.overlapWith(interval): return [newInterval] if sameStrand and self.getDirection() != interval.getDirection(): return [newInterval] intervals = [] if self.getStart() < interval.getStart(): newInterval = Interval() newInterval.copy(self) newInterval.setEnd(min(self.getEnd(), interval.getStart() - 1)) intervals.append(newInterval) if self.getEnd() > interval.getEnd(): newInterval = Interval() newInterval.copy(self) newInterval.setStart(max(self.getStart(), interval.getEnd() + 1)) intervals.append(newInterval) return intervals def getIntersection(self, interval): """ Get the intersection between this interval and another one @param interval: object to be compared to @type interval: class L{Interval<Interval>} @return: an other interval """ if not self.overlapWith(interval): return None newInterval = Interval() newInterval.setChromosome(self.getChromosome()) newInterval.setDirection(self.getDirection()) newInterval.setName("%s_intersect_%s" % (self.getName(), interval.getName())) newInterval.setStart(max(self.getStart(), interval.getStart())) newInterval.setEnd(min(self.getEnd(), interval.getEnd())) return newInterval def getDistance(self, interval): """ Get the distance between two intervals (a non-negative value) @param interval: another interval @type interval: class L{Interval<Interval>} """ if self.overlapWith(interval): return 0 if self.getChromosome() != interval.getChromosome(): raise Exception("Cannot get the distance between %s and %s" % (str(self), str(interval))) return min(abs(self.getStart() - interval.getEnd()), abs(self.getEnd() - interval.getStart())) def getRelativeDistance(self, interval): """ Get the distance between two intervals (negative if first interval is before) @param interval: another interval @type interval: class L{Interval<Interval>} """ if self.overlapWith(interval): return 0 if self.getChromosome() != interval.getChromosome(): raise Exception("Cannot get the distance between %s and %s" % (str(self), str(interval))) if self.getEnd() < interval.getStart(): distance = interval.getStart() - self.getEnd() else: distance = interval.getEnd() - self.getStart() distance *= self.getDirection() return distance def merge(self, interval, normalization = False): """ Merge two intervals @param interval: another interval @type interval: class L{Interval<Interval>} @param normalization: whether the sum of the merge should be normalized wrt the number of mappings of each elements @type normalization: boolean """ if self.getChromosome() != interval.getChromosome(): raise Exception("Cannot merge '%s' and '%s' for they are on different chromosomes." % (str(self), str(interval))) direction = None if self.getStart() == self.getEnd(): direction = interval.getDirection() elif interval.getStart() == interval.getEnd(): direction = self.getDirection() elif self.getDirection() != interval.getDirection(): raise Exception("Cannot merge '%s' and '%s' for they are on different strands." % (str(self), str(interval))) self.setStart(min(self.getStart(), interval.getStart())) self.setEnd(max(self.getEnd(), interval.getEnd())) if direction != None: self.setDirection(direction) nbElements = 0.0 for element in (self, interval): for tagName in ("nbElements", "nbOccurrences"): if tagName not in element.getTagNames(): element.setTagValue(tagName, 1) nbElements += float(element.getTagValue("nbElements")) / float(element.getTagValue("nbOccurrences")) if normalization else float(element.getTagValue("nbElements")) self.setTagValue("nbElements", nbElements) self.bin = None for tagName in ("identity", "nbOccurrences", "occurrence", "nbMismatches", "nbGaps", "rank", "evalue", "bestRegion"): if tagName in self.getTagNames(): del self.tags[tagName] def getBin(self): """ Get the bin of the interval Computed on the fly """ if self.bin == None: self.bin = getBin(self.getStart(), self.getEnd()) return self.bin def getBins(self): """ Get all the bin this interval could fall into """ return getOverlappingBins(self.getStart(), self.getEnd()) def getSqlVariables(cls): """ Get the properties of the object that should be saved in a database """ variables = ["name", "chromosome", "start", "end", "direction", "tags", "bin"] return variables getSqlVariables = classmethod(getSqlVariables) def setSqlValues(self, array): """ Set the values of the properties of this object as given by a results line of a SQL query """ self.id = array[0] self.name = array[1].strip("'") self.setChromosome(array[2].strip("'")) self.setStart(array[3]) self.setEnd(array[4]) self.setDirection(array[5]) self.setTagValues(array[6].strip("'"), ";", "=") self.bin = array[7] def getSqlValues(self): """ Get the values of the properties that should be saved in a database """ values = dict() values["name"] = self.name values["chromosome"] = self.getChromosome() values["start"] = self.getStart() values["end"] = self.getEnd() values["direction"] = self.getDirection() values["tags"] = self.getTagValues(";", "=") values["bin"] = self.getBin() return values def getSqlTypes(cls): """ Get the values of the properties that should be saved in a database """ types = dict() types["name"] = "varchar" types["chromosome"] = "varchar" types["start"] = "int" types["end"] = "int" types["direction"] = "tinyint" types["tags"] = "varchar" types["bin"] = "int" return types getSqlTypes = classmethod(getSqlTypes) def getSqlSizes(cls): """ Get the sizes of the properties that should be saved in a database """ sizes = dict() sizes["name"] = 255 sizes["chromosome"] = 255 sizes["start"] = 11 sizes["end"] = 11 sizes["direction"] = 4 sizes["tags"] = 1023 sizes["bin"] = 11 return sizes getSqlSizes = classmethod(getSqlSizes) def printCoordinates(self): """ Print start and end positions (depending on the direction of the interval) """ if self.getDirection() == 1: return "%d-%d" % (self.getStart(), self.getEnd()) else: return "%d-%d" % (self.getEnd(), self.getStart()) def extractSequence(self, parser): """ Get the sequence corresponding to this interval @param parser: a parser to a FASTA file @type parser: class L{SequenceListParser<SequenceListParser>} @return : a instance of L{Sequence<Sequence>} """ return parser.getSubSequence(self.getChromosome(), self.getStart(), self.getEnd(), self.getDirection(), self.name) def extractWigData(self, parser): """ Get the data retrieved from a wig file @param parser: a parser class to a WIG file @type parser: class L{WigParser<WigParser>} """ data = parser.getRange(self.getChromosome(), self.getStart(), self.getEnd()) if self.getDirection() == -1: if parser.strands: newData = {} for strand in data: data[strand].reverse() newData[-strand] = data[strand] data = newData else: data.reverse() return data def __str__(self): """ Output a simple representation of this interval """ direction = "+" if self.getDirection() == -1: direction = "-" string = "%s:%d-%d (%s)" % (self.getChromosome(), self.getStart(), self.getEnd(), direction) if self.name != "": string = "(%s) %s" % (self.name, string) return string