bag_analyzer.py

Go to the documentation of this file.

#!/usr/bin/env python
'''
bag_analyzer.py
Mac Mason <mac@cs.duke.edu>

Given a list of bagfiles, compute for each their (a) duration and size (in
bits).
'''

import sys
import os
import subprocess

if len(sys.argv) < 2:
  print "Usage: bag_analyzer.py <one or more bagfiles>"
  sys.exit(1)

def sum_tuple(a, b):
  ''' 
  Our tuples are (duration (sec.), compressed size, uncompressed size (GB),
  count)
  '''
  return (a[0] + b[0], a[1] + b[1], a[2] + b[2], a[3] + b[3])

def tuplize(f):
  '''
  Given a filename f, call rosbag info and parse out the three tuple
  values.
  '''
  lines = subprocess.Popen(['rosbag', 'info', f],
                           stdout=subprocess.PIPE,
                           stderr=subprocess.STDOUT,
                           env=os.environ).communicate()[0].split('\n')

  def parse_duration(lines):
    ''' Find the duration in second, given the lines.'''
    try:
      line = (line for line in lines if 'duration:' in line).next()  # OWW.
    except StopIteration:
      print lines
    return float(line.split()[-1].split('(')[1].split('s')[0])

  def parse_compressed(lines):
    '''
    Get the compressed value. Note the check for '%' to make sure we don't
    actually parse the 'uncompressed' line.
    '''
    line = (l for l in lines if 'compressed:' in l and '%' in l).next()
    x = float(line.split()[1])
    if 'GB' not in line:
      x /= 1024.0
    return x

  def parse_uncompressed(lines):
    '''
    And the uncompressed size.
    '''
    line = (line for line in lines if 'uncompressed:' in line).next()
    return float(line.split()[1])

  return (parse_duration(lines), 
          parse_compressed(lines),
          parse_uncompressed(lines),
          1)



print "I have", len(sys.argv[1:]), "bagfiles."
res = reduce(sum_tuple, (tuplize(x) for x in sys.argv[1:]))
print "Duration: %f" % res[0]
print "Compressed: %f GB" % res[1]
print "Uncompressed: %f GB " % res[2]
print "Count: %d" % res[3]