psudo testing of ParameterFunctionValidator

pfunc_validation.py

#!/usr/bin/env python

from coverage_model import *
import networkx as nx
import numpy as np
import os

'''

from scratch import pfunc_validation as pfv

for x in [x for x in dir(pfv) if x.startswith('test_')]:
    print 'Running %s' % x
    getattr(pfv, x)(True)
    print

'''

def _save_graph(dir, name, graph):
    if not name.lower().endswith('.dot'):
        name += '.dot'

    if not os.path.exists(dir):
        os.makedirs(dir)
    out_path = os.path.join(dir, name)
    if os.path.exists(out_path):
        os.remove(out_path)

    nx.write_dot(graph, out_path)

def test_good_1(save_graphs=False):
    print 'test_good_1: active_in==in_pc_set==IndParams; active_out==out_pc_set==DepParams'

    active_in = get_pc_set('TIME', 'LAT', 'LON', 'TEMPWAT_L0', 'CONDWAT_L0', 'PRESWAT_L0')
    in_pc_set = active_in
    out_pc_set = get_pc_set('DENSITY', 'PRACSAL', 'TEMPWAT_L1', 'CONDWAT_L1', 'PRESWAT_L1')
    active_out = [p.name for p in out_pc_set]

    pfv = ParameterFunctionValidator(active_in, in_pc_set, out_pc_set)

    # Test that we can make all the desired outputs
    ec = 0
    for t in active_out:
        print '  Validate {0}'.format(t)
        try:
            g = pfv.validate(t)
        except ValueError, ve:
            print 'ERROR!! %s' % ve.message
            ec += 1
            continue
        if save_graphs:
            _save_graph('pfunc_tests/test_good_1', t, g)

    print 'Passed !!' if ec == 0 else 'Failed %s tests!' % ec

def test_good_2(save_graphs=False):
    print 'test_good_2: active_in==in_pc_set==T,La,Lo,T0,C0,P1; active_out==R,PS,T1,C1; out_pc_set==R,PS,T1,C1,P1'

    active_in = get_pc_set('TIME', 'LAT', 'LON', 'TEMPWAT_L0', 'CONDWAT_L0', 'PRESWAT_L1')
    in_pc_set = active_in
    out_pc_set = get_pc_set('DENSITY', 'PRACSAL', 'TEMPWAT_L1', 'CONDWAT_L1')
    active_out = [p.name for p in out_pc_set] + ['PRESWAT_L1']

    pfv = ParameterFunctionValidator(active_in, in_pc_set, out_pc_set)

    # Test that we can make all the desired outputs
    ec = 0
    for t in active_out:
        print '  Validate {0}'.format(t)
        try:
            g = pfv.validate(t)
        except ValueError, ve:
            print 'ERROR!! %s' % ve.message
            ec += 1
            continue
        if save_graphs:
            _save_graph('pfunc_tests/test_good_2', t, g)

    print 'Passed !!' if ec == 0 else 'Failed %s tests!' % ec

def test_good_3(save_graphs=False):
    print 'test_good_3: active_in==in_pc_set==T,La,Lo,T1,P1,PS; active_out==out_pc_set==R,PS,T1,P1'

    active_in = get_pc_set('TIME', 'LAT', 'LON', 'TEMPWAT_L1', 'PRESWAT_L1', 'PRACSAL')
    in_pc_set = active_in
    out_pc_set = get_pc_set('DENSITY', 'PRACSAL', 'TEMPWAT_L1', 'PRESWAT_L1')
    active_out = [p.name for p in out_pc_set]

    pfv = ParameterFunctionValidator(active_in, in_pc_set, out_pc_set)

    # Test that we can make all the desired outputs
    ec = 0
    for t in active_out:
        print '  Validate {0}'.format(t)
        try:
            g = pfv.validate(t)
        except ValueError, ve:
            print 'ERROR!! %s' % ve.message
            ec += 1
            continue
        if save_graphs:
            _save_graph('pfunc_tests/test_good_3', t, g)

    print 'Passed !!' if ec == 0 else 'Failed %s tests!' % ec

