Select Git revision
grbadub.py 6.51 KiB
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
grbadub.py
==========
PURPOSE
Patch incorrect calibration values in GOES-16 preliminary L1b NetCDF4 files.
REQUIRES
Python 2.7 or newer
python netCDF4 module and its dependencies
See grbadub.sh wrapper, which can download and install a suitable relocatable linux python runtime if needed.
:author: R.K.Garcia <rayg@ssec.wisc.edu>
:copyright: 2017 by University of Wisconsin Regents, see AUTHORS for more details
:license: GPLv3, see LICENSE for more details
"""
import os, sys, socket
import logging, unittest, argparse
from hashlib import md5
from datetime import datetime
import netCDF4 as nc4
import numpy as np
__author__ = 'rkgarcia@wisc.edu'
__docformat__ = 'reStructuredText'
LOG = logging.getLogger(__name__)
BC1 = {
7: 0.43361,
8: 1.55228,
9: 0.34427,
10: 0.05651,
11: 0.18733,
12: 0.09102,
13: 0.07550,
14: 0.22516,
15: 0.21702,
16: 0.06266
}
BC2 = {
7: 0.99939,
8: 0.99667,
9: 0.99918,
10: 0.99986,
11: 0.99948,
12: 0.99971,
13: 0.99975,
14: 0.99920,
15: 0.99916,
16: 0.99974
}
FK1 = {
7: 2.02263E+05,
8: 5.06871E+04,
9: 3.58283E+04,
10: 3.01740E+04,
11: 1.97799E+04,
12: 1.34321E+04,
13: 1.08033E+04,
14: 8.51022E+03,
15: 6.45462E+03,
16: 5.10127E+03
}
FK2 = {
7: 3.69819E+03,
8: 2.33158E+03,
9: 2.07695E+03,
10: 1.96138E+03,
11: 1.70383E+03,
12: 1.49761E+03,
13: 1.39274E+03,
14: 1.28627E+03,
15: 1.17303E+03,
16: 1.08453E+03
}
ESUN = {
1: 2017.1648,
2: 1631.3351,
3: 957.0699,
4: 360.9018,
5: 242.5404,
6: 76.8999,
}
def calc_kappa0(d, e_sun):
return d * d * np.pi / e_sun
def bandaid(nc, band):
"""
Generic band-aid function using FK1, FK2, BC1, BC2, ESUN look-up tables
:param nc: netCDF4.Dataset file we're editing, in read-write mode
:param band: int band number from nc.band_id[:]
"""
LOG.debug("processing patches for band %d" % band)
bc1 = BC1.get(band, None)
if bc1 is not None:
LOG.debug("patching planck_bc1")
nc['planck_bc1'][:] = bc1
bc2 = BC2.get(band, None)
if bc2 is not None:
LOG.debug("patching planck_bc2")
nc['planck_bc2'][:] = bc2
fk1 = FK1.get(band, None)
if fk1 is not None:
LOG.debug("patching planck_fk1")
nc['planck_fk1'][:] = fk1
fk2 = FK2.get(band, None)
if fk2 is not None:
LOG.debug("patching planck_fk2")
nc['planck_fk2'][:] = fk2
d = float(nc['earth_sun_distance_anomaly_in_AU'][:])
e_sun = ESUN.get(band, None)
if e_sun is not None:
LOG.debug("patching esun")
nc['esun'][:] = e_sun
kappa0 = calc_kappa0(d, e_sun)
LOG.debug('kappa0 calculated as %f' % kappa0)
LOG.debug("patching kappa0 for reflectance band")
nc['kappa0'][:] = kappa0
# Table of which functions to use for which bands, should there be any variances
# Specialized per-band patching routines can be substituted here.
BAND_TABLE = {
1: bandaid,
2: bandaid,
3: bandaid,
4: bandaid,
5: bandaid,
6: bandaid,
7: bandaid,
8: bandaid,
9: bandaid,
10: bandaid,
11: bandaid,
12: bandaid,
13: bandaid,
14: bandaid,
15: bandaid,
16: bandaid,
}
def grbadub_attributes():
"""
return dictionary of attributes to add to file to show we've visited it
:return: dictionary of attributes
"""
# collect md5 of this file
mdobj = md5()
me = open(__file__, 'rb').read()
mdobj.update(me)
my_md5 = mdobj.hexdigest()
# collect my modification time
mtime = datetime.fromtimestamp(os.stat(__file__).st_mtime)
my_mtime = mtime.isoformat()
# host we're running on
my_host = socket.gethostname()
# collect current UTC time
now = datetime.utcnow()
my_runtime = now.isoformat()
return {
'grbadub_md5': my_md5,
'grbadub_last_modified': my_mtime,
'grbadub_exec_time': my_runtime,
'grbadub_exec_host': my_host
}
def grbadub(path, attributes):
"""
edit a specified GOES-16 L1b file using band_id variable as key
:param path: netCDF4 file to edit, must be writable
:param attributes: dictionary of global attributes to add to file
:return:
"""
LOG.debug('opening %s' % path)
nc = nc4.Dataset(path, 'r+')
band = int(nc['band_id'][:])
fn_band = BAND_TABLE[band]
LOG.debug('patching band %d' % band)
fn_band(nc, band)
for attr_name, attr_val in attributes.items():
LOG.debug('setting .%s=%s' % (attr_name, repr(attr_val)))
setattr(nc, attr_name, attr_val)
LOG.info('patched %s' % path)
nc.close()
# class tests(unittest.TestCase):
# data_file = os.environ.get('TEST_DATA', os.path.expanduser("~/Data/test_files/thing.dat"))
#
# def setUp(self):
# pass
#
# def test_something(self):
# pass
def _debug(typ, value, tb):
""" interactive debug on exception encountered
"""
if not sys.stdin.isatty():
sys.__excepthook__(typ, value, tb)
else:
import traceback, pdb
traceback.print_exception(typ, value, tb)
# …then start the debugger in post-mortem mode.
pdb.post_mortem(tb) # more “modern”
def main():
parser = argparse.ArgumentParser(
description="Patch calibration parameters in GOES-16 L1b NetCDF4 files",
epilog="",
fromfile_prefix_chars='@')
parser.add_argument('-v', '--verbose', dest='verbosity', action="count", default=0,
help='each occurrence increases verbosity 1 level through ERROR-WARNING-INFO-DEBUG')
parser.add_argument('-d', '--debug', dest='debug', action='store_true',
help="enable interactive PDB debugger on exception")
# http://docs.python.org/2.7/library/argparse.html#nargs
# parser.add_argument('--stuff', nargs='5', dest='my_stuff',
# help="one or more random things")
parser.add_argument('inputs', nargs='*',
help="input files to process")
args = parser.parse_args()
levels = [logging.ERROR, logging.WARN, logging.INFO, logging.DEBUG]
logging.basicConfig(level=levels[min(3, args.verbosity)])
if args.debug:
sys.excepthook = _debug
if not args.inputs:
unittest.main()
return 0
signature = grbadub_attributes()
LOG.debug("signature is %s" % repr(signature))
for pn in args.inputs:
grbadub(pn, signature)
return 0
if __name__ == '__main__':
sys.exit(main())