graphdeps.py

#!/usr/bin/env python
"""
Graph the DAG of tasks

Modified from https://gist.github.com/BrianHicks/2769821
"""

import argparse
import json
import logging
from subprocess import Popen, PIPE, CalledProcessError
import sys

logging.getLogger(__name__)


def call_subprocess(cmd, stdin=None):
    """Run subprocess with the given command and input."""
    proc = Popen(cmd, stdout=PIPE, stderr=PIPE, stdin=PIPE)
    result, err = proc.communicate(stdin)
    if err:
        logging.error(err)
    if proc.returncode != 0:
        raise CalledProcessError(result)
    return result


def call_taskwarrior(args):
    """call taskwarrior, returning output and error"""
    logging.info('Calling TaskWarrior')
    return call_subprocess(['task', 'export'] + args)


def call_dot(args):
    """call dot, returning stdout and stdout"""
    logging.info('Calling dot')
    lines = ['digraph dependencies {']
    lines.extend(args)
    lines.append('}')
    dot_input = ' '.join(lines)
    logging.debug('dot args: "%s"', dot_input)
    return call_subprocess('dot -T pdf'.split(), dot_input)


def parse_args(argv):
    """Parse the command-line arguments."""
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('query', nargs='*', default=['status:pending'],
                        help='the arguments to pass to "task"')
    parser.add_argument('--verbose', action='store_true',
                        help='Enable outputting all debugging statements')
    args = parser.parse_args(argv)
    return args


def compose_dot_input(tasks):
    """Generate graphviz input from a list of tasks dictionaries"""
    # first pass:
    logging.info('Printing Labels')
    nodes = []
    node_template = '"{0[uuid]}"[label="{0[id]}: {0[description]}"];'
    for datum in tasks:
        label = node_template.format(datum)
        nodes.append(label)
    # second pass:
    logging.info('Resolving Dependencies')
    dependencies = []
    for datum in tasks:
        for dep in datum.get('depends', '').split(','):
            if dep != '':
                dependencies.append('"%s" -> "%s";' % (dep, datum['uuid']))
    return nodes + dependencies


def main(argv=None):
    """TODO: Do stuff."""
    options = parse_args(argv)
    if options.verbose:
        logging.getLogger().setLevel(logging.DEBUG)
    # Do stuff:
    logging.info('%s', options)

    tw_result = call_taskwarrior(options.query)
    lines = [line.strip() for line in tw_result.split('\n')
             if line.strip()]

    tasks = [json.loads(t) for t in lines]
    tasks = sorted(tasks, key=lambda x: x['urgency'], reverse=True)

    dot_input = compose_dot_input(tasks)
    pdf = call_dot(dot_input)

    logging.info('Writing to deps.pdf')
    with open('deps.pdf', 'w') as pdf_file:
        pdf_file.write(pdf)


if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    sys.exit(main())