def test_good_4(save_graphs=False):
    print 'test_good_4: active_in==in_pc_set==T,La,Lo,T1,P1,C1; out_pc_set==R,PS,T1,P1; active_out==R,PS,T1,P1,C1'

    active_in = get_pc_set('TIME', 'LAT', 'LON', 'TEMPWAT_L0', 'PRESWAT_L1', 'CONDWAT_L1')
    in_pc_set = active_in
    out_pc_set = get_pc_set('DENSITY', 'PRACSAL', 'TEMPWAT_L1', 'PRESWAT_L1')
    active_out = [p.name for p in out_pc_set] + ['CONDWAT_L1']

    pfv = ParameterFunctionValidator(active_in, in_pc_set, out_pc_set)

    # Test that we can make all the desired outputs
    ec = 0
    for t in active_out:
        print '  Validate {0}'.format(t)
        try:
            g = pfv.validate(t)
        except ValueError, ve:
            print 'ERROR!! %s' % ve.message
            ec += 1
            continue
        if save_graphs:
            _save_graph('pfunc_tests/test_good_4', t, g)

    print 'Passed !!' if ec == 0 else 'Failed %s tests!' % ec

def test_good_5(save_graphs=False):
    print 'test_good_5: in_pc_set==T,La,Lo,T1,P1,C0,C1; active_in==T,La,Lo,T1,P1,C0; out_pc_set==R,PS,T1,P1; active_out==R,PS,T1,P1,C1'

    active_in = get_pc_set('TIME', 'LAT', 'LON', 'TEMPWAT_L0', 'PRESWAT_L1', 'CONDWAT_L0')
    in_pc_set = active_in + get_pc_set('CONDWAT_L1')
    out_pc_set = get_pc_set('DENSITY', 'PRACSAL', 'TEMPWAT_L1', 'PRESWAT_L1')
    active_out = [p.name for p in out_pc_set] + ['CONDWAT_L1']

    pfv = ParameterFunctionValidator(active_in, in_pc_set, out_pc_set)

    # Test that we can make all the desired outputs
    ec = 0
    for t in active_out:
        print '  Validate {0}'.format(t)
        try:
            g = pfv.validate(t)
        except ValueError, ve:
            print 'ERROR!! %s' % ve.message
            ec += 1
            continue
        if save_graphs:
            _save_graph('pfunc_tests/test_good_5', t, g)

    print 'Passed !!' if ec == 0 else 'Failed %s tests!' % ec

def test_bad_1(save_graphs=False):
    print 'test_good_3: active_in==in_pc_set==T,La,Lo,T1,P1,PS; active_out==out_pc_set==R,PS,T1,P1,C1'

    active_in = get_pc_set('TIME', 'LAT', 'LON', 'TEMPWAT_L1', 'PRESWAT_L1', 'PRACSAL')
    in_pc_set = active_in
    out_pc_set = get_pc_set('DENSITY', 'PRACSAL', 'TEMPWAT_L1', 'PRESWAT_L1', 'CONDWAT_L1')
    active_out = [p.name for p in out_pc_set]

    pfv = ParameterFunctionValidator(active_in, in_pc_set, out_pc_set)

    # Test that we can make all the desired outputs
    ec=0
    for t in active_out:
        print '  Validate {0}'.format(t)
        try:
            g = pfv.validate(t)
        except ValueError, ve:
            print 'ERROR!! %s' % ve.message
            ec += 1
            continue
        if save_graphs:
            _save_graph('pfunc_tests/test_bad_1', t, g)

    print 'Passed !!' if ec == 0 else 'Failed %s tests!' % ec

