iris-grib_save_equivalence.ipynb

iris-grib_save_equivalence.py

"""
Check save works the same between iris-grib and iris.fileformats.grib(strict),
when testing all the FF example files from iris-test-data.

"""

import os
import os.path
import shutil
from shutil import rmtree
import tempfile
from tempfile import mkdtemp

import iris
from iris.fileformats.grib import save_grib2 as old_irisffg_save
import iris.tests
from iris.tests import get_data_path as get_testdata_path
import iris_grib
from iris_grib import save_grib2 as new_irisgrib_save
from nose import exc

def filesin(dirpath):
    assert os.path.isdir(dirpath)
    for name in os.listdir(dirpath):
        subpath = os.path.join(dirpath, name)
        if os.path.isdir(subpath):
            for result in filesin(subpath):
                yield result
        else:
            yield subpath

def all_testdata_FF_files():
    ff_testfiles_path = get_testdata_path('FF')
    for filepath in filesin(ff_testfiles_path):
        yield filepath

def all_testdata_FF_cubes():
    for i_file, filepath in enumerate(all_testdata_FF_files()):
        try:
            cubes = iris.load(filepath)
        except Exception as e:
            cubes = [iris.cube.Cube([1], long_name="FAILED : " + str(e))]
        for i_cube, cube in enumerate(cubes):
            yield i_file, filepath, i_cube, cube

def save_result(cube, save_call):
    try:
        temp_output_dirpath = mkdtemp()
        temp_output_filepath = os.path.join(temp_output_dirpath, 'temp.grib2')

        try:
            save_call(cube, temp_output_filepath)
            exception = None
        except Exception as e:
            exception = e

        if exception is None:
            result_brief, exception_type, exception_text = ('ok', None, None)
        else:
            result_brief, exception_type, exception_text = \
                ('FAIL', type(exception), str(exception))
    finally:
        rmtree(temp_output_dirpath, ignore_errors=True)

    return result_brief, exception_type, exception_text


def report_on(cube):
    old_msg, old_type, old_text = save_result(cube, old_irisffg_save)
    new_msg, new_type, new_text = save_result(cube, new_irisgrib_save)
    if old_msg == 'ok' and new_msg == 'ok':
        report = '  BOTH ok.'
    else:
        if old_msg == new_msg:
            report = '  BOTH FAILED'
            if old_type == new_type and old_text == new_text:
                report += ': SAME - {} {}'.format(new_type, new_text)
            else:
                report += ': DIFFERENT..'
                msg = '\n    {} {} {}'
                report += msg.format('old', old_type, old_text)
                report += msg.format('new', new_type, new_text)
        else:
            # One failed
            report = '  DIFFERENT SUCCESSES'
            msg = '\n    {} "{}" {} {}'
            report += msg.format('old', old_msg, old_type, old_text)
            report += msg.format('new', new_msg, new_type, new_text)
    return report


def test():
    print 'Iris version : ', iris.__version__
    print 'Old save routine from : ', old_irisffg_save.__module__
    print 'New save routine from : ', new_irisgrib_save.__module__
    print
    for i_file, filepath, i_cube, cube in all_testdata_FF_cubes():
        msg = 'file#{:02d} "{}" : cube#{:02d} - {}'
        print msg.format(i_file, filepath.rjust(40), i_cube, cube.summary(shorten=True))
        print report_on(cube)


if __name__ == '__main__':
    test()