Skip to content

Data_objects

ASTRAtypes

Define basic types for ASTRA type hinting.

RV_measurement = Quantity module-attribute

Used by:

data_vector = Iterable[Quantity] module-attribute

Used by:

unitless_data_vector = Union[List[float], np.ndarray] module-attribute

unitless_data_matrix = np.ndarray module-attribute

UI_PATH = Union[str, Path] module-attribute

UI_DICT = Dict[str, Any] module-attribute

BASE

Base class for ASTRA objects.

Presents a common set of routines for all objects, mainly controlling:

1) internal link to disk paths 2) general options, such as the SAVE_DISK_SPACE and WORKING_MODE

Raises: custom_exceptions.InvalidConfiguration: description custom_exceptions.InvalidConfiguration: description

BASE

Base class, for almost all of SBART objects.

Inheriting from this class brings a common interface for SBART's User parameters and handling of disk paths.

Subclassed by:
Source code in src/ASTRA/utils/BASE.py 
 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
class BASE:
    """Base class, for almost all of SBART objects.

    Inheriting from this class brings a common interface for SBART's User parameters and handling of disk paths.

    """

    _object_type = ""
    _name = ""

    _default_params = DefaultValues(
        SAVE_DISK_SPACE=UserParam(
            DISK_SAVE_MODE.DISABLED,
            constraint=ValueFromIterable(DISK_SAVE_MODE),
            description="Save disk space in the outputs if different than None",
        ),
        WORKING_MODE=UserParam(
            default_value=WORKING_MODE.ONE_SHOT,
            constraint=ValueFromIterable(WORKING_MODE),
            description="How to store the output files. If one-shot, overwrites all files, otherwise updates products",
        ),
    )

    def __init__(
        self,
        user_configs: Optional[UI_DICT] = None,
        root_level_path: Optional[UI_PATH] = None,
        needed_folders: Optional[Dict[str, str]] = None,
        start_with_valid_status: bool = True,
        quiet_user_params: bool = False,
    ):
        """Instantiate a new object.

        Args:
            user_configs (Optional[UI_DICT], optional): User-provided dictionary with configurations.
                Defaults to None, which is interpreted as empty list.
            root_level_path (Optional[UI_PATH], optional): Main storage folder for this object. Defaults to None.
            needed_folders (Optional[Dict[str, str]], optional): dictionary of internal path names
                and relative (to root_level_path) paths. Defaults to None.
            start_with_valid_status (bool, optional): The object is instantiated with a valid status. Defaults to True.
            quiet_user_params (bool, optional): Avoid logging at run-time. Defaults to False.

        """
        self._internal_configs = InternalParameters(
            self.name,
            self._default_params,
            no_logs=quiet_user_params,
        )

        if user_configs is None:
            user_configs = {}

        self._internalPaths: Paths = Paths(root_level_path=root_level_path, preconfigured_paths=needed_folders)

        self._internal_configs.receive_user_inputs(user_configs)
        self._needed_folders = needed_folders
        self._status = Status(assume_valid=start_with_valid_status)  # BY DEFAULT IT IS A VALID ONE!

    def update_user_configs(self, new_configs: dict[str, Any]) -> None:
        """Update the current configurations with new values."""
        self._internal_configs.update_configs_with_values(new_configs)

    @property
    def disk_save_level(self) -> DISK_SAVE_MODE:
        """Return the current disk save level for this object."""
        return self._internal_configs["SAVE_DISK_SPACE"]

    @property
    def work_mode(self) -> WORKING_MODE:
        """Return the current working mode for this object."""
        return self._internal_configs["WORKING_MODE"]

    def update_work_mode_level(self, level: WORKING_MODE) -> None:
        """Update the disk save level."""
        if self.disk_save_level != level:
            self._internal_configs.update_configs_with_values({"WORKING_MODE": level})

    def update_disk_saving_level(self, level: DISK_SAVE_MODE) -> None:
        """Update the work level of the current object."""
        if self.disk_save_level != level:
            self._internal_configs.update_configs_with_values({"SAVE_DISK_SPACE": level})

    ###
    #   Data storage
    ###

    def trigger_data_storage(self, *args: Any, **kwargs: Any) -> None:
        """To be implemented by child classes."""
        ...

    def json_ready(self) -> Dict[str, Any]:
        """Convert current class into a json entry.

        Returns:
            Dict[str, Any]: json-compatible dictionary

        """
        return {}

    def generate_root_path(self, storage_path: Path, no_logs: bool = True) -> None:
        """Generate root storage folder in memory."""
        if not isinstance(storage_path, (str, Path)):
            raise custom_exceptions.InvalidConfiguration(
                f"The root path must be a string or Path object, instead of {storage_path}",
            )

        if not isinstance(storage_path, Path):
            storage_path = Path(storage_path)

        self._internalPaths.add_root_path(storage_path)

    def add_relative_path(self, path_name: str, relative_structure: str) -> None:
        """Add a new relative path to internal structure.

        Args:
            path_name (str): Keyword that will be used to specify the new path
            relative_structure (str): relative path in relation to the root one

        """
        self._internalPaths.add_relative_path(path_name, relative_structure)

    ###
    #   Handling the status of the sBART objects
    ###

    def add_to_status(self, new_flag: Flag) -> None:
        """Add a new flag to the status of this object.

        Args:
            new_flag (Flag): New flag to be added into the status

        """
        self._status.store_flag(new_flag=new_flag)

    def _data_access_checks(self) -> None:
        """Ensure that the status of the ASTRA object is valid.

        This is a very broad check of validity that is overloaded
        in multiple places in the code

        Raises
        ------
            InvalidConfiguration: If the object is not valid

        """
        if not self.is_valid:
            raise custom_exceptions.InvalidConfiguration(
                "Attempting to access data from sBART object that is not "
                f"valid. Check previous log messages for further information: {self._status}",
            )

    def load_from_file(self, root_path: Path, loading_path: str) -> None:
        """Load from file, to be overriden."""
        self.generate_root_path(root_path)

    ###
    #  Properties
    ###
    def is_object_type(self, type_to_check: str) -> bool:
        """Check if this object is of a given 'type'."""
        return self._object_type == type_to_check

    @property
    def is_valid(self) -> bool:
        """Check if object only has 'good' flags."""
        return self._status.is_valid

    @property
    def name(self) -> str:
        """Representation name."""
        return f"{self.__class__._object_type} - {self.__class__._name}"

    @property
    def storage_name(self) -> str:
        """Storage name of object."""
        return self.__class__._name

    @property
    def disk_save_enabled(self) -> bool:
        """True if there is any sort of disk savings active."""
        return self.disk_save_level != DISK_SAVE_MODE.DISABLED

    @classmethod
    def config_help(cls) -> None:
        """Print terminal all possible configurations and their constraints."""
        print(cls._default_params)

    @classmethod
    def control_parameters(cls) -> Iterable[str]:
        """Retrieve all default parameters of object."""
        return cls._default_params.keys()

disk_save_level: DISK_SAVE_MODE property

Return the current disk save level for this object.

work_mode: WORKING_MODE property

Return the current working mode for this object.

is_valid: bool property

Check if object only has 'good' flags.

name: str property

Representation name.

storage_name: str property

Storage name of object.

disk_save_enabled: bool property

True if there is any sort of disk savings active.

update_user_configs(new_configs: dict[str, Any]) -> None

Update the current configurations with new values.

Source code in src/ASTRA/utils/BASE.py 
86
87
88
def update_user_configs(self, new_configs: dict[str, Any]) -> None:
    """Update the current configurations with new values."""
    self._internal_configs.update_configs_with_values(new_configs)

update_work_mode_level(level: WORKING_MODE) -> None

Update the disk save level.

Source code in src/ASTRA/utils/BASE.py 
100
101
102
103
def update_work_mode_level(self, level: WORKING_MODE) -> None:
    """Update the disk save level."""
    if self.disk_save_level != level:
        self._internal_configs.update_configs_with_values({"WORKING_MODE": level})

update_disk_saving_level(level: DISK_SAVE_MODE) -> None

Update the work level of the current object.

Source code in src/ASTRA/utils/BASE.py 
105
106
107
108
def update_disk_saving_level(self, level: DISK_SAVE_MODE) -> None:
    """Update the work level of the current object."""
    if self.disk_save_level != level:
        self._internal_configs.update_configs_with_values({"SAVE_DISK_SPACE": level})

trigger_data_storage(*args: Any, **kwargs: Any) -> None

To be implemented by child classes.

Source code in src/ASTRA/utils/BASE.py 
114
115
116
def trigger_data_storage(self, *args: Any, **kwargs: Any) -> None:
    """To be implemented by child classes."""
    ...

json_ready() -> Dict[str, Any]

Convert current class into a json entry.

Returns: Dict[str, Any]: json-compatible dictionary

Source code in src/ASTRA/utils/BASE.py 
118
119
120
121
122
123
124
125
def json_ready(self) -> Dict[str, Any]:
    """Convert current class into a json entry.

    Returns:
        Dict[str, Any]: json-compatible dictionary

    """
    return {}

generate_root_path(storage_path: Path, no_logs: bool = True) -> None

Generate root storage folder in memory.

Source code in src/ASTRA/utils/BASE.py 
127
128
129
130
131
132
133
134
135
136
137
def generate_root_path(self, storage_path: Path, no_logs: bool = True) -> None:
    """Generate root storage folder in memory."""
    if not isinstance(storage_path, (str, Path)):
        raise custom_exceptions.InvalidConfiguration(
            f"The root path must be a string or Path object, instead of {storage_path}",
        )

    if not isinstance(storage_path, Path):
        storage_path = Path(storage_path)

    self._internalPaths.add_root_path(storage_path)

add_relative_path(path_name: str, relative_structure: str) -> None

Add a new relative path to internal structure.

Args: path_name (str): Keyword that will be used to specify the new path relative_structure (str): relative path in relation to the root one

Source code in src/ASTRA/utils/BASE.py 
139
140
141
142
143
144
145
146
147
def add_relative_path(self, path_name: str, relative_structure: str) -> None:
    """Add a new relative path to internal structure.

    Args:
        path_name (str): Keyword that will be used to specify the new path
        relative_structure (str): relative path in relation to the root one

    """
    self._internalPaths.add_relative_path(path_name, relative_structure)

add_to_status(new_flag: Flag) -> None

Add a new flag to the status of this object.

Args: new_flag (Flag): New flag to be added into the status

Source code in src/ASTRA/utils/BASE.py 
153
154
155
156
157
158
159
160
def add_to_status(self, new_flag: Flag) -> None:
    """Add a new flag to the status of this object.

    Args:
        new_flag (Flag): New flag to be added into the status

    """
    self._status.store_flag(new_flag=new_flag)

load_from_file(root_path: Path, loading_path: str) -> None

Load from file, to be overriden.

Source code in src/ASTRA/utils/BASE.py 
179
180
181
def load_from_file(self, root_path: Path, loading_path: str) -> None:
    """Load from file, to be overriden."""
    self.generate_root_path(root_path)

is_object_type(type_to_check: str) -> bool

Check if this object is of a given 'type'.

Source code in src/ASTRA/utils/BASE.py 
186
187
188
def is_object_type(self, type_to_check: str) -> bool:
    """Check if this object is of a given 'type'."""
    return self._object_type == type_to_check

config_help() -> None classmethod

Print terminal all possible configurations and their constraints.

Source code in src/ASTRA/utils/BASE.py 
210
211
212
213
@classmethod
def config_help(cls) -> None:
    """Print terminal all possible configurations and their constraints."""
    print(cls._default_params)

control_parameters() -> Iterable[str] classmethod

Retrieve all default parameters of object.

Source code in src/ASTRA/utils/BASE.py 
215
216
217
218
@classmethod
def control_parameters(cls) -> Iterable[str]:
    """Retrieve all default parameters of object."""
    return cls._default_params.keys()

Cubic_spline

Custom cubic spline, as defined in Silva et al, 2022.

CYTHON_UNAVAILABLE = False module-attribute

CustomCubicSpline

Source code in src/ASTRA/utils/Cubic_spline.py 
 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
class CustomCubicSpline:
    def __init__(
        self,
        old_wavelengths,
        original_data,
        original_errors,
        ignore_covariances=True,
        n_threads=4,
    ):
        if CYTHON_UNAVAILABLE:
            raise custom_exceptions.InternalError("Cython interface is not installed")
        self.old_wavelengths = old_wavelengths
        self.original_data = original_data
        self.original_errors = original_errors
        self.data_size = original_data.size

        self.ignore_covariances = ignore_covariances

        if ignore_covariances:
            self.cov_matrix = original_errors**2
        else:
            self.cov_matrix = np.zeros((original_data.size, original_data.size)) + original_errors**2

        self._cached_h = False
        self._inv_h = []
        self._partials = {}
        self.n_threads = n_threads
        self.inv_delta_wave = 1 / np.diff(old_wavelengths)

        # using memory layout C as the "second index" is the one that changes fastest
        self.cached_ones = np.ones((self.data_size - 2, self.data_size - 2), dtype=np.float64, order="C")

    def compute_h_matrix(self):
        """Create the H matrix."""
        if self._cached_h:
            return self._inv_h

        x = self.old_wavelengths

        diag_c = (np.roll(x, -2) - x)[0:-2] / 3
        # we remove the first element as we are interested in x_3 - x_2
        # the last one is removed for the same reason
        # the one before the last is removed as the second diagonal only goes up to N-1
        diag_r = (np.roll(x, -1) - x)[1:-2] / 6
        self._cached_h = True

        # avoid creating a new numpy array at each iteration!
        output = self.cached_ones
        output[:] = 1

        tridiagonal_inverter(diag_c, diag_r, output, n_threads=self.n_threads)
        # print("CYTHON vs Numpy EQUAL: ", np.allclose(np.linalg.inv(h), output))
        self._inv_h = output
        return self._inv_h

    def compute_partial(self, index_i):
        try:
            return self._partials[index_i]
        except KeyError:
            pass

        if index_i in (0, self.data_size - 1):
            # both edges have y'' equal to zero
            partials = np.zeros(self.data_size)
            self._partials[index_i] = partials
            return partials

        partials = np.empty(self.data_size)

        # in the expressions the array starts at one, instead of zero !!
        inv_h_matrix = self.compute_h_matrix()

        # the two edges have a value of zero!
        partials[0] = 0
        partials[-1] = 0

        partial_derivative(inv_h_matrix, self.inv_delta_wave, index_i, partials, self.n_threads)

        # avoid underflows
        partials[np.where(np.abs(partials) < 1e-100)] = 0
        self._partials[index_i] = partials
        return partials

    def _U_entry(self, mode, first_0, second_0):
        """Mode == '22':
            calculate u(y''_first_0, y''_second_0)
        if mode == '20':
            calculate u(y''_first_0, y_second_0)

        """
        partial_first = self.compute_partial(first_0)
        if mode == "22":  # calculate entry for cov between two 2nd derivatives
            partial_second = self.compute_partial(second_0)
            if self.ignore_covariances:
                first_part = np.multiply(partial_first, self.cov_matrix)
            else:
                first_part = np.matmul(partial_first, self.cov_matrix)
            return np.matmul(first_part, np.transpose(partial_second))
        if mode == "20":  # entry for 2nd derivative and data
            second = np.zeros(partial_first.shape)
            second[second_0] = 1

            if self.ignore_covariances:
                first_part = np.multiply(partial_first, self.cov_matrix)
                # print(np.min(first_part), np.max(first_part), partial_first.dtype, self.cov_matrix.dtype, first_part.dtype)

            else:
                first_part = np.matmul(partial_first, self.cov_matrix)
            return np.matmul(first_part, second)

    def build_U_matrix(self, index_i_0, index_j_0):
        """Create the "U" matrix for the covariance estimation; Only computes the relevant quadrant."""
        cov_mat = self.cov_matrix

        i = index_i_0
        j = index_j_0

        if self.ignore_covariances:
            cov_i_i = cov_mat[i]
            cov_i_j1 = 0
            cov_i1_j = 0
            cov_j_j = cov_mat[i + 1]
        else:
            cov_i_i = cov_mat[i, j]
            cov_i_j1 = cov_mat[i, j + 1]
            cov_i1_j = cov_mat[i + 1, j]
            cov_j_j = cov_mat[i + 1, j + 1]

        # we are only evaluating the diagonal, i == j
        u_03 = self._U_entry("20", j + 1, i)
        u_01 = self._U_entry("20", j, i)
        u_13 = self._U_entry("20", j + 1, i + 1)

        u_23 = self._U_entry("22", i, j + 1)

        U = np.array(
            [
                [cov_i_i, cov_i_j1, u_01, u_03],
                [cov_i1_j, cov_j_j, u_03, u_13],
                [u_01, u_03, self._U_entry("22", i, j), u_23],
                [u_03, u_13, u_23, self._U_entry("22", i + 1, j + 1)],
            ],
        )

        return U

    def compute_second_derivative(self):
        old_wavelengths = self.old_wavelengths
        original_data = self.original_data
        inv_h_matrix = self.compute_h_matrix()

        second_deriv_vals = np.zeros(original_data.shape, dtype=np.float64)

        second_derivative(
            np.diff(old_wavelengths),
            np.diff(original_data),
            inv_h_matrix,
            second_deriv_vals,
            self.n_threads,
        )

        return second_deriv_vals

    def interpolate(self, new_wavelengths):
        """Interpolate the data to the new wavelengths"""
        old_wavelengths = self.old_wavelengths
        original_data = self.original_data

        number_points = old_wavelengths.size
        new_data = np.empty(new_wavelengths.size)
        new_errors = np.empty(new_wavelengths.size)

        start_index = 0

        index_0 = 0
        index_1 = 0
        inv_h = self.compute_h_matrix()
        self.second_derivatives = self.compute_second_derivative()

        for new_index, new_wave_position in enumerate(new_wavelengths):
            found_exact_match = False

            for index in range(start_index, number_points):
                if old_wavelengths[index] == new_wave_position:
                    interpolated_value = old_wavelengths[index]
                    propagated_error = self.original_errors[index]
                    found_exact_match = True
                    break
                try:
                    if old_wavelengths[index] < new_wave_position and old_wavelengths[index + 1] >= new_wave_position:
                        index_0 = index
                        index_1 = index + 1
                        # wavelengths are sorted; can start searching after the i
                        start_index = index_0
                        break
                except Exception as e:
                    print(new_wave_position, old_wavelengths[-1])
                    raise e

            if not found_exact_match:
                x_0 = old_wavelengths[index_0]
                x_1 = old_wavelengths[index_1]
                delta_x = x_1 - x_0
                A = (x_1 - new_wave_position) / delta_x
                B = (new_wave_position - x_0) / delta_x
                C = (1 / 6) * (A**3 - A) * (delta_x) ** 2
                D = (1 / 6) * (B**3 - B) * (delta_x) ** 2

                interpolated_value = (
                    A * original_data[index_0]
                    + B * original_data[index_1]
                    + C * self.second_derivatives[index_0]
                    + D * self.second_derivatives[index_1]
                )

                first = [A, B, C, D]

                U = self.build_U_matrix(index_0, index_0)
                second = np.transpose(first)
                propagated_error = np.matmul(np.matmul(first, U), second)

            new_data[new_index] = interpolated_value
            new_errors[new_index] = abs(propagated_error)

        return new_data, np.sqrt(new_errors)