def test_bad_2(save_graphs=False):
    print 'test_bad_2: active_in==in_pc_set==T,La,Lo,T1,P1,PS; active_out==out_pc_set==R,PS,T1,P1,C1'

    active_in = get_pc_set('TIME', 'LAT', 'LON', 'TEMPWAT_L1', 'PRESWAT_L1', 'PRACSAL')
    in_pc_set = active_in
    out_pc_set = get_pc_set('DENSITY', 'PRACSAL', 'TEMPWAT_L1', 'PRESWAT_L1', 'CONDWAT_L1')
    active_out = [p.name for p in out_pc_set] + ['CONDWAT_L1']

    pfv = ParameterFunctionValidator(active_in, in_pc_set, out_pc_set)

    # Test that we can make all the desired outputs
    ec=0
    for t in active_out:
        print '  Validate {0}'.format(t)
        try:
            g = pfv.validate(t)
        except ValueError, ve:
            print 'ERROR!! %s' % ve.message
            ec += 1
            continue
        if save_graphs:
            _save_graph('pfunc_tests/test_bad_2', t, g)

    print 'Passed !!' if ec == 0 else 'Failed %s tests!' % ec


def get_pc_set(*pnames):
    all_pc = _create_all_params()
    return [all_pc[x] for x in pnames]

