/
spherical.py
executable file
·1824 lines (1383 loc) · 61.8 KB
/
spherical.py
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
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""
Copyright 2017-2019 Louis Moresi, Ben Mather
This file is part of Stripy.
Stripy is free software: you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as published by
the Free Software Foundation, either version 3 of the License, or any later version.
Stripy is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with Stripy. If not, see <http://www.gnu.org/licenses/>.
"""
#!/usr/bin/python
# -*- coding: utf-8 -*-
from multiprocessing import cpu_count
from . import _stripack
from . import _ssrfpack
import numpy as np
try: range = xrange
except: pass
_ier_codes = {0: "no errors were encountered.",
-1: "N < 3 on input.",
-2: "the first three nodes lie on a great circle.\nSet permute to True or reorder nodes manually.",
-3: "duplicate nodes were encountered.",
-4: "an error flag was returned by a call to SWAP in ADDNOD.\n \
This is an internal error and should be reported to the programmer.",
'L':"nodes L and M coincide for some M > L.\n \
The linked list represents a triangulation of nodes 1 to M-1 in this case.",
1: "NCC, N, NROW, or an LCC entry is outside its valid range on input.",
2: "the triangulation data structure (LIST,LPTR,LEND) is invalid.",
'K': 'NPTS(K) is not a valid index in the range 1 to N.',
9999: "Triangulation encountered duplicate nodes."}
def _auto_threads(n_in, n_out, order):
"""Attempt to automatically determine the best number of threads to use
for spherical interpolation.
Parameters
----------
n_in : int
Number of input points for triangulation.
n_out : int
Number of output points for interpolation.
order : {0, 1, 3}
Order of interpolation (nearest-neighbour, linear, cubic).
Returns
-------
threads : int
Number of threads to use.
"""
if order == 3:
use_threads = False
elif n_out >= 50000:
use_threads = True
elif n_out >= 5000 and n_in >= 10000:
use_threads = True
else:
use_threads = False
return cpu_count() if use_threads else 1
class sTriangulation(object):
"""
Define a Delaunay triangulation for given points on a sphere
where lons and lats are 1D numpy arrays of equal length.
Algorithm:
R. J. Renka (1997), Algorithm 772; STRIPACK: Delaunay triangulation
and Voronoi diagram on the surface of a sphere"
ACM Trans. Math. Softw., Vol. 23, No. 3, pp 416-434
doi:10.1145/275323.275329
Args:
lons : 1D array
longitudinal coordinates in radians
lats : 1D array
latitudinal coordinates in radians
refinement_levels : int
refine the number of points in the triangulation
(see uniformly_refine_triangulation)
permute : bool
randomises the order of lons and lats to improve
triangulation efficiency and eliminate issues where the first points lie on a great circle (see notes)
tree : bool
construct a cKDtree for efficient nearest-neighbour lookup
Attributes:
lons : array of floats, shape (n,)
stored longitudinal coordinates on a sphere
lats : array of floats, shape (n,)
stored latitudinal coordinates on a sphere
x : array of floats, shape (n,)
stored Cartesian x coordinates from input
y : array of floats, shape (n,)
stored Cartesian y coordinates from input
z : array of floats, shape (n,)
stored Cartesian z coordinates from input
simplices : array of ints, shape (nsimplex, 3)
indices of the points forming the simplices in the triangulation
points are ordered anticlockwise
lst : array of ints, shape (6n-12,)
nodal indices with lptr and lend, define the triangulation as a set of N
adjacency lists; counterclockwise-ordered sequences of neighboring nodes
such that the first and last neighbors of a boundary node are boundary
nodes (the first neighbor of an interior node is arbitrary). In order to
distinguish between interior and boundary nodes, the last neighbor of
each boundary node is represented by the negative of its index.
The indices are 1-based (as in Fortran), not zero based (as in python).
lptr : array of ints, shape (6n-12),)
set of pointers in one-to-one correspondence with the elements of lst.
lst(lptr(i)) indexes the node which follows lst(i) in cyclical
counterclockwise order (the first neighbor follows the last neighbor).
The indices are 1-based (as in Fortran), not zero based (as in python).
lend : array of ints, shape (n,)
N pointers to adjacency lists.
lend(k) points to the last neighbor of node K.
lst(lend(K)) < 0 if and only if K is a boundary node.
The indices are 1-based (as in Fortran), not zero based (as in python).
Notes:
Provided the nodes are randomly ordered, the algorithm
has expected time complexity O(N*log(N)) for most nodal
distributions. Note, however, that the complexity may be
as high as O(N**2) if, for example, the nodes are ordered
on increasing latitude.
if permute=True, lons and lats are randomised on input before
they are triangulated. The distribution of triangles will
differ between setting permute=True and permute=False,
however, the node ordering will remain identical.
"""
def __init__(self, lons, lats, refinement_levels=0, permute=False, tree=False):
# lons, lats = self._check_integrity(lons, lats)
self.permute = permute
self.tree = tree
self._update_triangulation(lons, lats)
for r in range(0,refinement_levels):
lons, lats = self.uniformly_refine_triangulation(faces=False, trisect=False)
self._update_triangulation(lons,lats)
return
def _generate_permutation(self, npoints):
"""
Create shuffle and deshuffle vectors
"""
i = np.arange(0, npoints)
# permutation
p = np.random.permutation(npoints)
ip = np.empty_like(p)
# inverse permutation
ip[p[i]] = i
return p, ip
def _is_collinear(self, lons, lats):
"""
Checks if first three points are collinear - in the spherical
case this corresponds to all points lying on a great circle
and, hence, all coordinate vectors being in a single plane.
"""
x, y, z = lonlat2xyz(lons[:3], lats[:3])
pts = np.column_stack([x, y, z])
collinearity = (np.linalg.det(pts.T) == 0.0)
return collinearity
def _update_triangulation(self, lons, lats):
npoints = len(lons)
# We do this spherical -> cartesian -> spherical conversion
# to protect against lons or lats being out of range.
# (Only really an issue for refined meshes)
xs, ys, zs = lonlat2xyz(lons, lats)
lons, lats = xyz2lonlat(xs, ys, zs)
# Deal with collinear issue
if self.permute:
# Store permutation vectors to shuffle/deshuffle lons and lats
niter = 0
ierr = -2
while ierr==-2 and niter < 5:
p, ip = self._generate_permutation(npoints)
lons = lons[p]
lats = lats[p]
# compute cartesian coords on unit sphere.
x, y, z = _stripack.trans(lats, lons)
lst, lptr, lend, ierr = _stripack.trmesh(x, y, z)
niter += 1
if niter >= 5:
raise ValueError(_ier_codes[-2])
else:
p = np.arange(0, npoints)
ip = p
# compute cartesian coords on unit sphere.
x, y, z = _stripack.trans(lats, lons)
lst, lptr, lend, ierr = _stripack.trmesh(x, y, z)
self._permutation = p
self._invpermutation = ip
if ierr > 0:
raise ValueError('ierr={} in trmesh\n{}'.format(ierr, _ier_codes[9999]))
if ierr != 0:
raise ValueError('ierr={} in trmesh\n{}'.format(ierr, _ier_codes[ierr]))
self.npoints = npoints
self._lons = lons
self._lats = lats
self._x = x
self._y = y
self._z = z
self._points = np.column_stack([x, y, z])
self.lst = lst
self.lptr = lptr
self.lend = lend
# initialise dummy sigma array with zero tension factors
self._sigma = np.zeros(self.lptr.size)
# Convert a triangulation to a triangle list form (human readable)
# Uses an optimised version of trlist that returns triangles
# without neighbours or arc indices
nt, ltri, ierr = _stripack.trlist2(lst, lptr, lend)
if ierr != 0:
raise ValueError('ierr={} in trlist2\n{}'.format(ierr, _ier_codes[ierr]))
# extract triangle list and convert to zero-based ordering
self._simplices = ltri.T[:nt] - 1
## np.ndarray.sort(self.simplices, axis=1)
## If scipy is installed, build a KDtree to find neighbour points
if self.tree:
self._build_cKDtree()
@property
def lons(self):
""" Longitudinal coordinates on a sphere """
return self._deshuffle_field(self._lons)
@property
def lats(self):
""" Latitudinal coordinates on a sphere """
return self._deshuffle_field(self._lats)
@property
def x(self):
""" Stored Cartesian x coordinates from triangulation """
return self._deshuffle_field(self._x)
@property
def y(self):
""" Stored Cartesian y coordinates from triangulation """
return self._deshuffle_field(self._y)
@property
def z(self):
""" Stored Cartesian x coordinates from triangulation """
return self._deshuffle_field(self._z)
@property
def points(self):
""" Stored Cartesian xyz coordinates from triangulation """
return self._deshuffle_field(self._points)
@property
def simplices(self):
""" Indices of the points forming the simplices in the triangulation.
Points are ordered anticlockwise """
return self._deshuffle_simplices(self._simplices)
def _shuffle_field(self, *args):
"""
Permute field
"""
p = self._permutation
fields = []
for arg in args:
fields.append( arg[p] )
if len(fields) == 1:
return fields[0]
else:
return fields
def _deshuffle_field(self, *args):
"""
Return to original ordering
"""
ip = self._invpermutation
fields = []
for arg in args:
fields.append( arg[ip] )
if len(fields) == 1:
return fields[0]
else:
return fields
def _shuffle_simplices(self, simplices):
"""
Permute ordering
"""
ip = self._invpermutation
return ip[simplices]
def _deshuffle_simplices(self, simplices):
"""
Return to original ordering
"""
p = self._permutation
return p[simplices]
def gradient_lonlat(self, data, nit=3, tol=1.0e-3, guarantee_convergence=False, sigma=None):
"""
Return the lon / lat components of the gradient
of a scalar field on the surface of the UNIT sphere.
(Note: the companion routine is derivatives_lonlat which returns
the components of the derivative in each direction - these differ by a factor of
1/cos(lat) in the first component)
The method consists of minimizing a quadratic functional Q(G) over
gradient vectors, where Q is an approximation to the linearized
curvature over the triangulation of a C-1 bivariate function F(x,y)
which interpolates the nodal values and gradients.
Args:
data : array of floats, shape (n,)
field over which to evaluate the gradient
nit : int (default: 3)
number of iterations to reach a convergence tolerance, tol
nit >= 1
tol : float (default: 1e-3)
maximum change in gradient between iterations.
convergence is reached when this condition is met.
sigma : array of floats, shape (6n-12)
precomputed array of spline tension factors from
`get_spline_tension_factors(zdata, tol=1e-3, grad=None)`
Returns:
dfdlon : array of floats, shape (n,)
derivative of f in the longitudinal direction
dfdlat : array of floats, shape (n,)
derivative of f in the lattitudinal direction
Notes:
The gradient is computed via the Cartesian components using
`spherical.sTriangulation.gradient_xyz` and the iteration parameters
controling the spline interpolation are passed directly to this
routine (See notes for `gradient_xyz` for more details).
The gradient operator in this geometry is not well defined at the poles
even if the scalar field is smooth and the Cartesian gradient is well defined.
The routine spherical.dxyz2dlonlat is available to convert the Cartesian
to lon/lat coordinates at any point on the unit sphere. This is helpful
to avoid recalculation if you need both forms.
"""
dfxs, dfys, dfzs = self.gradient_xyz(data, nit=nit, tol=tol, \
guarantee_convergence=guarantee_convergence, sigma=sigma)
# get deshuffled versions
lons = self.lons
lats = self.lats
z = self.z
dlon = -dfxs * np.cos(lats) * np.sin(lons) + dfys * np.cos(lats) * np.cos(lons) # no z dependence
dlat = -dfxs * np.sin(lats) * np.cos(lons) - dfys * np.sin(lats) * np.sin(lons) + dfzs * np.cos(lats)
corr = np.sqrt((1.0-z**2))
valid = ~np.isclose(corr,0.0)
dlon[valid] = dlon[valid] / corr[valid]
return dlon, dlat
def derivatives_lonlat(self, data, nit=3, tol=1.0e-3, guarantee_convergence=False, sigma=None):
"""
Return the lon / lat components of the derivatives
of a scalar field on the surface of the UNIT sphere.
(Note: the companion routine is gradient_lonlat which returns
the components of the surface gradient - these differ by a factor of
1/cos(lat) in the first component)
The method consists of minimizing a quadratic functional Q(G) over
gradient vectors, where Q is an approximation to the linearized
curvature over the triangulation of a C-1 bivariate function F(x,y)
which interpolates the nodal values and gradients.
Args:
data : array of floats, shape (n,)
field over which to evaluate the gradient
nit : int (default: 3)
number of iterations to reach a convergence tolerance, tol
nit >= 1
tol : float (default: 1e-3)
maximum change in gradient between iterations.
convergence is reached when this condition is met.
sigma : array of floats, shape (6n-12)
precomputed array of spline tension factors from
`get_spline_tension_factors(zdata, tol=1e-3, grad=None)`
Returns:
dfdlon : array of floats, shape (n,)
derivative of f in the longitudinal direction
dfdlat : array of floats, shape (n,)
derivative of f in the lattitudinal direction
Notes:
The gradient is computed via the Cartesian components using
`spherical.sTriangulation.gradient_xyz` and the iteration parameters
controling the spline interpolation are passed directly to this
routine (See notes for `gradient_xyz` for more details).
The gradient operator in this geometry is not well defined at the poles
even if the scalar field is smooth and the Cartesian gradient is well defined.
The routine spherical.dxyz2dlonlat is available to convert the Cartesian
to lon/lat coordinates at any point on the unit sphere. This is helpful
to avoid recalculation if you need both forms.
"""
dfxs, dfys, dfzs = self.gradient_xyz(data, nit=nit, tol=tol, \
guarantee_convergence=guarantee_convergence, sigma=sigma)
# get deshuffled versions
lons = self.lons
lats = self.lats
z = self.z
dlon = -dfxs * np.cos(lats) * np.sin(lons) + dfys * np.cos(lats) * np.cos(lons) # no z dependence
dlat = -dfxs * np.sin(lats) * np.cos(lons) - dfys * np.sin(lats) * np.sin(lons) + dfzs * np.cos(lats)
return dlon, dlat
def gradient_xyz(self, f, nit=3, tol=1e-3, guarantee_convergence=False, sigma=None):
"""
Return the cartesian components of the gradient
of a scalar field on the surface of the sphere.
The method consists of minimizing a quadratic functional Q(G) over
gradient vectors, where Q is an approximation to the linearized
curvature over the triangulation of a C-1 bivariate function F(x,y)
which interpolates the nodal values and gradients.
Args:
f : array of floats, shape (n,)
field over which to evaluate the gradient
nit : int (default: 3)
number of iterations to reach a convergence tolerance, tol
nit >= 1
tol : float (default: 1e-3)
maximum change in gradient between iterations.
convergence is reached when this condition is met.
sigma : array of floats, shape (6n-12)
precomputed array of spline tension factors from
`get_spline_tension_factors(zdata, tol=1e-3, grad=None)`
Returns:
dfdx : array of floats, shape (n,)
derivative of f in the x direction
dfdy : array of floats, shape (n,)
derivative of f in the y direction
dfdz : array of floats, shape (n,)
derivative of f in the z direction
Notes:
For SIGMA = 0, optimal efficiency was achieved in testing with
tol = 0, and nit = 3 or 4.
The restriction of F to an arc of the triangulation is taken to be
the Hermite interpolatory tension spline defined by the data values
and tangential gradient components at the endpoints of the arc, and
Q is the sum over the triangulation arcs, excluding interior
constraint arcs, of the linearized curvatures of F along the arcs --
the integrals over the arcs of D2F(T)**2, where D2F(T) is the second
derivative of F with respect to distance T along the arc.
"""
if f.size != self.npoints:
raise ValueError('f should be the same size as mesh')
# gradient = np.zeros((3,self.npoints), order='F', dtype=float32)
sigma, iflgs = self._check_sigma(sigma)
f = self._shuffle_field(f)
ierr = 1
while ierr == 1:
grad, ierr = _ssrfpack.gradg(self._x, self._y, self._z, f,\
self.lst, self.lptr, self.lend,\
iflgs, sigma, nit, tol)
if not guarantee_convergence:
break
import warnings
if ierr < 0:
import warnings
warnings.warn('ierr={} in gradg\n{}'.format(ierr, _ier_codes[ierr]))
return self._deshuffle_field(grad[0], grad[1], grad[2])
def smoothing(self, f, w, sm, smtol, gstol, sigma=None):
"""
Smooths a surface f by choosing nodal function values and gradients to
minimize the linearized curvature of F subject to a bound on the
deviation from the data values. This is more appropriate than interpolation
when significant errors are present in the data.
Args:
f : array of floats, shape (n,)
field to apply smoothing on
w : array of floats, shape (n,)
weights associated with data value in f
w[i] = 1/sigma_f^2 is a good rule of thumb.
sm : float
positive parameter specifying an upper bound on Q2(f).
generally n-sqrt(2n) <= sm <= n+sqrt(2n)
smtol : float
specifies relative error in satisfying the constraint
sm(1-smtol) <= Q2 <= sm(1+smtol) between 0 and 1.
gstol : float
tolerance for convergence.
gstol = 0.05*mean(sigma_f)^2 is a good rule of thumb.
sigma : array of floats, shape (6n-12)
precomputed array of spline tension factors from
`get_spline_tension_factors(zdata, tol=1e-3, grad=None)`
Returns:
f_smooth : array of floats, shape (n,)
smoothed version of f
derivatives : tuple of floats, shape (n,3)
(dfdx, dfdy, dfdz) first derivatives of f_smooth in the
x, y, and z directions
err : error indicator
0 indicates no error, +ve values indicate warnings, -ve values are errors
"""
if f.size != self.npoints or f.size != w.size:
raise ValueError('f and w should be the same size as mesh')
f, w = self._shuffle_field(f, w)
sigma, iflgs = self._check_sigma(sigma)
prnt = -1
f_smooth, df, ierr = _ssrfpack.smsurf(self._x, self._y, self._z, f,\
self.lst, self.lptr, self.lend,\
iflgs, sigma, w, sm, smtol, gstol, prnt)
import warnings
# Note - warnings are good because they can be 'upgraded' to exceptions by the
# user of the module. The warning text is usually something that we don't
# emit every time the error occurs. So here we emit a message about the problem
# and a warning that explains it (once) - and also serves as a hook for an exception trap.
if ierr < 0:
print('ierr={} in smooth routines\n{}'.format(ierr, _ier_codes[ierr]))
if ierr == 1:
warnings.warn("No errors were encountered but the constraint is not active --\n\
F, FX, and FY are the values and partials of a linear function which minimizes Q2(F), and Q1 = 0.")
if ierr == 2:
warnings.warn("The constraint could not be satisfied to within SMTOL due to ill-conditioned linear systems.")
return self._deshuffle_field(f_smooth), self._deshuffle_field(df[0], df[1], df[2]), ierr
def _check_integrity(self, lons, lats):
"""
Ensure lons and lats are:
- 1D numpy arrays
- equal size
- within the appropriate range in radians
"""
lons = np.array(lons).ravel()
lats = np.array(lats).ravel()
if len(lons.shape) != 1 or len(lats.shape) != 1:
raise ValueError('lons and lats must be 1D')
if lats.size != lons.size:
raise ValueError('lons and lats must have same length')
if (np.abs(lons)).max() > 2.*np.pi:
raise ValueError("lons must be in radians (-2*pi <= lon <= 2*pi)")
if (np.abs(lats)).max() > 0.5*np.pi:
raise ValueError("lats must be in radians (-pi/2 <= lat <= pi/2)")
return lons, lats
def _check_gradient(self, zdata, grad):
"""
Error checking on the gradient operator
`grad` must be (3,n) array that is permuted
iflgg = 0 if gradient should be estimated
iflgg = 1 if gradient is provided
"""
p = self._permutation
if grad is None:
grad = np.empty((3,self.npoints))
iflgg = 0
elif grad.shape == (3,self.npoints):
grad = grad[:,p] # permute
iflgg = 1
else:
raise ValueError("gradient should be 'None' or of shape (3,n).")
return grad, iflgg
def _check_sigma(self, sigma):
"""
Error checking on sigma
`sigma` must be of length 6n-12.
"""
if sigma is None:
iflgs = 0
sigma = self._sigma
else:
assert len(sigma) == 6*self.npoints-12, "sigma must be of length 6n-12"
iflgs = int(np.any(sigma))
return sigma, iflgs
def update_tension_factors(self, zdata, tol=1e-3, grad=None):
"""
WARNING: this is deprecated in favour of `get_spline_tension_factors`
"""
import warnings
message = "Use get_spline_tension_factors and supply tension factors to interpolation/gradient arrays"
message += "\nsigma stored on this mesh object no longer does anything as of v2.0"
warnings.warn(message, DeprecationWarning, stacklevel=2)
return self.get_spline_tension_factors(zdata, tol, grad)
def get_spline_tension_factors(self, zdata, tol=1e-3, grad=None):
"""
Determines, for each triangulation arc, the smallest (nonnegative) tension factor `sigma`
such that the Hermite interpolatory tension spline, defined by `sigma` and specified
endpoint data, preserves local shape properties (monotonicity and convexity) of `zdata`.
Args:
zdata : array of floats, shape(n,)
value at each point in the triangulation
must be the same size of the mesh
tol : float
tolerance of each tension factor to its optimal value
when nonzero finite tension is necessary.
grad : array of floats, shape(3,n)
precomputed gradient of zdata or if not provided,
the result of `self.gradient(zdata)`.
Returns:
sigma : array of floats, shape(6n-12)
tension factors which preserves the local properties of `zdata` on each
triangulation arc with the restriction that `sigma[i] <= 85`.
- `sigma[i] = 85` if infinite tension is required on an arc.
- `sigma[i] = 0` if the result should be cubic on the arc.
Notes:
Supply sigma to gradient, interpolate, derivative, or smoothing
methods for tensioned splines. Here is a list of compatible methods:
- `interpolate(lons, lats, zdata, order=3, grad=None, sigma=None)`
- `interpolate_cubic(lons, lats, zdata, grad=None, sigma=None)`
- `interpolate_to_grid(lons, lats, zdata, grad=None, sigma=None)`
- `gradient_xyz(f, nit=3, tol=1e-3, guarantee_convergence=False, sigma=None)`
- `gradient_lonlat(f, nit=3, tol=1e-3, guarantee_convergence=False, sigma=None)`
- `smoothing(f, w, sm, smtol, gstol, sigma=None)`
"""
if zdata.size != self.npoints:
raise ValueError("data must be of size {}".format(self.npoints))
p = self._permutation
zdata = self._shuffle_field(zdata)
if grad is None:
grad = np.vstack(self.gradient_xyz(zdata, tol=tol))
grad = grad[:,p] # permute
elif grad.shape == (3,self.npoints):
grad = grad[:,p] # permute
else:
raise ValueError("gradient should be 'None' or of shape (3,n).")
sigma, dsmax, ierr = _ssrfpack.getsig(self._x, self._y, self._z, zdata,\
self.lst, self.lptr, self.lend,\
grad, tol)
if ierr == -1:
import warnings
warnings.warn("sigma is not altered.")
# self.sigma = sigma
# self.iflgs = int(sigma.any())
return sigma
def interpolate_to_grid(self, lons, lats, zdata, grad=None, sigma=None):
"""
Interplates the data values to a uniform grid defined by
longitude and latitudinal arrays. The interpolant is once
continuously differentiable. Extrapolation is performed at
grid points exterior to the triangulation when the nodes
do not cover the entire sphere.
Args:
lons : array of floats, shape (ni,)
longitudinal coordinates in ascending order
lats : array of floats, shape (nj,)
latitudinal coordinates in ascending order
zdata : array of floats, shape(n,)
value at each point in the triangulation
must be the same size of the mesh
grad : array of floats, shape(3,n)
precomputed gradient of zdata or if not provided,
the result of `self.gradient(zdata)`.
sigma : array of floats, shape (6n-12)
precomputed array of spline tension factors from
`get_spline_tension_factors(zdata, tol=1e-3, grad=None)`
Returns:
zgrid : array of floats, shape(nj,ni)
interpolated values defined by gridded lons/lats
"""
_emsg = {-1: "n, ni, nj, or iflgg is outside its valid range.",\
-2: "nodes are collinear.",\
-3: "extrapolation failed due to the uniform grid extending \
too far beyond the triangulation boundary"}
if zdata.size != self.npoints:
raise ValueError("data must be of size {}".format(self.npoints))
zdata = self._shuffle_field(zdata)
grad, iflgg = self._check_gradient(zdata, grad)
sigma, iflgs = self._check_sigma(sigma)
nrow = len(lats)
ff, ierr = _ssrfpack.unif(self._x, self._y, self._z, zdata,\
self.lst, self.lptr, self.lend,\
iflgs, sigma, nrow, lats, lons,\
iflgg, grad)
if ierr < 0:
raise ValueError(_emsg[ierr])
return ff
def interpolate(
self,
lons,
lats,
zdata,
order=1,
grad=None,
sigma=None,
threads=1,
):
"""
Base class to handle nearest neighbour, linear, and cubic interpolation.
Given a triangulation of a set of nodes on the unit sphere, along with data
values at the nodes, this method interpolates (or extrapolates) the value
at a given longitude and latitude.
Args:
lons : float / array of floats, shape (l,)
longitudinal coordinate(s) on the sphere
lats : float / array of floats, shape (l,)
latitudinal coordinate(s) on the sphere
zdata : array of floats, shape (n,)
value at each point in the triangulation
must be the same size of the mesh
order : int (default=1)
order of the interpolatory function used
- `order=0` = nearest-neighbour
- `order=1` = linear
- `order=3` = cubic
sigma : array of floats, shape (6n-12)
precomputed array of spline tension factors from
`get_spline_tension_factors(zdata, tol=1e-3, grad=None)`
(only used in cubic interpolation)
threads : int or 'auto', optional; default : 1
Number of threads to use for linear and nearest-neighbour
interpolation (N.B. multi-threaded cubic interpolation is
not supported).
By default, only a single thread will be used. Use
`threads='auto'` to attempt to automatically determine how
many threads to use based on the size of the input and output
data.
Negative values count backwards, such that -1 is equivalent to
`multiprocessing.cpu_count()`, -2 to `cpu_count() - 1`, etc.
Returns:
zi : float / array of floats, shape (l,)
interpolated value(s) at (lons, lats)
err : int / array of ints, shape (l,)
whether interpolation (0), extrapolation (1) or error (other)
"""
shape = np.shape(lons)
lons, lats = self._check_integrity(lons, lats)
n_in = np.size(zdata)
n_out = np.size(lons)
if n_in != self.npoints:
raise ValueError("data must be of size {}".format(self.npoints))
zdata = self._shuffle_field(zdata)
if order not in {0, 1, 3}:
raise ValueError("order must be 0, 1, or 3")
if threads == 0:
raise ValueError("threads must not be zero")
if threads == "auto":
# Try to guess whether multiple threads would be helpful
threads = _auto_threads(n_in, n_out, order)
threads = int(threads)
if order == 3 and threads != 1:
import warnings
warnings.warn(
"Multithreading not supported for cubic interpolation",
RuntimeWarning,
)
threads = 1
if threads < 0:
# -1 corresponds to cpu_count(), etc.
threads = cpu_count() + threads + 1
if threads == 1:
if order == 3:
sigma, iflgs = self._check_sigma(sigma)
grad, iflgg = self._check_gradient(zdata, grad)
zi, zierr, ierr = _ssrfpack.interp_cubic(lats, lons,
self._x, self._y, self._z, zdata,
self.lst, self.lptr, self.lend,
iflgs, sigma, iflgg, grad)
else:
zi, zierr, ierr = _stripack.interp_n(order, lats, lons,
self._x, self._y, self._z, zdata,
self.lst, self.lptr, self.lend)
else:
# The following code is largely adapted from here:
# https://numpy.org/doc/stable/reference/random/multithreading.html
import concurrent.futures
size = lons.size
zi = np.full(size, np.nan)
zierr = np.full(size, np.nan)
ierr = np.full(threads, 0)
step = np.ceil(size / threads).astype(np.int_)
futures = {}
executor = concurrent.futures.ThreadPoolExecutor(threads)
def _f(
out_zi,
out_zierr,
out_ierr,
order,
lats,
lons,
x,
y,
z,
zdata,
lst,
lptr,
lend,
first,
last,
ierr_index,
):
tmp_zi, tmp_zierr, tmp_ierr = _stripack.interp_n(
order, lats[first:last], lons[first:last],
x, y, z, zdata,
lst, lptr, lend,
)
out_zi[first:last] = tmp_zi
out_zierr[first:last] = tmp_zierr
out_ierr[ierr_index] = tmp_ierr
for i in range(threads):
first = i * step
last = (i + 1) * step
args = (
_f,
zi,
zierr,
ierr,
order,
lats,
lons,
self._x,
self._y,
self._z,
zdata,
self.lst,
self.lptr,
self.lend,
first,
last,
i,
)
futures[executor.submit(*args)] = i
concurrent.futures.wait(futures)
executor.shutdown(False)
ierr = int((ierr != 0).any())
if ierr != 0:
import warnings
warnings.warn(
'Warning some points may have errors - check error array',
RuntimeWarning,
)
zi[zierr < 0] = np.nan
return zi.reshape(shape), zierr.reshape(shape)
def interpolate_nearest(self, lons, lats, data, threads=1):
"""
Interpolate using nearest-neighbour approximation
Returns the same as `interpolate(lons,lats,data,order=0)`
"""
return self.interpolate(lons, lats, data, order=0, threads=threads)
def interpolate_linear(self, lons, lats, data, threads=1):
"""
Interpolate using linear approximation
Returns the same as `interpolate(lons,lats,data,order=1)`
"""
return self.interpolate(lons, lats, data, order=1, threads=threads)
def interpolate_cubic(
self,
lons,
lats,
data,
grad=None,
sigma=None,
*,
threads=1, # currently unused; provided for API consistency
):
"""