import cf import numpy import os import itertools import time from operator import mul import unittest import inspect def reshape_array(a, axes): new_order = [i for i in range(a.ndim) if i not in axes] new_order.extend(axes) b = numpy.transpose(a, new_order) new_shape = b.shape[:b.ndim-len(axes)] new_shape += (reduce(mul, b.shape[b.ndim-len(axes):]),) b = b.reshape(new_shape) return b #--- End: def class DataTest(unittest.TestCase): print '\nRunning tests on cf.Data' print 'cf version', cf.__version__, 'running from', os.path.abspath(cf.__file__), '\n' filename = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'test_file.nc') chunk_sizes = (17, 34, 300, 100000)[::-1] do_test = () # do_test = ('test_Data__collapse_UNWEIGHTED_MASKED',) # do_test = ('test_Data__collapse_UNWEIGHTED_UNMASKED',) # do_test = ('test_Data__collapse_WEIGHTED_UNMASKED',) # do_test = ('test_Data__collapse_WEIGHTED_MASKED',) # Variables for _collapse a = numpy.arange(-100, 200., dtype=float).reshape(3, 4, 5, 5) ma = numpy.ma.arange(-100, 200., dtype=float).reshape(3, 4, 5, 5) ma[0, :, 2, 3] = numpy.ma.masked ma[1, 2, 3, :] = numpy.ma.masked ma[0, 3, :, 3] = numpy.ma.masked ma[:, 1, 4, 4] = numpy.ma.masked w = numpy.arange(1, 301., dtype=float).reshape(a.shape) w[-1, -1, ...] = w[-1, -1, ...]*2 w /= w.min() axes_combinations = [axes for n in range(1, a.ndim+1) for axes in itertools.permutations(range(a.ndim), n)] def test_Data___contains__(self): '''cf.Data.__contains__''' if self.do_test and inspect.stack()[0][3] not in self.do_test: return print for chunksize in self.chunk_sizes: cf.CHUNKSIZE(chunksize) d = cf.Data([[0.0, 1, 2], [3, 4, 5]], units='m') self.assertTrue(4 in d) self.assertFalse(40 in d) self.assertTrue(cf.Data(3) in d) self.assertTrue(cf.Data([[[[3]]]]) in d) value = d[1, 2] value.Units *= 2 value.squeeze(0) self.assertTrue(value in d) self.assertTrue(numpy.array([[[2]]]) in d) print "cf.Data.__contains__ passed", "pmshape =", d._pmshape #--- End: def def test_Data___setitem__(self): '''cf.Data.__setitem__''' if self.do_test and inspect.stack()[0][3] not in self.do_test: return print for chunksize in self.chunk_sizes: cf.CHUNKSIZE(chunksize) for hardmask in (False, True): a = numpy.ma.arange(3000).reshape(50, 60) if hardmask: a.harden_mask() else: a.soften_mask() d = cf.Data(a.filled(), 'm') d.hardmask = hardmask for n, (j, i) in enumerate(((34, 23), (0, 0), (-1, -1), (slice(40, 50), slice(58, 60)), (Ellipsis, Ellipsis), )): n = -n-1 for dvalue, avalue in ((n, n), (cf.masked, numpy.ma.masked), (n, n)): message = "cf.Data[%s, %s]=%s failed" % (j, i, dvalue) d[j, i] = dvalue a[j, i] = avalue self.assertTrue((d.array == a).all() in (True, numpy.ma.masked), message) self.assertTrue((d.mask.array == numpy.ma.getmaskarray(a)).all(), message) #--- End: for a = numpy.ma.arange(3000).reshape(50, 60) if hardmask: a.harden_mask() else: a.soften_mask() d = cf.Data(a.filled(), 'm') d.hardmask = hardmask (j, i) = (slice(0, 2), slice(0, 3)) array = numpy.array([[1, 2, 6],[3, 4, 5]])*-1 for dvalue in (array, numpy.ma.masked_where(array < -2, array), array): message = "cf.Data[%s, %s]=%s failed" % (j, i, dvalue) d[j, i] = dvalue a[j, i] = dvalue self.assertTrue((d.array == a).all() in (True, numpy.ma.masked), message) self.assertTrue((d.mask.array == numpy.ma.getmaskarray(a)).all(), message) #--- End: for print 'cf.Data.__setitem__ passed: hardmask =',hardmask,', pmshape =', d._pmshape #--- End: for #--- End: def def test_Data_all(self): '''cf.Data.all''' if self.do_test and inspect.stack()[0][3] not in self.do_test: return print for chunksize in self.chunk_sizes: cf.CHUNKSIZE(chunksize) d = cf.Data(numpy.array([[0] * 1000])) self.assertTrue(not d.all()) d[-1,-1] = 1 self.assertFalse(d.all()) d[...] = 1 self.assertTrue(d.all()) d[...] = cf.masked self.assertTrue(d.all()) print 'cf.Data.all passed', "pmshape =", d._pmshape #--- End: def def test_Data_any(self): '''cf.Data.any''' if self.do_test and inspect.stack()[0][3] not in self.do_test: return print for chunksize in self.chunk_sizes: cf.CHUNKSIZE(chunksize) d = cf.Data(numpy.array([[0] * 1000])) self.assertFalse(d.any()) d[-1,-1] = 1 self.assertTrue(d.any()) d[...] = 1 self.assertTrue(d.any()) d[...] = cf.masked self.assertFalse(d.any()) print 'cf.Data.any passed', "pmshape =", d._pmshape #--- End: def def test_Data_array(self): '''cf.Data.array''' if self.do_test and inspect.stack()[0][3] not in self.do_test: return print for chunksize in self.chunk_sizes: cf.CHUNKSIZE(chunksize) d = cf.Data(numpy.arange(10*15*19).reshape(10, 1, 15, 19), 'km') a = d.array a[0,0,0,0] = -999 self.assertTrue(d.array[0,0,0,0] == 0) print "cf.Data.array passed", "pmshape =", d._pmshape #--- End: def def test_Data_asdatetime_asreftime(self): '''cf.Data.asdatetime, cf.Data.asdatetime''' if self.do_test and inspect.stack()[0][3] not in self.do_test: return print for chunksize in self.chunk_sizes: cf.CHUNKSIZE(chunksize) d = cf.Data([[1.93, 5.17]], 'days since 2000-12-29') self.assertTrue(d.dtype == numpy.dtype(float)) self.assertFalse(d._isdt) d.asreftime(i=True) self.assertTrue(d.dtype == numpy.dtype(float)) self.assertFalse(d._isdt) d.asdatetime(i=True) self.assertTrue(d.dtype == numpy.dtype(object)) self.assertTrue(d._isdt) d.asdatetime(i=True) self.assertTrue(d.dtype == numpy.dtype(object)) self.assertTrue(d._isdt) d.asreftime(i=True) self.assertTrue(d.dtype == numpy.dtype(float)) self.assertFalse(d._isdt) print "cf.Data.asdatetime passed", "pmshape =", d._pmshape print "cf.Data.asreftime passed", "pmshape =", d._pmshape #--- End: def def test_Data_datum(self): '''cf.Data.datum''' if self.do_test and inspect.stack()[0][3] not in self.do_test: return print for chunksize in self.chunk_sizes: cf.CHUNKSIZE(chunksize) d = cf.Data(5, 'metre') self.assertTrue(d.datum() == 5) self.assertTrue(d.datum(0) == 5) self.assertTrue(d.datum(-1) == 5) for d in [cf.Data([4, 5, 6, 1, 2, 3], 'metre'), cf.Data([[4, 5, 6], [1, 2, 3]], 'metre')]: self.assertTrue(d.datum(0) == 4) self.assertTrue(d.datum(-1) == 3) for index in d.ndindex(): self.assertTrue(d.datum(index) == d.array[index].item()) self.assertTrue(d.datum(*index) == d.array[index].item()) #--- End: for d = cf.Data(5, 'metre') d[()] = cf.masked self.assertTrue(d.datum() is cf.masked) self.assertTrue(d.datum(0) is cf.masked) self.assertTrue(d.datum(-1) is cf.masked) d = cf.Data([[5]], 'metre') d[0, 0] = cf.masked self.assertTrue(d.datum() is cf.masked) self.assertTrue(d.datum(0) is cf.masked) self.assertTrue(d.datum(-1) is cf.masked) self.assertTrue(d.datum(0, 0) is cf.masked) self.assertTrue(d.datum(-1, 0) is cf.masked) self.assertTrue(d.datum([0, 0]) is cf.masked) self.assertTrue(d.datum([0, -1]) is cf.masked) self.assertTrue(d.datum(-1, -1) is cf.masked) print "cf.Data.datum passed", "pmshape =", d._pmshape #--- End: def def test_Data_flip(self): '''cf.Data.flip''' if self.do_test and inspect.stack()[0][3] not in self.do_test: return print for chunksize in self.chunk_sizes: cf.CHUNKSIZE(chunksize) array = numpy.arange(24000).reshape(120, 200) d = cf.Data(array.copy(), 'metre') for axes, indices in zip( (0, 1, [0, 1]), ((slice(None, None, -1), slice(None)), (slice(None) , slice(None, None, -1)), (slice(None, None, -1), slice(None, None, -1))) ): array = array[indices] d.flip(axes, i=True) #--- End: for self.assertTrue((d.array == array).all()) print "cf.Data.flip passed", "pmshape =", d._pmshape #--- End: def def test_Data_max(self): '''cf.Data.max''' if self.do_test and inspect.stack()[0][3] not in self.do_test: return print for chunksize in self.chunk_sizes: cf.CHUNKSIZE(chunksize) d = cf.Data([[4, 5, 6], [1, 2, 3]], 'metre') self.assertTrue(d.max() == cf.Data(6, 'metre')) self.assertTrue(d.max().datum() == 6) d[0, 2] = cf.masked self.assertTrue(d.max() == 5) self.assertTrue(d.max().datum() == 5) self.assertTrue(d.max() == cf.Data(0.005, 'km')) print "cf.Data.max passed", "pmshape =", d._pmshape #--- End: def def test_Data_min(self): '''cf.Data.min''' if self.do_test and inspect.stack()[0][3] not in self.do_test: return print for chunksize in self.chunk_sizes: cf.CHUNKSIZE(chunksize) d = cf.Data([[4, 5, 6], [1, 2, 3]], 'metre') self.assertTrue(d.min() == cf.Data(1, 'metre')) self.assertTrue(d.min().datum() == 1) d[1, 0] = cf.masked self.assertTrue(d.min() == 2) self.assertTrue(d.min().datum() == 2) self.assertTrue(d.min() == cf.Data(0.002, 'km')) print "cf.Data.min passed", "pmshape =", d._pmshape #--- End: def def test_Data_ndindex(self): '''cf.Data.ndindex''' if self.do_test and inspect.stack()[0][3] not in self.do_test: return print for chunksize in self.chunk_sizes: cf.CHUNKSIZE(chunksize) for d in (cf.Data(5, 'metre'), cf.Data([4, 5, 6, 1, 2, 3], 'metre'), cf.Data([[4, 5, 6], [1, 2, 3]], 'metre') ): for i, j in zip(d.ndindex(), numpy.ndindex(d.shape)): self.assertTrue(i == j) #--- End: for print "cf.Data.ndindex passed", "pmshape =", d._pmshape #--- End: def def test_Data_roll(self): '''cf.Data.roll''' if self.do_test and inspect.stack()[0][3] not in self.do_test: return print for chunksize in self.chunk_sizes: cf.CHUNKSIZE(chunksize) a = numpy.arange(10*15*19).reshape(10, 1, 15, 19) d = cf.Data(a) pmshape = d._pmshape e = d.roll(0, 4) e.roll(2, 120, i=True) e.roll(3, -77, i=True) a = numpy.roll(a, 4, 0) a = numpy.roll(a, 120, 2) a = numpy.roll(a, -77, 3) self.assertTrue(e.shape == a.shape) self.assertTrue((a == e.array).all()) f = e.roll(3, 77) f.roll(2, -120, i=True) f.roll(0, -4, i=True) self.assertTrue(f.shape == d.shape) self.assertTrue(f.equals(d)) print "cf.Data.roll passed", "pmshape =", pmshape #--- End: def def test_Data_sample_size(self): '''cf.Data.sample_size''' if self.do_test and inspect.stack()[0][3] not in self.do_test: return print for chunksize in self.chunk_sizes: cf.CHUNKSIZE(chunksize) d = cf.Data([[4, 5, 6], [1, 2, 3]], 'metre') self.assertTrue(d.sample_size() == 6) d[1, 0] = cf.masked self.assertTrue(d.sample_size() == cf.Data(50, '0.1')) print "cf.Data.sample_size passed", "pmshape =", d._pmshape #--- End: def def test_Data_swapaxes(self): '''cf.cf.Data.swapaxes''' if self.do_test and inspect.stack()[0][3] not in self.do_test: return print for chunksize in self.chunk_sizes: cf.CHUNKSIZE(chunksize) a = numpy.arange(10*15*19).reshape(10, 1, 15, 19) d = cf.Data(a.copy()) for i in range(-a.ndim, a.ndim): for j in range(-a.ndim, a.ndim): b = numpy.swapaxes(a.copy(), i, j) e = d.swapaxes(i, j) message = "cf.Data.swapaxes(%d, %d) failed" % (i, j) self.assertTrue(b.shape == e.shape, message) self.assertTrue((b == e.array).all(), message) #--- End: for #--- End: for print "cf.Data.swapaxes passed", "pmshape =", d._pmshape #--- End: def def test_Data_transpose(self): '''cf.cf.Data.transpose''' if self.do_test and inspect.stack()[0][3] not in self.do_test: return print for chunksize in self.chunk_sizes: cf.CHUNKSIZE(chunksize) a = numpy.arange(10*15*19).reshape(10, 1, 15, 19) d = cf.Data(a.copy()) for indices in (range(a.ndim), range(-a.ndim, 0)): for axes in itertools.permutations(indices): a = numpy.transpose(a, axes) d.transpose(axes, i=True) message = 'cf.Data.transpose(%s) failed: d.shape=%s, a.shape=%s' % \ (axes, d.shape, a.shape) self.assertTrue(d.shape == a.shape, message) self.assertTrue((d.array == a).all(), message) #--- End: for #--- End: for print "cf.Data.transpose passed", "pmshape =", d._pmshape #--- End: def def test_Data_unique(self): '''cf.Data.unique''' if self.do_test and inspect.stack()[0][3] not in self.do_test: return print for chunksize in self.chunk_sizes: cf.CHUNKSIZE(chunksize) d = cf.Data([[4, 2, 1], [1, 2, 3]], 'metre') self.assertTrue((d.unique == cf.Data([1, 2, 3, 4], 'metre')).all()) d[1, -1] = cf.masked self.assertTrue((d.unique == cf.Data([1, 2, 4], 'metre')).all()) print "cf.Data.unique passed", "pmshape =", d._pmshape #--- End: def def test_Data_varray(self): '''cf.Data.varray''' if self.do_test and inspect.stack()[0][3] not in self.do_test: return print for chunksize in self.chunk_sizes: cf.CHUNKSIZE(chunksize) d = cf.Data(numpy.arange(10*15*19).reshape(10, 1, 15, 19), 'km') e = d.copy() v = e.varray v[0,0,0,0] = -999 self.assertTrue(e.array[0,0,0,0] == -999) print "cf.Data.varray passed", "pmshape =", d._pmshape #--- End: def def test_Data_year_month_day_hour_minute_second(self): '''cf.Data.year, cf.Data.month, cf.Data.day, cf.Data.hour, cf.Data.minute, cf.Data.second''' if self.do_test and inspect.stack()[0][3] not in self.do_test: return print d = cf.Data([[1.93, 5.17]], 'days since 2000-12-29') self.assertTrue(d.year.equals(cf.Data([[2000, 2001]]))) self.assertTrue(d.month.equals(cf.Data([[12, 1]]))) self.assertTrue(d.day.equals(cf.Data([[30, 3]]))) self.assertTrue(d.hour.equals(cf.Data([[22, 4]]))) self.assertTrue(d.minute.equals(cf.Data([[19, 4]]))) self.assertTrue(d.second.equals(cf.Data([[12, 48]]))) d = cf.Data([[1.93, 5.17]], cf.Units('days since 2000-12-29', '360_day')) self.assertTrue(d.year.equals(cf.Data([[2000, 2001]]))) self.assertTrue(d.month.equals(cf.Data([[12, 1]]))) self.assertTrue(d.day.equals(cf.Data([[30, 4]]))) self.assertTrue(d.hour.equals(cf.Data([[22, 4]]))) self.assertTrue(d.minute.equals(cf.Data([[19, 4]]))) self.assertTrue(d.second.equals(cf.Data([[12, 48]]))) print 'cf.Data.year passed' , "pmshape =", d._pmshape print 'cf.Data.month passed' , "pmshape =", d._pmshape print 'cf.Data.day passed' , "pmshape =", d._pmshape print 'cf.Data.hour passed' , "pmshape =", d._pmshape print 'cf.Data.minute passed', "pmshape =", d._pmshape print 'cf.Data.second passed', "pmshape =", d._pmshape #--- End: def def test_Data_BINARY_AND_UNARY_OPERATORS(self): '''BINARY AND UNARY OPERATORS''' if self.do_test and inspect.stack()[0][3] not in self.do_test: return print for chunksize in self.chunk_sizes: cf.CHUNKSIZE(chunksize) array=numpy.arange(3*4*5).reshape(3, 4, 5) + 1 arrays = (numpy.arange(3*4*5).reshape(3, 4, 5) + 1.0, numpy.arange(3*4*5).reshape(3, 4, 5) + 1) for a0 in arrays: for a1 in arrays[::-1]: d = cf.Data(a0[(slice(None, None, -1),)*a0.ndim], 'metre') d.flip(i=True) x = cf.Data(a1, 'metre') self.assertTrue((d + x).equals(cf.Data(a0 + a1, 'm' ), traceback=1)), '%s+%s' % (repr(d), x) self.assertTrue((d * x).equals(cf.Data(a0 * a1, 'm2'), traceback=1)), '%s*%s' % (repr(d), x) self.assertTrue((d / x).equals(cf.Data(a0 / a1, '1' ), traceback=1)), '%s/%s' % (repr(d), x) self.assertTrue((d - x).equals(cf.Data(a0 - a1, 'm' ), traceback=1)), '%s-%s' % (repr(d), x) self.assertTrue((d // x).equals(cf.Data(a0 // a1, '1' ), traceback=1)), '%s//%s' % (repr(d), x) self.assertTrue(d.__truediv__(x).equals(cf.Data(array.__truediv__(array), '1'), traceback=1)), '%s.__truediv__(%s)' % (d, x) self.assertTrue(d.__rtruediv__(x).equals(cf.Data(array.__rtruediv__(array), '1'), traceback=1)) , '%s.__rtruediv__(%s)' % (d, x) try: d ** x except: pass else: self.assertTrue((d**x).all(), '%s**%s' % (d, repr(x))) print '%s {+, -, *, /, //, **, __truediv__, __rtruediv__} %s passed' % (d, x), \ "pmshape =", d._pmshape #--- End: for #--- End: for for a0 in arrays: d = cf.Data(a0, 'metre') for x in (2, 2.0): self.assertTrue((d + x).equals(cf.Data(a0 + x, 'm' ), traceback=1)), '%s+%s' % (repr(d), x) self.assertTrue((d * x).equals(cf.Data(a0 * x, 'm' ), traceback=1)), '%s*%s' % (repr(d), x) self.assertTrue((d / x).equals(cf.Data(a0 / x, 'm' ), traceback=1)), '%s/%s' % (repr(d), x) self.assertTrue((d - x).equals(cf.Data(a0 - x, 'm' ), traceback=1)), '%s-%s' % (repr(d), x) self.assertTrue((d // x).equals(cf.Data(a0 // x, 'm' ), traceback=1)), '%s//%s' % (repr(d), x) self.assertTrue((d ** x).equals(cf.Data(a0 ** x, 'm2'), traceback=1)), '%s**%s' % (repr(d), x) self.assertTrue(d.__truediv__(x).equals(cf.Data(a0.__truediv__(x), 'm'), traceback=1)), '%s.__truediv__(%s)' % (d, x) self.assertTrue(d.__rtruediv__(x).equals(cf.Data(a0.__rtruediv__(x), 'm-1'), traceback=1)) , '%s.__rtruediv__(%s)' % (d, x) print '%s {+, -, *, /, //, **, __truediv__, __rtruediv__} %s passed' % (d, x), \ "pmshape =", d._pmshape self.assertTrue((x + d).equals(cf.Data(x + a0, 'm' ), traceback=1)), '%s+%s' % (x, repr(d)) self.assertTrue((x * d).equals(cf.Data(x * a0, 'm' ), traceback=1)), '%s*%s' % (x, repr(d)) self.assertTrue((x / d).equals(cf.Data(x / a0, 'm-1'), traceback=1)), '%s/%s' % (x, repr(d)) self.assertTrue((x - d).equals(cf.Data(x - a0, 'm' ), traceback=1)), '%s-%s' % (x, repr(d)) self.assertTrue((x // d).equals(cf.Data(x // a0, 'm-1'), traceback=1)), '%s//%s' % (x, repr(d)) try: x ** d except: pass else: self.assertTrue((x**d).all(), '%s**%s' % (x, repr(d))) print '%s {+, -, *, /, //, **} %s passed' % (x, repr(d)), \ "pmshape =", d._pmshape e = d.copy() a = a0.copy() e += x a += x self.assertTrue(e.equals(cf.Data(a, 'm'), traceback=1)), '%s+=%s' % (repr(d), x) e = d.copy() a = a0.copy() e *= x a *= x self.assertTrue(e.equals(cf.Data(a, 'm'), traceback=1)), '%s*=%s' % (repr(d), x) e = d.copy() a = a0.copy() e /= x a /= x self.assertTrue(e.equals(cf.Data(a, 'm'), traceback=1)), '%s/=%s' % (repr(d), x) e = d.copy() a = a0.copy() e -= x a -= x self.assertTrue(e.equals(cf.Data(a, 'm'), traceback=1)), '%s-=%s' % (repr(d), x) e = d.copy() a = a0.copy() e //= x a //= x self.assertTrue(e.equals(cf.Data(a, 'm'), traceback=1)), '%s//=%s' % (repr(d), x) e = d.copy() a = a0.copy() e **= x a **= x self.assertTrue(e.equals(cf.Data(a, 'm2'), traceback=1)), '%s**=%s' % (repr(d), x) e = d.copy() a = a0.copy() e.__itruediv__(x) a.__itruediv__(x) self.assertTrue(e.equals(cf.Data(a, 'm'), traceback=1)), '%s.__itruediv__(%s)' % (d, x) print '%s {+=, -=, *=, /=, //=, **=, __itruediv__} %s passed' % (repr(d), x), \ "pmshape =", d._pmshape #--- End: for for x in (cf.Data(2, 'metre'), cf.Data(2.0, 'metre')): self.assertTrue((d + x).equals(cf.Data(a0 + x.datum(), 'm' ), traceback=1)) self.assertTrue((d * x).equals(cf.Data(a0 * x.datum(), 'm2'), traceback=1)) self.assertTrue((d / x).equals(cf.Data(a0 / x.datum(), '1' ), traceback=1)) self.assertTrue((d - x).equals(cf.Data(a0 - x.datum(), 'm' ), traceback=1)) self.assertTrue((d // x).equals(cf.Data(a0 // x.datum(), '1' ), traceback=1)) try: d ** x except: pass else: self.assertTrue((x**d).all(), '%s**%s' % (x, repr(d))) self.assertTrue(d.__truediv__(x).equals(cf.Data(a0.__truediv__(x.datum()), ''), traceback=1)) print '%s {+, -, *, /, //, **, __truediv__} %s passed' % (repr(d), repr(x)), \ "pmshape =", d._pmshape #--- End: for #--- End: for #--- End: def def test_Data_BROADCASTING(self): '''BROADCASTING''' if self.do_test and inspect.stack()[0][3] not in self.do_test: return print A = [numpy.array(3), numpy.array( [3]), numpy.array( [3]).reshape(1, 1), numpy.array( [3]).reshape(1, 1, 1), numpy.arange( 5).reshape(5, 1), numpy.arange( 5).reshape(1, 5), numpy.arange( 5).reshape(1, 5, 1), numpy.arange( 5).reshape(5, 1, 1), numpy.arange( 5).reshape(1, 1, 5), numpy.arange( 25).reshape(1, 5, 5), numpy.arange( 25).reshape(5, 1, 5), numpy.arange( 25).reshape(5, 5, 1), numpy.arange(125).reshape(5, 5, 5), ] for chunksize in self.chunk_sizes: cf.CHUNKSIZE(chunksize) for a in A: for b in A: d = cf.Data(a) e = cf.Data(b) ab = a*b de = d*e self.assertTrue(de.shape == ab.shape) self.assertTrue((de.array == ab).all()) #--- End: for #--- End: for print "cf.Data BROADCASTING passed" #--- End: for #--- End: def def test_Data__collapse_SHAPE(self): '''cf.Data._collapse SHAPE''' if self.do_test and inspect.stack()[0][3] not in self.do_test: return print a = numpy.arange(-100, 200., dtype=float).reshape(3, 4, 5, 5) ones = numpy.ones(a.shape, dtype=float) for chunksize in self.chunk_sizes: cf.CHUNKSIZE(chunksize) d = cf.Data(a[(slice(None, None, -1),) * a.ndim].copy()) d.flip(i=True) x = cf.Data(self.w.copy()) shape = list(d.shape) for h in ('sample_size', 'sum', 'min', 'max', 'mean', 'var', 'sd', 'mid_range', 'range'): for axes in self.axes_combinations: e = getattr(d, h)(axes=axes, squeeze=False) shape = list(d.shape) for i in axes: shape[i] = 1 shape = tuple(shape) self.assertTrue( e.shape == shape, "%s, axes=%s, not squeezed bad shape: %s != %s" % \ (h, axes, e.shape, shape)) #--- End: for for axes in self.axes_combinations: e = getattr(d, h)(axes=axes, squeeze=True) shape = list(d.shape) for i in sorted(axes, reverse=True): shape.pop(i) shape = tuple(shape) assert e.shape == shape, \ "%s, axes=%s, squeezed bad shape: %s != %s" % \ (h, axis, e.shape, shape) #--- End: for e = getattr(d, h)(squeeze=True) shape = () self.assertTrue( e.shape == shape, "%s, axes=%s, squeezed bad shape: %s != %s" % \ (h, None, e.shape, shape)) e = getattr(d, h)(squeeze=False) shape = (1,) * d.ndim self.assertTrue( e.shape == shape, "%s, axes=%s, not squeezed bad shape: %s != %s" % \ (h, None, e.shape, shape)) print 'cf.Data.%s SHAPE passed' % h, "pmshape =", d._pmshape #--- End: for #--- End: def def test_Data__collapse_UNWEIGHTED_MASKED(self): '''cf.Data._collapse UNWEIGHTED MASKED''' if self.do_test and inspect.stack()[0][3] not in self.do_test: return print a = numpy.ma.arange(-100, 200., dtype=float).reshape(3, 4, 5, 5) a[0, :, 2, 3] = numpy.ma.masked a[1, 2, 3, :] = numpy.ma.masked a[0, 3, :, 3] = numpy.ma.masked a[:, 1, 4, 4] = numpy.ma.masked ones = numpy.ones(a.shape, dtype=float) ones = numpy.ma.array(ones) for chunksize in self.chunk_sizes: cf.CHUNKSIZE(chunksize) d = cf.Data(self.ma.copy()) for c, h in zip((ones , ones), ('sum_of_weights', 'sum_of_weights2')): for axes in self.axes_combinations: c = c.copy() c.mask = self.ma.mask b = reshape_array(c, axes) b = numpy.ma.asanyarray(b.sum(axis=-1)) e = getattr(d, h)(axes=axes, squeeze=True) self.assertTrue( (e.mask.array == b.mask).all(), "%s, axis=%s, \ne.mask=%s, \nb.mask=%s, \ne.mask==b.mask=%s" % (h, axes, e.mask.array, b.mask, e.mask.array==b.mask)) self.assertTrue( e.allclose(b, rtol=1e-05, atol=1e-08), "%s, axis=%s, \ne=%s, \nb=%s, \ne-b=%s" % (h, axes, e.array, b, e.array-b)) #--- End: for print 'cf.Data.%s UNWEIGHTED, MASKED passed' % h, \ "pmshape =", d._pmshape #--- End: for for np, h in zip((numpy.ma.sum, numpy.ma.amin, numpy.ma.amax), ( 'sum', 'min', 'max')): for axes in self.axes_combinations: b = reshape_array(self.ma, axes) b = np(b, axis=-1) b = numpy.ma.asanyarray(b) e = getattr(d, h)(axes=axes, squeeze=True) self.assertTrue( (e.mask.array == b.mask).all(), "%s, axis=%s, \ne.mask=%s, \nb.mask=%s, \ne.mask==b.mask=%s" % (h, axes, e.mask.array, b.mask, e.mask.array==b.mask)) self.assertTrue( e.allclose(b, rtol=1e-05, atol=1e-08), "%s, axis=%s, unweighted, unmasked \ne=%s, \nb=%s, \ne-b=%s" % (h, axes, e.array, b, e.array-b)) #--- End: for print 'cf.Data.%s UNWEIGHTED, MASKED passed' % \ h, "pmshape =", d._pmshape #--- End: for h = 'mean' for axes in self.axes_combinations: b = reshape_array(self.ma, axes) b = numpy.ma.average(b, axis=-1) b = numpy.ma.asanyarray(b) e = getattr(d, h)(axes=axes, squeeze=True) self.assertTrue( (e.mask.array == b.mask).all(), "%s, axis=%s, \ne.mask=%s, \nb.mask=%s, \ne.mask==b.mask=%s" % (h, axes, e.mask.array, b.mask, e.mask.array==b.mask)) self.assertTrue( e.allclose(b, rtol=1e-05, atol=1e-08), "%s, axis=%s, \ne=%s, \nb=%s, \ne-b=%s" % (h, axes, e.array, b, e.array-b)) #--- End: for print 'cf.Data.%s UNWEIGHTED, MASKED passed' % h, \ "pmshape =", d._pmshape ddofs = (0, 1, 2) f = 2 for h in ('var', 'sd'): for axes in self.axes_combinations: for ddof in ddofs: e = getattr(d, h)(axes=axes, squeeze=True, ddof=ddof) b = reshape_array(a, axes) not_enough_data = numpy.ma.count(b, axis=-1) <= ddof avg = numpy.ma.average(b, axis=-1) if numpy.ndim(avg) < b.ndim: avg = numpy.ma.expand_dims(avg, -1) b, sv = numpy.ma.average((b-avg)**2, axis=-1, returned=True) b = numpy.ma.where(not_enough_data, numpy.ma.masked, b) b *= sv/(sv-ddof) if h == 'sd': b **= 0.5 b = numpy.ma.asanyarray(b) e = getattr(d, h)(axes=axes, squeeze=True, ddof=ddof) self.assertTrue( (e.mask.array == b.mask).all(), "%s, axes=%s, ddof=%s, \ne.mask=%s, \nb.mask=%s, \ne.mask==b.mask=%s" % (h, axes, ddof, e.mask.array, b.mask, e.mask.array==b.mask)) self.assertTrue( e.allclose(b, rtol=1e-05, atol=1e-08), "%s, axes=%s, ddof=%s, \ne=%s, \nb=%s, \ne-b=%s" % (h, axes, ddof, e.array, b, e.array-b)) #--- End: for #--- End: for print 'cf.Data.%s UNWEIGHTED, MASKED passed' % h, \ "pmshape =", d._pmshape #--- End: for #--- End: def def test_Data__collapse_UNWEIGHTED_UNMASKED(self): '''cf.Data._collapse UNWEIGHTED UNMASKED''' if self.do_test and inspect.stack()[0][3] not in self.do_test: return print a = numpy.arange(-100, 200., dtype=float).reshape(3, 4, 5, 5) ones = numpy.ones(a.shape, dtype=float) for chunksize in self.chunk_sizes: cf.CHUNKSIZE(chunksize) d = cf.Data(a[(slice(None, None, -1),) * a.ndim].copy()) d.flip(i=True) for h in ('sample_size', 'sum_of_weights', 'sum_of_weights2'): for axes in self.axes_combinations: b = reshape_array(ones, axes) b = b.sum(axis=-1) e = getattr(d, h)(axes=axes, squeeze=True) self.assertTrue( e.allclose(b, rtol=1e-05, atol=1e-08), "%s, axis=%s, \ne=%s, \nb=%s, \ne-b=%s" % \ (h, axes, e.array, b, e.array-b)) #--- End: for print 'cf.Data.%s UNWEIGHTED, UNMASKED passed' % h, \ "pmshape =", d._pmshape #--- End: for for np, h in zip((numpy.sum, numpy.amin, numpy.amax, numpy.mean), ( 'sum', 'min', 'max', 'mean')): for axes in self.axes_combinations: b = reshape_array(a, axes) b = np(b, axis=-1) e = getattr(d, h)(axes=axes, squeeze=True) self.assertTrue( e.allclose(b, rtol=1e-05, atol=1e-08), "%s, axis=%s, unweighted, unmasked \ne=%s, \nb=%s, \ne-b=%s" % (h, axes, e.array, b, e.array-b)) #--- End: for print 'cf.Data.%s UNWEIGHTED, UNMASKED passed' % h, \ "pmshape =", d._pmshape #--- End: for ddofs=(0, 1, 2) for np, h in zip((numpy.var, numpy.std), ('var' , 'sd')): for ddof in ddofs: for axes in self.axes_combinations: b = reshape_array(a, axes) b = np(b, axis=-1, ddof=ddof) e = getattr(d, h)(axes=axes, squeeze=True, ddof=ddof) self.assertTrue( e.allclose(b, rtol=1e-05, atol=1e-08), "%s, axis=%s, unweighted, unmasked \ne=%s, \nb=%s, \ne-b=%s" % (h, axes, e.array, b, e.array-b)) #--- End: for #--- End: for print 'cf.Data.%s UNWEIGHTED, UNMASKED passed' % h, \ "pmshape =", d._pmshape #--- End: for #--- End: def def test_Data__collapse_WEIGHTED_MASKED(self): '''cf.Data._collapse WEIGHTED MASKED''' if self.do_test and inspect.stack()[0][3] not in self.do_test: return print mw = numpy.ma.array(self.w) for chunksize in self.chunk_sizes: cf.CHUNKSIZE(chunksize) d = cf.Data(self.ma.copy()) x = cf.Data(self.w.copy()) for c, h in zip((mw , mw*mw), ('sum_of_weights', 'sum_of_weights2')): for axes in self.axes_combinations: c = c.copy() c.mask = self.ma.mask b = reshape_array(c, axes) b = numpy.ma.asanyarray(b.sum(axis=-1)) e = getattr(d, h)(axes=axes, weights=x, squeeze=True) self.assertTrue( (e.mask.array == b.mask).all(), "%s, axis=%s, \ne.mask=%s, \nb.mask=%s, \ne.mask==b.mask=%s" % (h, axes, e.mask.array, b.mask, e.mask.array==b.mask)) self.assertTrue( e.allclose(b, rtol=1e-05, atol=1e-08), "%s, axis=%s, \ne=%s, \nb=%s, \ne-b=%s" % (h, axes, e.array, b, e.array-b)) #--- End: for print 'cf.Data.%s WEIGHTED, MASKED passed' % h, \ "pmshape =", d._pmshape #--- End: for h = 'mean' for axes in self.axes_combinations: b = reshape_array(self.ma, axes) v = reshape_array(mw, axes) b = numpy.ma.average(b, axis=-1, weights=v) b = numpy.ma.asanyarray(b) e = getattr(d, h)(axes=axes, weights=x, squeeze=True) self.assertTrue( (e.mask.array == b.mask).all(), "%s, axis=%s, \ne.mask=%s, \nb.mask=%s, \ne.mask==b.mask=%s" % (h, axes, e.mask.array, b.mask, e.mask.array==b.mask)) self.assertTrue( e.allclose(b, rtol=1e-05, atol=1e-08), "%s, axis=%s, \ne=%s, \nb=%s, \ne-b=%s" % (h, axes, e.array, b, e.array-b)) #--- End: for print 'cf.Data.%s WEIGHTED, MASKED passed' % h, \ "pmshape =", d._pmshape ddofs = (0, 1, 2) f = 2 for h in ('var', 'sd'): for axes in self.axes_combinations: for ddof in ddofs: b = reshape_array(self.ma, axes) v = reshape_array(mw, axes) not_enough_data = numpy.ma.count(b, axis=-1) <= ddof avg = numpy.ma.average(b, axis=-1, weights=v) if numpy.ndim(avg) < b.ndim: avg = numpy.ma.expand_dims(avg, -1) b, sv = numpy.ma.average((b-avg)**2, axis=-1, weights=v, returned=True) b = numpy.ma.where(not_enough_data, numpy.ma.masked, b) b *= f*sv/(f*sv-ddof) if h == 'sd': b **= 0.5 b = numpy.ma.asanyarray(b) e = getattr(d, h)(axes=axes, weights=x, squeeze=True, ddof=ddof, a=f) self.assertTrue( (e.mask.array == b.mask).all(), "%s, axis=%s, \ne.mask=%s, \nb.mask=%s, \ne.mask==b.mask=%s" % (h, axes, e.mask.array, b.mask, e.mask.array==b.mask)) self.assertTrue( e.allclose(b, rtol=1e-05, atol=1e-08), "%s, axis=%s, ddof=%s, \ne=%s, \nb=%s, \ne-b=%s" % (h, axes, ddof, e.array, b, e.array-b)) #--- End: for #--- End: for print 'cf.Data.%s WEIGHTED, MASKED passed' % h, \ "pmshape =", d._pmshape #--- End: for #--- End: def def test_Data__collapse_WEIGHTED_UNMASKED(self): '''cf.Data._collapse WEIGHTED UNMASKED''' if self.do_test and inspect.stack()[0][3] not in self.do_test: return print a = numpy.arange(-100, 200., dtype=float).reshape(3, 4, 5, 5) for chunksize in self.chunk_sizes: cf.CHUNKSIZE(chunksize) d = cf.Data(a[(slice(None, None, -1),) * a.ndim].copy()) d.flip(i=True) x = cf.Data(self.w.copy()) for c, h in zip((self.w , self.w*self.w), ('sum_of_weights', 'sum_of_weights2')): for axes in self.axes_combinations: b = reshape_array(c, axes) b = b.sum(axis=-1) e = getattr(d, h)(axes=axes, weights=x, squeeze=True) self.assertTrue( e.allclose(b, rtol=1e-05, atol=1e-08), "%s, axis=%s, \ne=%s, \nb=%s, \ne-b=%s" % (h, axes, e.array, b, e.array-b)) #--- End: for print 'cf.Data.%s WEIGHTED, UNMASKED passed' % h, \ "pmshape =", d._pmshape #--- End: for h = 'mean' for axes in self.axes_combinations: b = reshape_array(a, axes) v = reshape_array(self.w, axes) b = numpy.average(b, axis=-1, weights=v) e = getattr(d, h)(axes=axes, weights=x, squeeze=True) self.assertTrue( e.allclose(b, rtol=1e-05, atol=1e-08), "%s, axis=%s, \ne=%s, \nb=%s, \ne-b=%s" % (h, axes, e.array, b, e.array-b)) #--- End: for print 'cf.Data.%s WEIGHTED, UNMASKED passed' % h, \ "pmshape =", d._pmshape ddofs = (0, 1, 2) f= 2 for h in ('var', 'sd'): for axes in self.axes_combinations: for ddof in ddofs: b = reshape_array(a, axes) v = reshape_array(self.w, axes) avg = numpy.average(b, axis=-1, weights=v) if numpy.ndim(avg) < b.ndim: avg = numpy.expand_dims(avg, -1) b, sv = numpy.average((b-avg)**2, axis=-1, weights=v, returned=True) b *= f*sv/(f*sv-ddof) if h == 'sd': b **= 0.5 e = getattr(d, h)(axes=axes, weights=x, squeeze=True, ddof=ddof, a=f) self.assertTrue( e.allclose(b, rtol=1e-05, atol=1e-08) , "%s, axis=%s, ddof=%s, \ne=%s, \nb=%s, \ne-b=%s" % \ (h, axes, ddof, e.array, b, e.array-b)) #--- End: for #--- End: for print 'cf.Data.%s WEIGHTED, UNMASKED passed' % h, \ "pmshape =", d._pmshape #--- End: for #--- End: def #--- End: class if __name__ == "__main__": original_chunksize = cf.CHUNKSIZE() unittest.main() cf.CHUNKSIZE(original_chunksize)