old_wavelengths = old_wavelengths instance-attribute

original_data = original_data instance-attribute

original_errors = original_errors instance-attribute

data_size = original_data.size instance-attribute

ignore_covariances = ignore_covariances instance-attribute

cov_matrix = original_errors ** 2 instance-attribute

n_threads = n_threads instance-attribute

inv_delta_wave = 1 / np.diff(old_wavelengths) instance-attribute

cached_ones = np.ones((self.data_size - 2, self.data_size - 2), dtype=np.float64, order='C') instance-attribute

compute_h_matrix()

Create the H matrix.

Source code in src/ASTRA/utils/Cubic_spline.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def compute_h_matrix(self):
    """Create the H matrix."""
    if self._cached_h:
        return self._inv_h

    x = self.old_wavelengths

    diag_c = (np.roll(x, -2) - x)[0:-2] / 3
    # we remove the first element as we are interested in x_3 - x_2
    # the last one is removed for the same reason
    # the one before the last is removed as the second diagonal only goes up to N-1
    diag_r = (np.roll(x, -1) - x)[1:-2] / 6
    self._cached_h = True

    # avoid creating a new numpy array at each iteration!
    output = self.cached_ones
    output[:] = 1

    tridiagonal_inverter(diag_c, diag_r, output, n_threads=self.n_threads)
    # print("CYTHON vs Numpy EQUAL: ", np.allclose(np.linalg.inv(h), output))
    self._inv_h = output
    return self._inv_h

compute_partial(index_i)

Source code in src/ASTRA/utils/Cubic_spline.py 
 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
def compute_partial(self, index_i):
    try:
        return self._partials[index_i]
    except KeyError:
        pass

    if index_i in (0, self.data_size - 1):
        # both edges have y'' equal to zero
        partials = np.zeros(self.data_size)
        self._partials[index_i] = partials
        return partials

    partials = np.empty(self.data_size)

    # in the expressions the array starts at one, instead of zero !!
    inv_h_matrix = self.compute_h_matrix()

    # the two edges have a value of zero!
    partials[0] = 0
    partials[-1] = 0

    partial_derivative(inv_h_matrix, self.inv_delta_wave, index_i, partials, self.n_threads)

    # avoid underflows
    partials[np.where(np.abs(partials) < 1e-100)] = 0
    self._partials[index_i] = partials
    return partials

build_U_matrix(index_i_0, index_j_0)

Create the "U" matrix for the covariance estimation; Only computes the relevant quadrant.

Source code in src/ASTRA/utils/Cubic_spline.py 
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
def build_U_matrix(self, index_i_0, index_j_0):
    """Create the "U" matrix for the covariance estimation; Only computes the relevant quadrant."""
    cov_mat = self.cov_matrix

    i = index_i_0
    j = index_j_0

    if self.ignore_covariances:
        cov_i_i = cov_mat[i]
        cov_i_j1 = 0
        cov_i1_j = 0
        cov_j_j = cov_mat[i + 1]
    else:
        cov_i_i = cov_mat[i, j]
        cov_i_j1 = cov_mat[i, j + 1]
        cov_i1_j = cov_mat[i + 1, j]
        cov_j_j = cov_mat[i + 1, j + 1]

    # we are only evaluating the diagonal, i == j
    u_03 = self._U_entry("20", j + 1, i)
    u_01 = self._U_entry("20", j, i)
    u_13 = self._U_entry("20", j + 1, i + 1)

    u_23 = self._U_entry("22", i, j + 1)

    U = np.array(
        [
            [cov_i_i, cov_i_j1, u_01, u_03],
            [cov_i1_j, cov_j_j, u_03, u_13],
            [u_01, u_03, self._U_entry("22", i, j), u_23],
            [u_03, u_13, u_23, self._U_entry("22", i + 1, j + 1)],
        ],
    )

    return U

compute_second_derivative()

Source code in src/ASTRA/utils/Cubic_spline.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def compute_second_derivative(self):
    old_wavelengths = self.old_wavelengths
    original_data = self.original_data
    inv_h_matrix = self.compute_h_matrix()

    second_deriv_vals = np.zeros(original_data.shape, dtype=np.float64)

    second_derivative(
        np.diff(old_wavelengths),
        np.diff(original_data),
        inv_h_matrix,
        second_deriv_vals,
        self.n_threads,
    )

    return second_deriv_vals

interpolate(new_wavelengths)

Interpolate the data to the new wavelengths

Source code in src/ASTRA/utils/Cubic_spline.py 
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
def interpolate(self, new_wavelengths):
    """Interpolate the data to the new wavelengths"""
    old_wavelengths = self.old_wavelengths
    original_data = self.original_data

    number_points = old_wavelengths.size
    new_data = np.empty(new_wavelengths.size)
    new_errors = np.empty(new_wavelengths.size)

    start_index = 0

    index_0 = 0
    index_1 = 0
    inv_h = self.compute_h_matrix()
    self.second_derivatives = self.compute_second_derivative()

    for new_index, new_wave_position in enumerate(new_wavelengths):
        found_exact_match = False

        for index in range(start_index, number_points):
            if old_wavelengths[index] == new_wave_position:
                interpolated_value = old_wavelengths[index]
                propagated_error = self.original_errors[index]
                found_exact_match = True
                break
            try:
                if old_wavelengths[index] < new_wave_position and old_wavelengths[index + 1] >= new_wave_position:
                    index_0 = index
                    index_1 = index + 1
                    # wavelengths are sorted; can start searching after the i
                    start_index = index_0
                    break
            except Exception as e:
                print(new_wave_position, old_wavelengths[-1])
                raise e

        if not found_exact_match:
            x_0 = old_wavelengths[index_0]
            x_1 = old_wavelengths[index_1]
            delta_x = x_1 - x_0
            A = (x_1 - new_wave_position) / delta_x
            B = (new_wave_position - x_0) / delta_x
            C = (1 / 6) * (A**3 - A) * (delta_x) ** 2
            D = (1 / 6) * (B**3 - B) * (delta_x) ** 2

            interpolated_value = (
                A * original_data[index_0]
                + B * original_data[index_1]
                + C * self.second_derivatives[index_0]
                + D * self.second_derivatives[index_1]
            )

            first = [A, B, C, D]

            U = self.build_U_matrix(index_0, index_0)
            second = np.transpose(first)
            propagated_error = np.matmul(np.matmul(first, U), second)

        new_data[new_index] = interpolated_value
        new_errors[new_index] = abs(propagated_error)

    return new_data, np.sqrt(new_errors)

UserConfigs

ASTRA configurations.

This defines the main configuration scheme of ASTRA:

1) Userparam, which defines the individual configurations 2) InternalParameters, which are used by ASTRA objects 3) DefaultValues, which agregates all possible configurations that are available to the users

UserParam

Define an individual configuration that the user can specify.

Source code in src/ASTRA/utils/UserConfigs.py 
 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
class UserParam:
    """Define an individual configuration that the user can specify."""

    __slots__ = (
        "_valueConstraint",
        "_default_value",
        "_mandatory",
        "quiet",
        "description",
    )

    def __init__(
        self,
        default_value: Optional[Any] = None,
        constraint: Optional[Constraint] = None,
        mandatory: bool = False,
        quiet: bool = False,
        description: Optional[str] = None,
    ) -> None:
        """Instantiate a user parameter and assoctied constraints and values.

        Args:
            default_value (Optional[Any], optional): Default value of the parameter. Defaults to None.
            constraint (Optional[Constraint], optional): Constraint that is used to
                validate the user input. Defaults to None.
            mandatory (bool, optional): If True, the configuration must always be provided by the
                user. Defaults to False.
            quiet (bool, optional): Avoid logs. Defaults to False.
            description (Optional[str], optional): Description of the parameter, to be printed
                to terminal when requested. Defaults to None.

        """
        self._valueConstraint = constraint if constraint is not None else Constraint("")
        self._default_value = default_value
        self._mandatory = mandatory
        self.quiet = quiet
        self.description = description

    def apply_constraints_to_value(self, param_name: str, value: Any) -> None:
        """Apply the constraints of this parameter to a given value.

        Args:
            param_name (str): Name of the parameter, used in the logs
            value (Any): Value that the user provided

        """
        self._valueConstraint.check_if_value_meets_constraint(value)

    @property
    def existing_constraints(self) -> Constraint:
        """Retrieve existing constraints of parameter."""
        return self._valueConstraint

    @property
    def is_mandatory(self) -> bool:
        """True if the parameter is mandatory."""
        return self._mandatory

    @property
    def quiet_output(self) -> bool:
        """True if the parameter should not log to terminal."""
        return self.quiet

    @property
    def default_value(self) -> Any:
        """Return the default value of the parameter.

        Raises
        ------
        InvalidConfiguration: if the parameter is mandatory, there is no default

        """
        if self.is_mandatory:
            raise InvalidConfiguration("Trying to use default value of a mandatory parameter")

        self.apply_constraints_to_value("default_value", self._default_value)
        return self._default_value

    def __repr__(self) -> str:  # noqa: D105
        return (
            "\n".join(
                (
                    f" Mandatory Flag: {self._mandatory}",
                    f"Default Value: {self._default_value}",
                    f"Constraints: {self._valueConstraint}",
                )
            )
            + "\n"
        )

    def get_terminal_output(self, indent_level: int = 1) -> str:
        """Generate terminal-formatted text from this UserParam.

        Args:
            indent_level (int, optional): How many tabs to add at the start of each line. Defaults to 1.

        Returns:
            str: Formatted message with the Description, mandatory status, default value and constraints

        """
        offset = indent_level * "\t"
        message = ""
        for name, value in [
            ("Description", self.description),
            ("Mandatory", self._mandatory),
            ("Default value", self._default_value),
            ("Constraints", self._valueConstraint),
        ]:
            message += offset + f"{name}:: {value}\n"
        return message

quiet = quiet instance-attribute

description = description instance-attribute

existing_constraints: Constraint property

Retrieve existing constraints of parameter.

is_mandatory: bool property

True if the parameter is mandatory.

quiet_output: bool property

True if the parameter should not log to terminal.

default_value: Any property

Return the default value of the parameter.

Raises:

Type Description
InvalidConfiguration: if the parameter is mandatory, there is no default

apply_constraints_to_value(param_name: str, value: Any) -> None

Apply the constraints of this parameter to a given value.

Args: param_name (str): Name of the parameter, used in the logs value (Any): Value that the user provided

Source code in src/ASTRA/utils/UserConfigs.py 
59
60
61
62
63
64
65
66
67
def apply_constraints_to_value(self, param_name: str, value: Any) -> None:
    """Apply the constraints of this parameter to a given value.

    Args:
        param_name (str): Name of the parameter, used in the logs
        value (Any): Value that the user provided

    """
    self._valueConstraint.check_if_value_meets_constraint(value)

get_terminal_output(indent_level: int = 1) -> str

Generate terminal-formatted text from this UserParam.

Args: indent_level (int, optional): How many tabs to add at the start of each line. Defaults to 1.

Returns: str: Formatted message with the Description, mandatory status, default value and constraints

Source code in src/ASTRA/utils/UserConfigs.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def get_terminal_output(self, indent_level: int = 1) -> str:
    """Generate terminal-formatted text from this UserParam.

    Args:
        indent_level (int, optional): How many tabs to add at the start of each line. Defaults to 1.

    Returns:
        str: Formatted message with the Description, mandatory status, default value and constraints

    """
    offset = indent_level * "\t"
    message = ""
    for name, value in [
        ("Description", self.description),
        ("Mandatory", self._mandatory),
        ("Default value", self._default_value),
        ("Constraints", self._valueConstraint),
    ]:
        message += offset + f"{name}:: {value}\n"
    return message

InternalParameters

Contain all configurations from a given ASTRA object.

This class will store all configurations (UserParam) from a given ASTRA object, as well as their current value. When updating a given parameter, the QC checks that are imposed by the UserParam object.

Source code in src/ASTRA/utils/UserConfigs.py 
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
class InternalParameters:
    """Contain all configurations from a given ASTRA object.

    This class will store all configurations (*UserParam*) from a given ASTRA object, as well as
    their current value. When updating a given parameter, the QC checks that are imposed
    by the UserParam object.
    """

    __slots__ = ("_default_params", "_user_configs", "_name_of_parent", "no_logs")

    def __init__(
        self,
        name_of_parent: str,
        default_params: DefaultValues,
        no_logs: bool = False,
    ):
        """Create new object.

        Args:
            name_of_parent (str): Name of the config
            default_params (Dict[str, UserParam]): Dict with text name of parameters
                and a UserParam object that describes it.
            no_logs (bool, optional): Avoid terminal logs. Defaults to False.

        """
        self._default_params = default_params
        self._user_configs: dict[str, Any] = {}
        self._name_of_parent = name_of_parent
        self.no_logs = no_logs

    def update_configs_with_values(self, user_configs: dict[str, Any]) -> None:
        """Update the parameters with a configuration dictionary.

        Args:
            user_configs (dict[str, Any]): Dictionary with user parameters.

        Raises:
            InternalError: If the provided value does not meet the imposed restrictions.

        """
        for key, value in user_configs.items():
            try:
                parameter_def_information = self._default_params[key]
            except KeyError:
                if not self.no_logs:
                    # The only object that will have this enabled are the Frames
                    # And we shall call one of the Frames with the User-Param logs enabled!
                    logger.warning(
                        "{} received a configuration flag that is not recognized: {}",
                        self._name_of_parent,
                        key,
                    )
                continue

            try:
                parameter_def_information.apply_constraints_to_value(key, value)
            except InvalidConfiguration as exc:
                logger.critical("User-given parameter {} does not meet the constraints", key)
                raise InternalError from exc

            self._user_configs[key] = value

            if not self.no_logs:
                if not self._default_params[key].quiet_output:
                    logger.debug("Configuration <{}> taking the value: {}", key, value)
                else:
                    logger.debug("Configuration <{}> was updated")

    def receive_user_inputs(self, user_configs: dict[str, Any] = None) -> None:
        """Parse the config dictionary from the user.

        Args:
            user_configs (Optional[Dict[str, Any]], optional): Dictionary where the keys are the parameter names, and
                the values are the new values of that parameter. The values will be checked by the validation
                layer defined in the corresponding UserParam. Defaults to None.

        Raises:
            InvalidConfiguration: If a given parameter is mandatory and it was not provided by the user
            InternalError: If the validation layer does not agree with the provided value.

        """
        if not self.no_logs:
            logger.debug("Generating internal configs of {}", self._name_of_parent)

        self.update_configs_with_values(user_configs)

        if not self.no_logs:
            logger.info("Checking for any parameter that will take default value")
        for key, default_param in self._default_params.items():
            if key not in self._user_configs:
                if default_param.is_mandatory:
                    raise InvalidConfiguration(f"SBART parameter <{key}> is mandatory.")

                if not self.no_logs:
                    logger.debug(
                        "Configuration <{}> using the default value: {}",
                        key,
                        default_param.default_value,
                    )
                try:
                    self._user_configs[key] = default_param.default_value
                except Exception as e:
                    logger.critical("Error in the generation of parameter: {}", key)
                    raise InternalError from e

    def text_pretty_description(self, indent_level: int) -> str:
        """Generate textual description of internal parameters."""
        file_offset = indent_level * "\t"
        to_write = ""

        to_write += f"\n{file_offset}Parameters in use:"
        for key, value in self._user_configs.items():
            to_write += f"\n{file_offset}\t{key} -> {value}"
        return to_write

    def __setitem__(self, key: str, value: Any) -> None:  # noqa: D105
        logger.warning(f"Internal configs are being updated in real time ({key=})")
        try:
            parameter_def_information = self._default_params[key]
        except KeyError:
            if not self.no_logs:
                # The only object that will have this enabled are the Frames
                # And we shall call one of the Frames with the User-Param logs enabled!
                logger.warning(
                    "{} received a configuration flag that is not recognized: {}",
                    self._name_of_parent,
                    key,
                )
        try:
            parameter_def_information.apply_constraints_to_value(key, value)
        except InvalidConfiguration as exc:
            logger.critical("User-given parameter {} does not meet the constraints", key)
            raise InternalError from exc
        self._user_configs[key] = value

    def __getitem__(self, item: str):  # noqa: D105
        try:
            return self._user_configs[item]
        except KeyError:
            msg = f"<{item}> is not a valid parameter of {self._name_of_parent}"
            logger.critical(msg)
            raise Exception(msg)

    def get_user_configs(self) -> Dict[str, Any]:
        """Get all user configs."""
        return self._user_configs

    def items(self):
        """Get the values."""
        return self._user_configs.items()

no_logs = no_logs instance-attribute

update_configs_with_values(user_configs: dict[str, Any]) -> None

Update the parameters with a configuration dictionary.

Args: user_configs (dict[str, Any]): Dictionary with user parameters.

Raises: InternalError: If the provided value does not meet the imposed restrictions.

Source code in src/ASTRA/utils/UserConfigs.py 
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
def update_configs_with_values(self, user_configs: dict[str, Any]) -> None:
    """Update the parameters with a configuration dictionary.

    Args:
        user_configs (dict[str, Any]): Dictionary with user parameters.

    Raises:
        InternalError: If the provided value does not meet the imposed restrictions.

    """
    for key, value in user_configs.items():
        try:
            parameter_def_information = self._default_params[key]
        except KeyError:
            if not self.no_logs:
                # The only object that will have this enabled are the Frames
                # And we shall call one of the Frames with the User-Param logs enabled!
                logger.warning(
                    "{} received a configuration flag that is not recognized: {}",
                    self._name_of_parent,
                    key,
                )
            continue

        try:
            parameter_def_information.apply_constraints_to_value(key, value)
        except InvalidConfiguration as exc:
            logger.critical("User-given parameter {} does not meet the constraints", key)
            raise InternalError from exc

        self._user_configs[key] = value

        if not self.no_logs:
            if not self._default_params[key].quiet_output:
                logger.debug("Configuration <{}> taking the value: {}", key, value)
            else:
                logger.debug("Configuration <{}> was updated")

