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
|
import warnings
import inspect
from functools import wraps
from astropy.utils.exceptions import AstropyUserWarning
bigdataurl = "https://spectral-cube.readthedocs.io/en/latest/big_data.html"
from tqdm.auto import tqdm
def ProgressBar(niter, **kwargs):
return tqdm(total=niter, **kwargs)
def cached(func):
"""
Decorator to cache function calls
"""
@wraps(func)
def wrapper(self, *args):
# The cache lives in the instance so that it gets garbage collected
if (func, args) not in self._cache:
self._cache[(func, args)] = func(self, *args)
return self._cache[(func, args)]
wrapper.wrapped_function = func
return wrapper
def warn_slow(function):
@wraps(function)
def wrapper(self, *args, **kwargs):
# if the function accepts a 'how', the 'cube' approach requires the whole cube in memory
argspec = inspect.getfullargspec(function)
accepts_how_keyword = 'how' in argspec.args or argspec.varkw == 'how'
warn_how = accepts_how_keyword and ((kwargs.get('how') == 'cube') or 'how' not in kwargs)
loads_whole_cube = not (kwargs.get('how') in ('slice', 'ray'))
if self._is_huge and not self.allow_huge_operations and loads_whole_cube:
warn_message = ("This function ({0}) requires loading the entire "
"cube into memory, and the cube is large ({1} "
"pixels), so by default we disable this operation. "
"To enable the operation, set "
"`cube.allow_huge_operations=True` and try again. ").format(str(function), self.size)
if warn_how:
warn_message += ("Alternatively, you may want to consider using an "
"approach that does not load the whole cube into "
"memory by specifying how='slice' or how='ray'. ")
warn_message += ("See {bigdataurl} for details.".format(bigdataurl=bigdataurl))
raise ValueError(warn_message)
elif warn_how and not self._is_huge and loads_whole_cube:
# TODO: add check for whether cube has been loaded into memory
warnings.warn("This function ({0}) requires loading the entire cube into "
"memory and may therefore be slow.".format(str(function)),
PossiblySlowWarning
)
return function(self, *args, **kwargs)
return wrapper
class SpectralCubeWarning(AstropyUserWarning):
pass
class UnsupportedIterationStrategyWarning(SpectralCubeWarning):
pass
class VarianceWarning(SpectralCubeWarning):
pass
class SliceWarning(SpectralCubeWarning):
pass
class BeamAverageWarning(SpectralCubeWarning):
pass
class BeamWarning(SpectralCubeWarning):
pass
class WCSCelestialError(Exception):
pass
class WCSMismatchWarning(SpectralCubeWarning):
pass
class NotImplementedWarning(SpectralCubeWarning):
pass
class StokesWarning(SpectralCubeWarning):
pass
class ExperimentalImplementationWarning(SpectralCubeWarning):
pass
class PossiblySlowWarning(SpectralCubeWarning):
pass
class SmoothingWarning(SpectralCubeWarning):
pass
class NonFiniteBeamsWarning(SpectralCubeWarning):
pass
class WCSWarning(SpectralCubeWarning):
pass
class FITSWarning(SpectralCubeWarning):
pass
class BadVelocitiesWarning(SpectralCubeWarning):
pass
class FITSReadError(Exception):
pass
class NoBeamError(Exception):
pass
class BeamUnitsError(Exception):
pass
|
