Convert MODIS MOD14 active fire HDFv4 to CSV

mod14hdf4_to_fireloc.py

#!/usr/bin/env python

#
# Author: [email protected]
# Version: 20130307c
# Source: https://gist.github.com/cynici/3223681
# Requires: pyHDF
#
# To install pyHDF on Ubuntu 11+, adapt these instructions accordingly:
# https://gist.github.com/1408088
#
import sys, os, re
import datetime
import csv
from pyhdf.SD import SD, SDC
import logging

help_text = """Read MODIS MOD14 HDF4 and output RSRU fireloc CSV
Optional output PATHSPEC goes through two-pass substitution:
- All strftime() parameters are substituted using fire datetime
- %%(yyyyjjj)s, %%(hhmm)s and %%(satellite)s placeholders
%prog mod14.hdf ... [options, -h for details]"""



#! Main
#! ----
def main(argv=None):
    if argv is None:
        argv = sys.argv

    debuglevelD = {
        'debug': logging.DEBUG,
        'info': logging.INFO,
        'warning': logging.WARNING,
        'error': logging.ERROR,
        'critical': logging.CRITICAL,
    }
    
    from optparse import OptionParser       
    parser = OptionParser(usage=help_text)
    parser.add_option("--output", "-O", dest="output", type="string", help="Output pathname"%debuglevelD.keys(), metavar='PATHSPEC')
    parser.add_option("--loglevel", dest="loglevel", type="string", help="Verbosity %s"%debuglevelD.keys(), metavar='LOGLEVEL')
    
    #parser.set_defaults(**defvals)
    (options, args) = parser.parse_args()

    if options.loglevel:
        if options.loglevel not in debuglevelD: raise AssertionError("Verbosity level must be one of: %s"%debuglevelD.keys())
        dbglvl = debuglevelD[options.loglevel]
    else:
        dbglvl = logging.INFO
    logger = logging.getLogger()
    logger.setLevel(dbglvl)
    ch = logging.StreamHandler()
    ch.setFormatter( logging.Formatter('%(asctime)s %(lineno)d %(name)s %(funcName)s %(message)s') )
    ch.setLevel(dbglvl)
    logger.addHandler(ch)

    for hdffile in args:
        filename = os.path.basename(hdffile)
        
        hhmm = re.search('\.(?P<HH>\d{2})(?P<MM>\d{2})\.', filename)
        if hhmm is None:
            parser.error("No .HHMM. in filename: %s" % filename)
        
        yyyyjjj = None
        y4j3 = re.search('\.(?P<yyyy>\d{4})(?P<jjj>\d{3})\.', filename)
        if y4j3:
            yyyyjjj = y4j3.group('yyyy') + y4j3.group('jjj')
        else:
            y2j3 = re.search('\.(?P<yy>\d{2})(?P<jjj>\d{3})\.', filename)
            yyyyjjj = '20' + y2j3.group('yy') + y2j3.group('jjj')
        if yyyyjjj is None:
            parser.error("No .YYYYJJJ. or .YYJJJ. in filename: %s" % filename)
        obsdatetime = datetime.datetime.strptime("%s %s:%s"%(yyyyjjj, hhmm.group('HH'), hhmm.group('MM')), "%Y%j %H:%M")
        
        satellite = None
        if re.search('^mod14\.', filename, re.I) or re.search('^t1\.', filename, re.I):
            satellite = 'T'
        elif re.search('^myd14\.', filename, re.I) or re.search('^a1\.', filename, re.I):
            satellite = 'A'
        else:
            parser.error('No satellite name in filename: %s' % hdffile)
        
        
        hdfpath_no_ext, ext = os.path.splitext(hdffile)
        if options.output:
            csvfile = obsdatetime.strftime(options.output) % dict(yyyyjjj=yyyyjjj, satellite=satellite, hhmm=hhmm)
        else:
            # Output save in same directory as input
            csvfile = hdfpath_no_ext + '.txt'        
        
        logger.debug("sat=%s csv=%s datetime=%s" % (satellite, csvfile, obsdatetime))

        hdfobj = SD(hdffile, SDC.READ)
        csvobj = csv.writer(open(csvfile, "w"))
        logger.info("creating %s ..." % csvfile)
        
        FP_latitude_ds_obj = hdfobj.select ("FP_latitude")
        FP_longitude_ds_obj = hdfobj.select ("FP_longitude")
        FP_line_ds_obj = hdfobj.select ("FP_line")
        FP_sample_ds_obj = hdfobj.select ("FP_sample")
        FP_T21_ds_obj = hdfobj.select ("FP_T21")
        FP_power_ds_obj = hdfobj.select ("FP_power")
        FP_confidence_ds_obj = hdfobj.select ("FP_confidence")
        
        number_samples_int = FP_latitude_ds_obj.info()[2]
        for i in range (number_samples_int):
            csv_row_data_tuple = ("%.3f" % (round (FP_latitude_ds_obj[i], 3), ), \
                                  "%.3f" % (round (FP_longitude_ds_obj[i], 3), ), \
                                  "%.1f" % (round (FP_T21_ds_obj[i], 1), ), \
                                  "%.1f" % (round (FP_power_ds_obj[i], 1), ), \
                                  1.0, \
                                  obsdatetime.strftime("%m/%d/%Y"), \
                                  obsdatetime.strftime("%H%M"), \
                                  satellite, \
                                  FP_confidence_ds_obj[i])
            csvobj.writerow(csv_row_data_tuple)
        csvobj = None
        hdfobj.end()
    return 0
        

if __name__ == "__main__":
    sys.exit(main())