<s> import itertools <EOL> import operator <EOL> import os <EOL> import struct <EOL> import sys <EOL> import time <EOL> izip = getattr ( itertools , '<STR_LIT>' , zip ) <EOL> ifilter = getattr ( itertools , '<STR_LIT>' , filter ) <EOL> if sys . version_info >= ( <NUM_LIT:3> , <NUM_LIT:0> ) : <EOL> xrange = range <EOL> try : <EOL> import fcntl <EOL> CAN_LOCK = True <EOL> except ImportError : <EOL> CAN_LOCK = False <EOL> try : <EOL> import ctypes <EOL> import ctypes . util <EOL> CAN_FALLOCATE = True <EOL> except ImportError : <EOL> CAN_FALLOCATE = False <EOL> fallocate = None <EOL> if CAN_FALLOCATE : <EOL> libc_name = ctypes . util . find_library ( '<STR_LIT:c>' ) <EOL> libc = ctypes . CDLL ( libc_name ) <EOL> c_off64_t = ctypes . c_int64 <EOL> c_off_t = ctypes . c_int <EOL> try : <EOL> _fallocate = libc . posix_fallocate64 <EOL> _fallocate . restype = ctypes . c_int <EOL> _fallocate . argtypes = [ ctypes . c_int , c_off64_t , c_off64_t ] <EOL> except AttributeError : <EOL> try : <EOL> _fallocate = libc . posix_fallocate <EOL> _fallocate . restype = ctypes . c_int <EOL> _fallocate . argtypes = [ ctypes . c_int , c_off_t , c_off_t ] <EOL> except AttributeError : <EOL> CAN_FALLOCATE = False <EOL> if CAN_FALLOCATE : <EOL> def _py_fallocate ( fd , offset , len_ ) : <EOL> res = _fallocate ( fd . fileno ( ) , offset , len_ ) <EOL> if res != <NUM_LIT:0> : <EOL> raise IOError ( res , '<STR_LIT>' ) <EOL> fallocate = _py_fallocate <EOL> del libc <EOL> del libc_name <EOL> LOCK = False <EOL> CACHE_HEADERS = False <EOL> AUTOFLUSH = False <EOL> __headerCache = { } <EOL> longFormat = "<STR_LIT>" <EOL> longSize = struct . calcsize ( longFormat ) <EOL> floatFormat = "<STR_LIT>" <EOL> floatSize = struct . calcsize ( floatFormat ) <EOL> valueFormat = "<STR_LIT>" <EOL> valueSize = struct . calcsize ( valueFormat ) <EOL> pointFormat = "<STR_LIT>" <EOL> pointSize = struct . calcsize ( pointFormat ) <EOL> metadataFormat = "<STR_LIT>" <EOL> metadataSize = struct . calcsize ( metadataFormat ) <EOL> archiveInfoFormat = "<STR_LIT>" <EOL> archiveInfoSize = struct . calcsize ( archiveInfoFormat ) <EOL> aggregationTypeToMethod = dict ( { <EOL> <NUM_LIT:1> : '<STR_LIT>' , <EOL> <NUM_LIT:2> : '<STR_LIT>' , <EOL> <NUM_LIT:3> : '<STR_LIT>' , <EOL> <NUM_LIT:4> : '<STR_LIT>' , <EOL> <NUM_LIT:5> : '<STR_LIT>' <EOL> } ) <EOL> aggregationMethodToType = dict ( [ [ v , k ] for k , v in aggregationTypeToMethod . items ( ) ] ) <EOL> aggregationMethods = aggregationTypeToMethod . values ( ) <EOL> debug = startBlock = endBlock = lambda * a , ** k : None <EOL> UnitMultipliers = { <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> * <NUM_LIT:7> , <EOL> '<STR_LIT>' : <NUM_LIT> * <NUM_LIT> <EOL> } <EOL> def getUnitString ( s ) : <EOL> if '<STR_LIT>' . startswith ( s ) : return '<STR_LIT>' <EOL> if '<STR_LIT>' . startswith ( s ) : return '<STR_LIT>' <EOL> if '<STR_LIT>' . startswith ( s ) : return '<STR_LIT>' <EOL> if '<STR_LIT>' . startswith ( s ) : return '<STR_LIT>' <EOL> if '<STR_LIT>' . startswith ( s ) : return '<STR_LIT>' <EOL> if '<STR_LIT>' . startswith ( s ) : return '<STR_LIT>' <EOL> raise ValueError ( "<STR_LIT>" % s ) <EOL> def parseRetentionDef ( retentionDef ) : <EOL> import re <EOL> ( precision , points ) = retentionDef . strip ( ) . split ( '<STR_LIT::>' ) <EOL> if precision . isdigit ( ) : <EOL> precision = int ( precision ) * UnitMultipliers [ getUnitString ( '<STR_LIT:s>' ) ] <EOL> else : <EOL> precision_re = re . compile ( r'<STR_LIT>' ) <EOL> match = precision_re . match ( precision ) <EOL> if match : <EOL> precision = int ( match . group ( <NUM_LIT:1> ) ) * UnitMultipliers [ getUnitString ( match . group ( <NUM_LIT:2> ) ) ] <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" % precision ) <EOL> if points . isdigit ( ) : <EOL> points = int ( points ) <EOL> else : <EOL> points_re = re . compile ( r'<STR_LIT>' ) <EOL> match = points_re . match ( points ) <EOL> if match : <EOL> points = int ( match . group ( <NUM_LIT:1> ) ) * UnitMultipliers [ getUnitString ( match . group ( <NUM_LIT:2> ) ) ] // precision <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" % points ) <EOL> return ( precision , points ) <EOL> class WhisperException ( Exception ) : <EOL> """<STR_LIT>""" <EOL> class InvalidConfiguration ( WhisperException ) : <EOL> """<STR_LIT>""" <EOL> class InvalidAggregationMethod ( WhisperException ) : <EOL> """<STR_LIT>""" <EOL> class InvalidTimeInterval ( WhisperException ) : <EOL> """<STR_LIT>""" <EOL> class TimestampNotCovered ( WhisperException ) : <EOL> """<STR_LIT>""" <EOL> class CorruptWhisperFile ( WhisperException ) : <EOL> def __init__ ( self , error , path ) : <EOL> Exception . __init__ ( self , error ) <EOL> self . error = error <EOL> self . path = path <EOL> def __repr__ ( self ) : <EOL> return "<STR_LIT>" % ( self . path , self . error ) <EOL> def __str__ ( self ) : <EOL> return "<STR_LIT>" % ( self . error , self . path ) <EOL> def enableDebug ( ) : <EOL> global open , debug , startBlock , endBlock <EOL> class open ( file ) : <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> file . __init__ ( self , * args , ** kwargs ) <EOL> self . writeCount = <NUM_LIT:0> <EOL> self . readCount = <NUM_LIT:0> <EOL> def write ( self , data ) : <EOL> self . writeCount += <NUM_LIT:1> <EOL> debug ( '<STR_LIT>' % ( len ( data ) , self . writeCount ) ) <EOL> return file . write ( self , data ) <EOL> def read ( self , bytes ) : <EOL> self . readCount += <NUM_LIT:1> <EOL> debug ( '<STR_LIT>' % ( bytes , self . readCount ) ) <EOL> return file . read ( self , bytes ) <EOL> def debug ( message ) : <EOL> print ( '<STR_LIT>' % message ) <EOL> __timingBlocks = { } <EOL> def startBlock ( name ) : <EOL> __timingBlocks [ name ] = time . time ( ) <EOL> def endBlock ( name ) : <EOL> debug ( "<STR_LIT>" % ( name , time . time ( ) - __timingBlocks . pop ( name ) ) ) <EOL> def __readHeader ( fh ) : <EOL> info = __headerCache . get ( fh . name ) <EOL> if info : <EOL> return info <EOL> originalOffset = fh . tell ( ) <EOL> fh . seek ( <NUM_LIT:0> ) <EOL> packedMetadata = fh . read ( metadataSize ) <EOL> try : <EOL> ( aggregationType , maxRetention , xff , archiveCount ) = struct . unpack ( metadataFormat , packedMetadata ) <EOL> except : <EOL> raise CorruptWhisperFile ( "<STR_LIT>" , fh . name ) <EOL> archives = [ ] <EOL> for i in xrange ( archiveCount ) : <EOL> packedArchiveInfo = fh . read ( archiveInfoSize ) <EOL> try : <EOL> ( offset , secondsPerPoint , points ) = struct . unpack ( archiveInfoFormat , packedArchiveInfo ) <EOL> except : <EOL> raise CorruptWhisperFile ( "<STR_LIT>" % i , fh . name ) <EOL> archiveInfo = { <EOL> '<STR_LIT>' : offset , <EOL> '<STR_LIT>' : secondsPerPoint , <EOL> '<STR_LIT>' : points , <EOL> '<STR_LIT>' : secondsPerPoint * points , <EOL> '<STR_LIT:size>' : points * pointSize , <EOL> } <EOL> archives . append ( archiveInfo ) <EOL> fh . seek ( originalOffset ) <EOL> info = { <EOL> '<STR_LIT>' : aggregationTypeToMethod . get ( aggregationType , '<STR_LIT>' ) , <EOL> '<STR_LIT>' : maxRetention , <EOL> '<STR_LIT>' : xff , <EOL> '<STR_LIT>' : archives , <EOL> } <EOL> if CACHE_HEADERS : <EOL> __headerCache [ fh . name ] = info <EOL> return info <EOL> def setAggregationMethod ( path , aggregationMethod , xFilesFactor = None ) : <EOL> """<STR_LIT>""" <EOL> fh = None <EOL> try : <EOL> fh = open ( path , '<STR_LIT>' ) <EOL> if LOCK : <EOL> fcntl . flock ( fh . fileno ( ) , fcntl . LOCK_EX ) <EOL> packedMetadata = fh . read ( metadataSize ) <EOL> try : <EOL> ( aggregationType , maxRetention , xff , archiveCount ) = struct . unpack ( metadataFormat , packedMetadata ) <EOL> except : <EOL> raise CorruptWhisperFile ( "<STR_LIT>" , fh . name ) <EOL> try : <EOL> newAggregationType = struct . pack ( longFormat , aggregationMethodToType [ aggregationMethod ] ) <EOL> except KeyError : <EOL> raise InvalidAggregationMethod ( "<STR_LIT>" % <EOL> aggregationMethod ) <EOL> if xFilesFactor is not None : <EOL> xff = struct . pack ( floatFormat , float ( xFilesFactor ) ) <EOL> else : <EOL> xff = struct . pack ( floatFormat , xff ) <EOL> maxRetention = struct . pack ( longFormat , maxRetention ) <EOL> archiveCount = struct . pack ( longFormat , archiveCount ) <EOL> packedMetadata = newAggregationType + maxRetention + xff + archiveCount <EOL> fh . seek ( <NUM_LIT:0> ) <EOL> fh . write ( packedMetadata ) <EOL> if AUTOFLUSH : <EOL> fh . flush ( ) <EOL> os . fsync ( fh . fileno ( ) ) <EOL> if CACHE_HEADERS and fh . name in __headerCache : <EOL> del __headerCache [ fh . name ] <EOL> finally : <EOL> if fh : <EOL> fh . close ( ) <EOL> return aggregationTypeToMethod . get ( aggregationType , '<STR_LIT>' ) <EOL> def validateArchiveList ( archiveList ) : <EOL> """<STR_LIT>""" <EOL> if not archiveList : <EOL> raise InvalidConfiguration ( "<STR_LIT>" ) <EOL> archiveList . sort ( key = lambda a : a [ <NUM_LIT:0> ] ) <EOL> for i , archive in enumerate ( archiveList ) : <EOL> if i == len ( archiveList ) - <NUM_LIT:1> : <EOL> break <EOL> nextArchive = archiveList [ i + <NUM_LIT:1> ] <EOL> if not archive [ <NUM_LIT:0> ] < nextArchive [ <NUM_LIT:0> ] : <EOL> raise InvalidConfiguration ( "<STR_LIT>" <EOL> "<STR_LIT>" % <EOL> ( i , archive , i + <NUM_LIT:1> , nextArchive ) ) <EOL> if nextArchive [ <NUM_LIT:0> ] % archive [ <NUM_LIT:0> ] != <NUM_LIT:0> : <EOL> raise InvalidConfiguration ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % <EOL> ( i , archive [ <NUM_LIT:0> ] , i + <NUM_LIT:1> , nextArchive [ <NUM_LIT:0> ] ) ) <EOL> retention = archive [ <NUM_LIT:0> ] * archive [ <NUM_LIT:1> ] <EOL> nextRetention = nextArchive [ <NUM_LIT:0> ] * nextArchive [ <NUM_LIT:1> ] <EOL> if not nextRetention > retention : <EOL> raise InvalidConfiguration ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % <EOL> ( i , retention , i + <NUM_LIT:1> , nextRetention ) ) <EOL> archivePoints = archive [ <NUM_LIT:1> ] <EOL> pointsPerConsolidation = nextArchive [ <NUM_LIT:0> ] // archive [ <NUM_LIT:0> ] <EOL> if not archivePoints >= pointsPerConsolidation : <EOL> raise InvalidConfiguration ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % <EOL> ( i + <NUM_LIT:1> , pointsPerConsolidation , i , archivePoints ) ) <EOL> def create ( path , archiveList , xFilesFactor = None , aggregationMethod = None , sparse = False , useFallocate = False ) : <EOL> """<STR_LIT>""" <EOL> if xFilesFactor is None : <EOL> xFilesFactor = <NUM_LIT:0.5> <EOL> if aggregationMethod is None : <EOL> aggregationMethod = '<STR_LIT>' <EOL> validateArchiveList ( archiveList ) <EOL> if os . path . exists ( path ) : <EOL> raise InvalidConfiguration ( "<STR_LIT>" % path ) <EOL> fh = None <EOL> try : <EOL> fh = open ( path , '<STR_LIT:wb>' ) <EOL> if LOCK : <EOL> fcntl . flock ( fh . fileno ( ) , fcntl . LOCK_EX ) <EOL> aggregationType = struct . pack ( longFormat , aggregationMethodToType . get ( aggregationMethod , <NUM_LIT:1> ) ) <EOL> oldest = max ( [ secondsPerPoint * points for secondsPerPoint , points in archiveList ] ) <EOL> maxRetention = struct . pack ( longFormat , oldest ) <EOL> xFilesFactor = struct . pack ( floatFormat , float ( xFilesFactor ) ) <EOL> archiveCount = struct . pack ( longFormat , len ( archiveList ) ) <EOL> packedMetadata = aggregationType + maxRetention + xFilesFactor + archiveCount <EOL> fh . write ( packedMetadata ) <EOL> headerSize = metadataSize + ( archiveInfoSize * len ( archiveList ) ) <EOL> archiveOffsetPointer = headerSize <EOL> for secondsPerPoint , points in archiveList : <EOL> archiveInfo = struct . pack ( archiveInfoFormat , archiveOffsetPointer , secondsPerPoint , points ) <EOL> fh . write ( archiveInfo ) <EOL> archiveOffsetPointer += ( points * pointSize ) <EOL> if CAN_FALLOCATE and useFallocate : <EOL> remaining = archiveOffsetPointer - headerSize <EOL> fallocate ( fh , headerSize , remaining ) <EOL> elif sparse : <EOL> fh . seek ( archiveOffsetPointer - <NUM_LIT:1> ) <EOL> fh . write ( '<STR_LIT:\x00>' ) <EOL> else : <EOL> remaining = archiveOffsetPointer - headerSize <EOL> chunksize = <NUM_LIT> <EOL> zeroes = b'<STR_LIT:\x00>' * chunksize <EOL> while remaining > chunksize : <EOL> fh . write ( zeroes ) <EOL> remaining -= chunksize <EOL> fh . write ( zeroes [ : remaining ] ) <EOL> if AUTOFLUSH : <EOL> fh . flush ( ) <EOL> os . fsync ( fh . fileno ( ) ) <EOL> finally : <EOL> if fh : <EOL> fh . close ( ) <EOL> def aggregate ( aggregationMethod , knownValues ) : <EOL> if aggregationMethod == '<STR_LIT>' : <EOL> return float ( sum ( knownValues ) ) / float ( len ( knownValues ) ) <EOL> elif aggregationMethod == '<STR_LIT>' : <EOL> return float ( sum ( knownValues ) ) <EOL> elif aggregationMethod == '<STR_LIT>' : <EOL> return knownValues [ len ( knownValues ) - <NUM_LIT:1> ] <EOL> elif aggregationMethod == '<STR_LIT>' : <EOL> return max ( knownValues ) <EOL> elif aggregationMethod == '<STR_LIT>' : <EOL> return min ( knownValues ) <EOL> else : <EOL> raise InvalidAggregationMethod ( "<STR_LIT>" % <EOL> aggregationMethod ) <EOL> def __propagate ( fh , header , timestamp , higher , lower ) : <EOL> aggregationMethod = header [ '<STR_LIT>' ] <EOL> xff = header [ '<STR_LIT>' ] <EOL> lowerIntervalStart = timestamp - ( timestamp % lower [ '<STR_LIT>' ] ) <EOL> lowerIntervalEnd = lowerIntervalStart + lower [ '<STR_LIT>' ] <EOL> fh . seek ( higher [ '<STR_LIT>' ] ) <EOL> packedPoint = fh . read ( pointSize ) <EOL> ( higherBaseInterval , higherBaseValue ) = struct . unpack ( pointFormat , packedPoint ) <EOL> if higherBaseInterval == <NUM_LIT:0> : <EOL> higherFirstOffset = higher [ '<STR_LIT>' ] <EOL> else : <EOL> timeDistance = lowerIntervalStart - higherBaseInterval <EOL> pointDistance = timeDistance // higher [ '<STR_LIT>' ] <EOL> byteDistance = pointDistance * pointSize <EOL> higherFirstOffset = higher [ '<STR_LIT>' ] + ( byteDistance % higher [ '<STR_LIT:size>' ] ) <EOL> higherPoints = lower [ '<STR_LIT>' ] // higher [ '<STR_LIT>' ] <EOL> higherSize = higherPoints * pointSize <EOL> relativeFirstOffset = higherFirstOffset - higher [ '<STR_LIT>' ] <EOL> relativeLastOffset = ( relativeFirstOffset + higherSize ) % higher [ '<STR_LIT:size>' ] <EOL> higherLastOffset = relativeLastOffset + higher [ '<STR_LIT>' ] <EOL> fh . seek ( higherFirstOffset ) <EOL> if higherFirstOffset < higherLastOffset : <EOL> seriesString = fh . read ( higherLastOffset - higherFirstOffset ) <EOL> else : <EOL> higherEnd = higher [ '<STR_LIT>' ] + higher [ '<STR_LIT:size>' ] <EOL> seriesString = fh . read ( higherEnd - higherFirstOffset ) <EOL> fh . seek ( higher [ '<STR_LIT>' ] ) <EOL> seriesString += fh . read ( higherLastOffset - higher [ '<STR_LIT>' ] ) <EOL> byteOrder , pointTypes = pointFormat [ <NUM_LIT:0> ] , pointFormat [ <NUM_LIT:1> : ] <EOL> points = len ( seriesString ) // pointSize <EOL> seriesFormat = byteOrder + ( pointTypes * points ) <EOL> unpackedSeries = struct . unpack ( seriesFormat , seriesString ) <EOL> neighborValues = [ None ] * points <EOL> currentInterval = lowerIntervalStart <EOL> step = higher [ '<STR_LIT>' ] <EOL> for i in xrange ( <NUM_LIT:0> , len ( unpackedSeries ) , <NUM_LIT:2> ) : <EOL> pointTime = unpackedSeries [ i ] <EOL> if pointTime == currentInterval : <EOL> neighborValues [ i // <NUM_LIT:2> ] = unpackedSeries [ i + <NUM_LIT:1> ] <EOL> currentInterval += step <EOL> knownValues = [ v for v in neighborValues if v is not None ] <EOL> if not knownValues : <EOL> return False <EOL> knownPercent = float ( len ( knownValues ) ) / float ( len ( neighborValues ) ) <EOL> if knownPercent >= xff : <EOL> aggregateValue = aggregate ( aggregationMethod , knownValues ) <EOL> myPackedPoint = struct . pack ( pointFormat , lowerIntervalStart , aggregateValue ) <EOL> fh . seek ( lower [ '<STR_LIT>' ] ) <EOL> packedPoint = fh . read ( pointSize ) <EOL> ( lowerBaseInterval , lowerBaseValue ) = struct . unpack ( pointFormat , packedPoint ) <EOL> if lowerBaseInterval == <NUM_LIT:0> : <EOL> fh . seek ( lower [ '<STR_LIT>' ] ) <EOL> fh . write ( myPackedPoint ) <EOL> else : <EOL> timeDistance = lowerIntervalStart - lowerBaseInterval <EOL> pointDistance = timeDistance // lower [ '<STR_LIT>' ] <EOL> byteDistance = pointDistance * pointSize <EOL> lowerOffset = lower [ '<STR_LIT>' ] + ( byteDistance % lower [ '<STR_LIT:size>' ] ) <EOL> fh . seek ( lowerOffset ) <EOL> fh . write ( myPackedPoint ) <EOL> return True <EOL> else : <EOL> return False <EOL> def update ( path , value , timestamp = None ) : <EOL> """<STR_LIT>""" <EOL> value = float ( value ) <EOL> fh = None <EOL> try : <EOL> fh = open ( path , '<STR_LIT>' ) <EOL> return file_update ( fh , value , timestamp ) <EOL> finally : <EOL> if fh : <EOL> fh . close ( ) <EOL> def file_update ( fh , value , timestamp ) : <EOL> if LOCK : <EOL> fcntl . flock ( fh . fileno ( ) , fcntl . LOCK_EX ) <EOL> header = __readHeader ( fh ) <EOL> now = int ( time . time ( ) ) <EOL> if timestamp is None : <EOL> timestamp = now <EOL> timestamp = int ( timestamp ) <EOL> diff = now - timestamp <EOL> if not ( ( diff < header [ '<STR_LIT>' ] ) and diff >= <NUM_LIT:0> ) : <EOL> raise TimestampNotCovered ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> for i , archive in enumerate ( header [ '<STR_LIT>' ] ) : <EOL> if archive [ '<STR_LIT>' ] < diff : continue <EOL> lowerArchives = header [ '<STR_LIT>' ] [ i + <NUM_LIT:1> : ] <EOL> break <EOL> myInterval = timestamp - ( timestamp % archive [ '<STR_LIT>' ] ) <EOL> myPackedPoint = struct . pack ( pointFormat , myInterval , value ) <EOL> fh . seek ( archive [ '<STR_LIT>' ] ) <EOL> packedPoint = fh . read ( pointSize ) <EOL> ( baseInterval , baseValue ) = struct . unpack ( pointFormat , packedPoint ) <EOL> if baseInterval == <NUM_LIT:0> : <EOL> fh . seek ( archive [ '<STR_LIT>' ] ) <EOL> fh . write ( myPackedPoint ) <EOL> baseInterval , baseValue = myInterval , value <EOL> else : <EOL> timeDistance = myInterval - baseInterval <EOL> pointDistance = timeDistance // archive [ '<STR_LIT>' ] <EOL> byteDistance = pointDistance * pointSize <EOL> myOffset = archive [ '<STR_LIT>' ] + ( byteDistance % archive [ '<STR_LIT:size>' ] ) <EOL> fh . seek ( myOffset ) <EOL> fh . write ( myPackedPoint ) <EOL> higher = archive <EOL> for lower in lowerArchives : <EOL> if not __propagate ( fh , header , myInterval , higher , lower ) : <EOL> break <EOL> higher = lower <EOL> if AUTOFLUSH : <EOL> fh . flush ( ) <EOL> os . fsync ( fh . fileno ( ) ) <EOL> def update_many ( path , points ) : <EOL> """<STR_LIT>""" <EOL> if not points : return <EOL> points = [ ( int ( t ) , float ( v ) ) for ( t , v ) in points ] <EOL> points . sort ( key = lambda p : p [ <NUM_LIT:0> ] , reverse = True ) <EOL> fh = None <EOL> try : <EOL> fh = open ( path , '<STR_LIT>' ) <EOL> return file_update_many ( fh , points ) <EOL> finally : <EOL> if fh : <EOL> fh . close ( ) <EOL> def file_update_many ( fh , points ) : <EOL> if LOCK : <EOL> fcntl . flock ( fh . fileno ( ) , fcntl . LOCK_EX ) <EOL> header = __readHeader ( fh ) <EOL> now = int ( time . time ( ) ) <EOL> archives = iter ( header [ '<STR_LIT>' ] ) <EOL> currentArchive = next ( archives ) <EOL> currentPoints = [ ] <EOL> for point in points : <EOL> age = now - point [ <NUM_LIT:0> ] <EOL> while currentArchive [ '<STR_LIT>' ] < age : <EOL> if currentPoints : <EOL> currentPoints . reverse ( ) <EOL> __archive_update_many ( fh , header , currentArchive , currentPoints ) <EOL> currentPoints = [ ] <EOL> try : <EOL> currentArchive = next ( archives ) <EOL> except StopIteration : <EOL> currentArchive = None <EOL> break <EOL> if not currentArchive : <EOL> break <EOL> currentPoints . append ( point ) <EOL> if currentArchive and currentPoints : <EOL> currentPoints . reverse ( ) <EOL> __archive_update_many ( fh , header , currentArchive , currentPoints ) <EOL> if AUTOFLUSH : <EOL> fh . flush ( ) <EOL> os . fsync ( fh . fileno ( ) ) <EOL> def __archive_update_many ( fh , header , archive , points ) : <EOL> step = archive [ '<STR_LIT>' ] <EOL> alignedPoints = [ ( timestamp - ( timestamp % step ) , value ) <EOL> for ( timestamp , value ) in points ] <EOL> alignedPoints = dict ( alignedPoints ) . items ( ) <EOL> packedStrings = [ ] <EOL> previousInterval = None <EOL> currentString = b"<STR_LIT>" <EOL> for ( interval , value ) in alignedPoints : <EOL> if ( not previousInterval ) or ( interval == previousInterval + step ) : <EOL> currentString += struct . pack ( pointFormat , interval , value ) <EOL> previousInterval = interval <EOL> else : <EOL> numberOfPoints = len ( currentString ) // pointSize <EOL> startInterval = previousInterval - ( step * ( numberOfPoints - <NUM_LIT:1> ) ) <EOL> packedStrings . append ( ( startInterval , currentString ) ) <EOL> currentString = struct . pack ( pointFormat , interval , value ) <EOL> previousInterval = interval <EOL> if currentString : <EOL> numberOfPoints = len ( currentString ) // pointSize <EOL> startInterval = previousInterval - ( step * ( numberOfPoints - <NUM_LIT:1> ) ) <EOL> packedStrings . append ( ( startInterval , currentString ) ) <EOL> fh . seek ( archive [ '<STR_LIT>' ] ) <EOL> packedBasePoint = fh . read ( pointSize ) <EOL> ( baseInterval , baseValue ) = struct . unpack ( pointFormat , packedBasePoint ) <EOL> if baseInterval == <NUM_LIT:0> : <EOL> baseInterval = packedStrings [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> for ( interval , packedString ) in packedStrings : <EOL> timeDistance = interval - baseInterval <EOL> pointDistance = timeDistance // step <EOL> byteDistance = pointDistance * pointSize <EOL> myOffset = archive [ '<STR_LIT>' ] + ( byteDistance % archive [ '<STR_LIT:size>' ] ) <EOL> fh . seek ( myOffset ) <EOL> archiveEnd = archive [ '<STR_LIT>' ] + archive [ '<STR_LIT:size>' ] <EOL> bytesBeyond = ( myOffset + len ( packedString ) ) - archiveEnd <EOL> if bytesBeyond > <NUM_LIT:0> : <EOL> fh . write ( packedString [ : - bytesBeyond ] ) <EOL> assert fh . tell ( ) == archiveEnd , "<STR_LIT>" % ( archiveEnd , fh . tell ( ) , bytesBeyond , len ( packedString ) ) <EOL> fh . seek ( archive [ '<STR_LIT>' ] ) <EOL> fh . write ( packedString [ - bytesBeyond : ] ) <EOL> else : <EOL> fh . write ( packedString ) <EOL> higher = archive <EOL> lowerArchives = [ arc for arc in header [ '<STR_LIT>' ] if arc [ '<STR_LIT>' ] > archive [ '<STR_LIT>' ] ] <EOL> for lower in lowerArchives : <EOL> fit = lambda i : i - ( i % lower [ '<STR_LIT>' ] ) <EOL> lowerIntervals = [ fit ( p [ <NUM_LIT:0> ] ) for p in alignedPoints ] <EOL> uniqueLowerIntervals = set ( lowerIntervals ) <EOL> propagateFurther = False <EOL> for interval in uniqueLowerIntervals : <EOL> if __propagate ( fh , header , interval , higher , lower ) : <EOL> propagateFurther = True <EOL> if not propagateFurther : <EOL> break <EOL> higher = lower <EOL> def info ( path ) : <EOL> """<STR_LIT>""" <EOL> fh = None <EOL> try : <EOL> fh = open ( path , '<STR_LIT:rb>' ) <EOL> return __readHeader ( fh ) <EOL> finally : <EOL> if fh : <EOL> fh . close ( ) <EOL> return None <EOL> def fetch ( path , fromTime , untilTime = None , now = None ) : <EOL> """<STR_LIT>""" <EOL> fh = None <EOL> try : <EOL> fh = open ( path , '<STR_LIT:rb>' ) <EOL> return file_fetch ( fh , fromTime , untilTime , now ) <EOL> finally : <EOL> if fh : <EOL> fh . close ( ) <EOL> def file_fetch ( fh , fromTime , untilTime , now = None ) : <EOL> header = __readHeader ( fh ) <EOL> if now is None : <EOL> now = int ( time . time ( ) ) <EOL> if untilTime is None : <EOL> untilTime = now <EOL> fromTime = int ( fromTime ) <EOL> untilTime = int ( untilTime ) <EOL> if ( fromTime > untilTime ) : <EOL> raise InvalidTimeInterval ( "<STR_LIT>" % ( fromTime , untilTime ) ) <EOL> oldestTime = now - header [ '<STR_LIT>' ] <EOL> if fromTime > now : <EOL> return None <EOL> if untilTime < oldestTime : <EOL> return None <EOL> if fromTime < oldestTime : <EOL> fromTime = oldestTime <EOL> if untilTime > now : <EOL> untilTime = now <EOL> diff = now - fromTime <EOL> for archive in header [ '<STR_LIT>' ] : <EOL> if archive [ '<STR_LIT>' ] >= diff : <EOL> break <EOL> return __archive_fetch ( fh , archive , fromTime , untilTime ) <EOL> def __archive_fetch ( fh , archive , fromTime , untilTime ) : <EOL> """<STR_LIT>""" <EOL> fromInterval = int ( fromTime - ( fromTime % archive [ '<STR_LIT>' ] ) ) + archive [ '<STR_LIT>' ] <EOL> untilInterval = int ( untilTime - ( untilTime % archive [ '<STR_LIT>' ] ) ) + archive [ '<STR_LIT>' ] <EOL> if fromInterval == untilInterval : <EOL> untilInterval += archive [ '<STR_LIT>' ] <EOL> fh . seek ( archive [ '<STR_LIT>' ] ) <EOL> packedPoint = fh . read ( pointSize ) <EOL> ( baseInterval , baseValue ) = struct . unpack ( pointFormat , packedPoint ) <EOL> if baseInterval == <NUM_LIT:0> : <EOL> step = archive [ '<STR_LIT>' ] <EOL> points = ( untilInterval - fromInterval ) // step <EOL> timeInfo = ( fromInterval , untilInterval , step ) <EOL> valueList = [ None ] * points <EOL> return ( timeInfo , valueList ) <EOL> timeDistance = fromInterval - baseInterval <EOL> pointDistance = timeDistance // archive [ '<STR_LIT>' ] <EOL> byteDistance = pointDistance * pointSize <EOL> fromOffset = archive [ '<STR_LIT>' ] + ( byteDistance % archive [ '<STR_LIT:size>' ] ) <EOL> timeDistance = untilInterval - baseInterval <EOL> pointDistance = timeDistance // archive [ '<STR_LIT>' ] <EOL> byteDistance = pointDistance * pointSize <EOL> untilOffset = archive [ '<STR_LIT>' ] + ( byteDistance % archive [ '<STR_LIT:size>' ] ) <EOL> fh . seek ( fromOffset ) <EOL> if fromOffset < untilOffset : <EOL> seriesString = fh . read ( untilOffset - fromOffset ) <EOL> else : <EOL> archiveEnd = archive [ '<STR_LIT>' ] + archive [ '<STR_LIT:size>' ] <EOL> seriesString = fh . read ( archiveEnd - fromOffset ) <EOL> fh . seek ( archive [ '<STR_LIT>' ] ) <EOL> seriesString += fh . read ( untilOffset - archive [ '<STR_LIT>' ] ) <EOL> byteOrder , pointTypes = pointFormat [ <NUM_LIT:0> ] , pointFormat [ <NUM_LIT:1> : ] <EOL> points = len ( seriesString ) // pointSize <EOL> seriesFormat = byteOrder + ( pointTypes * points ) <EOL> unpackedSeries = struct . unpack ( seriesFormat , seriesString ) <EOL> valueList = [ None ] * points <EOL> currentInterval = fromInterval <EOL> step = archive [ '<STR_LIT>' ] <EOL> for i in xrange ( <NUM_LIT:0> , len ( unpackedSeries ) , <NUM_LIT:2> ) : <EOL> pointTime = unpackedSeries [ i ] <EOL> if pointTime == currentInterval : <EOL> pointValue = unpackedSeries [ i + <NUM_LIT:1> ] <EOL> valueList [ i // <NUM_LIT:2> ] = pointValue <EOL> currentInterval += step <EOL> timeInfo = ( fromInterval , untilInterval , step ) <EOL> return ( timeInfo , valueList ) <EOL> def merge ( path_from , path_to ) : <EOL> """<STR_LIT>""" <EOL> fh_from = open ( path_from , '<STR_LIT:rb>' ) <EOL> fh_to = open ( path_to , '<STR_LIT>' ) <EOL> return file_merge ( fh_from , fh_to ) <EOL> def file_merge ( fh_from , fh_to ) : <EOL> headerFrom = __readHeader ( fh_from ) <EOL> headerTo = __readHeader ( fh_to ) <EOL> if headerFrom [ '<STR_LIT>' ] != headerTo [ '<STR_LIT>' ] : <EOL> raise NotImplementedError ( "<STR_LIT>" "<STR_LIT>" % ( fh_from . name , fh_to . name ) ) <EOL> archives = headerFrom [ '<STR_LIT>' ] <EOL> archives . sort ( key = operator . itemgetter ( '<STR_LIT>' ) ) <EOL> now = int ( time . time ( ) ) <EOL> untilTime = now <EOL> for archive in archives : <EOL> fromTime = now - archive [ '<STR_LIT>' ] <EOL> ( timeInfo , values ) = __archive_fetch ( fh_from , archive , fromTime , untilTime ) <EOL> ( start , end , archive_step ) = timeInfo <EOL> pointsToWrite = list ( ifilter ( <EOL> lambda points : points [ <NUM_LIT:1> ] is not None , <EOL> izip ( xrange ( start , end , archive_step ) , values ) ) ) <EOL> __archive_update_many ( fh_to , headerTo , archive , pointsToWrite ) <EOL> untilTime = fromTime <EOL> fh_from . close ( ) <EOL> fh_to . close ( ) <EOL> def diff ( path_from , path_to , ignore_empty = False ) : <EOL> """<STR_LIT>""" <EOL> fh_from = open ( path_from , '<STR_LIT:rb>' ) <EOL> fh_to = open ( path_to , '<STR_LIT:rb>' ) <EOL> diffs = file_diff ( fh_from , fh_to , ignore_empty ) <EOL> fh_to . close ( ) <EOL> fh_from . close ( ) <EOL> return diffs <EOL> def file_diff ( fh_from , fh_to , ignore_empty = False ) : <EOL> headerFrom = __readHeader ( fh_from ) <EOL> headerTo = __readHeader ( fh_to ) <EOL> if headerFrom [ '<STR_LIT>' ] != headerTo [ '<STR_LIT>' ] : <EOL> raise NotImplementedError ( "<STR_LIT>" "<STR_LIT>" % ( fh_from . name , fh_to . name ) ) <EOL> archives = headerFrom [ '<STR_LIT>' ] <EOL> archives . sort ( key = operator . itemgetter ( '<STR_LIT>' ) ) <EOL> archive_diffs = [ ] <EOL> now = int ( time . time ( ) ) <EOL> untilTime = now <EOL> for archive_number , archive in enumerate ( archives ) : <EOL> diffs = [ ] <EOL> startTime = now - archive [ '<STR_LIT>' ] <EOL> ( fromTimeInfo , fromValues ) = __archive_fetch ( fh_from , archive , startTime , untilTime ) <EOL> ( toTimeInfo , toValues ) = __archive_fetch ( fh_to , archive , startTime , untilTime ) <EOL> ( start , end , archive_step ) = ( min ( fromTimeInfo [ <NUM_LIT:0> ] , toTimeInfo [ <NUM_LIT:0> ] ) , max ( fromTimeInfo [ <NUM_LIT:1> ] , toTimeInfo [ <NUM_LIT:1> ] ) , min ( fromTimeInfo [ <NUM_LIT:2> ] , toTimeInfo [ <NUM_LIT:2> ] ) ) <EOL> points = map ( lambda s : ( s * archive_step + start , fromValues [ s ] , toValues [ s ] ) , xrange ( <NUM_LIT:0> , ( end - start ) // archive_step ) ) <EOL> if ignore_empty : <EOL> points = [ p for p in points if p [ <NUM_LIT:1> ] != None and p [ <NUM_LIT:2> ] != None ] <EOL> else : <EOL> points = [ p for p in points if p [ <NUM_LIT:1> ] != None or p [ <NUM_LIT:2> ] != None ] <EOL> diffs = [ p for p in points if p [ <NUM_LIT:1> ] != p [ <NUM_LIT:2> ] ] <EOL> archive_diffs . append ( ( archive_number , diffs , points . __len__ ( ) ) ) <EOL> untilTime = startTime <EOL> return archive_diffs </s>
