<s> import contextlib <EOL> import numpy as np <EOL> import pandas as pd <EOL> import warnings <EOL> from xarray import conventions , Variable , Dataset , open_dataset <EOL> from xarray . core import utils , indexing <EOL> from . import TestCase , requires_netCDF4 , unittest <EOL> from . test_backends import CFEncodedDataTest <EOL> from xarray . core . pycompat import iteritems <EOL> from xarray . backends . memory import InMemoryDataStore <EOL> from xarray . backends . common import WritableCFDataStore <EOL> from xarray . conventions import decode_cf <EOL> class TestMaskedAndScaledArray ( TestCase ) : <EOL> def test ( self ) : <EOL> x = conventions . MaskedAndScaledArray ( np . arange ( <NUM_LIT:3> ) , fill_value = <NUM_LIT:0> ) <EOL> self . assertEqual ( x . dtype , np . dtype ( '<STR_LIT:float>' ) ) <EOL> self . assertEqual ( x . shape , ( <NUM_LIT:3> , ) ) <EOL> self . assertEqual ( x . size , <NUM_LIT:3> ) <EOL> self . assertEqual ( x . ndim , <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( x ) , <NUM_LIT:3> ) <EOL> self . assertArrayEqual ( [ np . nan , <NUM_LIT:1> , <NUM_LIT:2> ] , x ) <EOL> x = conventions . MaskedAndScaledArray ( np . arange ( <NUM_LIT:3> ) , add_offset = <NUM_LIT:1> ) <EOL> self . assertArrayEqual ( np . arange ( <NUM_LIT:3> ) + <NUM_LIT:1> , x ) <EOL> x = conventions . MaskedAndScaledArray ( np . arange ( <NUM_LIT:3> ) , scale_factor = <NUM_LIT:2> ) <EOL> self . assertArrayEqual ( <NUM_LIT:2> * np . arange ( <NUM_LIT:3> ) , x ) <EOL> x = conventions . MaskedAndScaledArray ( np . array ( [ - <NUM_LIT> , - <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> ] ) , <EOL> - <NUM_LIT> , <NUM_LIT> , <NUM_LIT:1> ) <EOL> expected = np . array ( [ np . nan , <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> self . assertArrayEqual ( expected , x ) <EOL> def test_0d ( self ) : <EOL> x = conventions . MaskedAndScaledArray ( np . array ( <NUM_LIT:0> ) , fill_value = <NUM_LIT:0> ) <EOL> self . assertTrue ( np . isnan ( x ) ) <EOL> self . assertTrue ( np . isnan ( x [ ... ] ) ) <EOL> x = conventions . MaskedAndScaledArray ( np . array ( <NUM_LIT:0> ) , fill_value = <NUM_LIT:10> ) <EOL> self . assertEqual ( <NUM_LIT:0> , x [ ... ] ) <EOL> def test_multiple_fill_value ( self ) : <EOL> x = conventions . MaskedAndScaledArray ( <EOL> np . arange ( <NUM_LIT:4> ) , fill_value = np . array ( [ <NUM_LIT:0> , <NUM_LIT:1> ] ) ) <EOL> self . assertArrayEqual ( [ np . nan , np . nan , <NUM_LIT:2> , <NUM_LIT:3> ] , x ) <EOL> x = conventions . MaskedAndScaledArray ( <EOL> np . array ( <NUM_LIT:0> ) , fill_value = np . array ( [ <NUM_LIT:0> , <NUM_LIT:1> ] ) ) <EOL> self . assertTrue ( np . isnan ( x ) ) <EOL> self . assertTrue ( np . isnan ( x [ ... ] ) ) <EOL> class TestCharToStringArray ( TestCase ) : <EOL> def test_wrapper_class ( self ) : <EOL> array = np . array ( list ( '<STR_LIT:abc>' ) , dtype = '<STR_LIT:S>' ) <EOL> actual = conventions . CharToStringArray ( array ) <EOL> expected = np . array ( '<STR_LIT:abc>' , dtype = '<STR_LIT:S>' ) <EOL> self . assertEqual ( actual . dtype , expected . dtype ) <EOL> self . assertEqual ( actual . shape , expected . shape ) <EOL> self . assertEqual ( actual . size , expected . size ) <EOL> self . assertEqual ( actual . ndim , expected . ndim ) <EOL> with self . assertRaises ( TypeError ) : <EOL> len ( actual ) <EOL> self . assertArrayEqual ( expected , actual ) <EOL> with self . assertRaises ( IndexError ) : <EOL> actual [ : <NUM_LIT:2> ] <EOL> self . assertEqual ( str ( actual ) , '<STR_LIT:abc>' ) <EOL> array = np . array ( [ list ( '<STR_LIT:abc>' ) , list ( '<STR_LIT>' ) ] , dtype = '<STR_LIT:S>' ) <EOL> actual = conventions . CharToStringArray ( array ) <EOL> expected = np . array ( [ '<STR_LIT:abc>' , '<STR_LIT>' ] , dtype = '<STR_LIT:S>' ) <EOL> self . assertEqual ( actual . dtype , expected . dtype ) <EOL> self . assertEqual ( actual . shape , expected . shape ) <EOL> self . assertEqual ( actual . size , expected . size ) <EOL> self . assertEqual ( actual . ndim , expected . ndim ) <EOL> self . assertEqual ( len ( actual ) , len ( expected ) ) <EOL> self . assertArrayEqual ( expected , actual ) <EOL> self . assertArrayEqual ( expected [ : <NUM_LIT:1> ] , actual [ : <NUM_LIT:1> ] ) <EOL> with self . assertRaises ( IndexError ) : <EOL> actual [ : , : <NUM_LIT:2> ] <EOL> def test_char_to_string ( self ) : <EOL> array = np . array ( [ [ '<STR_LIT:a>' , '<STR_LIT:b>' , '<STR_LIT:c>' ] , [ '<STR_LIT:d>' , '<STR_LIT:e>' , '<STR_LIT:f>' ] ] ) <EOL> expected = np . array ( [ '<STR_LIT:abc>' , '<STR_LIT>' ] ) <EOL> actual = conventions . char_to_string ( array ) <EOL> self . assertArrayEqual ( actual , expected ) <EOL> expected = np . array ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> actual = conventions . char_to_string ( array . T ) <EOL> self . assertArrayEqual ( actual , expected ) <EOL> def test_string_to_char ( self ) : <EOL> array = np . array ( [ [ '<STR_LIT>' , '<STR_LIT>' ] , [ '<STR_LIT>' , '<STR_LIT>' ] ] ) <EOL> expected = np . array ( [ [ [ '<STR_LIT:a>' , '<STR_LIT:b>' ] , [ '<STR_LIT:c>' , '<STR_LIT:d>' ] ] , <EOL> [ [ '<STR_LIT:e>' , '<STR_LIT:f>' ] , [ '<STR_LIT:g>' , '<STR_LIT:h>' ] ] ] ) <EOL> actual = conventions . string_to_char ( array ) <EOL> self . assertArrayEqual ( actual , expected ) <EOL> expected = np . array ( [ [ [ '<STR_LIT:a>' , '<STR_LIT:b>' ] , [ '<STR_LIT:e>' , '<STR_LIT:f>' ] ] , <EOL> [ [ '<STR_LIT:c>' , '<STR_LIT:d>' ] , [ '<STR_LIT:g>' , '<STR_LIT:h>' ] ] ] ) <EOL> actual = conventions . string_to_char ( array . T ) <EOL> self . assertArrayEqual ( actual , expected ) <EOL> @ np . vectorize <EOL> def _ensure_naive_tz ( dt ) : <EOL> if hasattr ( dt , '<STR_LIT>' ) : <EOL> return dt . replace ( tzinfo = None ) <EOL> else : <EOL> return dt <EOL> class TestDatetime ( TestCase ) : <EOL> @ requires_netCDF4 <EOL> def test_cf_datetime ( self ) : <EOL> import netCDF4 as nc4 <EOL> for num_dates , units in [ <EOL> ( np . arange ( <NUM_LIT:10> ) , '<STR_LIT>' ) , <EOL> ( np . arange ( <NUM_LIT:10> ) . reshape ( <NUM_LIT:2> , <NUM_LIT:5> ) , '<STR_LIT>' ) , <EOL> ( <NUM_LIT> + np . arange ( <NUM_LIT:5> ) , '<STR_LIT>' ) , <EOL> ( <NUM_LIT> + np . arange ( <NUM_LIT:5> ) , '<STR_LIT>' ) , <EOL> ( <NUM_LIT> + np . arange ( <NUM_LIT:5> ) , u'<STR_LIT>' ) , <EOL> ( <NUM_LIT> + np . arange ( <NUM_LIT:5> ) , '<STR_LIT>' ) , <EOL> ( <NUM_LIT:10> , '<STR_LIT>' ) , <EOL> ( [ <NUM_LIT:10> ] , '<STR_LIT>' ) , <EOL> ( [ [ <NUM_LIT:10> ] ] , '<STR_LIT>' ) , <EOL> ( [ <NUM_LIT:10> , <NUM_LIT:10> ] , '<STR_LIT>' ) , <EOL> ( np . array ( <NUM_LIT:10> ) , '<STR_LIT>' ) , <EOL> ( <NUM_LIT:0> , '<STR_LIT>' ) , <EOL> ( [ <NUM_LIT:0> ] , '<STR_LIT>' ) , <EOL> ( [ [ <NUM_LIT:0> ] ] , '<STR_LIT>' ) , <EOL> ( np . arange ( <NUM_LIT:2> ) , '<STR_LIT>' ) , <EOL> ( np . arange ( <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT> ) , '<STR_LIT>' ) , <EOL> ( <NUM_LIT> , '<STR_LIT>' ) , <EOL> ( [ <NUM_LIT:0.5> , <NUM_LIT> ] , '<STR_LIT>' ) , <EOL> ( <NUM_LIT:0> , '<STR_LIT>' ) , <EOL> ( <NUM_LIT:0> , '<STR_LIT>' ) , <EOL> ] : <EOL> for calendar in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> expected = _ensure_naive_tz ( nc4 . num2date ( num_dates , units , calendar ) ) <EOL> print ( num_dates , units , calendar ) <EOL> with warnings . catch_warnings ( ) : <EOL> warnings . filterwarnings ( '<STR_LIT:ignore>' , <EOL> '<STR_LIT>' ) <EOL> actual = conventions . decode_cf_datetime ( num_dates , units , <EOL> calendar ) <EOL> if ( isinstance ( actual , np . ndarray ) and <EOL> np . issubdtype ( actual . dtype , np . datetime64 ) ) : <EOL> actual_cmp = actual . astype ( '<STR_LIT>' ) <EOL> else : <EOL> actual_cmp = actual <EOL> self . assertArrayEqual ( expected , actual_cmp ) <EOL> encoded , _ , _ = conventions . encode_cf_datetime ( actual , units , <EOL> calendar ) <EOL> if '<STR_LIT>' not in units : <EOL> self . assertArrayEqual ( num_dates , np . around ( encoded , <NUM_LIT:1> ) ) <EOL> if ( hasattr ( num_dates , '<STR_LIT>' ) and num_dates . ndim == <NUM_LIT:1> and <EOL> '<STR_LIT>' not in units ) : <EOL> encoded , _ , _ = conventions . encode_cf_datetime ( <EOL> pd . Index ( actual ) , units , calendar ) <EOL> self . assertArrayEqual ( num_dates , np . around ( encoded , <NUM_LIT:1> ) ) <EOL> def test_decoded_cf_datetime_array ( self ) : <EOL> actual = conventions . DecodedCFDatetimeArray ( <EOL> np . array ( [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> ] ) , '<STR_LIT>' , '<STR_LIT>' ) <EOL> expected = pd . date_range ( '<STR_LIT>' , periods = <NUM_LIT:3> ) . values <EOL> self . assertEqual ( actual . dtype , np . dtype ( '<STR_LIT>' ) ) <EOL> self . assertArrayEqual ( actual , expected ) <EOL> actual = conventions . DecodedCFDatetimeArray ( <EOL> np . array ( [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> ] ) , '<STR_LIT>' ) <EOL> self . assertEqual ( actual . dtype , np . dtype ( '<STR_LIT>' ) ) <EOL> self . assertArrayEqual ( actual , expected ) <EOL> def test_slice_decoded_cf_datetime_array ( self ) : <EOL> actual = conventions . DecodedCFDatetimeArray ( <EOL> np . array ( [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> ] ) , '<STR_LIT>' , '<STR_LIT>' ) <EOL> expected = pd . date_range ( '<STR_LIT>' , periods = <NUM_LIT:3> ) . values <EOL> self . assertEqual ( actual . dtype , np . dtype ( '<STR_LIT>' ) ) <EOL> self . assertArrayEqual ( actual [ slice ( <NUM_LIT:0> , <NUM_LIT:2> ) ] , expected [ slice ( <NUM_LIT:0> , <NUM_LIT:2> ) ] ) <EOL> actual = conventions . DecodedCFDatetimeArray ( <EOL> np . array ( [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> ] ) , '<STR_LIT>' , '<STR_LIT>' ) <EOL> expected = pd . date_range ( '<STR_LIT>' , periods = <NUM_LIT:3> ) . values <EOL> self . assertEqual ( actual . dtype , np . dtype ( '<STR_LIT>' ) ) <EOL> self . assertArrayEqual ( actual [ [ <NUM_LIT:0> , <NUM_LIT:2> ] ] , expected [ [ <NUM_LIT:0> , <NUM_LIT:2> ] ] ) <EOL> def test_decode_cf_datetime_non_standard_units ( self ) : <EOL> expected = pd . date_range ( periods = <NUM_LIT:100> , start = '<STR_LIT>' , freq = '<STR_LIT:h>' ) <EOL> units = '<STR_LIT>' <EOL> actual = conventions . decode_cf_datetime ( np . arange ( <NUM_LIT:100> ) , units ) <EOL> self . assertArrayEqual ( actual , expected ) <EOL> def test_decode_cf_with_conflicting_fill_missing_value ( self ) : <EOL> var = Variable ( [ '<STR_LIT:t>' ] , np . arange ( <NUM_LIT:10> ) , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:1> } ) <EOL> self . assertRaisesRegexp ( ValueError , "<STR_LIT>" , <EOL> lambda : conventions . decode_cf_variable ( var ) ) <EOL> @ requires_netCDF4 <EOL> def test_decode_cf_datetime_non_iso_strings ( self ) : <EOL> expected = pd . date_range ( periods = <NUM_LIT:100> , start = '<STR_LIT>' , freq = '<STR_LIT:h>' ) <EOL> cases = [ ( np . arange ( <NUM_LIT:100> ) , '<STR_LIT>' ) , <EOL> ( np . arange ( <NUM_LIT:100> ) , '<STR_LIT>' ) , <EOL> ( np . arange ( <NUM_LIT:100> ) , '<STR_LIT>' ) ] <EOL> for num_dates , units in cases : <EOL> actual = conventions . decode_cf_datetime ( num_dates , units ) <EOL> self . assertArrayEqual ( actual , expected ) <EOL> @ requires_netCDF4 <EOL> def test_decode_non_standard_calendar ( self ) : <EOL> import netCDF4 as nc4 <EOL> for calendar in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] : <EOL> units = '<STR_LIT>' <EOL> times = pd . date_range ( '<STR_LIT>' , end = '<STR_LIT>' , <EOL> freq = '<STR_LIT:H>' ) <EOL> noleap_time = nc4 . date2num ( times . to_pydatetime ( ) , units , <EOL> calendar = calendar ) <EOL> expected = times . values <EOL> with warnings . catch_warnings ( ) : <EOL> warnings . filterwarnings ( '<STR_LIT:ignore>' , '<STR_LIT>' ) <EOL> actual = conventions . decode_cf_datetime ( noleap_time , units , <EOL> calendar = calendar ) <EOL> self . assertEqual ( actual . dtype , np . dtype ( '<STR_LIT>' ) ) <EOL> abs_diff = abs ( actual - expected ) <EOL> self . assertTrue ( ( abs_diff <= np . timedelta64 ( <NUM_LIT:1> , '<STR_LIT:s>' ) ) . all ( ) ) <EOL> @ requires_netCDF4 <EOL> def test_decode_non_standard_calendar_single_element ( self ) : <EOL> units = '<STR_LIT>' <EOL> for calendar in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] : <EOL> for num_time in [ <NUM_LIT> , [ <NUM_LIT> ] , [ [ <NUM_LIT> ] ] ] : <EOL> with warnings . catch_warnings ( ) : <EOL> warnings . filterwarnings ( '<STR_LIT:ignore>' , <EOL> '<STR_LIT>' ) <EOL> actual = conventions . decode_cf_datetime ( num_time , units , <EOL> calendar = calendar ) <EOL> self . assertEqual ( actual . dtype , np . dtype ( '<STR_LIT>' ) ) <EOL> @ requires_netCDF4 <EOL> def test_decode_non_standard_calendar_single_element_fallback ( self ) : <EOL> import netCDF4 as nc4 <EOL> units = '<STR_LIT>' <EOL> dt = nc4 . netcdftime . datetime ( <NUM_LIT> , <NUM_LIT:2> , <NUM_LIT> ) <EOL> for calendar in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> num_time = nc4 . date2num ( dt , units , calendar ) <EOL> with self . assertWarns ( '<STR_LIT>' ) : <EOL> actual = conventions . decode_cf_datetime ( num_time , units , <EOL> calendar = calendar ) <EOL> expected = np . asarray ( nc4 . num2date ( num_time , units , calendar ) ) <EOL> print ( num_time , calendar , actual , expected ) <EOL> self . assertEqual ( actual . dtype , np . dtype ( '<STR_LIT:O>' ) ) <EOL> self . assertEqual ( expected , actual ) <EOL> @ requires_netCDF4 <EOL> def test_decode_non_standard_calendar_multidim_time ( self ) : <EOL> import netCDF4 as nc4 <EOL> calendar = '<STR_LIT>' <EOL> units = '<STR_LIT>' <EOL> times1 = pd . date_range ( '<STR_LIT>' , end = '<STR_LIT>' , freq = '<STR_LIT:D>' ) <EOL> times2 = pd . date_range ( '<STR_LIT>' , end = '<STR_LIT>' , freq = '<STR_LIT:D>' ) <EOL> noleap_time1 = nc4 . date2num ( times1 . to_pydatetime ( ) , units , <EOL> calendar = calendar ) <EOL> noleap_time2 = nc4 . date2num ( times2 . to_pydatetime ( ) , units , <EOL> calendar = calendar ) <EOL> mdim_time = np . empty ( ( len ( noleap_time1 ) , <NUM_LIT:2> ) , ) <EOL> mdim_time [ : , <NUM_LIT:0> ] = noleap_time1 <EOL> mdim_time [ : , <NUM_LIT:1> ] = noleap_time2 <EOL> expected1 = times1 . values <EOL> expected2 = times2 . values <EOL> with warnings . catch_warnings ( ) : <EOL> warnings . filterwarnings ( '<STR_LIT:ignore>' , '<STR_LIT>' ) <EOL> actual = conventions . decode_cf_datetime ( mdim_time , units , <EOL> calendar = calendar ) <EOL> self . assertEqual ( actual . dtype , np . dtype ( '<STR_LIT>' ) ) <EOL> self . assertArrayEqual ( actual [ : , <NUM_LIT:0> ] , expected1 ) <EOL> self . assertArrayEqual ( actual [ : , <NUM_LIT:1> ] , expected2 ) <EOL> @ requires_netCDF4 <EOL> def test_decode_non_standard_calendar_fallback ( self ) : <EOL> import netCDF4 as nc4 <EOL> for year in [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] : <EOL> for calendar in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> calendar = '<STR_LIT>' <EOL> units = '<STR_LIT>' . format ( year ) <EOL> num_times = np . arange ( <NUM_LIT:100> ) <EOL> expected = nc4 . num2date ( num_times , units , calendar ) <EOL> with warnings . catch_warnings ( record = True ) as w : <EOL> warnings . simplefilter ( '<STR_LIT>' ) <EOL> actual = conventions . decode_cf_datetime ( num_times , units , <EOL> calendar = calendar ) <EOL> self . assertEqual ( len ( w ) , <NUM_LIT:1> ) <EOL> self . assertIn ( '<STR_LIT>' , <EOL> str ( w [ <NUM_LIT:0> ] . message ) ) <EOL> self . assertEqual ( actual . dtype , np . dtype ( '<STR_LIT:O>' ) ) <EOL> self . assertArrayEqual ( actual , expected ) <EOL> def test_cf_datetime_nan ( self ) : <EOL> for num_dates , units , expected_list in [ <EOL> ( [ np . nan ] , '<STR_LIT>' , [ '<STR_LIT>' ] ) , <EOL> ( [ np . nan , <NUM_LIT:0> ] , '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] ) , <EOL> ( [ np . nan , <NUM_LIT:0> , <NUM_LIT:1> ] , '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) , <EOL> ] : <EOL> with warnings . catch_warnings ( ) : <EOL> warnings . filterwarnings ( '<STR_LIT:ignore>' , '<STR_LIT>' ) <EOL> actual = conventions . decode_cf_datetime ( num_dates , units ) <EOL> expected = np . array ( expected_list , dtype = '<STR_LIT>' ) <EOL> self . assertArrayEqual ( expected , actual ) <EOL> def test_infer_datetime_units ( self ) : <EOL> for dates , expected in [ ( pd . date_range ( '<STR_LIT>' , periods = <NUM_LIT:5> ) , <EOL> '<STR_LIT>' ) , <EOL> ( pd . date_range ( '<STR_LIT>' , freq = '<STR_LIT:H>' , <EOL> periods = <NUM_LIT:2> ) , <EOL> '<STR_LIT>' ) , <EOL> ( [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] , <EOL> '<STR_LIT>' ) , <EOL> ( pd . to_datetime ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' ) , <EOL> ( pd . to_datetime ( [ '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' ) , <EOL> ( pd . to_datetime ( [ '<STR_LIT>' , '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' ) , <EOL> ( pd . to_datetime ( [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' ) , <EOL> ] : <EOL> self . assertEqual ( expected , conventions . infer_datetime_units ( dates ) ) <EOL> def test_cf_timedelta ( self ) : <EOL> examples = [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' , np . int64 ( <NUM_LIT:1> ) ) , <EOL> ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , '<STR_LIT>' , np . array ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] , '<STR_LIT>' ) ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , np . int64 ( <NUM_LIT:1> ) ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , np . int64 ( <NUM_LIT:1> ) ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , np . int64 ( <NUM_LIT:1> ) ) , <EOL> ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , None , [ np . nan , <NUM_LIT:0> , <NUM_LIT:1> ] ) , <EOL> ( [ '<STR_LIT>' , '<STR_LIT>' ] , '<STR_LIT>' , [ <NUM_LIT:0.5> , <NUM_LIT:1.0> ] ) , <EOL> ] <EOL> if pd . __version__ >= '<STR_LIT>' : <EOL> examples . extend ( [ ( np . timedelta64 ( '<STR_LIT>' , '<STR_LIT>' ) , '<STR_LIT>' , np . nan ) , <EOL> ( [ '<STR_LIT>' , '<STR_LIT>' ] , '<STR_LIT>' , [ np . nan , np . nan ] ) ] ) <EOL> for timedeltas , units , numbers in examples : <EOL> timedeltas = pd . to_timedelta ( timedeltas , box = False ) <EOL> numbers = np . array ( numbers ) <EOL> expected = numbers <EOL> actual , _ = conventions . encode_cf_timedelta ( timedeltas , units ) <EOL> self . assertArrayEqual ( expected , actual ) <EOL> self . assertEqual ( expected . dtype , actual . dtype ) <EOL> if units is not None : <EOL> expected = timedeltas <EOL> actual = conventions . decode_cf_timedelta ( numbers , units ) <EOL> self . assertArrayEqual ( expected , actual ) <EOL> self . assertEqual ( expected . dtype , actual . dtype ) <EOL> expected = np . timedelta64 ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> actual = conventions . decode_cf_timedelta ( np . array ( np . nan ) , '<STR_LIT>' ) <EOL> self . assertArrayEqual ( expected , actual ) <EOL> def test_infer_timedelta_units ( self ) : <EOL> for deltas , expected in [ <EOL> ( pd . to_timedelta ( [ '<STR_LIT>' , '<STR_LIT>' ] ) , '<STR_LIT>' ) , <EOL> ( pd . to_timedelta ( [ '<STR_LIT>' , '<STR_LIT>' ] ) , '<STR_LIT>' ) , <EOL> ( pd . to_timedelta ( [ '<STR_LIT>' , '<STR_LIT>' , np . nan ] ) , '<STR_LIT>' ) , <EOL> ( pd . to_timedelta ( [ '<STR_LIT>' , '<STR_LIT>' ] ) , '<STR_LIT>' ) ] : <EOL> self . assertEqual ( expected , conventions . infer_timedelta_units ( deltas ) ) <EOL> def test_invalid_units_raises_eagerly ( self ) : <EOL> ds = Dataset ( { '<STR_LIT:time>' : ( '<STR_LIT:time>' , [ <NUM_LIT:0> , <NUM_LIT:1> ] , { '<STR_LIT>' : '<STR_LIT>' } ) } ) <EOL> with self . assertRaisesRegexp ( ValueError , '<STR_LIT>' ) : <EOL> decode_cf ( ds ) <EOL> @ requires_netCDF4 <EOL> def test_dataset_repr_with_netcdf4_datetimes ( self ) : <EOL> attrs = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } <EOL> with warnings . catch_warnings ( ) : <EOL> warnings . filterwarnings ( '<STR_LIT:ignore>' , '<STR_LIT>' ) <EOL> ds = decode_cf ( Dataset ( { '<STR_LIT:time>' : ( '<STR_LIT:time>' , [ <NUM_LIT:0> , <NUM_LIT:1> ] , attrs ) } ) ) <EOL> self . assertIn ( '<STR_LIT>' , repr ( ds ) ) <EOL> attrs = { '<STR_LIT>' : '<STR_LIT>' } <EOL> ds = decode_cf ( Dataset ( { '<STR_LIT:time>' : ( '<STR_LIT:time>' , [ <NUM_LIT:0> , <NUM_LIT:1> ] , attrs ) } ) ) <EOL> self . assertIn ( '<STR_LIT>' , repr ( ds ) ) <EOL> class TestNativeEndiannessArray ( TestCase ) : <EOL> def test ( self ) : <EOL> x = np . arange ( <NUM_LIT:5> , dtype = '<STR_LIT>' ) <EOL> expected = np . arange ( <NUM_LIT:5> , dtype = '<STR_LIT>' ) <EOL> a = conventions . NativeEndiannessArray ( x ) <EOL> assert a . dtype == expected . dtype <EOL> assert a . dtype == expected [ : ] . dtype <EOL> self . assertArrayEqual ( a , expected ) <EOL> @ requires_netCDF4 <EOL> class TestEncodeCFVariable ( TestCase ) : <EOL> def test_incompatible_attributes ( self ) : <EOL> invalid_vars = [ <EOL> Variable ( [ '<STR_LIT:t>' ] , pd . date_range ( '<STR_LIT>' , periods = <NUM_LIT:3> ) , <EOL> { '<STR_LIT>' : '<STR_LIT>' } ) , <EOL> Variable ( [ '<STR_LIT:t>' ] , pd . to_timedelta ( [ '<STR_LIT>' ] ) , { '<STR_LIT>' : '<STR_LIT>' } ) , <EOL> Variable ( [ '<STR_LIT:t>' ] , [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> ] , { '<STR_LIT>' : <NUM_LIT:0> } , { '<STR_LIT>' : <NUM_LIT:2> } ) , <EOL> Variable ( [ '<STR_LIT:t>' ] , [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> ] , { '<STR_LIT>' : <NUM_LIT:0> } , { '<STR_LIT>' : <NUM_LIT:2> } ) , <EOL> ] <EOL> for var in invalid_vars : <EOL> with self . assertRaises ( ValueError ) : <EOL> conventions . encode_cf_variable ( var ) <EOL> def test_missing_fillvalue ( self ) : <EOL> v = Variable ( [ '<STR_LIT:x>' ] , np . array ( [ np . nan , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) ) <EOL> v . encoding = { '<STR_LIT>' : '<STR_LIT>' } <EOL> with self . assertWarns ( '<STR_LIT>' ) : <EOL> conventions . encode_cf_variable ( v ) <EOL> @ requires_netCDF4 <EOL> class TestDecodeCF ( TestCase ) : <EOL> def test_dataset ( self ) : <EOL> original = Dataset ( { <EOL> '<STR_LIT:t>' : ( '<STR_LIT:t>' , [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> ] , { '<STR_LIT>' : '<STR_LIT>' } ) , <EOL> '<STR_LIT:foo>' : ( '<STR_LIT:t>' , [ <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ] , { '<STR_LIT>' : '<STR_LIT:y>' , '<STR_LIT>' : '<STR_LIT:bar>' } ) , <EOL> '<STR_LIT:y>' : ( '<STR_LIT:t>' , [ <NUM_LIT:5> , <NUM_LIT:10> , - <NUM_LIT> ] , { '<STR_LIT>' : - <NUM_LIT> } ) <EOL> } ) <EOL> expected = Dataset ( { '<STR_LIT:foo>' : ( '<STR_LIT:t>' , [ <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ] , { '<STR_LIT>' : '<STR_LIT:bar>' } ) } , <EOL> { '<STR_LIT:t>' : pd . date_range ( '<STR_LIT>' , periods = <NUM_LIT:3> ) , <EOL> '<STR_LIT:y>' : ( '<STR_LIT:t>' , [ <NUM_LIT> , <NUM_LIT> , np . nan ] ) } ) <EOL> actual = conventions . decode_cf ( original ) <EOL> self . assertDatasetIdentical ( expected , actual ) <EOL> def test_invalid_coordinates ( self ) : <EOL> original = Dataset ( { '<STR_LIT:foo>' : ( '<STR_LIT:t>' , [ <NUM_LIT:1> , <NUM_LIT:2> ] , { '<STR_LIT>' : '<STR_LIT>' } ) } ) <EOL> actual = conventions . decode_cf ( original ) <EOL> self . assertDatasetIdentical ( original , actual ) <EOL> def test_decode_coordinates ( self ) : <EOL> original = Dataset ( { '<STR_LIT:foo>' : ( '<STR_LIT:t>' , [ <NUM_LIT:1> , <NUM_LIT:2> ] , { '<STR_LIT>' : '<STR_LIT:x>' } ) , <EOL> '<STR_LIT:x>' : ( '<STR_LIT:t>' , [ <NUM_LIT:4> , <NUM_LIT:5> ] ) } ) <EOL> actual = conventions . decode_cf ( original ) <EOL> self . assertEqual ( actual . foo . encoding [ '<STR_LIT>' ] , '<STR_LIT:x>' ) <EOL> def test_0d_int32_encoding ( self ) : <EOL> original = Variable ( ( ) , np . int32 ( <NUM_LIT:0> ) , encoding = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> expected = Variable ( ( ) , np . int64 ( <NUM_LIT:0> ) ) <EOL> actual = conventions . maybe_encode_dtype ( original ) <EOL> self . assertDatasetIdentical ( expected , actual ) <EOL> def test_decode_cf_with_multiple_missing_values ( self ) : <EOL> original = Variable ( [ '<STR_LIT:t>' ] , [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> ] , <EOL> { '<STR_LIT>' : np . array ( [ <NUM_LIT:0> , <NUM_LIT:1> ] ) } ) <EOL> expected = Variable ( [ '<STR_LIT:t>' ] , [ np . nan , np . nan , <NUM_LIT:2> ] , { } ) <EOL> with warnings . catch_warnings ( record = True ) as w : <EOL> actual = conventions . decode_cf_variable ( original ) <EOL> self . assertDatasetIdentical ( expected , actual ) <EOL> self . assertIn ( '<STR_LIT>' , str ( w [ <NUM_LIT:0> ] . message ) ) <EOL> def test_decode_cf_with_drop_variables ( self ) : <EOL> original = Dataset ( { <EOL> '<STR_LIT:t>' : ( '<STR_LIT:t>' , [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> ] , { '<STR_LIT>' : '<STR_LIT>' } ) , <EOL> '<STR_LIT:x>' : ( "<STR_LIT:x>" , [ <NUM_LIT:9> , <NUM_LIT:8> , <NUM_LIT:7> ] , { '<STR_LIT>' : '<STR_LIT>' } ) , <EOL> '<STR_LIT:foo>' : ( ( '<STR_LIT:t>' , '<STR_LIT:x>' ) , [ [ <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ] , [ <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> ] , [ <NUM_LIT:2> , <NUM_LIT:2> , <NUM_LIT:2> ] ] , { '<STR_LIT>' : '<STR_LIT:bar>' } ) , <EOL> '<STR_LIT:y>' : ( '<STR_LIT:t>' , [ <NUM_LIT:5> , <NUM_LIT:10> , - <NUM_LIT> ] , { '<STR_LIT>' : - <NUM_LIT> } ) <EOL> } ) <EOL> expected = Dataset ( { <EOL> '<STR_LIT:t>' : pd . date_range ( '<STR_LIT>' , periods = <NUM_LIT:3> ) , <EOL> '<STR_LIT:x>' : ( "<STR_LIT:x>" , [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> ] ) , <EOL> '<STR_LIT:foo>' : ( ( '<STR_LIT:t>' , '<STR_LIT:x>' ) , [ [ <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ] , [ <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> ] , [ <NUM_LIT:2> , <NUM_LIT:2> , <NUM_LIT:2> ] ] , { '<STR_LIT>' : '<STR_LIT:bar>' } ) , <EOL> '<STR_LIT:y>' : ( '<STR_LIT:t>' , [ <NUM_LIT:5> , <NUM_LIT:10> , np . nan ] ) <EOL> } ) <EOL> actual = conventions . decode_cf ( original , drop_variables = ( "<STR_LIT:x>" , ) ) <EOL> actual2 = conventions . decode_cf ( original , drop_variables = "<STR_LIT:x>" ) <EOL> self . assertDatasetIdentical ( expected , actual ) <EOL> self . assertDatasetIdentical ( expected , actual2 ) <EOL> class CFEncodedInMemoryStore ( WritableCFDataStore , InMemoryDataStore ) : <EOL> pass <EOL> class NullWrapper ( utils . NDArrayMixin ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , array ) : <EOL> self . array = array <EOL> def __getitem__ ( self , key ) : <EOL> return self . array [ indexing . orthogonal_indexer ( key , self . shape ) ] <EOL> def null_wrap ( ds ) : <EOL> """<STR_LIT>""" <EOL> variables = dict ( ( k , Variable ( v . dims , NullWrapper ( v . values ) , v . attrs ) ) <EOL> for k , v in iteritems ( ds ) ) <EOL> return InMemoryDataStore ( variables = variables , attributes = ds . attrs ) <EOL> @ requires_netCDF4 <EOL> class TestCFEncodedDataStore ( CFEncodedDataTest , TestCase ) : <EOL> @ contextlib . contextmanager <EOL> def create_store ( self ) : <EOL> yield CFEncodedInMemoryStore ( ) <EOL> @ contextlib . contextmanager <EOL> def roundtrip ( self , data , save_kwargs = { } , open_kwargs = { } ) : <EOL> store = CFEncodedInMemoryStore ( ) <EOL> data . dump_to_store ( store , ** save_kwargs ) <EOL> yield open_dataset ( store , ** open_kwargs ) <EOL> def test_roundtrip_coordinates ( self ) : <EOL> raise unittest . SkipTest ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_invalid_dataarray_names_raise ( self ) : <EOL> pass <EOL> def test_encoding_kwarg ( self ) : <EOL> pass </s>