def _create_all_params():
    '''
     [
     'DENSITY',
     'TIME',
     'LON',
     'TEMPWAT_L1',
     'TEMPWAT_L0',
     'CONDWAT_L1',
     'CONDWAT_L0',
     'PRESWAT_L1',
     'PRESWAT_L0',
     'LAT',
     'PRACSAL'
     ]
    @return:
    '''

    contexts = {}

    t_ctxt = ParameterContext('TIME', param_type=QuantityType(value_encoding=np.dtype('int64')))
    t_ctxt.uom = 'seconds since 01-01-1900'
    contexts['TIME'] = t_ctxt

    lat_ctxt = ParameterContext('LAT', param_type=ConstantType(QuantityType(value_encoding=np.dtype('float32'))), fill_value=-9999)
    lat_ctxt.axis = AxisTypeEnum.LAT
    lat_ctxt.uom = 'degree_north'
    contexts['LAT'] = lat_ctxt

    lon_ctxt = ParameterContext('LON', param_type=ConstantType(QuantityType(value_encoding=np.dtype('float32'))), fill_value=-9999)
    lon_ctxt.axis = AxisTypeEnum.LON
    lon_ctxt.uom = 'degree_east'
    contexts['LON'] = lon_ctxt

    # Independent Parameters

    # Temperature - values expected to be the decimal results of conversion from hex
    temp_ctxt = ParameterContext('TEMPWAT_L0', param_type=QuantityType(value_encoding=np.dtype('float32')), fill_value=-9999)
    temp_ctxt.uom = 'deg_C'
    contexts['TEMPWAT_L0'] = temp_ctxt

    # Conductivity - values expected to be the decimal results of conversion from hex
    cond_ctxt = ParameterContext('CONDWAT_L0', param_type=QuantityType(value_encoding=np.dtype('float32')), fill_value=-9999)
    cond_ctxt.uom = 'S m-1'
    contexts['CONDWAT_L0'] = cond_ctxt

    # Pressure - values expected to be the decimal results of conversion from hex
    press_ctxt = ParameterContext('PRESWAT_L0', param_type=QuantityType(value_encoding=np.dtype('float32')), fill_value=-9999)
    press_ctxt.uom = 'dbar'
    contexts['PRESWAT_L0'] = press_ctxt


    # Dependent Parameters

    # TEMPWAT_L1 = (TEMPWAT_L0 / 10000) - 10
    tl1_func = '(T_L0 / 10000) - 10'
    tl1_pmap = {'T_L0':'TEMPWAT_L0'}
    expr = NumexprExpression('TEMPWAT_L1', tl1_func, tl1_pmap)
    tempL1_ctxt = ParameterContext('TEMPWAT_L1', param_type=ParameterFunctionType(expression=expr), variability=VariabilityEnum.TEMPORAL)
    tempL1_ctxt.uom = 'deg_C'
    contexts['TEMPWAT_L1'] = tempL1_ctxt

    # CONDWAT_L1 = (CONDWAT_L0 / 100000) - 0.5
    cl1_func = '(C_L0 / 100000) - 0.5'
    cl1_pmap = {'C_L0':'CONDWAT_L0'}
    expr = NumexprExpression('CONDWAT_L1', cl1_func, cl1_pmap)
    condL1_ctxt = ParameterContext('CONDWAT_L1', param_type=ParameterFunctionType(expression=expr), variability=VariabilityEnum.TEMPORAL)
    condL1_ctxt.uom = 'S m-1'
    contexts['CONDWAT_L1'] = condL1_ctxt

    # Equation uses p_range, which is a calibration coefficient - Fixing to 679.34040721
    #   PRESWAT_L1 = (PRESWAT_L0 * p_range / (0.85 * 65536)) - (0.05 * p_range)
    pl1_func = '(P_L0 * 679.34040721 / (0.85 * 65536)) - (0.05 * 679.34040721)'
    pl1_pmap = {'P_L0':'PRESWAT_L0'}
    expr = NumexprExpression('PRESWAT_L1', pl1_func, pl1_pmap)
    presL1_ctxt = ParameterContext('PRESWAT_L1', param_type=ParameterFunctionType(expression=expr), variability=VariabilityEnum.TEMPORAL)
    presL1_ctxt.uom = 'S m-1'
    contexts['PRESWAT_L1'] = presL1_ctxt

    # Density & practical salinity calucluated using the Gibbs Seawater library - available via python-gsw project:
    #       https://code.google.com/p/python-gsw/ & http://pypi.python.org/pypi/gsw/3.0.1

    # PRACSAL = gsw.SP_from_C((CONDWAT_L1 * 10), TEMPWAT_L1, PRESWAT_L1)
    owner = 'gsw'
    sal_func = 'SP_from_C'
    sal_arglist = [NumexprExpression('CONDWAT_L1*10', 'C*10', {'C':'CONDWAT_L1'}), 'TEMPWAT_L1', 'PRESWAT_L1']
    sal_kwargmap = None
    expr = PythonExpression('PRACSAL', owner, sal_func, sal_arglist, sal_kwargmap)
    sal_ctxt = ParameterContext('PRACSAL', param_type=ParameterFunctionType(expr), variability=VariabilityEnum.TEMPORAL)
    sal_ctxt.uom = 'g kg-1'
    contexts['PRACSAL'] = sal_ctxt

    # absolute_salinity = gsw.SA_from_SP(PRACSAL, PRESWAT_L1, longitude, latitude)
    # conservative_temperature = gsw.CT_from_t(absolute_salinity, TEMPWAT_L1, PRESWAT_L1)
    # DENSITY = gsw.rho(absolute_salinity, conservative_temperature, PRESWAT_L1)
    owner = 'gsw'
    abs_sal_expr = PythonExpression('abs_sal', owner, 'SA_from_SP', ['PRACSAL', 'PRESWAT_L1', 'LON','LAT'], None)
    cons_temp_expr = PythonExpression('cons_temp', owner, 'CT_from_t', [abs_sal_expr, 'TEMPWAT_L1', 'PRESWAT_L1'], None)
    dens_expr = PythonExpression('DENSITY', owner, 'rho', [abs_sal_expr, cons_temp_expr, 'PRESWAT_L1'], None)
    dens_ctxt = ParameterContext('DENSITY', param_type=ParameterFunctionType(dens_expr), variability=VariabilityEnum.TEMPORAL)
    dens_ctxt.uom = 'kg m-3'
    contexts['DENSITY'] = dens_ctxt

    return contexts