File: Aggregate.py

package info (click to toggle)
cf-python 1.3.2+dfsg1-4
  • links: PTS, VCS
  • area: main
  • in suites: sid, stretch
  • size: 7,996 kB
  • sloc: python: 51,733; ansic: 2,736; makefile: 78; sh: 2
file content (84 lines) | stat: -rw-r--r-- 2,848 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
import cf
import os
import time

def test(chunk_sizes=(17, 34, 60, 300, 100000)):
    start_time = time.time()

    print '----------------------------------------------------------'
    print 'cf.aggregate'
    print '----------------------------------------------------------'

    #import create_field
    
    #f = create_field.test()
    filename = os.path.join(os.path.dirname(os.path.abspath(__file__)),
                            "test_file.nc")
    original_chunksize = cf.CHUNKSIZE()

    cf.CHUNKSIZE(10000000)
    f = cf.read(filename, squeeze=True)[0]
    print f

    f.transform('at').inspect()

    for chunksize in chunk_sizes[::-1]:
        f.aux('aux0').id = 'atmosphere_hybrid_height_coordinate_ak'
        f.aux('aux1').id = 'atmosphere_hybrid_height_coordinate_bk'

        g = cf.FieldList(f.subspace[0])
        g.append(f.subspace[1:3])
        g.append(f.subspace[3])
        g[-1].flip(0, i=True)
        g.append(f.subspace[4:7])
        g[-1].flip(0, i=True)
        g.extend([f.subspace[i] for i in range(7, f.shape[0])])

        g0 = g.copy()

        assert g.equals(g0, traceback=True), "g != itself"
        
        h = cf.aggregate(g, info=1)
        
        assert h[0].shape == (10, 9), 'h[0].shape is '+repr(h[0].shape)

        assert h[0].equals(f, traceback=True), 'asdasds'
        
        assert g.equals(g0, traceback=True), "g != itself after aggregation"
        
        i = cf.aggregate(g, info=1)
        
        assert i.equals(h, traceback=True), "The second aggregation != the first"
        
        assert g.equals(g0, traceback=True), "g != itself after the second aggregation"
        
        i = cf.aggregate(g, info=1, axes='grid_latitude')
        
        assert i.equals(h, traceback=True), "The third aggregation != the first"
        
        assert g.equals(g0, traceback=True), "g !=itself after the third aggregation"
        
        assert i[0].shape == (10,9), 'i[0].shape is '+repr(i[0].shape)    
        
        i = cf.aggregate(g, info=1, axes='grid_latitude', donotchecknonaggregatingaxes=1)
        
        assert i.equals(h, traceback=True), "The fourth aggregation != the first"
        
        assert g.equals(g0, traceback=True), "g != itself after the fourth aggregation"
        
        assert i[0].shape == (10,9), 'i[0].shape is '+repr(i[0].shape)    
    #--- End: for

    print  h

    time_elapsed = (time.time() - start_time)/60.0
    print
    print '---------------------------------------------------------------------------'
    print 'All aggregation tests passed for cf version', cf.__version__
    print 'Running from', os.path.abspath(cf.__file__)
    print 'Time elapsed: %f minutes' % time_elapsed
    print '---------------------------------------------------------------------------'
#--- End: def

if __name__ == "__main__":
    test()