-
-
Notifications
You must be signed in to change notification settings - Fork 92
Expand file tree
/
Copy pathlog_meta.py
More file actions
154 lines (128 loc) · 4.98 KB
/
Copy pathlog_meta.py
File metadata and controls
154 lines (128 loc) · 4.98 KB
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
# Licensed under a 3-clause BSD style license - see LICENSE.rst
import inspect
from functools import wraps
from itertools import chain
from astropy import units as u
from astropy.io import fits
from astropy.nddata import NDData
import ccdproc # Really only need Keyword from ccdproc
__all__ = []
_LOG_ARGUMENT = "add_keyword"
_LOG_ARG_HELP = f"""
{_LOG_ARGUMENT} : str, `~ccdproc.Keyword` or dict-like, optional
Item(s) to add to metadata of result. Set to False or None to
completely disable logging.
Default is to add a dictionary with a single item:
The key is the name of this function and the value is a string
containing the arguments the function was called with, except the
value of this argument.
"""
def _insert_in_metadata_fits_safe(ccd, key, value):
from .core import _short_names
if key in _short_names:
# This keyword was (hopefully) added by autologging but the
# combination of it and its value not FITS-compliant in two
# ways: the keyword name may be more than 8 characters and
# the value may be too long. FITS cannot handle both of
# those problems at once, so this fixes one of those
# problems...
# Shorten, sort of...
short_name = _short_names[key]
if isinstance(ccd.meta, fits.Header):
ccd.meta[f"HIERARCH {key.upper()}"] = (
short_name,
"Shortened name for ccdproc command",
)
else:
ccd.meta[key] = (short_name, "Shortened name for ccdproc command")
ccd.meta[short_name] = value
else:
ccd.meta[key] = value
def log_to_metadata(func):
"""
Decorator that adds logging to ccdproc functions.
The decorator adds the optional argument _LOG_ARGUMENT to function
signature and updates the function's docstring to reflect that.
It also sets the default value of the argument to the name of the function
and the arguments it was called with.
"""
func.__doc__ = func.__doc__.format(log=_LOG_ARG_HELP)
argspec = inspect.getfullargspec(func)
original_args, _, _, defaults = (
argspec.args,
argspec.varargs,
argspec.varkw,
argspec.defaults,
)
# Add logging keyword and its default value for docstring
original_args.append(_LOG_ARGUMENT)
try:
defaults = list(defaults)
except TypeError:
defaults = []
defaults.append(True)
signature_with_arg_added = inspect.signature(func)
signature_with_arg_added = f"{func.__name__}{signature_with_arg_added}"
func.__doc__ = "\n".join([signature_with_arg_added, func.__doc__])
@wraps(func)
def wrapper(*args, **kwd):
# Grab the logging keyword, if it is present.
log_result = kwd.pop(_LOG_ARGUMENT, True)
result = func(*args, **kwd)
if not log_result:
# No need to add metadata....
meta_dict = {}
elif log_result is not True:
meta_dict = _metadata_to_dict(log_result)
else:
# Logging is not turned off, but user did not provide a value
# so construct one unless the config parameter auto_logging is set to False
if ccdproc.conf.auto_logging:
key = func.__name__
# Get names of arguments, which may or may not have
# been called as keywords.
positional_args = original_args[: len(args)]
all_args = chain(zip(positional_args, args, strict=True), kwd.items())
all_args = [
f"{name}={_replace_array_with_placeholder(val)}"
for name, val in all_args
]
log_val = ", ".join(all_args)
log_val = log_val.replace("\n", "")
meta_dict = {key: log_val}
else:
meta_dict = {}
for k, v in meta_dict.items():
_insert_in_metadata_fits_safe(result, k, v)
return result
return wrapper
def _metadata_to_dict(arg):
if isinstance(arg, str):
# add the key, no value
return {arg: None}
elif isinstance(arg, ccdproc.Keyword):
return {arg.name: arg.value}
else:
return arg
def _replace_array_with_placeholder(value):
return_type_not_value = False
if isinstance(value, u.Quantity):
return_type_not_value = not value.isscalar
elif isinstance(value, NDData) or hasattr(
value, "__array_namespace__"
): # noqa: UP038
try:
length = len(value)
except TypeError:
# Value has no length...
try:
# ...but if it is NDData its .data will have a length
length = len(value.data)
except TypeError:
# No idea what this data is, assume length is not 1
length = 42
return_type_not_value = length > 1
if return_type_not_value:
return f"<{value.__class__.__name__}>"
else:
return value