Google Gantt Functional Refactoring

googlegantt2.py

from pysistence import make_dict
import datetime
d = datetime.date

GOOGLE_CHARTS_API_URL = 'https://chart.googleapis.com/chart'
DEFAULT_COLOR = '4D89F9FF' # A kind of nice blue.

def uniquify(seq, idfun=None):
    """
    Order-preserving uniquify of a list.
    Lifted directly from http://www.peterbe.com/plog/uniqifiers-benchmark
    """

    if idfun is None:
        idfun = lambda x: x

    seen = {}
    result = []
    for item in seq:
        marker = idfun(item)

        if marker in seen:
            continue

        seen[marker] = 1

        result.append(item)

    return result


def parse_color(color):
    "Accept various input formats for color and convert to 8-byte hex RGBA"
    if len(color) == 3:
        color = color[0]*2 + color[1]*2 + color[2]*2 + 'FF'
    if len(color) == 4:
        color = color[0]*2 + color[1]*2 + color[2]*2 + color[3]*2
    if len(color) == 6:
        color = color + 'FF'

    return color.upper()


def duration(chart_or_task):
    return (chart_or_task['end_date'] - chart_or_task['start_date']).days


def gantt_chart(name, width=650, height=200, progress=None, tasks=()):
    return make_dict({
        'name': name,
        'width': 650,
        'height': 200,
        'start_date': min(*(t['start_date'] for t in tasks)) if tasks else None,
        'end_date': max(*(t['end_date'] for t in tasks)) if tasks else None,
        'progress': progress or d.today(),
        'tasks': tasks
    })


def gantt_category(name, color=None):
    if color is None:
        color = DEFAULT_COLOR

    return make_dict({
        'name': name,
        'color': parse_color(color)
    })


def gantt_task(name, start_date=None, end_date=None, **kwargs):
    color = parse_color(kwargs.get('color', DEFAULT_COLOR))
    category = kwargs.get('category', None)
    if category is None:
        category = gantt_category('', color)

    depends_on = kwargs.get('depends_on')
    if depends_on is not None:
        start_date = depends_on['end_date']

    duration = kwargs.get('duration')
    if duration is not None:
        end_date = start_date + datetime.timedelta(days=int(duration))

    return make_dict({
        'name': name,
        'start_date': start_date,
        'end_date': end_date,
        'category': category
    })


def add_task(gc, task):
    return gc.using(
        start_date=min(gc['start_date'], task['start_date'])
                       if gc['start_date'] else task['start_date'],
        end_date=max(gc['end_date'], task['end_date'])
                     if gc['end_date'] else task['end_date'],
        tasks=gc['tasks'] + (task,)
    )


def add_tasks(gc, tasks):
    return reduce(add_task, tasks, gc)
        

def day_series(gc, format='%d/%m'):
    "Get the list of date labels for this chart"
    start_date = gc['start_date']
    dur = duration(gc) + 1 # Add 1 because we also label 0 days.


    if gc['width'] / dur > 80:
        skip_n_labels = 1
    else:
        skip_n_labels =  int(1. / (float(gc['width']) / float(dur) / 80.))

    for i in range(dur):
        if i % skip_n_labels == 0:
            yield (start_date + datetime.timedelta(days=i)).strftime(format)
        else:
            yield ' '


def chart_params(gc, day_format='%d/%m'):
    if len(gc['tasks']) == 0:
        raise Exception("Must have at least one task")
        
    # Compute the bar width for a desired height
    task_size = int((gc['height'] - 50.) / len(gc['tasks'])) - 4

    # Compute the grid spacing
    axis_step = 100. / duration(gc)

    categories = uniquify((t['category'] for t in gc['tasks']), lambda c: c['name'])
    colors = (c['color'] for c in categories)


    params = {
            'cht': 'bhs', #Chart Type: Horizontal Bar
            'chco': 'FFFFFF00,' + ','.join(colors), #Colors: Transparent, blue
            'chtt': gc['name'],
            'chs': '%sx%s' % (int(gc['width']), int(gc['height'])), #Chart size
            'chds': '0,%s' % duration(gc), # Data duration
            'chbh': '%s,4,0' % task_size, #Bar Width
            'chg': '%s,0' % axis_step, # Grid size
            'chxt': 'x,y',
            'chxl': '0:|' + '|'.join(day_series(gc, format=day_format)) + '|1:|' + '|'.join(t['name'] for t in reversed(gc['tasks'])), # Axes labels
    }


    # Add category labels if necessary
    if reduce(lambda acc, c: acc or c['name'], categories, False): # If any category has a title
        params['chdl'] = '|' + '|'.join([c['name'] for c in categories if c['name']]) # Legend


    # Add a progress indicator if progress was passed
    if gc['progress'] and gc['progress'] >= gc['start_date'] and gc['progress'] <= gc['end_date']:
        days = (gc['progress'] - gc['start_date']).days
        params['chm'] = 'r,%s33,0,0,%s' % (DEFAULT_COLOR[:6], float(days) / duration(gc))

    # Hold offsets to each task and the duration of each task in days.
    offsets = [str((t['start_date'] - gc['start_date']).days) for t in gc['tasks']]
    data = []


    # Create a separate set of data for each color, but preserve order.
    for c in categories:
        zeroed_data = ['0'] * len(gc['tasks'])

        for i, t in enumerate(gc['tasks']):
            if t['category'] != c:
                continue

            zeroed_data[i] = str(duration(t))
        data.append(zeroed_data)

    params['chd'] = 't:'+','.join(offsets) + '|' + '|'.join([','.join(d) for d in data])

    return params

def get_url(gc):
    "Returns a GET url for simple image embedding"
    params = chart_params(gc)

    return GOOGLE_CHARTS_API_URL + '?' + '&'.join(['%s=%s' % (key, params[key]) for key in params])


if __name__ == "__main__":
    late = gantt_category('Late', 'c00')
    on_time = gantt_category('On Time', '0c0')
    upcoming = gantt_category('Upcoming', '00c')

    task1 = gantt_task('Late Task', d(2011, 2, 20), d(2011, 2, 25), category=late)
    task2 = gantt_task('On Time Task', depends_on=task1, duration=3, category=on_time)
    task3 = gantt_task('Future Task', depends_on=task2, duration=7, category=upcoming)

    gc = gantt_chart('Test Chart', width=650, height=200, progress=d(2011, 2, 27), tasks=())



    print get_url(add_tasks(gc, (task1, task2, task3)))