aggregation_ymd_mongo.py

import pymongo
from datetime import datetime, timedelta, time
from calendar import monthrange

db = pymongo.MongoClient().test

def generate_events(n, step):
    date = datetime.now()

    for i in range(n):
        event = {
            'date' : date,
            'value' : i
        }
        date = date + step
        yield event

def metric_query(metric, venue, date, updated=None):
    q = {
        'metric' : metric,
        'venue' : venue,
        'date' : date,
    }
    if updated:
        q['updated'] = updated
    return q

def preallocate_year(metric, venue, date):
    d = datetime.combine(date, time.min).replace(month=1, day=1)
    doc = {
        'm' : {str(m): 0 for m in  range(1,13)},
        'h' : {str(h): 0 for h in range(1,25)},
        'dw' : {str(dw): 0 for dw in range(1,8)},
    }
    db.yearly.update(metric_query(metric, venue, d, d), {'$set' : doc}, upsert=True, safe=True)

def preallocate_month(metric, venue, date):
    d = datetime.combine(date, time.min).replace(day=1)
    doc = {
        'd' : {str(m): 0 for m in  range(1, 1+monthrange(year=date.year, month=date.month)[1])},
        'h' : {str(h): 0 for h in range(1,25)},
        'dw' : {str(dw): 0 for dw in range(1,8)},
    }

    db.monthly.update(metric_query(metric, venue, d, d), {'$set' : doc}, upsert=True, safe=True)

def preallocate_day(metric, venue, date):
    d = datetime.combine(date, time.min)
    doc = {
        'h' : {str(h): 0 for h in range(1,25)},
    }

    db.daily.update(metric_query(metric, venue, d, d), {'$set' : doc}, upsert=True, safe=True)

def update_yearly(metric, venue, event, previous_event):

    d = datetime.combine(event['date'], time.min).replace(month=1, day=1)
    month = event['date'].month
    hour = event['date'].hour
    dow = event['date'].isoweekday()

    if not previous_event:
        delta = event['value']
    else:
        delta = event['value'] - previous_event['value']

    q = metric_query(metric, venue, d, {'$lt' : event['date']})
    u = {
        '$inc' : {
            'm.%d' % month : delta,
            'h.%d' % hour : delta,
            'dw.%d' % dow : delta,
        },
        '$set' : {
            'updated' : event['date'],
        }
    }
    db.yearly.update(q, u, upsert=True, safe=True)

def update_monthly(metric, venue, event, previous_event):

    d = datetime.combine(event['date'], time.min).replace(day=1)
    day = event['date'].day
    hour = event['date'].hour
    dow = event['date'].isoweekday()

    if not previous_event:
        delta = event['value']
    else:
        delta = event['value'] - previous_event['value']

    q = metric_query(metric, venue, d, {'$lt' : event['date']})
    u = {
        '$inc' : {
            'd.%d' % day : delta,
            'h.%d' % hour : delta,
            'dw.%d' % dow : delta,
        },
        '$set' : {
            'updated' : event['date'],
        }
    }
    db.monthly.update(q, u, upsert=True, safe=True)

def update_daily(metric, venue, event, previous_event):

    d = datetime.combine(event['date'], time.min)
    hour = event['date'].hour

    q = metric_query(metric, venue, d, {'$lt' : event['date']})
    u = {
        '$set' : {
            'h.%d' % hour : event['value'],
            'updated' : event['date'],
        }
    }
    db.daily.update(q, u, upsert=True, safe=True)

if __name__ == '__main__':
    
    db.yearly.drop()
    db.monthly.drop()
    db.daily.drop()

    today = datetime.utcnow()
    metric = 'bananas'
    venue = 1

    # preallocate_year(metric, venue, today)
    # preallocate_month(metric, venue, today)
    # preallocate_day(metric, venue, today)

    events = generate_events(n=500, step=timedelta(minutes=50))

    previous_event = None
    for event in events:
        update_yearly(metric, venue, event, previous_event)
        update_monthly(metric, venue, event, previous_event)
        update_daily(metric, venue, event, previous_event)

        previous_event = event

    d1 = datetime.utcnow()
    d2 = d1 + timedelta(days=6)

    days = db.daily.find({
        'date' : {
            '$gt' : d1,
            '$lt' : d2 + timedelta(days=1),
        }
    })
    n = days.count()
    
    data = []
    for day in days:
        for hour, value in day['h'].items():
            date = day['date'].replace(hour=int(hour))
            data.append((date, value))

    print 'querying day granularity between %s and %s : %s objects' % (d1, d2, n)
    # data to plot:
    # print data