receive_user_inputs(user_configs: dict[str, Any] = None) -> None

Parse the config dictionary from the user.

Args: user_configs (Optional[Dict[str, Any]], optional): Dictionary where the keys are the parameter names, and the values are the new values of that parameter. The values will be checked by the validation layer defined in the corresponding UserParam. Defaults to None.

Raises: InvalidConfiguration: If a given parameter is mandatory and it was not provided by the user InternalError: If the validation layer does not agree with the provided value.

Source code in src/ASTRA/utils/UserConfigs.py 
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
def receive_user_inputs(self, user_configs: dict[str, Any] = None) -> None:
    """Parse the config dictionary from the user.

    Args:
        user_configs (Optional[Dict[str, Any]], optional): Dictionary where the keys are the parameter names, and
            the values are the new values of that parameter. The values will be checked by the validation
            layer defined in the corresponding UserParam. Defaults to None.

    Raises:
        InvalidConfiguration: If a given parameter is mandatory and it was not provided by the user
        InternalError: If the validation layer does not agree with the provided value.

    """
    if not self.no_logs:
        logger.debug("Generating internal configs of {}", self._name_of_parent)

    self.update_configs_with_values(user_configs)

    if not self.no_logs:
        logger.info("Checking for any parameter that will take default value")
    for key, default_param in self._default_params.items():
        if key not in self._user_configs:
            if default_param.is_mandatory:
                raise InvalidConfiguration(f"SBART parameter <{key}> is mandatory.")

            if not self.no_logs:
                logger.debug(
                    "Configuration <{}> using the default value: {}",
                    key,
                    default_param.default_value,
                )
            try:
                self._user_configs[key] = default_param.default_value
            except Exception as e:
                logger.critical("Error in the generation of parameter: {}", key)
                raise InternalError from e

text_pretty_description(indent_level: int) -> str

Generate textual description of internal parameters.

Source code in src/ASTRA/utils/UserConfigs.py 
238
239
240
241
242
243
244
245
246
def text_pretty_description(self, indent_level: int) -> str:
    """Generate textual description of internal parameters."""
    file_offset = indent_level * "\t"
    to_write = ""

    to_write += f"\n{file_offset}Parameters in use:"
    for key, value in self._user_configs.items():
        to_write += f"\n{file_offset}\t{key} -> {value}"
    return to_write

get_user_configs() -> Dict[str, Any]

Get all user configs.

Source code in src/ASTRA/utils/UserConfigs.py 
276
277
278
def get_user_configs(self) -> Dict[str, Any]:
    """Get all user configs."""
    return self._user_configs

items()

Get the values.

Source code in src/ASTRA/utils/UserConfigs.py 
280
281
282
def items(self):
    """Get the values."""
    return self._user_configs.items()

DefaultValues

Defines all of the user parameters that SBART has available for any given object.

We can sum two DefaultValues objects to expand the possible configurable parameters of the ASTRA object.

Source code in src/ASTRA/utils/UserConfigs.py 
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
class DefaultValues:
    """Defines all of the user parameters that SBART has available for any given object.

    We can sum two DefaultValues objects to expand the possible configurable parameters
    of the ASTRA object.
    """

    def __init__(self, **kwargs: UserParam) -> None:
        """Map of str to UserParam to describe configurations."""
        self.default_mapping: dict[str, UserParam] = kwargs

    def update(self, item: str, new_value: UserParam) -> None:
        """Update the default value of a stored parameter, if it exists.

        Args:
            item (str): Param name
            new_value (UserParam): New configuration

        Raises:
            Exception: If the parameter that we want to update does not exist

        """
        if item not in self.default_mapping:
            raise Exception
        self.default_mapping[item] = new_value

    def __add__(self, other: DefaultValues) -> DefaultValues:  # noqa: D105
        new_default_mapping = {**self.default_mapping, **other.default_mapping}
        return DefaultValues(**new_default_mapping)

    def __radd__(self, other: DefaultValues) -> DefaultValues:  # noqa: D105
        return self.__add__(other)

    def __getitem__(self, item: str) -> Any:  # noqa: D105
        return self.default_mapping[item]

    def __str__(self) -> str:  # noqa: D105
        return self.__repr__()

    def __repr__(self) -> str:  # noqa: D105
        representation = "Configurations:\n\n"

        for key, value in self.default_mapping.items():
            representation += f"Name:: {key}\n{value.get_terminal_output()} \n"

        return representation

    ### Map the inside dict properties to the outside!
    def items(self):  # noqa: D102
        return self.default_mapping.items()

    def keys(self) -> Iterable[str]:  # noqa: D102
        return self.default_mapping.keys()

    def values(self) -> Iterable[UserParam]:  # noqa: D102
        return self.default_mapping.values()

    def __iter__(self) -> NoReturn:  # noqa: D105
        raise TypeError("no!")

default_mapping: dict[str, UserParam] = kwargs instance-attribute

update(item: str, new_value: UserParam) -> None

Update the default value of a stored parameter, if it exists.

Args: item (str): Param name new_value (UserParam): New configuration

Raises: Exception: If the parameter that we want to update does not exist

Source code in src/ASTRA/utils/UserConfigs.py 
296
297
298
299
300
301
302
303
304
305
306
307
308
309
def update(self, item: str, new_value: UserParam) -> None:
    """Update the default value of a stored parameter, if it exists.

    Args:
        item (str): Param name
        new_value (UserParam): New configuration

    Raises:
        Exception: If the parameter that we want to update does not exist

    """
    if item not in self.default_mapping:
        raise Exception
    self.default_mapping[item] = new_value

items()

Source code in src/ASTRA/utils/UserConfigs.py 
333
334
def items(self):  # noqa: D102
    return self.default_mapping.items()

keys() -> Iterable[str]

Source code in src/ASTRA/utils/UserConfigs.py 
336
337
def keys(self) -> Iterable[str]:  # noqa: D102
    return self.default_mapping.keys()

values() -> Iterable[UserParam]

Source code in src/ASTRA/utils/UserConfigs.py 
339
340
def values(self) -> Iterable[UserParam]:  # noqa: D102
    return self.default_mapping.values()

air_to_vac

airtovac(wave_air)

