averager.py

Go to the documentation of this file.

import logging, time, os, signal
log = logging.getLogger(__name__)
from datetime import timedelta
from pprint import pformat
from decimal import Decimal
from propagator import Propagator
from sourcevector import SourceVector
from territory import Territory

class Averager(Propagator):
    """
    Base class holding common averager features, such as the scrape logic
    which assumes:

    #. source instances correspond to fixed time measurement *snapshots*
    #. target instances represent source measurements over time ranges 
    #. 2 source instances are required to form one target instance, the target validity is derived from the datetimes of two source instances

    Initialisation in `Propagator` superclass
    """ 
    def changed(self, sv ):
        """
        Override in subclasses to return if a significant 
        change in source instances is observed.  This together
        with age checks is used to decide is the propagate method is called.

        :param sv:  source vector containing two source instances to interrogate for changes
        """
        return True

    def propagate(self, sv ):
        """
        Override this method in subclasses to yield one or more 
        write ready target dicts derived from the `sv[-1]` source instance or `sv[-1].aggd` aggregate dict 

        :param sv:  source vector containing two source instances to propagate to one target write  
        """
        yield {}

    def __call__(self):
        """
        Note the differences between averager and scraper:

        #. single pass rather than infinite loop is typically used

        #. `group_by` aggregation is used causing potentially many aggregated dicts to be propagated

        #. in an averager the writer must be instanciated within the propagate loop as 
           each aggregate dict corresponds to a different time range.

        #. time bins of typically 1 day are externally applied, via setting tcursor to a 2 element list  

        """
        i,w = 0,0
        while i<self.maxiter or self.maxiter==0:
            i += 1
            log.debug("i %s " % i )
            for sv in self:
                log.info("sv.source.xtn %r", sv.source.xtn ) 
                floor = sv.tcursor     ## tcursor is initialized based on target last validity endtime

                ## assumes single shot running
                if not getattr(sv,'territory',None) and self.aggregate_group_by_territory:
                    sv.territory = Territory( sv.source , by=self.aggregate_group_by_territory , suby=self.aggregate_group_by, att=self.aggregate_group_by_att , floor=floor  )
                    sv.territory.dump()
                    if len(sv.territory) == 0:
                        log.warn("no bins : insufficient time range for period\n%s\n" % sv.territory )
                    else:
                        log.info("looping over %s territory bins performing grouped aggregate queries in each bin  " % len(sv.territory) )

                ## this assumes independence of sources, maybe need to align territorial time ranges 
                ## so clients have option of mangling them together 
                nbin = len(sv.territory)
                for ibin,bin in enumerate(sv.territory): 
                    sv.tcursor = bin
                    log.debug( "averager ibin %s bin %s " % ( ibin, map(lambda _:_.ctime(),bin)) )
                    proceed = sv()  
                    if not proceed:
                        continue 
                    writes = []
                    for td in self.propagate( sv ):           ## iterate over "yield" supplier of target dicts   

                        localstart = td.get( "_%s_min" % self.aggregate_group_by_att , None)  # typically date_time
                        localend   = td.get( "_%s_max" % self.aggregate_group_by_att , None )
                        log.debug( "averager within propagate loop writer for range %s %s " % ( localstart, localend ))
                        wrt = self.target.writer( sv , localstart=localstart, localend=localend ) 

                        tdi = self.target.instance(**td)   
                        log.debug("tdi %r " % tdi )
                        wrt.Write(tdi)    

                        seqno = wrt.Close() ## hits target DB here
                        if seqno:
                            log.debug("write %s succeeded %s seqno %s  " % ( w, sv, seqno ) )
                            writes.append(dict(seqno=seqno,timestart=wrt.ctx.contextrange.timestart,timeend=wrt.ctx.contextrange.timeend,localstart=localstart,localend=localend))
                            w += 1
                        else:
                            log.fatal("writing failed %s " %  sv )
                            assert 0
                    pass
                    if len(writes)>0:
                        starts = map(lambda _:_['localstart'], writes)    
                        ends   = map(lambda _:_['localend'], writes)
                        seqnos = map(lambda _:_['seqno'], writes)
                        log.debug("propagated %s with seqnos %s:%s time range %s %s  " % ( len(writes), min(seqnos),max(seqnos), min(starts), max(ends) ))
                    else:
                        log.debug("no writes for bin %s %s " % ( ibin, map(lambda _:_.ctime(),bin)))
                    pass 
                    # end bin
                pass
                # end sv  
            pass 
            # end while
            self.handle_signal()
        pass

if __name__ == '__main__':
    pass