Adapted from idl astrolib (https://idlastro.gsfc.nasa.gov/ftp/pro/astro/airtovac.pro).

;+ ; NAME: ; AIRTOVAC ; PURPOSE: ; Convert air wavelengths to vacuum wavelengths ; EXPLANATION: ; Wavelengths are corrected for the index of refraction of air under ; standard conditions. Wavelength values below 2000 A will not be ; altered. Uses relation of Ciddor (1996). ; ; CALLING SEQUENCE: ; AIRTOVAC, WAVE_AIR, [ WAVE_VAC] ; ; INPUT/OUTPUT: ; WAVE_AIR - Wavelength in Angstroms, scalar or vector ; If this is the only parameter supplied, it will be updated on ; output to contain double precision vacuum wavelength(s). ; OPTIONAL OUTPUT: ; WAVE_VAC - Vacuum wavelength in Angstroms, same number of elements as ; WAVE_AIR, double precision ; ; EXAMPLE: ; If the air wavelength is W = 6056.125 (a Krypton line), then ; AIRTOVAC, W yields an vacuum wavelength of W = 6057.8019 ; ; METHOD: ; Formula from Ciddor 1996, Applied Optics 62, 958 ; ; NOTES: ; Take care within 1 A of 2000 A. Wavelengths below 2000 A in air are ; not altered. ; REVISION HISTORY ; Written W. Landsman November 1991 ; Use Ciddor (1996) formula for better accuracy in the infrared ; Added optional output vector, W Landsman Mar 2011 ; Iterate for better precision W.L./D. Schlegel Mar 2011 ;-

Source code in src/ASTRA/utils/air_to_vac.py 
 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
def airtovac(wave_air):
    """Adapted from idl astrolib (https://idlastro.gsfc.nasa.gov/ftp/pro/astro/airtovac.pro).

    ;+
    ; NAME:
    ;       AIRTOVAC
    ; PURPOSE:
    ;       Convert air wavelengths to vacuum wavelengths
    ; EXPLANATION:
    ;       Wavelengths are corrected for the index of refraction of air under
    ;       standard conditions.  Wavelength values below 2000 A will not be
    ;       altered.  Uses relation of Ciddor (1996).
    ;
    ; CALLING SEQUENCE:
    ;       AIRTOVAC, WAVE_AIR, [ WAVE_VAC]
    ;
    ; INPUT/OUTPUT:
    ;       WAVE_AIR - Wavelength in Angstroms, scalar or vector
    ;               If this is the only parameter supplied, it will be updated on
    ;               output to contain double precision vacuum wavelength(s).
    ; OPTIONAL OUTPUT:
    ;        WAVE_VAC - Vacuum wavelength in Angstroms, same number of elements as
    ;                 WAVE_AIR, double precision
    ;
    ; EXAMPLE:
    ;       If the air wavelength is  W = 6056.125 (a Krypton line), then
    ;       AIRTOVAC, W yields an vacuum wavelength of W = 6057.8019
    ;
    ; METHOD:
    ;	Formula from Ciddor 1996, Applied Optics 62, 958
    ;
    ; NOTES:
    ;       Take care within 1 A of 2000 A.   Wavelengths below 2000 A *in air* are
    ;       not altered.
    ; REVISION HISTORY
    ;       Written W. Landsman                November 1991
    ;       Use Ciddor (1996) formula for better accuracy in the infrared
    ;           Added optional output vector, W Landsman Mar 2011
    ;       Iterate for better precision W.L./D. Schlegel  Mar 2011
    ;-
    """
    wave_vac = wave_air * 1.0
    g = wave_vac > 2000  # Only modify above 2000 A

    if np.all(g):
        for _ in range(2):
            sigma2 = np.divide(1e4, wave_vac) ** 2.0  # Convert to wavenumber squared
            # Compute conversion factor
            fact = 1.0 + 5.792105e-2 / (238.0185 - sigma2) + 1.67917e-3 / (57.362 - sigma2)
            wave_vac = wave_air * fact  # Convert Wavelength

    return wave_vac

choices

Defines that availabe choices of ASTRA.

WORKING_MODE

Bases: Enum

Enumerator to represent the working mode of the ASRA object.

ONE_SHOT - All observations are processed at the same time

Source code in src/ASTRA/utils/choices.py 
 6
 7
 8
 9
10
11
12
13
class WORKING_MODE(Enum):
    """Enumerator to represent the working mode of the ASRA object.

    ONE_SHOT - All observations are processed at the same time

    """

    ONE_SHOT = "ONE_SHOT"

ONE_SHOT = 'ONE_SHOT' class-attribute instance-attribute

DISK_SAVE_MODE

Bases: Enum

Enumerator to represent the DISK save mode of a given ASTRA object.

Source code in src/ASTRA/utils/choices.py 
16
17
18
19
20
21
22
23
class DISK_SAVE_MODE(Enum):
    """Enumerator to represent the DISK save mode of a given ASTRA object."""

    DISABLED = 1

    BASIC = 2

    EXTREME = 3

DISABLED = 1 class-attribute instance-attribute

BASIC = 2 class-attribute instance-attribute

EXTREME = 3 class-attribute instance-attribute

TELLURIC_EXTENSION

Bases: Enum

Method for the extension of telluric template.

Source code in src/ASTRA/utils/choices.py 
26
27
28
29
30
class TELLURIC_EXTENSION(Enum):
    """Method for the extension of telluric template."""

    LINES = "LINES"
    WINDOW = "WINDOW"

LINES = 'LINES' class-attribute instance-attribute

WINDOW = 'WINDOW' class-attribute instance-attribute

TELLURIC_CREATION_MODE

Bases: Enum

Method for the construction of telluric template.

Source code in src/ASTRA/utils/choices.py 
33
34
35
36
37
38
class TELLURIC_CREATION_MODE(Enum):
    """Method for the construction of telluric template."""

    tapas = "tapas"
    telfit = "telfit"
    OHemission = "OHemission"

tapas = 'tapas' class-attribute instance-attribute

telfit = 'telfit' class-attribute instance-attribute

OHemission = 'OHemission' class-attribute instance-attribute

STELLAR_CREATION_MODE

Bases: Enum

Method for the construction of stellar template.

Source code in src/ASTRA/utils/choices.py 
41
42
43
44
45
46
47
class STELLAR_CREATION_MODE(Enum):
    """Method for the construction of stellar template."""

    Sum = "Sum"
    OBSERVATION = "OBSERVATION"
    Median = "Median"
    PHOENIX = "PHOENIX"

Sum = 'Sum' class-attribute instance-attribute

OBSERVATION = 'OBSERVATION' class-attribute instance-attribute

Median = 'Median' class-attribute instance-attribute

PHOENIX = 'PHOENIX' class-attribute instance-attribute

TELLURIC_APPLICATION_MODE

Bases: Enum

Approach to handle telluric features.

Source code in src/ASTRA/utils/choices.py 
50
51
52
53
54
class TELLURIC_APPLICATION_MODE(Enum):
    """Approach to handle telluric features."""

    removal = "removal"
    correction = "correction"

removal = 'removal' class-attribute instance-attribute

correction = 'correction' class-attribute instance-attribute

SPECTRA_INTERPOL_MODE

Bases: Enum

Enumerator to represent the DISK save mode of a given ASTRA object.

Source code in src/ASTRA/utils/choices.py 
57
58
59
60
61
class SPECTRA_INTERPOL_MODE(Enum):
    """Enumerator to represent the DISK save mode of a given ASTRA object."""

    SPLINES = 1
    GP = 2

SPLINES = 1 class-attribute instance-attribute

GP = 2 class-attribute instance-attribute

SPLINE_INTERPOL_MODE

Bases: Enum

Enumerator to represent the interpolation mode of a given ASTRA object.

Possible values: CUBIC_SPLINE QUADRATIC_SPLINE PCHIP NEAREST RBF SMOOTH_CUBIC_SPLINE AKIMA BARYCENTRIC_INTERPOL

Source code in src/ASTRA/utils/choices.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
class SPLINE_INTERPOL_MODE(Enum):
    """Enumerator to represent the interpolation mode of a given ASTRA object.

    Possible values:
        CUBIC_SPLINE
        QUADRATIC_SPLINE
        PCHIP
        NEAREST
        RBF
        SMOOTH_CUBIC_SPLINE
        AKIMA
        BARYCENTRIC_INTERPOL
    """

    CUBIC_SPLINE = 1
    QUADRATIC_SPLINE = 2
    PCHIP = 3
    NEAREST = 4
    RBF = 5
    SMOOTH_CUBIC_SPLINE = 6
    AKIMA = 7
    BARYCENTRIC_INTERPOL = 8

CUBIC_SPLINE = 1 class-attribute instance-attribute

QUADRATIC_SPLINE = 2 class-attribute instance-attribute

PCHIP = 3 class-attribute instance-attribute

NEAREST = 4 class-attribute instance-attribute

RBF = 5 class-attribute instance-attribute

SMOOTH_CUBIC_SPLINE = 6 class-attribute instance-attribute

AKIMA = 7 class-attribute instance-attribute

BARYCENTRIC_INTERPOL = 8 class-attribute instance-attribute

FLUX_SMOOTH_CONFIGS

Bases: Enum

Enumerator to flux smoothing applied at template construction.

Source code in src/ASTRA/utils/choices.py 
88
89
90
91
92
class FLUX_SMOOTH_CONFIGS(Enum):
    """Enumerator to flux smoothing applied at template construction."""

    NONE = 1
    SAVGOL = 2

NONE = 1 class-attribute instance-attribute

SAVGOL = 2 class-attribute instance-attribute

FLUX_SMOOTH_ORDER

Bases: Enum

The order in which the smooth is applied.

Source code in src/ASTRA/utils/choices.py 
 95
 96
 97
 98
 99
100
class FLUX_SMOOTH_ORDER(Enum):
    """The order in which the smooth is applied."""

    BEFORE = 1
    AFTER = 2
    BOTH = 3

BEFORE = 1 class-attribute instance-attribute

AFTER = 2 class-attribute instance-attribute

BOTH = 3 class-attribute instance-attribute

INTERPOLATION_ERR_PROP

Bases: Enum

how should we propagate uncertainties.

Source code in src/ASTRA/utils/choices.py 
103
104
105
106
107
108
class INTERPOLATION_ERR_PROP(Enum):
    """how should we propagate uncertainties."""

    interpolation = 1
    propagation = 2
    none = 3

interpolation = 1 class-attribute instance-attribute

propagation = 2 class-attribute instance-attribute

none = 3 class-attribute instance-attribute

NORMALIZATION_SOURCES

Bases: Enum

how should we propagate uncertainties.

Source code in src/ASTRA/utils/choices.py 
111
112
113
114
115
116
class NORMALIZATION_SOURCES(Enum):
    """how should we propagate uncertainties."""

    RASSINE = "RASSINE"
    SNT = "SNT"
    POLY_FIT = "POLY_FIT"

RASSINE = 'RASSINE' class-attribute instance-attribute

SNT = 'SNT' class-attribute instance-attribute

POLY_FIT = 'POLY_FIT' class-attribute instance-attribute

compute_berv_and_herv_poet

Compute BERV (Barycentric-Earth RV) and HERV (HEliocentric RV) correction.

BARYCORR_AVAILABLE = True module-attribute

compute_berv_and_herv_POET(header) -> tuple[RV_measurement, RV_measurement]

Compute barycentric and heliocentric correction for Paranal Observatory

Source code in src/ASTRA/utils/compute_berv_and_herv_poet.py 
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
def compute_berv_and_herv_POET(header) -> tuple[RV_measurement, RV_measurement]:
    """Compute barycentric and heliocentric correction for Paranal Observatory"""
    if not BARYCORR_AVAILABLE:
        raise custom_exceptions.InternalError(
            "Barycorrpy is not available. Please install it to compute BERV and HERV."
        )

    obsname = "Paranal Observatory"
    # Loop observations
    date_obs = header["DATE-OBS"]
    exp_time = header["EXPTIME"]
    photocen = header["HIERARCH ESO QC TMMEAN USED"]
    time_jdb = header["HIERARCH ESO QC BJD"]

    # Compute gravitational redshift of the Sun
    gr_sun = (ac.G.value * ac.M_sun.value) / (ac.R_sun.value * ac.c.value)

    # Compute BERV and HERV
    jd_utc_cen = Time(date_obs, format="isot").jd + exp_time * photocen / (60 * 60 * 24)
    berv_val = barycorrpy.get_BC_vel(
        JDUTC=jd_utc_cen, obsname=obsname, SolSystemTarget="Sun"
    )[0][0]

    herv_val = (
        barycorrpy.get_BC_vel(
            JDUTC=jd_utc_cen,
            obsname=obsname,
            SolSystemTarget="Sun",
            predictive=True,
        )[0][0]
        * (-1)
        - berv_val
        + gr_sun
    )

    # Convert from m/s to km/s
    berv_val *= 1e-3
    herv_val *= 1e-3

    if not np.isfinite((berv_val + herv_val)):
        raise custom_exceptions.InternalError(
            f"Got non-finite values for the BERV calculation ({berv_val=}; {herv_val=})"
        )

    return berv_val * kilometer_second, herv_val * kilometer_second

concurrent_tools

close_interfaces

close_buffers(shared_buffers)

Source code in src/ASTRA/utils/concurrent_tools/close_interfaces.py 
4
5
6
def close_buffers(shared_buffers):
    for buffer in shared_buffers:
        buffer.close()

kill_workers(shared_buffers, in_queue, number_workers)

Source code in src/ASTRA/utils/concurrent_tools/close_interfaces.py 
 9
10
11
12
13
14
15
16
17
def kill_workers(shared_buffers, in_queue, number_workers):
    close_buffers(shared_buffers)

    print("Emptying the working queue and shutting down other workers")
    while not in_queue.empty():
        in_queue.get()  # as docs say: Remove and return an item from the queue.

    for _ in range(number_workers):
        in_queue.put(np.inf)

create_shared_arr

create_shared_array(data: np.ndarray) -> tuple[list, np.ndarray]

Create a numpy array using a shared memory block as the buffer.

Source code in src/ASTRA/utils/concurrent_tools/create_shared_arr.py 
 8
 9
10
11
12
13
14
15
16
17
18
def create_shared_array(data: np.ndarray) -> tuple[list, np.ndarray]:
    """Create a numpy array using a shared memory block as the buffer."""
    array_size = data.shape
    buffer_info = [
        shared_memory.SharedMemory(create=True, size=data.nbytes),
        array_size,
        data.dtype,
    ]
    shared_data = np.ndarray(array_size, dtype=data.dtype, buffer=buffer_info[0].buf)
    shared_data[:] = data[:]
    return buffer_info, shared_data

evaluate_worker_shutdown

evaluate_shutdown(queue: Queue) -> Tuple[int, int]

Evaluate the shutdown messages sent by the workers, counting the number of good and bad shutdowns

Parameters:

Name Type Description Default
queue Queue

Queue in which the workers place their

 required

Returns:

Type Description
Tuple[int, int]

Goog, bad shutdowns
Source code in src/ASTRA/utils/concurrent_tools/evaluate_worker_shutdown.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
def evaluate_shutdown(queue: Queue) -> Tuple[int, int]:
    """Evaluate the shutdown messages sent by the workers, counting the number of good and bad shutdowns

    Parameters
    ----------
    queue : Queue
        Queue in which the workers place their

    Returns
    -------
    Tuple[int, int]
        Goog, bad shutdowns

    """
    good_shutdown = 0
    bad_shutdown = 0
    while not queue.empty():
        pkg = queue.get()
        if pkg == WORKER_ERROR:
            bad_shutdown += 1
        else:
            good_shutdown += 1
    return good_shutdown, bad_shutdown

open_buffers

open_buffer(buffer_info: dict, open_type: str, buffers: list)

Source code in src/ASTRA/utils/concurrent_tools/open_buffers.py 
 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
def open_buffer(buffer_info: dict, open_type: str, buffers: list):
    if open_type == "template":
        open_order = [
            "template",
            "template_errors",
            "template_wavelength",
            "template_counts",
        ]
    elif open_type == "BayesianCache":
        open_order = ["mask_cache", "cached_orders"]
    else:
        raise Exception(f"{open_type=} does not exist")

    data = []

    try:
        for array_name in open_order:
            shm = shared_memory.SharedMemory(name=buffer_info[array_name][0].name)
            data_array = ndarray(
                buffer_info[array_name][1],
                dtype=buffer_info[array_name][2],
                buffer=shm.buf,
            )
            buffers.append(shm)
            data.append(data_array)

        return *data, buffers

    except Exception as e:
        print(f"Opening buffers failed in  {array_name=}  due to {e}")
        print("Available buffers: ", buffer_info.keys(), "worker_outputs" in buffer_info)
        for buff in buffers:
            buff.close()
        raise Exception

proxyObjects

DataClassManager

Bases: BaseManager

This will be the server that manages the shared objects

Parameters:

Name Type Description Default
BaseManager [type]

[description]

 required
Source code in src/ASTRA/utils/concurrent_tools/proxyObjects.py 
 4
 5
 6
 7
 8
 9
10
11
12
class DataClassManager(BaseManager):
    """This will be the server that manages the shared objects

    Parameters
    ----------
    BaseManager : [type]
        [description]

    """

proxy_object

DataClassManager

Bases: BaseManager

This will be the server that manages the shared objects

Parameters:

Name Type Description Default
BaseManager [type]

[description]

 required
Source code in src/ASTRA/utils/concurrent_tools/proxy_object.py 
 4
 5
 6
 7
 8
 9
10
11
12
class DataClassManager(BaseManager):
    """This will be the server that manages the shared objects

    Parameters
    ----------
    BaseManager : [type]
        [description]

    """

create_logger

Setup the logger, handling terminal and disk output.

astra_logger = logger.bind(module='ASTRA') module-attribute

setup_ASTRA_logger(storage_path=None, level='INFO', log_to_terminal=True, write_to_file=True, append_to_file=True)

Configure ASTRA's logger dynamically.

Parameters:

Name Type Description Default
log_path str or Path or None

Path to the log file. If None, file logging is skipped.

 required
level str

Minimum level to log ("DEBUG", "INFO", "WARNING", etc.)

 'INFO'
console bool

Whether to also log to stdout.

 required
Source code in src/ASTRA/utils/create_logger.py 
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
def setup_ASTRA_logger(
    storage_path=None,
    level="INFO",
    log_to_terminal=True,
    write_to_file=True,
    append_to_file=True,
):
    """Configure ASTRA's logger dynamically.

    Parameters
    ----------
    log_path : str or Path or None
        Path to the log file. If None, file logging is skipped.
    level : str
        Minimum level to log ("DEBUG", "INFO", "WARNING", etc.)
    console : bool
        Whether to also log to stdout.
    """
    global _AST_INITIALIZED, _AST_SINKS

    if _AST_INITIALIZED:
        # Clean up previous sinks (safe for reconfiguration)
        for sink_id in _AST_SINKS:
            logger.remove(sink_id)
        _AST_SINKS.clear()

    # Define filter so only ASTRA messages appear
    def astra_filter(record):
        return record["extra"].get("module") == "ASTRA"

    astra_logger = logger.bind(module="ASTRA")
    logger.enable("ASTRA")
    # Optional console logging

    if log_to_terminal:
        _AST_SINKS.append(
            astra_logger.add(
                sys.stdout,
                level=level,
                filter=astra_filter,
                format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> {name} <level>{message}</level>",
            )
        )

    # Optional file logging
    if storage_path is not None and write_to_file:
        storage_path = Path(storage_path)
        _AST_SINKS.append(
            astra_logger.add(
                (storage_path / "ASTRA.log").as_posix(),
                level=level,
                filter=astra_filter,
                format="{time:YYYY-MM-DD HH:mm:ss} | {name} {level} | {message}",
                enqueue=True,
                mode="a" if append_to_file else "w",
            )
        )

    _AST_INITIALIZED = True
    return astra_logger

create_spectral_blocks

Compute start and end of all continuous regions.

build_blocks(indexes)

Evaluate the output of a np.where to find the continuous regions in it.

This returns a numpy array, where each entry is a list of a 'set' of ones.

E.g.

arr = np.array([0,0,1,1,1,1,1,1,0,1,0]) print(build_blocks(arr)) [[2, 3, 4, 5, 6, 7], [9]]

Parameters:

Name Type Description Default
indexes [type]

[description]

 required

Returns:

Type Description
[type]cd

[description]
Source code in src/ASTRA/utils/create_spectral_blocks.py 
 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
def build_blocks(indexes):
    """Evaluate the output of a np.where to find the continuous regions in it.

    This returns a numpy array, where each entry is a list of a 'set' of ones.

    E.g.
    >>> arr = np.array([0,0,1,1,1,1,1,1,0,1,0])
    >>> print(build_blocks(arr))
    [[2, 3, 4, 5, 6, 7], [9]]


    Parameters
    ----------
    indexes : [type]
        [description]

    Returns
    -------
    [type]cd
        [description]

    """
    diffs = np.diff(indexes)

    if len(indexes[0]) == 0:
        return []
    grouped_indexes = [[indexes[0][0]]]
    for index, dif in enumerate(diffs[0]):  # diffs are offseted by 1 index
        if dif == 1:
            grouped_indexes[-1].append(indexes[0][index + 1])
        else:
            grouped_indexes.append([indexes[0][index + 1]])

    return grouped_indexes

custom_exceptions

Defines the exceptions that ASTRA will raise.

Error

Bases: Exception

Base class for exceptions in this module.

Subclassed by:
Source code in src/ASTRA/utils/custom_exceptions.py 
6
7
class Error(Exception):
    """Base class for exceptions in this module."""

FrameError

Bases: Error

Raised when a INVALID S2D file is opened.

Source code in src/ASTRA/utils/custom_exceptions.py 
10
11
class FrameError(Error):
    """Raised when a INVALID S2D file is opened."""

AlreadyLoaded

Bases: Error

Raised when we try to load something that was already loaded.

Source code in src/ASTRA/utils/custom_exceptions.py 
14
15
class AlreadyLoaded(Error):
    """Raised when we try to load something that was already loaded."""

BadOrderError

Bases: Error

Raised when we access a bad order.

Source code in src/ASTRA/utils/custom_exceptions.py 
18
19
class BadOrderError(Error):
    """Raised when we access a bad order."""

NoConvergenceError

Bases: Error

Raised when we access data from a fit that did not converge.

Source code in src/ASTRA/utils/custom_exceptions.py 
22
23
class NoConvergenceError(Error):
    """Raised when we access data from a fit that did not converge."""

NoComputedRVsError

Bases: Error

Used when trying to access to a RV cue that was not computed.

Source code in src/ASTRA/utils/custom_exceptions.py 
26
27
class NoComputedRVsError(Error):
    """Used when trying to access to a RV cue that was not computed."""

DeadWorkerError

Bases: Error

Used when a worker finds a problem.

Source code in src/ASTRA/utils/custom_exceptions.py 
30
31
class DeadWorkerError(Error):
    """Used when a worker finds a problem."""

NoDataError

Bases: Error

Used when all loaded frames are blocked/invalid.

Raised by:
Source code in src/ASTRA/utils/custom_exceptions.py 
34
35
class NoDataError(Error):
    """Used when all loaded frames are blocked/invalid."""

InvalidConfiguration

Bases: Error

Used when a given configuration falls outside the accepted values.

Raised by:
Source code in src/ASTRA/utils/custom_exceptions.py 
38
39
class InvalidConfiguration(Error):
    """Used when a given configuration falls outside the accepted values."""

TemplateNotExistsError

Bases: Error

Used whenever we try to access a template that does not exist.

Source code in src/ASTRA/utils/custom_exceptions.py 
42
43
class TemplateNotExistsError(Error):
    """Used whenever we try to access a template that does not exist."""

BadTemplateError

Bases: Error

Used whenever we try to access a template that failed its computation.

Source code in src/ASTRA/utils/custom_exceptions.py 
46
47
class BadTemplateError(Error):
    """Used whenever we try to access a template that failed its computation."""

FailedStorage

Bases: Error

Used whenever we try to access a template that failed its computation.

Source code in src/ASTRA/utils/custom_exceptions.py 
50
51
class FailedStorage(Error):
    """Used whenever we try to access a template that failed its computation."""

InternalError

Bases: Error

Used whenever we have an error with the non-user part of the code.

Raised by:
Source code in src/ASTRA/utils/custom_exceptions.py 
54
55
class InternalError(Error):
    """Used whenever we have an error with the non-user part of the code."""

MissingRootPath

Bases: Error

Used whenever we have an error with the non-user part of the code.

Source code in src/ASTRA/utils/custom_exceptions.py 
58
59
class MissingRootPath(Error):
    """Used whenever we have an error with the non-user part of the code."""

StopComputationError

Bases: Error

Used whenever we have an error with the non-user part of the code.

Source code in src/ASTRA/utils/custom_exceptions.py 
62
63
class StopComputationError(Error):
    """Used whenever we have an error with the non-user part of the code."""

ensure_invalid_template(func)

Mark template as invalid if Exception is raised.

Source code in src/ASTRA/utils/custom_exceptions.py 
66
67
68
69
70
71
72
73
74
75
76
def ensure_invalid_template(func):
    """Mark template as invalid if Exception is raised."""

    def inner1(self, *args, **kwargs):
        try:
            func(self, *args, **kwargs)
        except Exception:
            self.mark_as_invalid()
            logger.opt(exception=True).critical("Template creation failed")

    return inner1

cython_codes

setup

targets = {'inverter': 'inverter.pyx', 'partial_derivative': 'partial_derivative.pyx', 'second_derivative': 'second_derivative.pyx'} module-attribute

ext_modules = [Extension(key, [value], extra_compile_args=['-fopenmp'], extra_link_args=['-fopenmp'], include_dirs=[np.get_include()]) for key, value in targets.items()] module-attribute

definitions

Enums that are used to define different properties.

DETECTOR_DEFINITION

Bases: Enum

Represent the cromatic intervals (full spectra, red and blue det.).

Source code in src/ASTRA/utils/definitions.py 
 6
 7
 8
 9
10
11
class DETECTOR_DEFINITION(Enum):
    """Represent the cromatic intervals (full spectra, red and blue det.)."""

    WHITE_LIGHT = auto()
    RED_DET = auto()
    BLUE_DET = auto()

WHITE_LIGHT = auto() class-attribute instance-attribute

RED_DET = auto() class-attribute instance-attribute

BLUE_DET = auto() class-attribute instance-attribute

handle_SDO_image

SUNPY_available = True module-attribute

configs = {'SDO_DOWNLOAD_REPEAT_INTEVAL': 5, 'SDO_FITS_NAME': 'hmi.ic_nolimbdark_720s_nrt.{}_TAI.3.continuum.fits', 'SDO_FITS_QUERY': 'hmi.ic_nolimbdark_720s_nrt[{}_TAI/{}]{{continuum}}', 'SDO_SERVER_NAME': 'JSOC', 'JSOC_EMAIL_ADDRESS': 'Andre.Silva@astro.up.pt', 'SDO_NUMBER_DOWNLOAD_REPEATS': 50} module-attribute

single_column_width = 85 / 25.4 module-attribute

double_column_width = 175 / 25.4 module-attribute

height = single_column_width * 0.8 module-attribute

download_from_SDO(export_request: drms.ExportRequest, image_name: str, storage_location: Path) -> bool

Base interface to download an image from SDO.

Creates an export request on JSOC servers, and downloads the latest image that was returned. If the image already exists on disk, the download does not proceed

Args: export_request (drms.ExportRequest): export request to use for the download image_index (int): index of the desired file in the request configs (ConfigHolder): description storage_location (Path): Disk location of the stored images

Returns: bool: returns true if it succeeds or false otherwise

Notes/References: - Documentation for the drms download can be found in here

- Description of the SDO data products: http://jsoc.stanford.edu/HMI/Continuum.html

- The email must be registered in [JSOC](http://jsoc.stanford.edu/ajax/register_email.html)

- This functions will retrieve the information from the SDO_SERVER_NAME that

is specified in the configs.

Source code in src/ASTRA/utils/handle_SDO_image.py 
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
def download_from_SDO(  # pylint: disable=invalid-name
    export_request: "drms.ExportRequest",
    image_name: str,
    storage_location: Path,
) -> bool:
    """Base interface to download an image from SDO.

    Creates an export request on JSOC servers, and downloads the latest image that was returned.
    If the image already exists on disk, the download does not proceed

    Args:
        export_request (drms.ExportRequest): export request to use for the download
        image_index (int): index of the desired file in the request
        configs (ConfigHolder): _description_
        storage_location (Path): Disk location of the stored images

    Returns:
        bool: returns true if it succeeds or false otherwise

    Notes/References:
        - Documentation for the drms download can be found in
         [here](https://docs.sunpy.org/projects/drms/en/stable/tutorial.html#url-quick-as-is)

        - Description of the SDO data products: http://jsoc.stanford.edu/HMI/Continuum.html

        - The email must be registered in [JSOC](http://jsoc.stanford.edu/ajax/register_email.html)

        - This functions will retrieve the information from the SDO_SERVER_NAME that
    is specified in the configs.

    """

    if not SUNPY_available:
        raise custom_exceptions.InternalError("Sunpy is not installed")

    server_name = configs["SDO_SERVER_NAME"]
    if server_name == "JSOC":
        return _JSOC_download_from_SDO(
            export_request=export_request,
            image_name=image_name,
            storage_location=storage_location,
        )
    logger.critical("Can't recognize the server name (%s)", server_name)
    return False

construct_JSOC_SDO_query_string(date: Time, time_range: timedelta) -> str

Construct the SDO query for a given date, image identifier and time range

Args: date (Time): Date at which we want to start our "search" image_identifier (str): SDO image idenfitifer. time_range (timedelta): Time range for the search (will search all available fits between date and date + time_range)

Returns: str: Query string to pass into SDO functions

Source code in src/ASTRA/utils/handle_SDO_image.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def construct_JSOC_SDO_query_string(  # pylint: disable=invalid-name
    date: Time,
    time_range: timedelta,
) -> str:
    """Construct the SDO query for a given date, image identifier and time range

    Args:
        date (Time): Date at which we want to start our "search"
        image_identifier (str): SDO image idenfitifer.
        time_range (timedelta): Time range for the search (will search all
            available fits between date and date + time_range)

    Returns:
        str: Query string to pass into SDO functions
    """
    image_identifier: str = configs.get("SDO_FITS_QUERY", "")
    return image_identifier.format(
        date.strftime("%Y.%m.%d_%H:%M:%S"), str(time_range.seconds) + "s"
    )

plot_sdo_with_multiple_crosses(fits_file: str | Path, tx_arcsec_list: list[float], ty_arcsec_list: list[float], zoom_center_tx: float, zoom_center_ty: float, plot_options: dict[str, str], zoom_width: u.Quantity = 300 * u.arcsec, zoom_height: u.Quantity = 300 * u.arcsec, zoom_map=None, fig=None, ax=None)

Source code in src/ASTRA/utils/handle_SDO_image.py 
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
def plot_sdo_with_multiple_crosses(
    fits_file: str | Path,
    tx_arcsec_list: list[float],
    ty_arcsec_list: list[float],
    zoom_center_tx: float,
    zoom_center_ty: float,
    plot_options: dict[str, str],
    zoom_width: u.Quantity = 300 * u.arcsec,
    zoom_height: u.Quantity = 300 * u.arcsec,
    zoom_map=None,
    fig=None,
    ax=None,
):
    sdo_map = sunpy.map.Map(fits_file)

    zoom_center = SkyCoord(
        Tx=zoom_center_tx * u.arcsec,
        Ty=zoom_center_ty * u.arcsec,
        frame=sdo_map.coordinate_frame,
    )

    bottom_left = SkyCoord(
        Tx=zoom_center.Tx - zoom_width / 2,
        Ty=zoom_center.Ty - zoom_height / 2,
        frame=sdo_map.coordinate_frame,
    )

    top_right = SkyCoord(
        Tx=zoom_center.Tx + zoom_width / 2,
        Ty=zoom_center.Ty + zoom_height / 2,
        frame=sdo_map.coordinate_frame,
    )

    if zoom_map is None:
        zoom_map = sdo_map.submap(bottom_left, top_right=top_right)

        fig = plt.figure(figsize=(1.4 * single_column_width, 1.6 * height))
        ax = fig.add_subplot(projection=zoom_map)

        zoom_map.plot(
            axes=ax,
            cmap="copper",
            # clip_interval=(1, 99.8) * u.percent,
        )
        if zoom_center_ty < 0:
            ax.invert_yaxis()
        if zoom_center_tx > 0:
            ax.invert_xaxis()

    for tx, ty in zip(tx_arcsec_list, ty_arcsec_list):
        coord = SkyCoord(
            Tx=tx * u.arcsec,
            Ty=ty * u.arcsec,
            frame=sdo_map.coordinate_frame,
        ).transform_to(zoom_map.coordinate_frame)

        ax.plot_coord(coord, **plot_options)

    return zoom_map, fig, ax

json_ready_converter

Utility function to convert any dictionary into one that can be saved to disk with json.

json_ready_converter(value: Any) -> Any

Convert a given value to a json-serializable format.

TODO: also prepare SBART objects that have the to_json attribute!

Parameters:

Name Type Description Default
value [type]

[description]

 required

Returns:

Type Description
[type]

[description]
Source code in src/ASTRA/utils/json_ready_converter.py 
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
def json_ready_converter(value: Any) -> Any:
    """Convert a given value to a json-serializable format.

    TODO: also prepare SBART objects that have the to_json attribute!

    Parameters
    ----------
    value : [type]
        [description]

    Returns
    -------
    [type]
        [description]

    """
    if isinstance(value, set):
        value = list(value)
    elif isinstance(value, Quantity):
        value = value.to(meter_second).value
    elif isinstance(value, np.ndarray):
        value = value.tolist()
    elif isinstance(value, np.integer):
        value = int(value)
    elif isinstance(value, np.floating):
        value = float(value)

    if isinstance(value, (list, tuple)) and len(value) > 0:
        if isinstance(value[0], Quantity):
            # TODO: this will fail if we mix QUantity and unitless in the same list!
            # TODO: this will also fail on lists of lists!
            value = convert_data(value, new_units=meter_second, as_value=True)
    if isinstance(value, dict):
        new_dict = {}
        for key, val in value.items():
            new_dict[key] = json_ready_converter(val)

        value = new_dict
    return value

parameter_validators

Validators for the configuration options.

We can combine multiple validation conditions by summing them, thus creating a AND condition

Raises: InvalidConfiguration: If passed value does not meet its imposed conditions InternalError: Passing unknown configuration of the methods

constraint_map = {'ValueInInterval': ValueInInterval, 'ValueFromDtype': ValueFromDtype, 'IterableMustHave': IterableMustHave, 'ValueFromList': ValueFromIterable} module-attribute

Positive_Value_Constraint = ValueInInterval((0, np.inf), include_edges=True) module-attribute

StringValue = ValueFromDtype((str,)) module-attribute

PathValue = ValueFromDtype((str, Path)) + PathExists() module-attribute

NumericValue = ValueFromDtype((int, float)) module-attribute

IntegerValue = ValueFromDtype((int,)) module-attribute

BooleanValue = ValueFromDtype((bool,)) module-attribute

predefined_constraints = {'Positive_Value_Constraint': Positive_Value_Constraint, 'StringValue': StringValue, 'PathValue': PathValue, 'NumericValue': NumericValue, 'IntegerValue': IntegerValue, 'BooleanValue': BooleanValue} module-attribute

Constraint

Base class to define a constraint on possible values that a parameter can take.

Serves as parent class, allows the "sum" of two constraints to make "AND" operations

Attributes: constraint_text: String that describes the constraint

Raises: InvalidConfiguration: If the constraint on the parameter value is not met

Subclassed by:
Source code in src/ASTRA/utils/parameter_validators.py 
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
class Constraint:
    """Base class to define a constraint on possible values that a parameter can take.

    Serves as parent class, allows the "sum" of two constraints to make "AND" operations

    Attributes:
        constraint_text: String that describes the constraint

    Raises:
        InvalidConfiguration: If the constraint on the parameter value is not met

    """

    def __init__(self, const_text: str):  # noqa: D107
        self._constraint_list = [self._evaluate]
        self.constraint_text = const_text

    def __add__(self, other: Constraint) -> Constraint:
        """Create AND condition of current and other."""
        new_const = Constraint(self.constraint_text)
        # ensure that we don't propagate changes to all existing constraints
        new_const._constraint_list = deepcopy(self._constraint_list)
        new_const._constraint_list.append(other._evaluate)
        new_const.constraint_text += " and " + other.constraint_text

        return new_const

    def __radd__(self, other: Constraint) -> Constraint:  # noqa: D105
        return self.__add__(other)

    def _evaluate(self, value: Any) -> None:
        del value

    def check_if_value_meets_constraint(self, value: Any) -> None:
        """Compare the provided value against all constraint.

        Args:
            value (Any): Value to be compared

        Raises:
            InvalidConfiguration: If the value is not met

        """
        for evaluator in self._constraint_list:
            evaluator(value)

    def __str__(self) -> str:  # noqa: D105
        return self.constraint_text

    def __repr__(self) -> str:  # noqa: D105
        return self.constraint_text

    def __call__(self, value: Any) -> None:
        """Call the self.check_if_value_meets_constraint."""
        self.check_if_value_meets_constraint(value)

constraint_text = const_text instance-attribute

check_if_value_meets_constraint(value: Any) -> None

Compare the provided value against all constraint.

Args: value (Any): Value to be compared

Raises: InvalidConfiguration: If the value is not met

Source code in src/ASTRA/utils/parameter_validators.py 
56
57
58
59
60
61
62
63
64
65
66
67
def check_if_value_meets_constraint(self, value: Any) -> None:
    """Compare the provided value against all constraint.

    Args:
        value (Any): Value to be compared

    Raises:
        InvalidConfiguration: If the value is not met

    """
    for evaluator in self._constraint_list:
        evaluator(value)

ValueInInterval

Bases: Constraint

Constraint that imposes the provided value to be inside a given interval.

Source code in src/ASTRA/utils/parameter_validators.py 
 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
class ValueInInterval(Constraint):
    """Constraint that imposes the provided value to be inside a given interval."""

    def __init__(self, interval: tuple[Any, Any], include_edges: bool = False):
        """Instantiate the ValueInInterval class.

        Args:
            interval (Tuple): Tuple of the two edges of the interval
            include_edges (bool, optional): Include the edge in the comparison.
                Defaults to False.

        """
        super().__init__(
            const_text=f"Value inside interval <{interval}>; Edges: {include_edges}",
        )
        self._interval = interval
        self._include_edges = include_edges

    def _evaluate(self, value: Any) -> None:
        good_value = False
        try:
            if self._include_edges:
                if self._interval[0] <= value <= self._interval[1]:
                    good_value = True
            elif self._interval[0] < value < self._interval[1]:
                good_value = True
        except TypeError as exc:
            msg = f"Config value can't be compared with the the interval: {value} ({type(value)}) vs {self._interval}"
            raise InvalidConfiguration(msg) from exc

        if not good_value:
            msg = f"Config value is not inside the interval: {value} vs {self._interval}"
            raise InvalidConfiguration(msg)

ValueFromDtype

Bases: Constraint

Constraint that limits the datatype of the input.

Source code in src/ASTRA/utils/parameter_validators.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
class ValueFromDtype(Constraint):
    """Constraint that limits the datatype of the input."""

    def __init__(self, dtype_list: tuple[type, ...]) -> None:
        """Constraint that limits the datatype of the input.

        Args:
            dtype_list (List[type]): Valid data types

        """
        super().__init__(const_text=f"Value from dtype <{dtype_list}>")
        self.valid_dtypes = dtype_list

        if not isinstance(dtype_list, tuple):
            msg = "Dtype list must be a tuple"
            raise InternalError(msg)

    def _evaluate(self, value: Any) -> None:
        if not isinstance(value, self.valid_dtypes):
            msg = f"Config value ({value}) not from" f"the valid dtypes: {type(value)} vs {self.valid_dtypes}"
            raise InvalidConfiguration(
                msg,
            )

valid_dtypes = dtype_list instance-attribute

ValueFromIterable

Bases: Constraint

Limits the possible values to be inside a list.

Source code in src/ASTRA/utils/parameter_validators.py 
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
class ValueFromIterable(Constraint):
    """Limits the possible values to be inside a list."""

    def __init__(self, available_options: Iterable[Any]) -> None:
        """Limits the possible values that the input can take.

        Args:
            available_options (Tuple[Any]): Tuple with the allowed values for this input

        """
        super().__init__(const_text=f"Value from list <{available_options}>")
        self.available_options = available_options
        if not isinstance(available_options, Iterable):
            msg = "The available options must be an iterable"
            raise InternalError(msg)

    def _evaluate(self, value: Any) -> None:
        bad_value = False
        if isinstance(value, (list, tuple)):
            for element in value:
                if element not in self.available_options:
                    bad_value = True
                    break
        elif isinstance(self.available_options, type) and not isinstance(
            value,
            self.available_options,
        ):
            # The value in Enum does not work properly before python3.12
            bad_value = True
        elif value not in self.available_options:
            bad_value = True

        if bad_value:
            msg = f"Config value not one of the valid ones: {value} vs {self.available_options}"
            raise InvalidConfiguration(
                msg,
            )

available_options = available_options instance-attribute

IterableMustHave

Bases: Constraint

Imposes that certain values must be inside a given iterable.

Source code in src/ASTRA/utils/parameter_validators.py 
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
class IterableMustHave(Constraint):
    """Imposes that certain values must be inside a given iterable."""

    def __init__(self, available_options: tuple[Any, ...], mode: str = "all") -> None:
        """Imposes that certain values must be inside the passed iterable object.

        Args:
            available_options (Tuple[Any]): Tuple with values that must be present in the iterable
            mode (str, optional): "all" if we want all values present in the iterable,
            "any" if we want at least one of them. Defaults to "all".

        """
        super().__init__(const_text=f"Must have value from list <{available_options}>")
        self.available_options = available_options
        self.mode = mode

        if mode not in ["all", "any"]:
            msg = "Using the wrong mode"
            raise InternalError(msg)

    def _evaluate(self, value: Any) -> None:
        if not isinstance(value, Iterable):
            msg = "Constraint needs a iterable object"
            raise InvalidConfiguration(msg)

        evaluation = [i in value for i in self.available_options]

        good_value = False

        if self.mode == "all":
            good_value = all(evaluation)
        elif self.mode == "any":
            good_value = any(evaluation)

        if not good_value:
            raise InvalidConfiguration(
                f"Config value {value} does not" " have {self.mode} of {self.available_options}",
            )

available_options = available_options instance-attribute

mode = mode instance-attribute

PathExists

Bases: Constraint

Imposes that a given path must exist.

Source code in src/ASTRA/utils/parameter_validators.py 
219
220
221
222
223
224
225
226
227
228
class PathExists(Constraint):
    """Imposes that a given path must exist."""

    def __init__(self) -> None:
        """Imposes that a given path exists."""
        super().__init__(const_text="The path must exist")

    def _evaluate(self, value: Any) -> None:
        if not Path(value).exists():
            raise InvalidConfiguration(f"Path {value} does not exist")

paths_tools

PathsHandler

Handle the interface with disk paths.

Store and generate paths from relative locations, after a given root path is set. It allows to define tags to relative path on disk, ensuring that ASTRA objects are agnostic to the storage path.

Raises: custom_exceptions.MissingRootPath: If accessing disk paths without setting the root one.

Paths

ASTRA internal class to store disk paths for storage.

Source code in src/ASTRA/utils/paths_tools/PathsHandler.py 
 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
class Paths:
    """ASTRA internal class to store disk paths for storage."""

    def __init__(
        self,
        root_level_path: Optional[Union[str, Path]] = None,
        preconfigured_paths: Optional[Dict[str, str]] = None,
    ):
        """Instantiate object.

        Args:
            root_level_path (Optional[Union[str, Path]], optional): If not None, sets the
                root path. Defaults to None.
            preconfigured_paths (Optional[Dict[str, str]], optional): Original mapping of
                tags and paths. Defaults to None.

        """
        self._folder_mappings = preconfigured_paths if preconfigured_paths is not None else {}

        if isinstance(root_level_path, str):
            root_level_path = Path(root_level_path)

        self._root_path: Optional[Path] = root_level_path if root_level_path is not None else None

        # For lazy creation of the folders and to avoid multiple creation attempts
        self._constructed_folders: set[str] = set()

    def add_relative_path(self, folder_KW: str, rel_path: str) -> None:
        """Add relative (to self._root_path) folder, associated with a given Keyword.

        Args:
            folder_KW (str): keyword that will be associated with the relative path
            rel_path (str): relative (to root path) path

        """
        self._folder_mappings[folder_KW] = rel_path

    def get_path_to(self, folder_tag: str, absolute: bool = True, as_posix: bool = True) -> Union[str, Path]:
        """Get the path from a relative "tag".

        Args:
            folder_tag (str): Internal path name
            absolute (bool, optional): returns absolute path. Defaults to True.
            as_posix (bool, optional): return string if True. Defaults to True.

        Raises:
            custom_exceptions.MissingRootPath: If the root path was not previously set

        Returns:
            Union[str, Path]: Path

        """
        if self._root_path is None:
            raise custom_exceptions.MissingRootPath("Must provide the root level path before asking for other paths")

        if folder_tag == "ROOT":
            out_path = self._root_path.absolute()
        else:
            out_path = self._folder_mappings[folder_tag]

            if absolute:
                out_path = self._root_path / out_path

        if folder_tag not in self._constructed_folders:
            out_path.mkdir(parents=True, exist_ok=True)
            self._constructed_folders.add(folder_tag)

        if as_posix:
            out_path = out_path.as_posix()

        return out_path

    def add_root_path(self, path: Union[str, Path], current_folder_name: Optional[str] = None) -> None:
        """Set the current root path."""
        if isinstance(path, str):
            path = Path(path)

        if current_folder_name is not None:
            path = path / current_folder_name

        # ensure that the folder exists
        path.mkdir(exist_ok=True, parents=True)

        self._root_path = path

    @property
    def root_storage_path(self) -> Path:
        """Retrieve the root storage path."""
        return self._root_path
root_storage_path: Path property

Retrieve the root storage path.

add_relative_path(folder_KW: str, rel_path: str) -> None

Add relative (to self._root_path) folder, associated with a given Keyword.

Args: folder_KW (str): keyword that will be associated with the relative path rel_path (str): relative (to root path) path

Source code in src/ASTRA/utils/paths_tools/PathsHandler.py 
47
48
49
50
51
52
53
54
55
def add_relative_path(self, folder_KW: str, rel_path: str) -> None:
    """Add relative (to self._root_path) folder, associated with a given Keyword.

    Args:
        folder_KW (str): keyword that will be associated with the relative path
        rel_path (str): relative (to root path) path

    """
    self._folder_mappings[folder_KW] = rel_path
get_path_to(folder_tag: str, absolute: bool = True, as_posix: bool = True) -> Union[str, Path]

Get the path from a relative "tag".

Args: folder_tag (str): Internal path name absolute (bool, optional): returns absolute path. Defaults to True. as_posix (bool, optional): return string if True. Defaults to True.

Raises: custom_exceptions.MissingRootPath: If the root path was not previously set

Returns: Union[str, Path]: Path

Source code in src/ASTRA/utils/paths_tools/PathsHandler.py 
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
def get_path_to(self, folder_tag: str, absolute: bool = True, as_posix: bool = True) -> Union[str, Path]:
    """Get the path from a relative "tag".

    Args:
        folder_tag (str): Internal path name
        absolute (bool, optional): returns absolute path. Defaults to True.
        as_posix (bool, optional): return string if True. Defaults to True.

    Raises:
        custom_exceptions.MissingRootPath: If the root path was not previously set

    Returns:
        Union[str, Path]: Path

    """
    if self._root_path is None:
        raise custom_exceptions.MissingRootPath("Must provide the root level path before asking for other paths")

    if folder_tag == "ROOT":
        out_path = self._root_path.absolute()
    else:
        out_path = self._folder_mappings[folder_tag]

        if absolute:
            out_path = self._root_path / out_path

    if folder_tag not in self._constructed_folders:
        out_path.mkdir(parents=True, exist_ok=True)
        self._constructed_folders.add(folder_tag)

    if as_posix:
        out_path = out_path.as_posix()

    return out_path
add_root_path(path: Union[str, Path], current_folder_name: Optional[str] = None) -> None

Set the current root path.

Source code in src/ASTRA/utils/paths_tools/PathsHandler.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def add_root_path(self, path: Union[str, Path], current_folder_name: Optional[str] = None) -> None:
    """Set the current root path."""
    if isinstance(path, str):
        path = Path(path)

    if current_folder_name is not None:
        path = path / current_folder_name

    # ensure that the folder exists
    path.mkdir(exist_ok=True, parents=True)

    self._root_path = path

build_filename

Path creation.

build_filename(og_path: Path, filename: str, fmt: str, ASTRA_version: Optional[str] = None, skip_version: bool = False) -> str

Standardize the filenames of ASTRA outputs.

Args: og_path (Path): Folder in which the file will be stored filename (str): stem of the filename fmt (str): format of the file ASTRA_version (Optional[str], optional): version of ASTRA. Defaults to None.

Returns: str: Posix version of the file that want to store

Source code in src/ASTRA/utils/paths_tools/build_filename.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
def build_filename(
    og_path: Path, filename: str, fmt: str, ASTRA_version: Optional[str] = None, skip_version: bool = False
) -> str:
    """Standardize the filenames of ASTRA outputs.

    Args:
        og_path (Path): Folder in which the file will be stored
        filename (str): stem of the filename
        fmt (str): format of the file
        ASTRA_version (Optional[str], optional): version of ASTRA. Defaults to None.

    Returns:
        str: Posix version of the file that want to store

    """
    ASTRA_version = ASTRA_version if ASTRA_version is not None else __version__
    if skip_version:
        return (og_path / f"{filename}_TM.{fmt}").as_posix()
    else:
        return (og_path / f"{filename}_TM_{ASTRA_version}.{fmt}").as_posix()

ensure_path

Ensure that a given input path is a pathlib.Path object.

ensure_path_from_input(input_path: Union[str, pathlib.Path], ensure_existence: bool = False) -> pathlib.Path

Ensure that a given path is a pathlib.Path object and (optionally) if it exists.

Args: input_path (Union[str, pathlib.Path]): User-provided path ensure_existence (bool): If True, raises an Exception if the input_path does not exists. By default False

Raises: InvalidConfiguration: If the path is neither a string nor a pathlib.Path object NoDataError: If the ensure_existence flag is set to True and input_path does not exist

Source code in src/ASTRA/utils/paths_tools/ensure_path.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def ensure_path_from_input(input_path: Union[str, pathlib.Path], ensure_existence: bool = False) -> pathlib.Path:
    """Ensure that a given path is a pathlib.Path object and (optionally) if it exists.

    Args:
        input_path (Union[str, pathlib.Path]): User-provided path
        ensure_existence (bool): If True, raises an Exception if the input_path does not exists. By default False

    Raises:
        InvalidConfiguration: If the path is neither a string nor a pathlib.Path object
        NoDataError: If the ensure_existence flag is set to True and input_path does not exist

    """
    if isinstance(input_path, str):
        input_path = pathlib.Path(input_path)

    if not isinstance(input_path, pathlib.Path):
        raise InvalidConfiguration("The user-provided path is neither a string nor a pathlib.Path")

    if ensure_existence:
        if not input_path.exists():
            raise NoDataError(f"The user-provided path <{input_path}> does not exist")

    return input_path

file_older_than

file_older_than(file: str, days: int) -> bool

Check if file has been modified more than days days ago

Source code in src/ASTRA/utils/paths_tools/file_older_than.py 
 7
 8
 9
10
11
def file_older_than(file: str, days: int) -> bool:
    """Check if `file` has been modified more than `days` days ago"""
    threshold_date = datetime.now() - relativedelta(days=days)
    file_time = datetime.fromtimestamp(os.path.getmtime(file))
    return file_time < threshold_date

find_latest_version

Search for latest version in file products.

find_latest_version(path: Path, enable_logs: bool = True) -> str

Search, inside a directory, for all files with SBART versions. Returns the latest version found on disk.

Source code in src/ASTRA/utils/paths_tools/find_latest_version.py 
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
def find_latest_version(path: Path, enable_logs: bool = True) -> str:
    """Search, inside a directory, for all files with SBART versions. Returns the latest version found on disk."""
    available_cubes = []
    versions_full = []
    version_sum = []

    for filename in os.listdir(path):
        if os.path.isdir(os.path.join(path, filename)) or "_" not in filename or filename.startswith("."):
            continue

        available_cubes.append(filename)
        cube_ver = filename.split("_")[-1].split(".")[0]
        versions_full.append(cube_ver)
        version_sum.append(sum([i * j for i, j in zip([100, 10, 1], map(int, cube_ver.split("-")))]))
    try:
        highest_ver = max(version_sum)
    except ValueError as exc:  # no version number in the list
        raise NoDataError(f"There are no SBART outputs in {path}") from exc

    if highest_ver != sum([i * j for i, j in zip([100, 10, 1], __version_info__)]) and enable_logs:
        logger.warning(
            (
                f"\tRV cube is not the most recently installed version ({__version__})."
                f"Using data from {versions_full[version_sum.index(highest_ver)]}"
            )
        )

    return versions_full[version_sum.index(highest_ver)]

poet_SDO_plotter

SUNPY_available = True module-attribute

plot_SDO_image(d: DataClass, include_fine_point_offset: bool, storage_location: Path, JSOC_EMAIL_ADDRESS: str, zoom_in=False, label_map: dict[str, str] | None = None, legend_options: dict[str, Any] | None = None)

Make plot of the pointing positions on SDO image

Parameters:

Name Type Description Default
d DataClass

DataClass object

 required
include_fine_point_offset bool

Include into the final coordinates the fine-pointing offsets (from the MCS)

 required
storage_location Path

Folder to store the image (pointing_locations.png)

 required
zoom_in bool

If True, zooms the plot to a region close to the pointing position, by default False

 False
label_map dict[str, str] | None

If not None, dictionary to connect PoET OB names to a legend label

 None

Raises:

Type Description
InternalError

If Sunpy is not installed
Source code in src/ASTRA/utils/poet_SDO_plotter.py 
 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
def plot_SDO_image(
    d: DataClass,
    include_fine_point_offset: bool,
    storage_location: Path,
    JSOC_EMAIL_ADDRESS: str,
    zoom_in=False,
    label_map: dict[str, str] | None = None,
    legend_options: dict[str, Any] | None = None,
):
    """Make plot of the pointing positions on SDO image

    Parameters
    ----------
    d : DataClass
        DataClass object
    include_fine_point_offset : bool
        Include into the final coordinates the fine-pointing offsets (from the MCS)
    storage_location : Path
        Folder to store the image (pointing_locations.png)
    zoom_in : bool, optional
        If True, zooms the plot to a region close to the pointing position, by default False
    label_map: dict[str,str], optional
        If not None, dictionary to connect PoET OB names to a legend label

    Raises
    ------
    custom_exceptions.InternalError
        If Sunpy is not installed
    """
    if not SUNPY_available:
        raise custom_exceptions.InternalError("Missing SUNPY imports")

    poet_loc = EarthLocation.from_geodetic(-70.404375, -24.627672222222223, 2635.43)
    ob_times = []
    Tx, Ty = [], []
    coord_locs = {}

    for subInst in d.get_subInstruments_with_valid_frames():
        logger.debug("Processing subInstrument: " + subInst)
        coord_locs[subInst] = {"Tx": [], "Ty": []}
        frameIDs = d.get_frameIDs_from_subInst(subInst)
        ref_frame = d.get_frame_by_ID(frameIDs[0])

        delay_in_secs = 7.08  # Add or subtract a time delay in seconds
        DateTime = ref_frame.get_KW_value("ISO-DATE")
        hr = DateTime[11:13]
        minu = DateTime[14:16]
        secs = DateTime[17:]
        time_decimal = (
            float(hr)
            + float(minu) / 60.0
            + float(secs) / 3600.0
            + delay_in_secs / 3600.0
        )
        nhr = math.floor(time_decimal)
        nmin = math.floor(((time_decimal - nhr) * 60.0))
        nsecs = (time_decimal - nhr) * 3600.0 - nmin * 60.0
        ob_time = Time(DateTime[:11] + str(nhr) + ":" + str(nmin) + ":" + str(nsecs))

        ob_times.append(ob_time)

        # Altitude delta from fine pointing
        Alt_offset = ref_frame.get_KW_value("ALT_OFF")
        Az_offset = ref_frame.get_KW_value("AZ_OFF")

        # Tx of target before fine pointing
        Tx_target = ref_frame.get_KW_value("TX")
        # Ty of target before fine pointing
        Ty_target = ref_frame.get_KW_value("TY")

        # Convert Tx and Ty to actual ALT/Az
        pointing_coords = SkyCoord(
            Tx_target,
            Ty_target,
            observer="earth",
            obstime=ob_time,
            frame=frames.Helioprojective,
        )

        local_frame = AltAz(
            obstime=ob_time,
            location=poet_loc,
        )

        # Compute distance between observer and Sun
        sun_altaz = pointing_coords.transform_to(local_frame)

        # Add dynamic offsets
        point_alt = sun_altaz.alt
        point_az = sun_altaz.az

        if include_fine_point_offset:
            # This will include the fine-pointing offsets in the final coordinates
            # This will not give the true pointing coordinates for dates between XX and YY
            logger.debug(
                f"Add fine point offsets to the pointing coordinates {Alt_offset}, {Az_offset}"
            )

            point_alt = point_alt + float(Alt_offset) * u.deg
            point_az = point_az + float(Az_offset) * u.deg

        skycoord = SkyCoord(
            az=point_az,
            alt=point_alt,
            distance=sunpy.coordinates.sun.earth_distance(ob_time),
            frame=AltAz(
                obstime=ob_time,
                location=poet_loc,
            ),
        )

        hpc = skycoord.transform_to(
            frames.Helioprojective(
                observer="earth",
            )
        )

        coord_locs[subInst]["Tx"].append(hpc.Tx.to(u.arcsec).value)
        coord_locs[subInst]["Ty"].append(hpc.Ty.to(u.arcsec).value)

    date_to_get_image = min(ob_times)

    hmi_files = list(storage_location.glob("hmi.ic_no*"))
    if not hmi_files:
        filenames, export = _JSOC_get_OBS_list_from_SDO(
            since=date_to_get_image - TimeDelta(1000, format="sec"),
            up_to=(date_to_get_image + TimeDelta(1000, format="sec")),
            JSOC_EMAIL_ADDRESS=JSOC_EMAIL_ADDRESS,
        )

        SDO_fname = download_from_SDO(
            export_request=export,
            image_name=filenames.iloc[-1],
            storage_location=storage_location,
        )
    else:
        SDO_fname = hmi_files[0]

    zoom_map = None
    fig = None
    ax = None

    found = False

    for key in coord_locs:
        if "_umbra" in key:
            center_tx = coord_locs[key]["Tx"][0]
            center_ty = coord_locs[key]["Ty"][0]
            found = True
    all_Tx = list(i["Tx"] for i in coord_locs.values())
    all_Ty = list(i["Ty"] for i in coord_locs.values())

    if not found:
        center_tx = np.median(all_Tx)
        center_ty = np.median(all_Ty)

    cmap = plt.get_cmap("Set1")
    if zoom_in:
        width = max(1.5 * (np.max(all_Tx) - np.min(all_Tx)), 210)
        height = max(1.5 * (np.max(all_Ty) - np.min(all_Ty)), 210)
    else:
        center_tx = 0
        center_ty = 0
        width = 40 * 60
        height = 40 * 60

    for index, stem in enumerate(coord_locs):

        label = label_map.get(stem, stem) if label_map is not None else stem
        zoom_map, fig, ax = plot_sdo_with_multiple_crosses(
            fits_file=SDO_fname,
            tx_arcsec_list=coord_locs[stem]["Tx"],
            ty_arcsec_list=coord_locs[stem]["Ty"],
            zoom_center_tx=center_tx,
            zoom_center_ty=center_ty,
            zoom_width=width * u.arcsec,
            zoom_height=height * u.arcsec,
            plot_options={
                "label": label,
                "marker": "x",
                "color": cmap(index / len(coord_locs)),
                "markersize": 6,
                "markeredgewidth": 2.2,
                "linestyle": "None",
            },
            zoom_map=zoom_map,
            fig=fig,
            ax=ax,
        )

    from matplotlib.lines import Line2D

    fig.tight_layout()
    fig.subplots_adjust(bottom=0.2)
    if legend_options is None:
        legend_options = {
        "ncol": 2,
        "loc": 4,
        "bbox_to_anchor": (1.05, -0.27),
        }

    ax.legend(**legend_options)

    fig.savefig(storage_location / "pointing_locations.png")

ranges

ranges(a)

Source code in src/ASTRA/utils/ranges.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
def ranges(a):
    if len(a) == 0:
        return a
    if isinstance(a, np.ndarray) and a.ndim == 0:
        return a

    a = np.sort(a)
    _none_1 = a[np.ediff1d(a, None, -1) != 1]
    _1_none = a[np.ediff1d(a, -1, None) != 1]
    starts = np.setdiff1d(_none_1, _1_none)
    ends = np.setdiff1d(_1_none, _none_1)
    uniques = np.intersect1d(_none_1, _1_none)
    R, nums = [], []
    for start, end in zip(starts, ends):
        R.append(f"{start}-{end}")
        nums.append(start)
    for u in uniques:
        R.append(str(u))
        nums.append(u)
    R = np.array(R)[np.argsort(nums)]
    return ", ".join(R)

secular_acceleration

Compute secular acceleration.

as_yr = u.arcsec / u.year module-attribute

mas_yr = u.milliarcsecond / u.year module-attribute

mas = u.milliarcsecond module-attribute

handler(signum, frame)

Source code in src/ASTRA/utils/secular_acceleration.py 
19
20
21
def handler(signum, frame):
    print("Timeout!")
    raise Exception

build_query(star: str)

Source code in src/ASTRA/utils/secular_acceleration.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def build_query(star: str):
    star = _de_escape(star)
    with warnings.catch_warnings():
        warnings.simplefilter("ignore")
    aSimbad = Simbad()
    aSimbad.add_votable_fields("rv_value", "pm", "pmdec", "pmra", "plx", "ra(d)", "dec(d)")

    r = aSimbad.query_object(star)
    if r is None:
        raise ValueError(f"star {star} not found in Simbad")
    r = r[0]
    query = f"""SELECT TOP 1 * FROM gaiadr2.gaia_source \
    WHERE CONTAINS(
                POINT('ICRS',gaiadr2.gaia_source.ra,gaiadr2.gaia_source.dec),
                CIRCLE('ICRS',
                    COORD1(EPOCH_PROP_POS({r["RA_d"]},{r["DEC_d"]},{r["PLX_VALUE"]},{r["PMRA"]},{r["PMDEC"]},{r["RV_VALUE"]},2000,2015.5)),
                    COORD2(EPOCH_PROP_POS({r["RA_d"]},{r["DEC_d"]},{r["PLX_VALUE"]},{r["PMRA"]},{r["PMDEC"]},{r["RV_VALUE"]},2000,2015.5)),
                    0.001388888888888889))=1
    """

    return r, query

run_job(query, timeout=5)

Source code in src/ASTRA/utils/secular_acceleration.py 
53
54
55
56
57
58
59
def run_job(query, timeout=5):
    # signal.signal(signal.SIGALRM, handler)
    # signal.alarm(timeout)
    from astroquery.gaia import Gaia  # avoid print at import time

    job = Gaia.launch_job_async(query)
    return job.get_results()

mu(pmra, pmdec)

Source code in src/ASTRA/utils/secular_acceleration.py 
62
63
def mu(pmra, pmdec):
    return np.sqrt(pmra**2 + pmdec**2)

secular_acceleration(star)

Source code in src/ASTRA/utils/secular_acceleration.py 
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
def secular_acceleration(star):
    star = _de_escape(star)
    conn = DB_connection()
    try:
        params = conn.get_star_params(star)
        pmra = params["pmra"]
        pmdec = params["pmdec"]
        p = params["parallax"]
        logger.info("Using cached information for SA calculation")
    except FileNotFoundError:
        logger.info("Querying SIMBAD for information to run SA calculation")

        rsimbad, query = build_query(star)

        r = run_job(query)[0]
        pmra = (r["pmra"] * mas_yr).to(as_yr).value
        pmdec = (r["pmdec"] * mas_yr).to(as_yr).value
        p = (r["parallax"] * mas).to(u.arcsec).value
        if p == 0.0:
            # print(f'Warning: No measured GAIA parallax for {star}. Using Simbad')
            pmra = (rsimbad["PMRA"] * mas_yr).to(as_yr).value
            pmdec = (rsimbad["PMDEC"] * mas_yr).to(as_yr).value
            p = (rsimbad["PLX_VALUE"] * mas).to(u.arcsec).value

        conn.add_new_star(star, pmra=pmra, pmdec=pmdec, parallax=p)

    sa = 0.0229 * mu(pmra, pmdec) ** 2 / p  # m/s/yr
    return (
        sa * meter_second
    )  # not really the proper units, but close enough!! The year is taken into account in the RV correction

shift_spectra

Functions to shift spectra.

Implements utility functions to correct for stellar RV and BERV

SPEED_OF_LIGHT = 299792.458 module-attribute

apply_RVshift(wave: np.ndarray, stellar_RV: float) -> np.ndarray

Apply RV shift to spectra.

Args: wave (np.ndarray): Wavelength array stellar_RV (float): RV shift to apply, in km/s

Returns: np.ndarray: New wavelength array, after RV shift

Source code in src/ASTRA/utils/shift_spectra.py 
11
12
13
14
15
16
17
18
19
20
21
22
def apply_RVshift(wave: np.ndarray, stellar_RV: float) -> np.ndarray:
    """Apply RV shift to spectra.

    Args:
        wave (np.ndarray): Wavelength array
        stellar_RV (float): RV shift to apply, in km/s

    Returns:
        np.ndarray: New wavelength array, after RV shift

    """
    return np.multiply(wave, (1 + stellar_RV / SPEED_OF_LIGHT))

remove_RVshift(wave: np.ndarray, stellar_RV: float) -> np.ndarray

Remove RV shift from wavelength vector.

Args: wave (np.ndarray): Wavelength vector stellar_RV (float): stellar RV, in km/s

Returns: np.ndarray: New wavelength array

Source code in src/ASTRA/utils/shift_spectra.py 
25
26
27
28
29
30
31
32
33
34
35
36
def remove_RVshift(wave: np.ndarray, stellar_RV: float) -> np.ndarray:
    """Remove RV shift from wavelength vector.

    Args:
        wave (np.ndarray): Wavelength vector
        stellar_RV (float): stellar RV, in km/s

    Returns:
        np.ndarray: New wavelength array

    """
    return wave / (1 + stellar_RV / SPEED_OF_LIGHT)

apply_approximated_BERV_correction(wave: np.ndarray, BERV: float) -> np.ndarray

Apply the approximated BERV correction that is implemented in ESO DRS 3.2.

Args: wave (np.ndarray): wavelength array BERV (float): BERV value, with the 1.55 * 10**-8 approximation

Returns: np.ndarray: BERV-corrected wavelength array

Source code in src/ASTRA/utils/shift_spectra.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
def apply_approximated_BERV_correction(wave: np.ndarray, BERV: float) -> np.ndarray:
    """Apply the approximated BERV correction that is implemented in ESO DRS 3.2.

    Args:
        wave (np.ndarray): wavelength array
        BERV (float): BERV value, with the 1.55 * 10**-8 approximation

    Returns:
        np.ndarray: BERV-corrected wavelength array

    """
    # Addint the new term that we found in the ESPRESSO DRS. Source of this is unclear.....
    return wave * (1 + 1.55e-8) * (1 + BERV / SPEED_OF_LIGHT)

remove_approximated_BERV_correction(wave: np.ndarray, BERV: float) -> np.ndarray

Remove the approximated BERV correction implemented in ESO DRS 3.2.

Args: wave (np.ndarray): wavelength array BERV (float): BERV value, with the 1.55 * 10**-8 approximation

Returns: np.ndarray: BERV-corrected wavelength array

Source code in src/ASTRA/utils/shift_spectra.py 
54
55
56
57
58
59
60
61
62
63
64
65
def remove_approximated_BERV_correction(wave: np.ndarray, BERV: float) -> np.ndarray:
    """Remove the approximated BERV correction implemented in ESO DRS 3.2.

    Args:
        wave (np.ndarray): wavelength array
        BERV (float): BERV value, with the 1.55 * 10**-8 approximation

    Returns:
        np.ndarray: BERV-corrected wavelength array

    """
    return wave / ((1 + 1.55e-8) * (1 + BERV / SPEED_OF_LIGHT))

apply_BERV_correction(wave: np.ndarray, BERV: float, BERV_factor: float | None = None) -> np.ndarray

Apply BERV correction to data.

Args: wave (np.ndarray): wavelength array BERV (float): BERV value BERV_factor (float | None): If not None, it will be taken as the value of (1 + BERV/C)

Returns: np.ndarray: BERV-corrected wavelength array

Source code in src/ASTRA/utils/shift_spectra.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def apply_BERV_correction(wave: np.ndarray, BERV: float, BERV_factor: float | None = None) -> np.ndarray:
    """Apply BERV correction to data.

    Args:
        wave (np.ndarray): wavelength array
        BERV (float): BERV value
        BERV_factor (float | None): If not None, it will be taken as the value of (1 + BERV/C)

    Returns:
        np.ndarray: BERV-corrected wavelength array

    """
    if BERV_factor is not None and np.isfinite(BERV_factor):
        return wave * BERV_factor
    return wave * (1 + BERV / SPEED_OF_LIGHT)

remove_BERV_correction(wave: np.ndarray, BERV: float, BERV_factor: float | None = None) -> np.ndarray

Remove BERV correction to data.

Args: wave (np.ndarray): wavelength array BERV (float): BERV value BERV_factor (float | None): If not None, it will be taken as the value of (1 + BERV/C)

Returns: np.ndarray: BERV-corrected wavelength array

Source code in src/ASTRA/utils/shift_spectra.py 
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def remove_BERV_correction(wave: np.ndarray, BERV: float, BERV_factor: float | None = None) -> np.ndarray:
    """Remove BERV correction to data.

    Args:
        wave (np.ndarray): wavelength array
        BERV (float): BERV value
        BERV_factor (float | None): If not None, it will be taken as the value of (1 + BERV/C)

    Returns:
        np.ndarray: BERV-corrected wavelength array

    """
    if BERV_factor is not None and np.isfinite(BERV_factor):
        return wave / BERV_factor
    return wave / (1 + BERV / SPEED_OF_LIGHT)

spectral_conditions

Selection of observations based on conditions.

When using ASTRA we can reject observations in two different ways:

  • globally - they are effectivelly discarded for all SBART operations
  • temporarily -Only disable them for the creation of the Stellar Templates

Even though the two rejection types have completely different effects, we always use the same format, as defined in :py:class:~SBART.utils.spectral_conditions.

Based on Header values

  • Allows to define bounds for the "valid" values that a given header value can take.
  • The bounds argument takes a list of list, with each entry defining a "region" that the value can take.
  • If we want to place no upper/lower limits, we can place a None

.. code-block:: python

# Only use observations that can have AIRMASS smaller than 1.5 or larger than 1.6
valid_AIRM_condition = KEYWORD_condition(KW = "airmass",
                                      bounds=[[None, 1.5], [1.6, None]]
                                     )

# Only select observations that have a previous RV error (the one from the CCF) smaller than 50 meter_second
# Note1: conditions for 'RV related' values (previousRV, drift, BERV) must be in meter_second OR kilometer_second
# Note2: units will be converted later on, so either "unit" can be chosen

valid_error_condition = KEYWORD_condition(KW="previousRV_ERR",
                                          bounds = [0, 50*meter_second])

  • It is possible to match the Keyword values to a specific one: .. code-block:: python

    valid_error_condition = KEYWORD_value(KW="OBS NAME", value = "OB NAME as defined in the header" )

Rejection based on filenames

Provide a list of bad filenames that will be rejected.

Note: Filename only, do NOT pass a path

.. code-block:: python

bad_filename_condition = FNAME_condition(filename_list=["filename.fits"])

Rejecting based on the "subInstrument"

Reject all observations from ESPRESSO21

.. code-block:: python

bad_subInst_condition = SubInstrument_condition("ESPRESSO21")

Rejecting based on Warning Flags when loading the Frames

Reject observation if a given warning flag was set when loading the spectra

.. code-block:: python

bad_subInst_condition = WarningFlag_Notset("HIERARCH ESO QC SCIRED DRIFT CHECK")

Combining conditions

To combine multiple conditions, sum them together. The observations will be rejected if any Condition decides to reject it

.. code-block:: python

full_conditions = valid_AIRM_condition + valid_error_condition + bad_filename_condition + bad_subInst_condition

Applying the conditions

For a :py:class:~SBART.data_objects.DataClass.DataClass object, we can use its :obj:~SBART.data_objects.DataClass.DataClass.reject_observations method

.. code-block:: python

data.reject_observations(full_conditions)
Referenced by:

ConditionModel

Defines the Base Condition Class, from which we can generate new conditions.

The conditions represent a boolean check that is applied to a given spectra, which is then rejected or kept, depending on the result from the check.

The ConditionModel is implemented in such a way that summing two conditions together creates a new, independent, condition that combines them all. This new, merged, conditions works as a sequential application of all conditions.

Subclassed by: Used by:
Source code in src/ASTRA/utils/spectral_conditions.py 
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
class ConditionModel:
    """Defines the Base Condition Class, from which we can generate new conditions.

    The conditions represent a boolean check that is applied to a given spectra, which is then rejected or kept,
        depending on the result from the check.

    The ConditionModel is implemented in such a way that summing two conditions together creates a new, independent,
    condition that combines them all. This new, merged, conditions works as a sequential application of all conditions.

    """

    def __init__(self):  # noqa: D107
        self._condition_list = [self.select_spectra]
        self._cond_information = [self.cond_info]

    def __add__(self, other: ConditionModel) -> ConditionModel:  # noqa: D105
        self._condition_list.append(other.select_spectra)
        self._cond_information.append(other.cond_info)
        return self

    def __radd__(self, other: ConditionModel) -> ConditionModel:  # noqa: D105
        return self.__add__(other)

    def evaluate(self, frame: Frame) -> Tuple[bool, List[Flag]]:
        """Apply all boolean checks to a given frame, returning a boolean value and a list of Flags with the results.

        Parameters
        ----------
        frame:
            frame to be validated against the conditions

        Returns
        -------
        valid_OBS:
            Boolean result of the comparison
        flags:
            List of flags, one for each condition that was applied

        """
        valid_OBS = True

        flags = []
        for condition in self._condition_list:
            output_flag = condition(frame)
            if output_flag != VALID:
                flags.append(output_flag)
                valid_OBS = False
        return valid_OBS, flags

    def select_spectra(self, frame: Frame) -> Flag:
        """To be implemented by child classes."""
        # must return a given flag and message
        return [], ""

    def __str__(self) -> str:  # noqa: D105
        return self.__repr__()

    def __repr__(self) -> str:  # noqa: D105
        return f"{self._cond_information}"

    @property
    def set_conditions(self) -> list[str]:  # noqa: D102
        return self._cond_information

    @property
    def cond_info(self) -> str:  # noqa: D102
        return "Base spectral condition"

    def write_to_disk(self, file: UI_PATH) -> None:
        """Write constraint to file."""
        file.write("Constraints on observations:\n")
        for condition in self._cond_information:
            file.write("\t" + "".join(condition) + "\n")

set_conditions: list[str] property

cond_info: str property

evaluate(frame: Frame) -> Tuple[bool, List[Flag]]

Apply all boolean checks to a given frame, returning a boolean value and a list of Flags with the results.

Parameters:

Name Type Description Default
frame Frame

frame to be validated against the conditions

 required

Returns:

Name Type Description
valid_OBS bool

Boolean result of the comparison
flags List[Flag]

List of flags, one for each condition that was applied
Source code in src/ASTRA/utils/spectral_conditions.py 
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
def evaluate(self, frame: Frame) -> Tuple[bool, List[Flag]]:
    """Apply all boolean checks to a given frame, returning a boolean value and a list of Flags with the results.

    Parameters
    ----------
    frame:
        frame to be validated against the conditions

    Returns
    -------
    valid_OBS:
        Boolean result of the comparison
    flags:
        List of flags, one for each condition that was applied

    """
    valid_OBS = True

    flags = []
    for condition in self._condition_list:
        output_flag = condition(frame)
        if output_flag != VALID:
            flags.append(output_flag)
            valid_OBS = False
    return valid_OBS, flags

select_spectra(frame: Frame) -> Flag

To be implemented by child classes.

Source code in src/ASTRA/utils/spectral_conditions.py 
161
162
163
164
def select_spectra(self, frame: Frame) -> Flag:
    """To be implemented by child classes."""
    # must return a given flag and message
    return [], ""

write_to_disk(file: UI_PATH) -> None

Write constraint to file.

Source code in src/ASTRA/utils/spectral_conditions.py 
180
181
182
183
184
def write_to_disk(self, file: UI_PATH) -> None:
    """Write constraint to file."""
    file.write("Constraints on observations:\n")
    for condition in self._cond_information:
        file.write("\t" + "".join(condition) + "\n")

KEYWORD_in_iterable

Bases: ConditionModel

Limit the kW to be a specific value.

Source code in src/ASTRA/utils/spectral_conditions.py 
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
class KEYWORD_in_iterable(ConditionModel):
    """Limit the kW to be a specific value."""

    def __init__(self, KW: str, possible_value: Iterable) -> None:
        self.KW = KW
        self.possible_value = possible_value
        super().__init__()

    def select_spectra(self, frame) -> Flag:
        """Reject if from a given sub-instrument."""

        KW_val = frame.get_KW_value(self.KW)
        if KW_val not in self.possible_value:
            flag = USER_BLOCKED(
                f"{self.KW} value ({KW_val}) is not inside({self.possible_value})"
            )
        else:
            flag = VALID
        return flag

    @property
    def cond_info(self) -> str:  # noqa: D102
        return f"Keyword {self.KW} has value {self.possible_value}"

KW = KW instance-attribute

possible_value = possible_value instance-attribute

cond_info: str property

select_spectra(frame) -> Flag

Reject if from a given sub-instrument.

Source code in src/ASTRA/utils/spectral_conditions.py 
195
196
197
198
199
200
201
202
203
204
205
def select_spectra(self, frame) -> Flag:
    """Reject if from a given sub-instrument."""

    KW_val = frame.get_KW_value(self.KW)
    if KW_val not in self.possible_value:
        flag = USER_BLOCKED(
            f"{self.KW} value ({KW_val}) is not inside({self.possible_value})"
        )
    else:
        flag = VALID
    return flag

KEYWORD_value

Bases: ConditionModel

Limit the kW to be a specific value.

Source code in src/ASTRA/utils/spectral_conditions.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
class KEYWORD_value(ConditionModel):
    """Limit the kW to be a specific value."""

    def __init__(self, KW: str, value: Any) -> None:
        self.KW = KW
        self.value = value
        super().__init__()

    def select_spectra(self, frame) -> Flag:
        """Reject if from a given sub-instrument."""

        KW_val = frame.get_KW_value(self.KW)
        if KW_val != self.value:
            flag = USER_BLOCKED(
                f"{self.KW} value ({KW_val}) does not match the required one ({self.value})"
            )
        else:
            flag = VALID
        return flag

    @property
    def cond_info(self) -> str:  # noqa: D102
        return f"Keyword {self.KW} has value {self.value}"

KW = KW instance-attribute

value = value instance-attribute

cond_info: str property

select_spectra(frame) -> Flag

Reject if from a given sub-instrument.

Source code in src/ASTRA/utils/spectral_conditions.py 
219
220
221
222
223
224
225
226
227
228
229
def select_spectra(self, frame) -> Flag:
    """Reject if from a given sub-instrument."""

    KW_val = frame.get_KW_value(self.KW)
    if KW_val != self.value:
        flag = USER_BLOCKED(
            f"{self.KW} value ({KW_val}) does not match the required one ({self.value})"
        )
    else:
        flag = VALID
    return flag

KEYWORD_condition

Bases: ConditionModel

Limit the KW to be inside the defined interval (edges included).

Source code in src/ASTRA/utils/spectral_conditions.py 
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
class KEYWORD_condition(ConditionModel):
    """Limit the KW to be inside the defined interval (edges included)."""

    def __init__(
        self, KW: str, bounds: List[tuple[Any, Any]], include_edges: bool = True
    ):
        """Define conditions.

        Args:
            KW (str): Keyword as defined by the Frames
            bounds (List[tuple[Any, Any]]): List of lower and upper bounds.
                If any value is None, assume that it goes up to +- np.inf
            include_edges (bool, optional): Include edges in comparison. Defaults to True.

        Raises:
            InvalidConfiguration: Inproper bounds
            InvalidConfiguration: Lower bound bigger than upper

        """
        self.KW = KW
        self.bounds = bounds
        super().__init__()

        if len(bounds) == 0:
            raise InvalidConfiguration("No bounds were provided")

        self._include_edges = include_edges

        if isinstance(bounds[0], (list, tuple)):
            bounds_to_check = self.bounds
            self._multiple_bounds = True
        else:
            bounds_to_check = [self.bounds]
            self._multiple_bounds = False

        self._standardize_bounds()
        for bound_entry in bounds_to_check:
            if bound_entry[0] > bound_entry[1]:
                raise InvalidConfiguration(
                    "The lower bound must be smaller than the upper one"
                )

    def _standardize_bounds(self) -> None:
        """Transform None into np.infs."""
        new_bounds = []
        for bound in self._bounds_to_check:
            new_entry = bound
            if bound[0] is None:
                new_entry[0] = -np.inf
            if bound[1] is None:
                new_entry[1] = np.inf
            new_bounds.append(new_entry)

        if not self._multiple_bounds:
            self.bounds = new_bounds[0]
        else:
            self.bounds = new_bounds

    def select_spectra(self, frame: Frame) -> Flag:
        """Apply conditions to spectra."""
        keep = False
        KW_val = frame.get_KW_value(self.KW)
        for bound_elem in self._bounds_to_check:
            if KW_val is None or not np.isfinite(KW_val):
                keep = True
                logger.warning(
                    "Frame has a NaN value for the KW: {}. Not applying the spectral condition",
                    self.KW,
                )
                break
            if self._include_edges:
                if bound_elem[0] <= KW_val <= bound_elem[1]:
                    keep = True
            elif bound_elem[0] < KW_val < bound_elem[1]:
                keep = True

            if keep:
                break

        message = f"KW {self.KW} inside the boundary {self.bounds}"
        if not keep:
            message = f"KW {self.KW} outside the boundary {self.bounds} : {KW_val}"

        flag = VALID if keep else USER_BLOCKED(message)

        return flag

    @property
    def _bounds_to_check(self) -> List[list]:
        if self._multiple_bounds:
            bounds_to_check = self.bounds
        else:
            bounds_to_check = [self.bounds]
        return bounds_to_check

    @property
    def cond_info(self) -> str:  # noqa: D102
        return f"KW {self.KW} inside window: {self.bounds}"

KW = KW instance-attribute

bounds = bounds instance-attribute

cond_info: str property

select_spectra(frame: Frame) -> Flag

Apply conditions to spectra.

Source code in src/ASTRA/utils/spectral_conditions.py 
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
def select_spectra(self, frame: Frame) -> Flag:
    """Apply conditions to spectra."""
    keep = False
    KW_val = frame.get_KW_value(self.KW)
    for bound_elem in self._bounds_to_check:
        if KW_val is None or not np.isfinite(KW_val):
            keep = True
            logger.warning(
                "Frame has a NaN value for the KW: {}. Not applying the spectral condition",
                self.KW,
            )
            break
        if self._include_edges:
            if bound_elem[0] <= KW_val <= bound_elem[1]:
                keep = True
        elif bound_elem[0] < KW_val < bound_elem[1]:
            keep = True

        if keep:
            break

    message = f"KW {self.KW} inside the boundary {self.bounds}"
    if not keep:
        message = f"KW {self.KW} outside the boundary {self.bounds} : {KW_val}"

    flag = VALID if keep else USER_BLOCKED(message)

    return flag

SubInstrument_condition

Bases: ConditionModel

Flag the observations that are from the defined subInstrument.

Source code in src/ASTRA/utils/spectral_conditions.py 
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
class SubInstrument_condition(ConditionModel):
    """Flag the observations that are from the defined subInstrument."""

    def __init__(self, subInst: str):
        self._bad_subInst = subInst
        super().__init__()

    def select_spectra(self, frame) -> Flag:
        """Reject if from a given sub-instrument."""
        if frame.is_SubInstrument(self._bad_subInst):
            message = f"Removed subInstrument: {self._bad_subInst}"
            flag = USER_BLOCKED(message)
        else:
            flag = VALID
        return flag

    @property
    def cond_info(self) -> str:  # noqa: D102
        return f"Removed subInstrument {self._bad_subInst}"

cond_info: str property

select_spectra(frame) -> Flag

Reject if from a given sub-instrument.

Source code in src/ASTRA/utils/spectral_conditions.py 
343
344
345
346
347
348
349
350
def select_spectra(self, frame) -> Flag:
    """Reject if from a given sub-instrument."""
    if frame.is_SubInstrument(self._bad_subInst):
        message = f"Removed subInstrument: {self._bad_subInst}"
        flag = USER_BLOCKED(message)
    else:
        flag = VALID
    return flag

Empty_condition

Bases: ConditionModel

Place no Condition.

Source code in src/ASTRA/utils/spectral_conditions.py 
357
358
359
360
361
362
363
364
365
366
367
368
369
class Empty_condition(ConditionModel):
    """Place no Condition."""

    def __init__(self):
        super().__init__()

    def select_spectra(self, frame: Frame) -> Flag:
        """Evaluate to valid."""
        return VALID

    @property
    def cond_info(self) -> str:
        return "No conditions"

cond_info: str property

select_spectra(frame: Frame) -> Flag

Evaluate to valid.

Source code in src/ASTRA/utils/spectral_conditions.py 
363
364
365
def select_spectra(self, frame: Frame) -> Flag:
    """Evaluate to valid."""
    return VALID

WarningFlag_Notset

Bases: ConditionModel

Reject the observation if the given warning flag is True, as set through the KW_WARNING flag of QC routines.

Source code in src/ASTRA/utils/spectral_conditions.py 
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
class WarningFlag_Notset(ConditionModel):
    """Reject the observation if the given warning flag is True, as set through the KW_WARNING flag of QC routines."""

    def __init__(self, flag_name: str, full_flag=False):
        """WarningFlag_Notset.

        Args:
            flag_name (str): Name of the flag
            full_flag (bool, optional): If True, applies the condition to search for the exact flag_name (
                i.e. KW_WARNING(flag_name)) Otherwise, searches for something in the form of
                msg = f"QC flag {self.flag_name} meets the bad value". Defaults to False.

        """
        self.flag_name = flag_name
        self.full_flag = full_flag
        super().__init__()

    @override
    def select_spectra(self, frame: Frame) -> Flag:
        """Keep observations that don't have a given warning."""
        if self.full_flag:
            msg = self.flag_name
        else:
            msg = f"QC flag {self.flag_name} meets the bad value"
        KW_flag = KW_WARNING(msg)
        if frame.status.check_if_warning_exists(KW_flag):
            message = f"Frame has the KW warning flag {self.flag_name} active"
            flag = USER_BLOCKED(message)
        else:
            flag = VALID

        return flag

    @property
    def cond_info(self) -> str:
        return f"Warning KW flag {self.flag_name} was raised"

flag_name = flag_name instance-attribute

full_flag = full_flag instance-attribute

cond_info: str property

select_spectra(frame: Frame) -> Flag

Keep observations that don't have a given warning.

Source code in src/ASTRA/utils/spectral_conditions.py 
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
@override
def select_spectra(self, frame: Frame) -> Flag:
    """Keep observations that don't have a given warning."""
    if self.full_flag:
        msg = self.flag_name
    else:
        msg = f"QC flag {self.flag_name} meets the bad value"
    KW_flag = KW_WARNING(msg)
    if frame.status.check_if_warning_exists(KW_flag):
        message = f"Frame has the KW warning flag {self.flag_name} active"
        flag = USER_BLOCKED(message)
    else:
        flag = VALID

    return flag

FNAME_condition

Bases: ConditionModel

Flag the observations that have filename inside the filename_list list.

Parameters:

Name Type Description Default
filename_list list

List of files to either be outright used or list of paths to .txt files from which we can select observations

 required
only_keep_filenames

Only keep the filenames that were selected. Default: False

 False
load_from_file bool

If True (default False), use the filename_list parameter to provide a list of files that will be selected

 False
Source code in src/ASTRA/utils/spectral_conditions.py 
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
class FNAME_condition(ConditionModel):
    """Flag the observations that have filename inside the filename_list list.

    Parameters
    ----------
    filename_list: list
        List of files to either be outright used or list of paths to .txt files from which we can select
        observations
    only_keep_filenames: bool
        Only keep the **filenames** that were selected. Default: False
    load_from_file: bool
        If True (default False), use the filename_list parameter to provide a list of files that will be
        selected

    """

    def __init__(
        self,
        filename_list: list,
        only_keep_filenames=False,
        load_from_file: bool = False,
    ):
        self._load_from_file = load_from_file
        if self._load_from_file:
            logger.info(
                f"Loading files to 'condition' from a disk file: {filename_list}"
            )
            files_to_reject = []
            for entry in filename_list:
                if not isinstance(entry, Path):
                    entry = Path(entry)
                if not entry.name.endswith("txt"):
                    raise custom_exceptions.InvalidConfiguration(
                        "File to load must be txt"
                    )
                with open(entry) as file:
                    files_to_reject.extend(file.readlines())

            filename_list = list(map(lambda x: x.replace("\n", ""), files_to_reject))

        self._filename_list = filename_list
        self._only_keep_filenames = only_keep_filenames
        super().__init__()

    @override
    def select_spectra(self, frame) -> Flag:
        if not self._only_keep_filenames:
            if frame.fname in self._filename_list:
                message = "Filename rejected"
                flag = USER_BLOCKED(message)
            else:
                flag = VALID
        elif frame.fname in self._filename_list:
            flag = VALID
        else:
            message = "Filename rejected"
            flag = USER_BLOCKED(message)
        return flag

    @property
    def cond_info(self) -> str:
        return f"Filename list {self._filename_list} - only keep: {self._only_keep_filenames}"

cond_info: str property

select_spectra(frame) -> Flag

Source code in src/ASTRA/utils/spectral_conditions.py 
454
455
456
457
458
459
460
461
462
463
464
465
466
467
@override
def select_spectra(self, frame) -> Flag:
    if not self._only_keep_filenames:
        if frame.fname in self._filename_list:
            message = "Filename rejected"
            flag = USER_BLOCKED(message)
        else:
            flag = VALID
    elif frame.fname in self._filename_list:
        flag = VALID
    else:
        message = "Filename rejected"
        flag = USER_BLOCKED(message)
    return flag

SNR_condition

Bases: ConditionModel

Reject observations based on the order-wise SNR.

Compares the SNR of the valid orders against the minimum SNR that is provided to this object.

Mostly useful for the creation of the stellar template

Source code in src/ASTRA/utils/spectral_conditions.py 
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
class SNR_condition(ConditionModel):
    """Reject observations based on the order-wise SNR.

    Compares the SNR of the valid orders against the minimum SNR that is provided to this
    object.

    Mostly useful for the creation of the stellar template
    """

    def __init__(self, minimum_SNR: float):
        self.minimum_SNR = minimum_SNR
        super().__init__()

        if minimum_SNR <= 0:
            raise InvalidConfiguration(f"SNR must be >= 0 ({minimum_SNR})")

    @override
    def select_spectra(self, frame):
        KW = np.asarray(frame.get_KW_value("orderwise_snrs"))
        valid_orders = list(frame.valid_orders())
        message = f"Did not pass SNR cutoff: {self.minimum_SNR:}"
        flag = (
            VALID
            if np.any(KW[valid_orders] < self.minimum_SNR)
            else USER_BLOCKED(message)
        )
        return flag

    @property
    def cond_info(self):
        return f"Minimum order-wise SNR of: {self.minimum_SNR}"

minimum_SNR = minimum_SNR instance-attribute

cond_info property

select_spectra(frame)

Source code in src/ASTRA/utils/spectral_conditions.py 
490
491
492
493
494
495
496
497
498
499
500
@override
def select_spectra(self, frame):
    KW = np.asarray(frame.get_KW_value("orderwise_snrs"))
    valid_orders = list(frame.valid_orders())
    message = f"Did not pass SNR cutoff: {self.minimum_SNR:}"
    flag = (
        VALID
        if np.any(KW[valid_orders] < self.minimum_SNR)
        else USER_BLOCKED(message)
    )
    return flag

telluric_utilities

binary_mask_from_template

create_binary_template(transmittance, continuum_level, percentage_drop) -> None

Transform the transmittance spectra into a binary template.

This is based on percentage deviations from the continuum level.

Parameters:

Name Type Description Default
continuum_level ndarray()

Continuum level that is estimated by the children classes

 required
Source code in src/ASTRA/utils/telluric_utilities/binary_mask_from_template.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
def create_binary_template(transmittance, continuum_level, percentage_drop) -> None:
    """Transform the transmittance spectra into a binary template.

    This is based on percentage deviations from the continuum level.

    Parameters
    ----------
    continuum_level : np.ndarray()
        Continuum level that is estimated by the children classes

    """
    logger.info("Converting from transmittance spectra to binary mask!")

    # Find a decrease of 1% in relation to the continuum level; Positive
    # gains (against the continuum value) are not considered as tellurics
    percentages = (transmittance - continuum_level) / continuum_level
    telluric_indexes = np.where(percentages < -percentage_drop / 100)

    # We want a binary template
    telluric_mask = np.zeros_like(transmittance, dtype=int)
    telluric_mask[telluric_indexes] = 1

    return telluric_mask

calculate_telluric_indexes

Compute telluric indexes.

calculate_telluric_mask(data_class: DataClass, telluric_temp: TelluricTemplate) -> np.ndarray

Construct the telluric mask.

Source code in src/ASTRA/utils/telluric_utilities/calculate_telluric_indexes.py 
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
def calculate_telluric_mask(data_class: DataClass, telluric_temp: TelluricTemplate) -> np.ndarray:
    """Construct the telluric mask."""
    if telluric_temp is None:
        return None
    tell_template = telluric_temp.template
    tell_waves = telluric_temp.wavelengths

    spectra_wavelengths, _ = data_class.wavelengths

    new_mask = np.zeros(spectra_wavelengths.shape, dtype=bool)
    for epoch in range(data_class.number_of_epochs):
        for order in range(data_class.mat_size[0]):
            spectra_wave_order = spectra_wavelengths[epoch][order]
            temp_order = tell_template[order]
            ratios = np.zeros(temp_order.shape)

            telluric_blocks = build_blocks(np.where(temp_order == 1))

            for block in telluric_blocks:
                lower_wave = tell_waves[order][block[0]]
                higher_wave = tell_waves[order][block[-1]]

                ratios[
                    np.where(
                        np.logical_and(
                            spectra_wave_order >= lower_wave,
                            spectra_wave_order <= higher_wave,
                        ),
                    )
                ] = 1

            new_mask[epoch][order][np.where(ratios >= 0.5)] = True

    return new_mask

clean_wings

clean_wings_order(current_order_wavelengths, telluric_bin_temp, template_base_berv, BERV_window)

Around each telluric feature, remove a window that spans the entire BERV motion over the year.

Parameters:

Name Type Description Default
current_order_wavelengths ndarray

Wavelengths associated with the template

 required
telluric_bin_temp ndarray

Telluric (binary) template

 required
template_base_berv float

BERV value from the template

 required
BERV_window tuple / list

Minimum and Maximum BERV

 required

Returns:

Type Description
ndarray

Binary template with the increased span
Source code in src/ASTRA/utils/telluric_utilities/clean_wings.py 
 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
def clean_wings_order(current_order_wavelengths, telluric_bin_temp, template_base_berv, BERV_window):
    """Around each telluric feature, remove a window that spans the entire BERV motion over the year.

    Parameters
    ----------
    current_order_wavelengths : np.ndarray
        Wavelengths associated with the template
    telluric_bin_temp : np.ndarray
        Telluric (binary) template
    template_base_berv : float
        BERV value from the template
    BERV_window : tuple/list
        Minimum and Maximum BERV

    Returns
    -------
    np.ndarray
        Binary template with the increased span

    """
    telluric_positions = np.where(telluric_bin_temp == 1)

    blocks = build_blocks(telluric_positions)
    # set 5 points to each side as a telluric point
    c = 299792.458  # km/s

    for BERV_block in blocks:
        start, end = BERV_block[0], BERV_block[-1]

        # removes the previous BERV correction and calculates the edges based on the MAXBERV window!
        # The template was created for the
        lowest_wavelength = current_order_wavelengths[start] * (c - BERV_window) / (c + template_base_berv)
        highest_wavelength = current_order_wavelengths[end] * (c + BERV_window) / (c + template_base_berv)
        telluric_bin_temp[
            np.where(
                np.logical_and(
                    current_order_wavelengths >= lowest_wavelength,
                    current_order_wavelengths <= highest_wavelength,
                ),
            )
        ] = 1

    return telluric_bin_temp

compute_common_blocks

Compute overlap between multiple elements.

others = [[[1, 6.1], [7, 10]], [[0, 4], [6, 8], [9, 11]], [[1, 12.2]]] module-attribute

out = compute_common(others) module-attribute

off = 0.1 * obs_index module-attribute

check_if_overlap(block_A, block_B)

Source code in src/ASTRA/utils/telluric_utilities/compute_common_blocks.py 
4
5
def check_if_overlap(block_A, block_B):
    return block_B[0] <= block_A[0] <= block_B[1] or block_A[0] <= block_B[0] <= block_A[1]

compute_common(values)

Source code in src/ASTRA/utils/telluric_utilities/compute_common_blocks.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
def compute_common(values):
    basis = values[0]

    for OBS_data in values[1:]:
        new_basis = []

        for base_index, base in enumerate(basis):
            for wave_block in OBS_data:
                if check_if_overlap(base, wave_block):
                    new_basis.append(base.copy())

                    new_basis[-1][0] = max(wave_block[0], base[0])
                    new_basis[-1][1] = min(wave_block[1], base[1])

            basis = new_basis
    return basis

compute_overlaps_blocks

Compute overlaps between multiple lists.

check_if_overlap(block_A, block_B) -> bool

Check if there is an overlap between two blocks.

Source code in src/ASTRA/utils/telluric_utilities/compute_overlaps_blocks.py 
6
7
8
def check_if_overlap(block_A, block_B) -> bool:
    """Check if there is an overlap between two blocks."""
    return block_B[0] <= block_A[0] <= block_B[1] or block_A[0] <= block_B[0] <= block_A[1]

find_overlaps(input_blocks: List[List[float]]) -> List[List[float]]

Loop over a list with 1D regions and find intersections among them.

Used to decrease the number of "standalone" blocks in the telluric template

E.g:

This:

|----| |----| |----| Will be merged into a single "block", as they are all intersected

Parameters:

Name Type Description Default
input_blocks List[List[float]]

[description]

 required

Returns:

Type Description
List[List[float]]

[description]
TODO
  • [ ] See if there is a need of porting this function to cython
Source code in src/ASTRA/utils/telluric_utilities/compute_overlaps_blocks.py 
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
def find_overlaps(input_blocks: List[List[float]]) -> List[List[float]]:
    """Loop over a list with 1D regions and find intersections among them.

    Used to decrease the number of "standalone" blocks in the telluric template

    E.g:

    This:

    |----| |----|
        |----|
    Will be merged into a single "block", as they are all intersected

    Parameters
    ----------
    input_blocks : List[List[float]]
        [description]

    Returns
    -------
    List[List[float]]
        [description]

    TODO
    ------

    - [ ] See if there is a need of porting this function to cython

    """
    found_overlap = False
    new_blocks = []
    found_matches = set()
    for block_index, block in enumerate(input_blocks):
        min_loc, max_loc = block

        if block_index in found_matches:
            # if the block has already been merged into another, ignore it!
            continue
        for comp_index, block_under_comp in enumerate(input_blocks[block_index + 1 :]):
            current_index = block_index + comp_index + 1

            if check_if_overlap(block, block_under_comp):
                found_overlap = True
                found_matches.add(current_index)

                min_loc = min(min_loc, block_under_comp[0])
                max_loc = max(max_loc, block_under_comp[1])

        if block_index not in found_matches:
            found_matches.add(block_index)
            new_blocks.append((min_loc, max_loc))

    if found_overlap:
        return find_overlaps(new_blocks)

    return input_blocks

units

Collection of utilities to handle units throughout the code base.

Aside from the data conversion function, it also provides three RV "measurements", which are astropy Quantities:

  • meter_second
  • kilometer_second
  • centimeter_second

meter_second = 1 * units.m / units.second module-attribute

kilometer_second = 1 * units.km / units.second module-attribute

centimeter_second = 1 * units.cm / units.second module-attribute

test_list = [meter_second, centimeter_second] module-attribute

convert_data(data: Union[data_vector, RV_measurement], new_units: Optional[RV_measurement] = None, as_value: bool = False) -> data_vector | RV_measurement

Accept input data that are astropy.units and converts it to a new unit and/or to a numerical value.

Parameters:

Name Type Description Default
data Union[data_vector, RV_measurement]

Input data

 required
new_units Optional[RV_measurement]

Units to which each element will be converted to

 None
as_value bool

Switch the list entries to numerical values

 False

Returns:

Name Type Description
converted_list list

Data in the specified format
Source code in src/ASTRA/utils/units.py 
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
def convert_data(
    data: Union[data_vector, RV_measurement],
    new_units: Optional[RV_measurement] = None,
    as_value: bool = False,
) -> data_vector | RV_measurement:
    """Accept input data that are astropy.units and converts it to a new unit and/or to a numerical value.

    Parameters
    ----------
    data :
        Input data
    new_units :
        Units to which each element will be converted to
    as_value :
        Switch the list entries to numerical values

    Returns
    -------
    converted_list : list
        Data in the specified format

    """
    single_elem = False
    data_to_process = data

    if not isinstance(data, (list, tuple)):
        single_elem = True
        data_to_process = [data]

    if new_units is not None:
        converted_list = [i.to(new_units) for i in data_to_process]
    else:
        converted_list = data_to_process

    if as_value:
        converted_list = [i.value for i in converted_list]

    return converted_list if not single_elem else converted_list[0]