Data_objects

DataClass

Used by:

• API
  • API
    • template_creation telluric_templates Telluric_Template TelluricTemplate load_information_from_DataClass
    • utils poet_SDO_plotter plot_SDO_image
  • Construction of templates Telluric_Template TelluricTemplate load_information_from_DataClass
  • Data_objects poet_SDO_plotter plot_SDO_image

DataClass

Bases: BASE

Standard interface to handle data.

The user-facing object that handles the loading and data access to the spectral data, independently of the instrument. Furthermore, this must be launched as a proxyObject (insert docs here) in order to avoid problems with data syncronization and optimize the speed of the code.

To use this class in SBART RV extraction routines, we place it in shared memory, allowing all processes to easily access it. This is done with a proxyObject <https://docs.python.org/3.8/library/multiprocessing.html>_.

SBART already provides a DataClass object that is wrapped by a proxyObject:

    from ASTRAdata_objects import DataClassManager
    manager = DataClassManager()
    manager.start()
    data_object = manager.DataClass(*args, **kwargs)

This data_object has all the functions that the DataClass object implements!

Used by:

• API
  • API template_creation telluric_templates telluric_from_telfit TelfitTelluric create_telluric_template
  • Construction of templates telluric_from_telfit TelfitTelluric create_telluric_template

Source code in src/ASTRA/data_objects/DataClass.py

class DataClass(BASE):
    """Standard interface to handle data.

    The user-facing object that handles the loading and data access to the spectral data, independently of the instrument.
    Furthermore, this must be launched as a proxyObject (insert docs here) in order to avoid problems with data syncronization
    and optimize the speed of the code.

    To use this class in SBART RV extraction routines, we place it in shared memory, allowing all processes to easily access
        it. This is done with a `proxyObject <https://docs.python.org/3.8/library/multiprocessing.html>`_.

        SBART already provides a DataClass object that is wrapped by a proxyObject:

    ```py

        from ASTRAdata_objects import DataClassManager
        manager = DataClassManager()
        manager.start()
        data_object = manager.DataClass(*args, **kwargs)
    ```

    This *data_object* has all the functions that the DataClass object implements!
    """

    extra_loading_functions: Dict[str, Callable] = {
        "CARMENES": load_CARMENES_extra_information,
        "POET": load_POET_extra_information,
    }

    def __init__(
        self,
        input_files: Iterable[UI_PATH],
        storage_path: UI_PATH,
        instrument: Type[Frame],
        instrument_options: dict[str, Any] | None = None,
        reject_subInstruments: Iterable[str] | None = None,
        target_name: str = None,
        sigma_clip_RVs: float | None = None,
        target_dictionary_path: UI_PATH | None = None,
    ):
        """Parameters
        ----------
        input_files:
            Either a path to a txt file, or a list of S2d files
        storage_path:
            Root path of the SBART outputs
        instrument:
            Instrument that we will be loading data from. Must be an object of type ASTRAInstruments
        instrument_options
        reject_subInstruments
            Iterable with names of subInstruments to automatically reject
        target_name:
            Original name of the target. To be deprecated
        sigma_clip_RVs:
            If it is a positive integer, reject frames with a sigma clip on the DRS RVs
        target_dictionary_path:
            File with two columns: first, target names; second: simbad resolvable mapping

        """
        instrument_options = {} if instrument_options is None else instrument_options
        super().__init__()
        self.sigma_clip_RVs = sigma_clip_RVs

        self._inst_type: Frame = instrument
        self.input_file = input_files

        # Hold all of the frames
        self.observations: List[Frame, ...] = []

        self.metaData = MetaData()

        if reject_subInstruments is not None:
            logger.warning(f"Rejecting subInstruments: {reject_subInstruments}")

        self.generate_root_path(storage_path)
        self.frameID_map = defaultdict(list)

        self.add_data_from_instrument(
            input_files=input_files,
            instrument=instrument,
            reject_subInstruments=reject_subInstruments,
            instrument_options=instrument_options,
        )

        N_files = len(self.observations)
        logger.debug("Selected {} observations from disk", N_files)
        if N_files == 0:
            raise InvalidConfiguration("Input has zero files")

        self._collect_MetaData()

        #  TODO: find a better way of doing this!
        self.Target = Target(
            self.collect_KW_observations(
                "OBJECT", self._inst_type.sub_instruments, include_invalid=False
            ),
            original_name=target_name,
            target_dictionary_path=target_dictionary_path,
        )

        self._validate_loaded_observations()
        self.show_loadedData_table()

        self._applied_telluric_removal = False

        self.StellarModel = None

        self.load_instrument_extra_information()

        for frame in self.observations:
            frame.initialize_modelling_interface()
            frame.finalize_data_load()

    def construct_subInstruments_by_obID(self) -> None:
        """Create new sub-Instruments with the obID of the observations"""
        logger.warning("Re-definning sub-Instruments based on the OBID")

        ob_ids, frameIDS = self.collect_KW_observations(
            KW="OBS NAME",
            subInstruments=self.get_subInstruments_with_valid_frames(),
            return_frameIDs=True,
        )
        mapa = defaultdict(list)
        for OB, ids in zip(ob_ids, frameIDS):
            mapa[OB].append(ids)

        new_submap = defaultdict(list)
        for new_inst, all_frameID in mapa.items():
            for frameID in all_frameID:
                frame = self.get_frame_by_ID(frameID)
                frame.overwrite_sub_instrument(new_inst)
                new_submap[new_inst].append(frameID)
        self.frameID_map = new_submap

    def add_data_from_instrument(
        self,
        input_files: Iterable[UI_PATH],
        instrument: Type[Frame],
        reject_subInstruments: Iterable[str] | None = None,
        custom_subIns_name: str | None = None,
        instrument_options: dict[str, Any] | None = None,
    ) -> None:

        OBS_list = []
        if isinstance(input_files, (str, Path)):
            logger.info("DataClass loading data from {}", self.input_file)
            with open(input_files) as input_file:
                for line in input_file:
                    OBS_list.append(Path(line.split("\n")[0]))

        elif isinstance(input_files, Iterable):
            logger.info(
                "DataClass opening {} files from a list/tuple", len(input_files)
            )

            OBS_list = [Path(i) if isinstance(i, str) else i for i in input_files]
        else:
            raise TypeError

        names = [i.name for i in OBS_list]

        if len(set(names)) != len(names):
            msg = "Ingesting repeated observations into s-BART"
            logger.critical(msg)
            raise InvalidConfiguration(msg)

        def build_obs(args):
            index, filepath = args
            frameID = hash_path_to_int(filepath.stem)
            try:
                return self._inst_type(
                    filepath,
                    instrument_options,
                    reject_subInstruments,
                    frameID=frameID,
                    quiet_user_params=frameID
                    != 0,  # Only the first frame will output logs
                )
            except Exception as e:
                logger.critical(f"Failed in {filepath} due to {e}")
                raise e

        if instrument is not MAROONX:
            with ThreadPoolExecutor() as ex:
                observations = list(ex.map(build_obs, enumerate(OBS_list)))
        else:
            observations = list(map(build_obs, enumerate(OBS_list)))
        all_ids = [i.frameID for i in self.observations]
        if len(all_ids) != len(set(all_ids)):
            raise custom_exceptions.InternalError("Clash in frameID, repeated numbers")

        # The Frames don't store data inside the Iteration folder!
        frame_root_path = self._internalPaths.root_storage_path
        for frame in observations:
            if custom_subIns_name is not None:
                frame.overwrite_sub_instrument(custom_subIns_name)

            frame.generate_root_path(frame_root_path)
        self.observations.extend(observations)
        self._build_frameID_map()

        if custom_subIns_name is not None:
            self.show_loadedData_table()

    ########################
    #    Operate on Data   #
    ########################
    def get_extra_loading_functions(self):
        return self.extra_loading_functions

    def load_previous_SBART_results(self, RV_holder, use_merged_cube: bool = False):
        """Load results of SBART run, store RV and uncertainty inside corresponding Frame object."""
        logger.info("Loading RVs from previous SBART run as the starting-RVs")
        for _, frameID in enumerate(self.get_valid_frameIDS()):
            frame = self.get_frame_by_ID(frameID)
            cube = RV_holder.get_RV_cube(frame.sub_instrument, merged=use_merged_cube)

            try:
                _, sbart_rv, sbart_uncert = cube.get_RV_from_ID(
                    frameID=frameID,
                    which="SBART",
                    apply_SA_corr=False,
                    as_value=False,
                    units=None,
                    apply_drift_corr=False,
                )
                previous_filename = cube.cached_info["date_folders"][
                    cube.frameIDs.index(frameID)
                ]
                if previous_filename != frame.file_path:
                    msg = f"Loading RVs from cube with different frameID layouts of {frame.sub_instrument} ({previous_filename} vs {frame.file_path})"
                    logger.critical(msg)
                    raise InvalidConfiguration(msg)
            except ValueError:
                sbart_rv = 0 * meter_second
                sbart_uncert = 10 * meter_second

            frame.store_previous_SBART_result(RV=sbart_rv, RV_err=sbart_uncert)
        logger.info("Finished loading of previous s-BART RVs")

    def reject_order_region_from_frame(self, frameID: int, order: int, region):
        frame = self.get_frame_by_ID(frameID)
        frame.reject_wavelength_region_from_order(order, region)
        frame.close_arrays()  # ensure that the arrays are closed and that the next call will load data

    def remove_activity_lines(self, lines: Indicators) -> None:
        """Find the wavelength windows in which activity-related lines are expected to appear.

        This is done for all valid observations

        Parameters
        ----------
        lines : Indicators
            Object with the wavelength blocks "original" positions.

        # TODO: also allow to use the previous SBART RVs for this!!!!

        """
        logger.info("Computing activity windows for each RV measurements")
        for frameID in self.get_valid_frameIDS():
            frame = self.get_frame_by_ID(frameID)
            blocked_regions = lines.compute_forbidden_wavelengths(
                frame.previous_RV_measurements[0]
            )
            frame.mark_wavelength_region(
                reason=ACTIVITY_LINE, wavelength_blocks=blocked_regions
            )

    def remove_telluric_features(self, Telluric_Template: TelluricModel) -> None:
        for subInstrument in self.get_subInstruments_with_valid_frames():
            valid_frameIDS = self.get_frameIDs_from_subInst(subInstrument)

            subInst_template = Telluric_Template.request_data(subInstrument)

            for frameID in valid_frameIDS:
                if subInst_template.for_feature_removal:
                    self.get_frame_by_ID(frameID).mark_wavelength_region(
                        reason=TELLURIC,
                        wavelength_blocks=subInst_template.contaminated_regions,
                    )
                elif subInst_template.for_feature_correction:
                    (
                        wavelengths,
                        model,
                        model_uncertainty,
                    ) = subInst_template.fit_telluric_model_to_frame(
                        self.get_frame_by_ID(frameID)
                    )

                    self.get_frame_by_ID(frameID).apply_telluric_correction(
                        wavelengths=wavelengths,
                        model=model,
                        model_uncertainty=model_uncertainty,
                    )
                else:
                    raise custom_exceptions.InternalError(
                        "Why aren't we correcting or removing tellurics?"
                    )
        self._applied_telluric_removal = True

    def replace_frames_with_S2D_version(
        self, new_shape: Optional[Tuple[int, int]] = None
    ):
        """In-place substitution of all frames with their S2D-compatible shapes."""
        logger.warning("Transforming the frames to have a S2D-compatible shape")
        for index, frame in enumerate(self.observations):
            s2d_frame = frame.copy_into_S2D(new_S2D_size=new_shape)
            s2d_frame.build_mask()
            self.observations[index] = s2d_frame
            del frame

    def ingest_StellarModel(self, Stellar_Model: StellarModel) -> None:
        logger.debug("Ingesting StellarModel into the DataClass")
        if self.StellarModel is not None:
            logger.warning(
                "Stellar template has already been ingested. Switching old template by the new one"
            )

        # Empty update just to ensure initialization of the modelling interfaces
        Stellar_Model.initialize_modelling_interfaces({})

        self.StellarModel = Stellar_Model

    def reject_observations(self, conditions: CondModel) -> None:
        """Apply the conditions to evaluate if the VALID frame meets the specified conditions or not.

        Parameters
        ----------
        conditions:
            Conditions set by the user

        """
        if conditions is None:
            return

        blocked_frames = 0

        for frameID in self.get_valid_frameIDS():
            frame = self.get_frame_by_ID(frameID)
            keep, flags = conditions.evaluate(frame)

            if not keep:
                for flag in flags:
                    frame.add_to_status(flag)
                blocked_frames += 1

        logger.info(
            "User conditions removed {} / {} frames",
            blocked_frames,
            len(self.observations),
        )

        logger.info("Updated observation on disk!")
        self.show_loadedData_table()

        if len(self.get_valid_frameIDS()) == 0:
            raise NoDataError("All observations have been blocked")

    ########################
    #    Sanity Control    #
    ########################
    def _validate_loaded_observations(self) -> None:
        """Check if the same DRS version is used across all loaded files."""
        for equal_KW in [
            "DRS-VERSION",
            "SPEC_TYPE",
            "DRS_CCF_MASK",
            "DRS_FLUX_CORRECTION_TEMPLATE",
        ]:
            collected_KW = set(
                self.collect_KW_observations(equal_KW, self._inst_type.sub_instruments)
            )

            if len(collected_KW) != 1:
                logger.warning(
                    "Different values for the KW value <{}>: {}", equal_KW, collected_KW
                )

        mapping_values = defaultdict(list)
        for fID in self.get_valid_frameIDS():
            frame = self.get_frame_by_ID(fID)
            mapping_values["BLAZE"].append(frame.is_blaze_corrected)
            mapping_values["SKYSUB"].append(frame.is_skysub)

        for key, value in mapping_values.items():
            if len(set(value)) != 1:
                logger.critical(
                    f"We have a different correction type in {key}:{set(value)}"
                )

        if self.sigma_clip_RVs is not None:
            logger.info(
                f"Rejecting frames that are more than {self.sigma_clip_RVs} sigma away from mean RV"
            )

            for subInstrument in self.get_subInstruments_with_valid_frames():
                RV = self.collect_RV_information(
                    KW="DRS_RV",
                    subInst=subInstrument,
                    include_invalid=False,
                    as_value=True,
                )
                err = self.collect_RV_information(
                    KW="DRS_RV_ERR",
                    subInst=subInstrument,
                    include_invalid=False,
                    as_value=True,
                )
                mean_RV = np.median(RV)
                metric = np.std(
                    RV,
                )  # using the same sigma clip as https://docs.scipy.org/doc/scipy/reference/generated/scipy.stats.sigmaclip.html

                bounds = [
                    mean_RV - self.sigma_clip_RVs * metric,
                    mean_RV + self.sigma_clip_RVs * metric,
                ]

                bad_indexes = np.where(np.logical_or(bounds[0] > RV, bounds[1] < RV))
                valid_frameIDs = self.get_frameIDs_from_subInst(subInstrument)
                bad_IDS = np.asarray(valid_frameIDs)[bad_indexes]

                for frameID_to_reject in bad_IDS:
                    bad_frame = self.get_frame_by_ID(frameID_to_reject)
                    logger.warning(
                        f"{bad_frame} rejected due to sigma clipping of DRS RVs"
                    )
                    bad_frame.add_to_status(SIGMA_CLIP_REJECTION)

                logger.info(
                    f"Sigma clip rejected {len(bad_IDS)} frames of {subInstrument}"
                )

    def _collect_MetaData(self) -> None:
        """Collect information from the individual (valid) observations to store inside the MetaData object"""
        logger.info("Collecting MetaData from the observations")
        for subInst in self.get_subInstruments_with_valid_frames():
            meta_search = {
                "DRS-VERSION": [],
                "MD5-CHECK": [],
            }

            for key in meta_search:
                meta_search[key] = self.collect_KW_observations(key, [subInst])

            count = len(self.get_frameIDs_from_subInst(subInst))

            for key, items in meta_search.items():
                self.metaData.add_info(key, items, subInst)

            self.metaData.add_info("N_OBS", [count], subInst)
            self.metaData.add_info("SBART_VERSION", [__version__], subInst)

    ########################
    #    access data       #
    ########################

    def get_Target(self) -> Target:
        return self.Target

    def load_frame_by_ID(self, frameID: int) -> int:
        frame = self.get_frame_by_ID(frameID)
        frame.load_data()
        return 1

    def set_all_as_Zscore_frames(self) -> None:
        logger.warning("Setting all frames as a zero-mean unit variance")
        for frameID in self.get_valid_frameIDS():
            frame = self.get_frame_by_ID(frameID)
            frame.set_frame_as_Zscore()

    def normalize_all(self) -> None:
        """Launch the normalization for all (valid) frames.

        Returns
        -------

        """
        logger.info("Normalizing all frames")
        for frameID in self.get_valid_frameIDS():
            frame = self.get_frame_by_ID(frameID)
            frame._never_close = True

        for subInst in self.get_subInstruments_with_valid_frames():
            self.normalize_all_from_subInst(subInst)

    def update_uncertainties_of_frame(self, frameID: int, new_uncerts) -> None:
        """Update the flux uncertainties of a given frame.

        .. warning::
            This will change your measured data, be careful with this!!!
        """
        logger.info(f"Setting uncertainties of frame (ID = {frameID}")

        frame = self.get_frame_by_ID(frameID)
        frame.update_uncertainties(new_uncerts)
        frame._never_close = True
        logger.warning("This frame will never close until SBART finished!")

    def normalize_all_from_subInst(self, subInst: str) -> None:
        """Normalize all (valid) frames from a given subInstrument.

        Parameters
        ----------
        subInst

        """
        logger.debug(f"Normalizing all frames from {subInst}")
        for fId in self.get_frameIDs_from_subInst(subInst):
            frame = self.get_frame_by_ID(fId)
            frame.normalize_spectra()

    def scale_up_all_observations(self, factor: float) -> None:
        """Multiply the flux and uncertainties by a given flux level (avoid possible SNR issues)."""
        logger.warning(f"Scaling up all spectra by a factor of {factor}")
        for fId in self.get_valid_frameIDS():
            frame = self.get_frame_by_ID(fId)
            frame.scale_spectra(factor)

    def set_stellar_RV_to_zero(self) -> None:
        """Allow to override the stellar RV so that it is fixed to zero."""
        for fID in self.get_valid_frameIDS():
            f = self.get_frame_by_ID(fID)
            f.observation_info["DRS_RV"] = 0 * kilometer_second

    def load_all_from_subInst(self, subInst: str) -> int:
        """Load all valid frames from a given subInstrument."""
        logger.debug("Opening all frames from {}", subInst)
        IDs_from_subInst = self.get_frameIDs_from_subInst(
            subInstrument=subInst, include_invalid=False
        )
        for frameID in IDs_from_subInst:
            try:
                self.load_frame_by_ID(frameID)
            except FrameError:
                logger.warning("Trying to open invalid frame")
        return 1

    def close_frame_by_ID(self, frameID: int) -> int:
        frame = self.get_frame_by_ID(frameID)
        frame.close_arrays()
        return 1

    def min_pixel_in_order(self) -> int:
        return self.observations[0].min_pixel_in_order

    def get_subInst_from_frameID(self, frameID: int) -> str:
        return self.get_frame_by_ID(frameID).sub_instrument

    def get_frame_OBS_order(
        self, frameID: int, order: int, include_invalid: bool = False
    ):
        """Request the data from one spectral order.

        Parameters
        ----------
        frameID
            FrameID of the observation
        order
            NUmber of order
        include_invalid
            Orders can also be flagged as "invalid". If False, the frame will raise an error when requesting the data

        Returns
        -------

        """
        frame = self.get_frame_by_ID(frameID)
        return frame.get_data_from_spectral_order(order, include_invalid)

    def update_interpol_properties_of_all_frames(self, new_properties: Dict[str, Any]):
        if not isinstance(new_properties, dict):
            raise custom_exceptions.InvalidConfiguration(
                "The interpolation properties must be passed as a dictionary"
            )

        for frame in self.observations:
            frame.update_user_configs(new_properties)

    def update_interpol_properties_of_stellar_model(
        self, new_properties: Dict[str, Any]
    ):
        if not isinstance(new_properties, dict):
            raise custom_exceptions.InvalidConfiguration(
                "The interpolation properties must be passed as a dictionary"
            )

        if self.StellarModel is None:
            raise custom_exceptions.NoDataError("The Stellar Model wasn't ingested")
        self.StellarModel.update_user_configs(new_properties)

    def update_frame_interpol_properties(self, frameID, new_properties) -> None:
        """Allow to update the interpolation settings from the outside, so that any object can configure
        the interpolation as it wishes
        """
        frame = self.get_frame_by_ID(frameID)
        frame.set_interpolation_properties(new_properties)

    def interpolate_frame_order(
        self,
        frameID,
        order,
        new_wavelengths,
        shift_RV_by,
        RV_shift_mode,
        include_invalid=False,
    ):
        """Interpolate a given order to a new wavelength solution"""
        frame = self.get_frame_by_ID(frameID)
        return frame.interpolate_spectrum_to_wavelength(
            order=order,
            new_wavelengths=new_wavelengths,
            shift_RV_by=shift_RV_by,
            RV_shift_mode=RV_shift_mode,
            include_invalid=include_invalid,
        )

    def get_frame_arrays_by_ID(self, frameID: int):
        """Access data from the entire spectral range (i.e. all orders come as a matrix)

        Parameters
        ----------
        frameID

        Returns
        -------

        """
        frame = self.get_frame_by_ID(frameID)
        return frame.get_data_from_full_spectrum()

    def get_KW_from_frameID(
        self, KW: str, frameID: int, use_header: bool = False
    ) -> Any:
        """Get Keywrod value from a given frame

        Parameters
        ----------
        KW : str
            ASTRA compatible keyword, as defined in the [Frame][ASTRA.base_models.Frame] class
        frameID : int
            Selected ID of teh frame
        use_header : bool, optional
            If True, searches for the KW in the header of the fits file, by default False

        Returns
        -------
        Any
            Value from the header
        """

        frame = self.get_frame_by_ID(frameID)
        if use_header:
            return frame.get_header_value(KW)
        else:
            return frame.get_KW_value(KW)

    def get_filename_from_frameID(
        self, frameID: int, full_path: bool = False
    ) -> str | Path:
        """Get the filename from a frame

        Parameters
        ----------
        frameID : int
            ID of the frame that we want to select
        full_path : bool, optional
            If True, returns the full disk file, by default False

        Returns
        -------
        str | Path
            Either the filename (string) of the full path (Path)
        """
        frame = self.get_frame_by_ID(frameID)
        if full_path:
            return frame.file_path
        return frame.fname

    def get_status_by_frameID(self, frameID: int) -> Status:
        """Get status of a given frame

        Parameters
        ----------
        frameID : int
            FrameID that we want to check the status of

        Returns
        -------
        Status
            Object that reflect the status
        """
        frame = self.get_frame_by_ID(frameID)
        return frame.status

    def collect_KW_observations(
        self,
        KW: str,
        subInstruments: Union[tuple, list],
        include_invalid: bool = False,
        conditions: CondModel = None,
        return_frameIDs: bool = False,
        from_header: bool = False,
        pad_condition_rejected: bool = True,
    ) -> Union[list, Tuple[List[float], List[int]]]:
        """Parse through the loaded observations and retrieve a specific KW from all of them.

        There is no sort of the files. The output will follow the
        order of the files loaded in memory.

        Parameters
        ----------
        return_frameIDs
        KW : str
            KW from the Frame.observation_info dictionary
        subInstruments : Union[tuple, list]
            List of the subInstruments for which we want to retrieve the KW. If
            there are multiple entries, the output will follow the order of this
            list
        include_invalid : bool, optional
            If True, also retrieve the KWs of Frames deemed to not be valid. By
            default False
        from_header: bool, optional
            If True, searches the fits header, rather than the internal keywords.
            By default, False

        Returns
        -------
        list
            List of the KW

        """
        output = []
        all_frameIDs = []

        for subInst in subInstruments:
            try:
                available_frameIDs = self.get_frameIDs_from_subInst(
                    subInst, include_invalid=include_invalid
                )
            except NoDataError:
                continue

            for frameID in available_frameIDs:
                if conditions is not None:
                    keep, flags = conditions.evaluate(self.get_frame_by_ID(frameID))
                    if not keep:
                        if pad_condition_rejected:
                            output.append(None)
                        continue

                frame = self.get_frame_by_ID(frameID)

                if from_header:
                    value = frame.get_header_value(KW)
                else:
                    value = frame.get_KW_value(KW)

                output.append(value)
                all_frameIDs.append(frameID)

        if return_frameIDs:
            return output, all_frameIDs

        return output

    def collect_RV_information(
        self,
        KW: str,
        subInst: str,
        frameIDs=None,
        include_invalid: bool = False,
        units=None,
        as_value: bool = True,
    ) -> list:
        """Return the RV measurements (or BERV) from the observations of a given sub-Instrument.

        Parameters
        ----------
        KW : KW to request
            previousRV, previousRV_ERR or BERV
        subInst : str
            Name of the sub-Instrument
        frameIDs : None | list
            If not None, retrive only for the frameIDs inside the list. If they are not from subInst, an exception
                will be raised
        include_invalid : bool, optional
            Include invalid observations, by default False
        units : [type], optional
            units to convert the measurements, by default None
        as_value : bool, optional
            provide a value instead of "units", by default True

        Returns
        -------
        list
            [description]

        Raises
        ------
        InvalidConfiguration
            [description]

        """
        if KW not in [
            "BERV",
            "DRS_RV",
            "DRS_RV_ERR",
            "previous_SBART_RV",
            "previous_SBART_RV_ERR",
        ]:
            msg = f"Asking for a non-RV KW: {KW}"
            logger.critical(msg)
            raise InvalidConfiguration(msg)

        if frameIDs is None:
            values = self.collect_KW_observations(
                KW, [subInst], include_invalid=include_invalid
            )
        else:
            values = []
            for frameID in frameIDs:
                frame = self.get_frame_by_ID(frameID)
                if not frame.is_SubInstrument(subInst):
                    msg = f"Frame ({frame.sub_instrument}) is not from the selected subInstrument ({subInst})"
                    logger.critical(msg)
                    raise InvalidConfiguration(msg)
                if (not frame.is_valid) and (not include_invalid):
                    msg = "Requesting invalid frame whilst the include_invalid argument is False"
                    logger.critical(msg)
                    raise InvalidConfiguration(msg)
                values.append(frame.get_KW_value(KW))

        if units is not None:
            values = [i.to(units) for i in values]
        if as_value:
            values = [i.value for i in values]

        return values

    def get_valid_frameIDS(self) -> list[int]:
        """Get list of all available frame IDs (across all subInstruments).

        Returns
        -------
        list
            [description]

        """
        out = []
        for subInst in self.frameID_map.keys():
            try:
                out.extend(
                    self.get_frameIDs_from_subInst(subInst, include_invalid=False)
                )
            except NoDataError:
                continue

        return out

    def get_invalid_frameIDs(
        self, subinstrument: None | str = None
    ) -> list[int]:  # noqa: N802
        """Get a list of the invalid frameIDs (by default, all of them).

        Parameters
        ----------
        subinstrument : None | str, optional
             subinstrument (None | str): If not None, returrns only for this
                    subInstrument, by default None

        Returns
        -------
        list[int]
            List of invalid frameIDs
        """
        out = []
        for subInst in self._inst_type.sub_instruments:  # noqa: N806
            if subinstrument is not None and subInst != subinstrument:
                continue
            try:
                good = set(
                    self.get_frameIDs_from_subInst(subInst, include_invalid=False)
                )
                all_f = set(
                    self.get_frameIDs_from_subInst(subInst, include_invalid=True)
                )
                out.extend(all_f - good)
            except NoDataError:
                continue
        return out

    def get_frameIDs_from_subInst(
        self, subInstrument: str, include_invalid: bool = False
    ) -> list[int]:  # noqa: FBT001, FBT002, N802
        """Get all frameIDs associated with a given instrument. By default, only returns the valid ones.

        Parameters
        ----------
        subInstrument : str
            SubInstrument name
        include_invalid : bool, optional
            If True, also return IDs associated with the invalid frames, by default False

        Returns
        -------
        list
            List of frameIDs from the instrument

        """
        frameIDS = [
            i
            for i in self.frameID_map[subInstrument]
            if self.get_frame_by_ID(i).is_valid or include_invalid
        ]
        if len(frameIDS) == 0:
            msg = f"There is no available observation in {subInstrument}"
            raise NoDataError(msg)

        return list(frameIDS)

    def get_frame_by_ID(self, frameID: int) -> Frame:
        """Get a frame by its ID

        Parameters
        ----------
        frameID : int
            ID of the frame

        Returns
        -------
        Frame
            Desired Frame

        Raises
        ------
        custom_exceptions.NoDataError
            If the frameID doesn't exist
        """
        for i in self.observations:
            if i.frameID == frameID:
                return i
        msg = f"{frameID} doesn't exist"
        raise custom_exceptions.NoDataError(msg)

    def get_subInstruments_with_valid_frames(self) -> list[str]:
        """Find all subInstruments that have at least one valid observation."""
        out = []
        for subInst in set(
            list(self._inst_type.sub_instruments.keys()) + list(self.frameID_map.keys())
        ):
            try:
                _ = self.get_frameIDs_from_subInst(subInst, include_invalid=False)
                out.append(subInst)
            except NoDataError:
                continue

        return out

    def get_available_subInstruments(self) -> list[str]:
        """Get sub-instruments with available data."""
        return self.frameID_map.keys()

    def get_instrument_information(self) -> dict:
        """Get info from the instrument."""
        return self.observations[0].instrument_properties

    def get_stellar_template(self, subInst: str) -> StellarTemplate:
        """Get stellar template for a given sub-Instrument."""
        return self.StellarModel.request_data(subInst)

    def get_stellar_model(self) -> StellarModel:
        """Get StellarModel."""
        return self.StellarModel

    def get_metaData(self) -> MetaData:
        """Get metadata."""
        return self.metaData

    ########################
    #    Data storage      #
    ########################

    def get_internalPaths(self):  # noqa: D102
        return self._internalPaths

    def trigger_data_storage(self) -> None:  # noqa: D102
        output_path = self._internalPaths.root_storage_path
        logger.debug("DataClass storing Data to {}", output_path)
        if self.disk_save_level != DISK_SAVE_MODE.EXTREME:
            self.metaData.store_json(output_path)
        logger.debug("DataClass finished data storage")

        for frame in self.observations:
            frame.trigger_data_storage()

    ########################
    #          MISC        #
    ########################

    def has_instrument_data(self, instrument: str) -> bool:
        """Check if the first loaded frame is of a given Instrument."""
        return self.observations[0].is_Instrument(instrument)

    def _build_frameID_map(self) -> None:
        """Populate the self.frameID_map with the frameIDs available for each subInstrument."""
        for frame in self.observations:
            if frame.sub_instrument is None:
                logger.critical(f"{frame} does not have a subInstrument")
                continue
            self.frameID_map[frame.sub_instrument].append(frame.frameID)

    def show_loadedData_table(self) -> Table:
        """Compute the number of observations available in each subInstrument.

        Find the number of valid (with and without warnings) and the invalid ones!
        """

        cols = ["subInstrument", "Total OBS", "Valid OBS [warnings]", "INVALID OBS"]
        is_solar = self.observations[0].is_solar
        if is_solar:
            cols += ["COMMENT"]

        tab = Table(cols)
        total_warnings = 0
        total = 0
        total_valid = 0
        total_invalid = 0
        for subinst, frames in self.frameID_map.items():
            row = [subinst, len(frames)]
            valid = 0
            with_warnings = 0
            invalid = 0
            comms = []

            for frameID in frames:
                frame = self.get_frame_by_ID(frameID)

                if frame.is_valid:
                    valid += 1
                    if frame.has_warnings:
                        with_warnings += 1
                else:
                    invalid += 1
                comms.append(frame.get_KW_value("COMMENT"))

            total_valid += valid
            total_warnings += with_warnings
            total_invalid += invalid
            total += len(frames)
            row.extend([f"{valid} [{with_warnings}]", invalid])
            if is_solar:
                if valid + invalid > 0:
                    if len(set(comms)) == 1:
                        row.extend([frame.get_KW_value("COMMENT")])
                    else:
                        row.extend(["Multiple OBs"])
                else:
                    row.extend(["---"])
            tab.add_row(row)

        row = ["Total", total, f"{total_valid} [{total_warnings}]", total_invalid]
        tab.add_row(row)
        logger.info(tab)

        return tab

    def plot_quantity(
        self,
        xx_var: str,
        yy_var: str,
        axis=None,
        return_fig_and_axis: bool = False,
        show: bool = False,
        condition: None | CondModel = None,
    ):
        """Plot any quantity from the frames against each other."""

        if axis is None:
            N = len(self.get_subInstruments_with_valid_frames())
            fig, axis = plt.subplots(
                nrows=N,
                sharex=True,
            )
            if N == 1:
                axis = [axis]

        for index, inst in enumerate(self.get_subInstruments_with_valid_frames()):
            xx = self.collect_KW_observations(
                xx_var,
                [inst],
                conditions=condition,
                pad_condition_rejected=False,
            )
            yy = self.collect_KW_observations(
                yy_var,
                [inst],
                conditions=condition,
                pad_condition_rejected=False,
            )

            if len(xx) > 0:
                if isinstance(xx[0], u.Quantity):
                    xx = convert_data(xx, as_value=True)
            if len(yy) > 0:
                if isinstance(yy[0], u.Quantity):
                    yy = convert_data(yy, as_value=True)

            axis[index].set_title(inst)
            axis[index].scatter(xx, yy)
            axis[index].set_xlabel(xx_var)
            axis[index].set_ylabel(yy_var)

        fig.tight_layout()
        if show:
            plt.show()

        if return_fig_and_axis:
            return fig, axis

    def load_instrument_extra_information(self) -> None:
        """See if the given instrument is one of the ones that has extra information to load.

        If so, then calls the relevant function.
        """
        logger.info("Checking if the instrument has extra data to load")

        for key, load_func in self.get_extra_loading_functions().items():
            if self.has_instrument_data(key):
                logger.info(
                    f"Dataclass has {key} data. Extra loading is being triggered"
                )
                load_func(self)

        self.show_loadedData_table()

    @classmethod
    def add_extra_loading_information(cls, name: str, loader: Callable) -> None:
        """Add a new loader function to run after the Instrument loads its data.

        Args:
            name (str): Instrument name
            loader (Callable): function to run afterwards

        """
        if name in cls.extra_loading_functions:
            logger.warning(f"Overloading extra-loading information from {name}")
        cls.extra_loading_functions[name] = loader

    def __repr__(self) -> str:  # noqa: D105
        return f"Data Class from {self._inst_type} holding " + ", ".join(
            [f"{len(IDS)} OBS from {name}" for name, IDS in self.frameID_map.items()],
        )

extra_loading_functions: Dict[str, Callable] = {'CARMENES': load_CARMENES_extra_information, 'POET': load_POET_extra_information} class-attribute instance-attribute

sigma_clip_RVs = sigma_clip_RVs instance-attribute

input_file = input_files instance-attribute

observations: List[Frame, ...] = [] instance-attribute

metaData = MetaData() instance-attribute

frameID_map = defaultdict(list) instance-attribute

Target = Target(self.collect_KW_observations('OBJECT', self._inst_type.sub_instruments, include_invalid=False), original_name=target_name, target_dictionary_path=target_dictionary_path) instance-attribute

StellarModel = None instance-attribute

construct_subInstruments_by_obID() -> None

Create new sub-Instruments with the obID of the observations

Source code in src/ASTRA/data_objects/DataClass.py

def construct_subInstruments_by_obID(self) -> None:
    """Create new sub-Instruments with the obID of the observations"""
    logger.warning("Re-definning sub-Instruments based on the OBID")

    ob_ids, frameIDS = self.collect_KW_observations(
        KW="OBS NAME",
        subInstruments=self.get_subInstruments_with_valid_frames(),
        return_frameIDs=True,
    )
    mapa = defaultdict(list)
    for OB, ids in zip(ob_ids, frameIDS):
        mapa[OB].append(ids)

    new_submap = defaultdict(list)
    for new_inst, all_frameID in mapa.items():
        for frameID in all_frameID:
            frame = self.get_frame_by_ID(frameID)
            frame.overwrite_sub_instrument(new_inst)
            new_submap[new_inst].append(frameID)
    self.frameID_map = new_submap

add_data_from_instrument(input_files: Iterable[UI_PATH], instrument: Type[Frame], reject_subInstruments: Iterable[str] | None = None, custom_subIns_name: str | None = None, instrument_options: dict[str, Any] | None = None) -> None

Source code in src/ASTRA/data_objects/DataClass.py

def add_data_from_instrument(
    self,
    input_files: Iterable[UI_PATH],
    instrument: Type[Frame],
    reject_subInstruments: Iterable[str] | None = None,
    custom_subIns_name: str | None = None,
    instrument_options: dict[str, Any] | None = None,
) -> None:

    OBS_list = []
    if isinstance(input_files, (str, Path)):
        logger.info("DataClass loading data from {}", self.input_file)
        with open(input_files) as input_file:
            for line in input_file:
                OBS_list.append(Path(line.split("\n")[0]))

    elif isinstance(input_files, Iterable):
        logger.info(
            "DataClass opening {} files from a list/tuple", len(input_files)
        )

        OBS_list = [Path(i) if isinstance(i, str) else i for i in input_files]
    else:
        raise TypeError

    names = [i.name for i in OBS_list]

    if len(set(names)) != len(names):
        msg = "Ingesting repeated observations into s-BART"
        logger.critical(msg)
        raise InvalidConfiguration(msg)

    def build_obs(args):
        index, filepath = args
        frameID = hash_path_to_int(filepath.stem)
        try:
            return self._inst_type(
                filepath,
                instrument_options,
                reject_subInstruments,
                frameID=frameID,
                quiet_user_params=frameID
                != 0,  # Only the first frame will output logs
            )
        except Exception as e:
            logger.critical(f"Failed in {filepath} due to {e}")
            raise e

    if instrument is not MAROONX:
        with ThreadPoolExecutor() as ex:
            observations = list(ex.map(build_obs, enumerate(OBS_list)))
    else:
        observations = list(map(build_obs, enumerate(OBS_list)))
    all_ids = [i.frameID for i in self.observations]
    if len(all_ids) != len(set(all_ids)):
        raise custom_exceptions.InternalError("Clash in frameID, repeated numbers")

    # The Frames don't store data inside the Iteration folder!
    frame_root_path = self._internalPaths.root_storage_path
    for frame in observations:
        if custom_subIns_name is not None:
            frame.overwrite_sub_instrument(custom_subIns_name)

        frame.generate_root_path(frame_root_path)
    self.observations.extend(observations)
    self._build_frameID_map()

    if custom_subIns_name is not None:
        self.show_loadedData_table()

get_extra_loading_functions()

Source code in src/ASTRA/data_objects/DataClass.py

def get_extra_loading_functions(self):
    return self.extra_loading_functions

load_previous_SBART_results(RV_holder, use_merged_cube: bool = False)

Load results of SBART run, store RV and uncertainty inside corresponding Frame object.

Source code in src/ASTRA/data_objects/DataClass.py

def load_previous_SBART_results(self, RV_holder, use_merged_cube: bool = False):
    """Load results of SBART run, store RV and uncertainty inside corresponding Frame object."""
    logger.info("Loading RVs from previous SBART run as the starting-RVs")
    for _, frameID in enumerate(self.get_valid_frameIDS()):
        frame = self.get_frame_by_ID(frameID)
        cube = RV_holder.get_RV_cube(frame.sub_instrument, merged=use_merged_cube)

        try:
            _, sbart_rv, sbart_uncert = cube.get_RV_from_ID(
                frameID=frameID,
                which="SBART",
                apply_SA_corr=False,
                as_value=False,
                units=None,
                apply_drift_corr=False,
            )
            previous_filename = cube.cached_info["date_folders"][
                cube.frameIDs.index(frameID)
            ]
            if previous_filename != frame.file_path:
                msg = f"Loading RVs from cube with different frameID layouts of {frame.sub_instrument} ({previous_filename} vs {frame.file_path})"
                logger.critical(msg)
                raise InvalidConfiguration(msg)
        except ValueError:
            sbart_rv = 0 * meter_second
            sbart_uncert = 10 * meter_second

        frame.store_previous_SBART_result(RV=sbart_rv, RV_err=sbart_uncert)
    logger.info("Finished loading of previous s-BART RVs")

reject_order_region_from_frame(frameID: int, order: int, region)

Source code in src/ASTRA/data_objects/DataClass.py

def reject_order_region_from_frame(self, frameID: int, order: int, region):
    frame = self.get_frame_by_ID(frameID)
    frame.reject_wavelength_region_from_order(order, region)
    frame.close_arrays()  # ensure that the arrays are closed and that the next call will load data

remove_activity_lines(lines: Indicators) -> None

Find the wavelength windows in which activity-related lines are expected to appear.

This is done for all valid observations

Parameters:

Name	Type	Description	Default
lines	Indicators	Object with the wavelength blocks "original" positions.	required

Source code in src/ASTRA/data_objects/DataClass.py

def remove_activity_lines(self, lines: Indicators) -> None:
    """Find the wavelength windows in which activity-related lines are expected to appear.

    This is done for all valid observations

    Parameters
    ----------
    lines : Indicators
        Object with the wavelength blocks "original" positions.

    # TODO: also allow to use the previous SBART RVs for this!!!!

    """
    logger.info("Computing activity windows for each RV measurements")
    for frameID in self.get_valid_frameIDS():
        frame = self.get_frame_by_ID(frameID)
        blocked_regions = lines.compute_forbidden_wavelengths(
            frame.previous_RV_measurements[0]
        )
        frame.mark_wavelength_region(
            reason=ACTIVITY_LINE, wavelength_blocks=blocked_regions
        )

remove_telluric_features(Telluric_Template: TelluricModel) -> None

Source code in src/ASTRA/data_objects/DataClass.py

def remove_telluric_features(self, Telluric_Template: TelluricModel) -> None:
    for subInstrument in self.get_subInstruments_with_valid_frames():
        valid_frameIDS = self.get_frameIDs_from_subInst(subInstrument)

        subInst_template = Telluric_Template.request_data(subInstrument)

        for frameID in valid_frameIDS:
            if subInst_template.for_feature_removal:
                self.get_frame_by_ID(frameID).mark_wavelength_region(
                    reason=TELLURIC,
                    wavelength_blocks=subInst_template.contaminated_regions,
                )
            elif subInst_template.for_feature_correction:
                (
                    wavelengths,
                    model,
                    model_uncertainty,
                ) = subInst_template.fit_telluric_model_to_frame(
                    self.get_frame_by_ID(frameID)
                )

                self.get_frame_by_ID(frameID).apply_telluric_correction(
                    wavelengths=wavelengths,
                    model=model,
                    model_uncertainty=model_uncertainty,
                )
            else:
                raise custom_exceptions.InternalError(
                    "Why aren't we correcting or removing tellurics?"
                )
    self._applied_telluric_removal = True

replace_frames_with_S2D_version(new_shape: Optional[Tuple[int, int]] = None)

In-place substitution of all frames with their S2D-compatible shapes.

Source code in src/ASTRA/data_objects/DataClass.py

def replace_frames_with_S2D_version(
    self, new_shape: Optional[Tuple[int, int]] = None
):
    """In-place substitution of all frames with their S2D-compatible shapes."""
    logger.warning("Transforming the frames to have a S2D-compatible shape")
    for index, frame in enumerate(self.observations):
        s2d_frame = frame.copy_into_S2D(new_S2D_size=new_shape)
        s2d_frame.build_mask()
        self.observations[index] = s2d_frame
        del frame

ingest_StellarModel(Stellar_Model: StellarModel) -> None

Source code in src/ASTRA/data_objects/DataClass.py

def ingest_StellarModel(self, Stellar_Model: StellarModel) -> None:
    logger.debug("Ingesting StellarModel into the DataClass")
    if self.StellarModel is not None:
        logger.warning(
            "Stellar template has already been ingested. Switching old template by the new one"
        )

    # Empty update just to ensure initialization of the modelling interfaces
    Stellar_Model.initialize_modelling_interfaces({})

    self.StellarModel = Stellar_Model

reject_observations(conditions: CondModel) -> None

Apply the conditions to evaluate if the VALID frame meets the specified conditions or not.

Parameters:

Name	Type	Description	Default
conditions	ConditionModel	Conditions set by the user	required

Source code in src/ASTRA/data_objects/DataClass.py

def reject_observations(self, conditions: CondModel) -> None:
    """Apply the conditions to evaluate if the VALID frame meets the specified conditions or not.

    Parameters
    ----------
    conditions:
        Conditions set by the user

    """
    if conditions is None:
        return

    blocked_frames = 0

    for frameID in self.get_valid_frameIDS():
        frame = self.get_frame_by_ID(frameID)
        keep, flags = conditions.evaluate(frame)

        if not keep:
            for flag in flags:
                frame.add_to_status(flag)
            blocked_frames += 1

    logger.info(
        "User conditions removed {} / {} frames",
        blocked_frames,
        len(self.observations),
    )

    logger.info("Updated observation on disk!")
    self.show_loadedData_table()

    if len(self.get_valid_frameIDS()) == 0:
        raise NoDataError("All observations have been blocked")

get_Target() -> Target

Source code in src/ASTRA/data_objects/DataClass.py

def get_Target(self) -> Target:
    return self.Target

load_frame_by_ID(frameID: int) -> int

Source code in src/ASTRA/data_objects/DataClass.py

def load_frame_by_ID(self, frameID: int) -> int:
    frame = self.get_frame_by_ID(frameID)
    frame.load_data()
    return 1

set_all_as_Zscore_frames() -> None

Source code in src/ASTRA/data_objects/DataClass.py

def set_all_as_Zscore_frames(self) -> None:
    logger.warning("Setting all frames as a zero-mean unit variance")
    for frameID in self.get_valid_frameIDS():
        frame = self.get_frame_by_ID(frameID)
        frame.set_frame_as_Zscore()

normalize_all() -> None

Launch the normalization for all (valid) frames.

Source code in src/ASTRA/data_objects/DataClass.py

def normalize_all(self) -> None:
    """Launch the normalization for all (valid) frames.

    Returns
    -------

    """
    logger.info("Normalizing all frames")
    for frameID in self.get_valid_frameIDS():
        frame = self.get_frame_by_ID(frameID)
        frame._never_close = True

    for subInst in self.get_subInstruments_with_valid_frames():
        self.normalize_all_from_subInst(subInst)

update_uncertainties_of_frame(frameID: int, new_uncerts) -> None

Update the flux uncertainties of a given frame.

.. warning:: This will change your measured data, be careful with this!!!

Source code in src/ASTRA/data_objects/DataClass.py

def update_uncertainties_of_frame(self, frameID: int, new_uncerts) -> None:
    """Update the flux uncertainties of a given frame.

    .. warning::
        This will change your measured data, be careful with this!!!
    """
    logger.info(f"Setting uncertainties of frame (ID = {frameID}")

    frame = self.get_frame_by_ID(frameID)
    frame.update_uncertainties(new_uncerts)
    frame._never_close = True
    logger.warning("This frame will never close until SBART finished!")

normalize_all_from_subInst(subInst: str) -> None

Normalize all (valid) frames from a given subInstrument.

Parameters:

Name	Type	Description	Default
subInst	str		required

Source code in src/ASTRA/data_objects/DataClass.py

def normalize_all_from_subInst(self, subInst: str) -> None:
    """Normalize all (valid) frames from a given subInstrument.

    Parameters
    ----------
    subInst

    """
    logger.debug(f"Normalizing all frames from {subInst}")
    for fId in self.get_frameIDs_from_subInst(subInst):
        frame = self.get_frame_by_ID(fId)
        frame.normalize_spectra()

scale_up_all_observations(factor: float) -> None

Multiply the flux and uncertainties by a given flux level (avoid possible SNR issues).

Source code in src/ASTRA/data_objects/DataClass.py

def scale_up_all_observations(self, factor: float) -> None:
    """Multiply the flux and uncertainties by a given flux level (avoid possible SNR issues)."""
    logger.warning(f"Scaling up all spectra by a factor of {factor}")
    for fId in self.get_valid_frameIDS():
        frame = self.get_frame_by_ID(fId)
        frame.scale_spectra(factor)

set_stellar_RV_to_zero() -> None

Allow to override the stellar RV so that it is fixed to zero.

Source code in src/ASTRA/data_objects/DataClass.py

def set_stellar_RV_to_zero(self) -> None:
    """Allow to override the stellar RV so that it is fixed to zero."""
    for fID in self.get_valid_frameIDS():
        f = self.get_frame_by_ID(fID)
        f.observation_info["DRS_RV"] = 0 * kilometer_second

load_all_from_subInst(subInst: str) -> int

Load all valid frames from a given subInstrument.

Source code in src/ASTRA/data_objects/DataClass.py

def load_all_from_subInst(self, subInst: str) -> int:
    """Load all valid frames from a given subInstrument."""
    logger.debug("Opening all frames from {}", subInst)
    IDs_from_subInst = self.get_frameIDs_from_subInst(
        subInstrument=subInst, include_invalid=False
    )
    for frameID in IDs_from_subInst:
        try:
            self.load_frame_by_ID(frameID)
        except FrameError:
            logger.warning("Trying to open invalid frame")
    return 1

close_frame_by_ID(frameID: int) -> int

Source code in src/ASTRA/data_objects/DataClass.py

def close_frame_by_ID(self, frameID: int) -> int:
    frame = self.get_frame_by_ID(frameID)
    frame.close_arrays()
    return 1

min_pixel_in_order() -> int

Source code in src/ASTRA/data_objects/DataClass.py

def min_pixel_in_order(self) -> int:
    return self.observations[0].min_pixel_in_order

get_subInst_from_frameID(frameID: int) -> str

Source code in src/ASTRA/data_objects/DataClass.py

def get_subInst_from_frameID(self, frameID: int) -> str:
    return self.get_frame_by_ID(frameID).sub_instrument

get_frame_OBS_order(frameID: int, order: int, include_invalid: bool = False)

Request the data from one spectral order.

Parameters:

Name	Type	Description	Default
frameID	int	FrameID of the observation	required
order	int	NUmber of order	required
include_invalid	bool	Orders can also be flagged as "invalid". If False, the frame will raise an error when requesting the data	False

Source code in src/ASTRA/data_objects/DataClass.py

def get_frame_OBS_order(
    self, frameID: int, order: int, include_invalid: bool = False
):
    """Request the data from one spectral order.

    Parameters
    ----------
    frameID
        FrameID of the observation
    order
        NUmber of order
    include_invalid
        Orders can also be flagged as "invalid". If False, the frame will raise an error when requesting the data

    Returns
    -------

    """
    frame = self.get_frame_by_ID(frameID)
    return frame.get_data_from_spectral_order(order, include_invalid)

update_interpol_properties_of_all_frames(new_properties: Dict[str, Any])

Source code in src/ASTRA/data_objects/DataClass.py

def update_interpol_properties_of_all_frames(self, new_properties: Dict[str, Any]):
    if not isinstance(new_properties, dict):
        raise custom_exceptions.InvalidConfiguration(
            "The interpolation properties must be passed as a dictionary"
        )

    for frame in self.observations:
        frame.update_user_configs(new_properties)

update_interpol_properties_of_stellar_model(new_properties: Dict[str, Any])

Source code in src/ASTRA/data_objects/DataClass.py

def update_interpol_properties_of_stellar_model(
    self, new_properties: Dict[str, Any]
):
    if not isinstance(new_properties, dict):
        raise custom_exceptions.InvalidConfiguration(
            "The interpolation properties must be passed as a dictionary"
        )

    if self.StellarModel is None:
        raise custom_exceptions.NoDataError("The Stellar Model wasn't ingested")
    self.StellarModel.update_user_configs(new_properties)

update_frame_interpol_properties(frameID, new_properties) -> None

Allow to update the interpolation settings from the outside, so that any object can configure the interpolation as it wishes

Source code in src/ASTRA/data_objects/DataClass.py

def update_frame_interpol_properties(self, frameID, new_properties) -> None:
    """Allow to update the interpolation settings from the outside, so that any object can configure
    the interpolation as it wishes
    """
    frame = self.get_frame_by_ID(frameID)
    frame.set_interpolation_properties(new_properties)

interpolate_frame_order(frameID, order, new_wavelengths, shift_RV_by, RV_shift_mode, include_invalid=False)

Interpolate a given order to a new wavelength solution

Source code in src/ASTRA/data_objects/DataClass.py

def interpolate_frame_order(
    self,
    frameID,
    order,
    new_wavelengths,
    shift_RV_by,
    RV_shift_mode,
    include_invalid=False,
):
    """Interpolate a given order to a new wavelength solution"""
    frame = self.get_frame_by_ID(frameID)
    return frame.interpolate_spectrum_to_wavelength(
        order=order,
        new_wavelengths=new_wavelengths,
        shift_RV_by=shift_RV_by,
        RV_shift_mode=RV_shift_mode,
        include_invalid=include_invalid,
    )

get_frame_arrays_by_ID(frameID: int)

Access data from the entire spectral range (i.e. all orders come as a matrix)

Parameters:

Name	Type	Description	Default
frameID	int		required

Source code in src/ASTRA/data_objects/DataClass.py

def get_frame_arrays_by_ID(self, frameID: int):
    """Access data from the entire spectral range (i.e. all orders come as a matrix)

    Parameters
    ----------
    frameID

    Returns
    -------

    """
    frame = self.get_frame_by_ID(frameID)
    return frame.get_data_from_full_spectrum()

get_KW_from_frameID(KW: str, frameID: int, use_header: bool = False) -> Any

Get Keywrod value from a given frame

Parameters:

Name	Type	Description	Default
KW	str	ASTRA compatible keyword, as defined in the Frame class	required
frameID	int	Selected ID of teh frame	required
use_header	bool	If True, searches for the KW in the header of the fits file, by default False	False

Returns:

Type	Description
Any	Value from the header

Source code in src/ASTRA/data_objects/DataClass.py

def get_KW_from_frameID(
    self, KW: str, frameID: int, use_header: bool = False
) -> Any:
    """Get Keywrod value from a given frame

    Parameters
    ----------
    KW : str
        ASTRA compatible keyword, as defined in the [Frame][ASTRA.base_models.Frame] class
    frameID : int
        Selected ID of teh frame
    use_header : bool, optional
        If True, searches for the KW in the header of the fits file, by default False

    Returns
    -------
    Any
        Value from the header
    """

    frame = self.get_frame_by_ID(frameID)
    if use_header:
        return frame.get_header_value(KW)
    else:
        return frame.get_KW_value(KW)

get_filename_from_frameID(frameID: int, full_path: bool = False) -> str | Path

Get the filename from a frame

Parameters:

Name	Type	Description	Default
frameID	int	ID of the frame that we want to select	required
full_path	bool	If True, returns the full disk file, by default False	False

Returns:

Type	Description
str | Path	Either the filename (string) of the full path (Path)

Source code in src/ASTRA/data_objects/DataClass.py

def get_filename_from_frameID(
    self, frameID: int, full_path: bool = False
) -> str | Path:
    """Get the filename from a frame

    Parameters
    ----------
    frameID : int
        ID of the frame that we want to select
    full_path : bool, optional
        If True, returns the full disk file, by default False

    Returns
    -------
    str | Path
        Either the filename (string) of the full path (Path)
    """
    frame = self.get_frame_by_ID(frameID)
    if full_path:
        return frame.file_path
    return frame.fname

get_status_by_frameID(frameID: int) -> Status

Get status of a given frame

Parameters:

Name	Type	Description	Default
frameID	int	FrameID that we want to check the status of	required

Returns:

Type	Description
Status	Object that reflect the status

Source code in src/ASTRA/data_objects/DataClass.py

def get_status_by_frameID(self, frameID: int) -> Status:
    """Get status of a given frame

    Parameters
    ----------
    frameID : int
        FrameID that we want to check the status of

    Returns
    -------
    Status
        Object that reflect the status
    """
    frame = self.get_frame_by_ID(frameID)
    return frame.status

collect_KW_observations(KW: str, subInstruments: Union[tuple, list], include_invalid: bool = False, conditions: CondModel = None, return_frameIDs: bool = False, from_header: bool = False, pad_condition_rejected: bool = True) -> Union[list, Tuple[List[float], List[int]]]

Parse through the loaded observations and retrieve a specific KW from all of them.

There is no sort of the files. The output will follow the order of the files loaded in memory.

Parameters:

Name	Type	Description	Default
return_frameIDs	bool		False
KW	str	KW from the Frame.observation_info dictionary	required
subInstruments	Union[tuple, list]	List of the subInstruments for which we want to retrieve the KW. If there are multiple entries, the output will follow the order of this list	required
include_invalid	bool	If True, also retrieve the KWs of Frames deemed to not be valid. By default False	False
from_header	bool	If True, searches the fits header, rather than the internal keywords. By default, False	False

Returns:

Type	Description
list	List of the KW

Source code in src/ASTRA/data_objects/DataClass.py

def collect_KW_observations(
    self,
    KW: str,
    subInstruments: Union[tuple, list],
    include_invalid: bool = False,
    conditions: CondModel = None,
    return_frameIDs: bool = False,
    from_header: bool = False,
    pad_condition_rejected: bool = True,
) -> Union[list, Tuple[List[float], List[int]]]:
    """Parse through the loaded observations and retrieve a specific KW from all of them.

    There is no sort of the files. The output will follow the
    order of the files loaded in memory.

    Parameters
    ----------
    return_frameIDs
    KW : str
        KW from the Frame.observation_info dictionary
    subInstruments : Union[tuple, list]
        List of the subInstruments for which we want to retrieve the KW. If
        there are multiple entries, the output will follow the order of this
        list
    include_invalid : bool, optional
        If True, also retrieve the KWs of Frames deemed to not be valid. By
        default False
    from_header: bool, optional
        If True, searches the fits header, rather than the internal keywords.
        By default, False

    Returns
    -------
    list
        List of the KW

    """
    output = []
    all_frameIDs = []

    for subInst in subInstruments:
        try:
            available_frameIDs = self.get_frameIDs_from_subInst(
                subInst, include_invalid=include_invalid
            )
        except NoDataError:
            continue

        for frameID in available_frameIDs:
            if conditions is not None:
                keep, flags = conditions.evaluate(self.get_frame_by_ID(frameID))
                if not keep:
                    if pad_condition_rejected:
                        output.append(None)
                    continue

            frame = self.get_frame_by_ID(frameID)

            if from_header:
                value = frame.get_header_value(KW)
            else:
                value = frame.get_KW_value(KW)

            output.append(value)
            all_frameIDs.append(frameID)

    if return_frameIDs:
        return output, all_frameIDs

    return output

collect_RV_information(KW: str, subInst: str, frameIDs=None, include_invalid: bool = False, units=None, as_value: bool = True) -> list

Return the RV measurements (or BERV) from the observations of a given sub-Instrument.

Parameters:

Name	Type	Description	Default
KW	KW to request	previousRV, previousRV_ERR or BERV	required
subInst	str	Name of the sub-Instrument	required
frameIDs	None | list	If not None, retrive only for the frameIDs inside the list. If they are not from subInst, an exception will be raised	None
include_invalid	bool	Include invalid observations, by default False	False
units	[type]	units to convert the measurements, by default None	None
as_value	bool	provide a value instead of "units", by default True	True

Returns:

Type	Description
list	[description]

Raises:

Type	Description
InvalidConfiguration	[description]

Source code in src/ASTRA/data_objects/DataClass.py

def collect_RV_information(
    self,
    KW: str,
    subInst: str,
    frameIDs=None,
    include_invalid: bool = False,
    units=None,
    as_value: bool = True,
) -> list:
    """Return the RV measurements (or BERV) from the observations of a given sub-Instrument.

    Parameters
    ----------
    KW : KW to request
        previousRV, previousRV_ERR or BERV
    subInst : str
        Name of the sub-Instrument
    frameIDs : None | list
        If not None, retrive only for the frameIDs inside the list. If they are not from subInst, an exception
            will be raised
    include_invalid : bool, optional
        Include invalid observations, by default False
    units : [type], optional
        units to convert the measurements, by default None
    as_value : bool, optional
        provide a value instead of "units", by default True

    Returns
    -------
    list
        [description]

    Raises
    ------
    InvalidConfiguration
        [description]

    """
    if KW not in [
        "BERV",
        "DRS_RV",
        "DRS_RV_ERR",
        "previous_SBART_RV",
        "previous_SBART_RV_ERR",
    ]:
        msg = f"Asking for a non-RV KW: {KW}"
        logger.critical(msg)
        raise InvalidConfiguration(msg)

    if frameIDs is None:
        values = self.collect_KW_observations(
            KW, [subInst], include_invalid=include_invalid
        )
    else:
        values = []
        for frameID in frameIDs:
            frame = self.get_frame_by_ID(frameID)
            if not frame.is_SubInstrument(subInst):
                msg = f"Frame ({frame.sub_instrument}) is not from the selected subInstrument ({subInst})"
                logger.critical(msg)
                raise InvalidConfiguration(msg)
            if (not frame.is_valid) and (not include_invalid):
                msg = "Requesting invalid frame whilst the include_invalid argument is False"
                logger.critical(msg)
                raise InvalidConfiguration(msg)
            values.append(frame.get_KW_value(KW))

    if units is not None:
        values = [i.to(units) for i in values]
    if as_value:
        values = [i.value for i in values]

    return values

get_valid_frameIDS() -> list[int]

Get list of all available frame IDs (across all subInstruments).

Returns:

Type	Description
list	[description]

Source code in src/ASTRA/data_objects/DataClass.py

def get_valid_frameIDS(self) -> list[int]:
    """Get list of all available frame IDs (across all subInstruments).

    Returns
    -------
    list
        [description]

    """
    out = []
    for subInst in self.frameID_map.keys():
        try:
            out.extend(
                self.get_frameIDs_from_subInst(subInst, include_invalid=False)
            )
        except NoDataError:
            continue

    return out

get_invalid_frameIDs(subinstrument: None | str = None) -> list[int]

Get a list of the invalid frameIDs (by default, all of them).

Parameters:

Name	Type	Description	Default
subinstrument	None | str	subinstrument (None | str): If not None, returrns only for this subInstrument, by default None	None

Returns:

Type	Description
list[int]	List of invalid frameIDs

Source code in src/ASTRA/data_objects/DataClass.py

def get_invalid_frameIDs(
    self, subinstrument: None | str = None
) -> list[int]:  # noqa: N802
    """Get a list of the invalid frameIDs (by default, all of them).

    Parameters
    ----------
    subinstrument : None | str, optional
         subinstrument (None | str): If not None, returrns only for this
                subInstrument, by default None

    Returns
    -------
    list[int]
        List of invalid frameIDs
    """
    out = []
    for subInst in self._inst_type.sub_instruments:  # noqa: N806
        if subinstrument is not None and subInst != subinstrument:
            continue
        try:
            good = set(
                self.get_frameIDs_from_subInst(subInst, include_invalid=False)
            )
            all_f = set(
                self.get_frameIDs_from_subInst(subInst, include_invalid=True)
            )
            out.extend(all_f - good)
        except NoDataError:
            continue
    return out

get_frameIDs_from_subInst(subInstrument: str, include_invalid: bool = False) -> list[int]

Get all frameIDs associated with a given instrument. By default, only returns the valid ones.

Parameters:

Name	Type	Description	Default
subInstrument	str	SubInstrument name	required
include_invalid	bool	If True, also return IDs associated with the invalid frames, by default False	False

Returns:

Type	Description
list	List of frameIDs from the instrument

Source code in src/ASTRA/data_objects/DataClass.py

def get_frameIDs_from_subInst(
    self, subInstrument: str, include_invalid: bool = False
) -> list[int]:  # noqa: FBT001, FBT002, N802
    """Get all frameIDs associated with a given instrument. By default, only returns the valid ones.

    Parameters
    ----------
    subInstrument : str
        SubInstrument name
    include_invalid : bool, optional
        If True, also return IDs associated with the invalid frames, by default False

    Returns
    -------
    list
        List of frameIDs from the instrument

    """
    frameIDS = [
        i
        for i in self.frameID_map[subInstrument]
        if self.get_frame_by_ID(i).is_valid or include_invalid
    ]
    if len(frameIDS) == 0:
        msg = f"There is no available observation in {subInstrument}"
        raise NoDataError(msg)

    return list(frameIDS)

get_frame_by_ID(frameID: int) -> Frame

Get a frame by its ID

Parameters:

Name	Type	Description	Default
frameID	int	ID of the frame	required

Returns:

Type	Description
Frame	Desired Frame

Raises:

Type	Description
NoDataError	If the frameID doesn't exist

Source code in src/ASTRA/data_objects/DataClass.py

def get_frame_by_ID(self, frameID: int) -> Frame:
    """Get a frame by its ID

    Parameters
    ----------
    frameID : int
        ID of the frame

    Returns
    -------
    Frame
        Desired Frame

    Raises
    ------
    custom_exceptions.NoDataError
        If the frameID doesn't exist
    """
    for i in self.observations:
        if i.frameID == frameID:
            return i
    msg = f"{frameID} doesn't exist"
    raise custom_exceptions.NoDataError(msg)

get_subInstruments_with_valid_frames() -> list[str]

Find all subInstruments that have at least one valid observation.

Source code in src/ASTRA/data_objects/DataClass.py

def get_subInstruments_with_valid_frames(self) -> list[str]:
    """Find all subInstruments that have at least one valid observation."""
    out = []
    for subInst in set(
        list(self._inst_type.sub_instruments.keys()) + list(self.frameID_map.keys())
    ):
        try:
            _ = self.get_frameIDs_from_subInst(subInst, include_invalid=False)
            out.append(subInst)
        except NoDataError:
            continue

    return out

get_available_subInstruments() -> list[str]

Get sub-instruments with available data.

Source code in src/ASTRA/data_objects/DataClass.py

def get_available_subInstruments(self) -> list[str]:
    """Get sub-instruments with available data."""
    return self.frameID_map.keys()

get_instrument_information() -> dict

Get info from the instrument.

Source code in src/ASTRA/data_objects/DataClass.py

def get_instrument_information(self) -> dict:
    """Get info from the instrument."""
    return self.observations[0].instrument_properties

get_stellar_template(subInst: str) -> StellarTemplate

Get stellar template for a given sub-Instrument.

Source code in src/ASTRA/data_objects/DataClass.py

def get_stellar_template(self, subInst: str) -> StellarTemplate:
    """Get stellar template for a given sub-Instrument."""
    return self.StellarModel.request_data(subInst)

get_stellar_model() -> StellarModel

Get StellarModel.

Source code in src/ASTRA/data_objects/DataClass.py

def get_stellar_model(self) -> StellarModel:
    """Get StellarModel."""
    return self.StellarModel

get_metaData() -> MetaData

Get metadata.

Source code in src/ASTRA/data_objects/DataClass.py

def get_metaData(self) -> MetaData:
    """Get metadata."""
    return self.metaData

get_internalPaths()

Source code in src/ASTRA/data_objects/DataClass.py

def get_internalPaths(self):  # noqa: D102
    return self._internalPaths

trigger_data_storage() -> None

Source code in src/ASTRA/data_objects/DataClass.py

def trigger_data_storage(self) -> None:  # noqa: D102
    output_path = self._internalPaths.root_storage_path
    logger.debug("DataClass storing Data to {}", output_path)
    if self.disk_save_level != DISK_SAVE_MODE.EXTREME:
        self.metaData.store_json(output_path)
    logger.debug("DataClass finished data storage")

    for frame in self.observations:
        frame.trigger_data_storage()

has_instrument_data(instrument: str) -> bool

Check if the first loaded frame is of a given Instrument.

Source code in src/ASTRA/data_objects/DataClass.py

def has_instrument_data(self, instrument: str) -> bool:
    """Check if the first loaded frame is of a given Instrument."""
    return self.observations[0].is_Instrument(instrument)

show_loadedData_table() -> Table

Compute the number of observations available in each subInstrument.

Find the number of valid (with and without warnings) and the invalid ones!

Source code in src/ASTRA/data_objects/DataClass.py

def show_loadedData_table(self) -> Table:
    """Compute the number of observations available in each subInstrument.

    Find the number of valid (with and without warnings) and the invalid ones!
    """

    cols = ["subInstrument", "Total OBS", "Valid OBS [warnings]", "INVALID OBS"]
    is_solar = self.observations[0].is_solar
    if is_solar:
        cols += ["COMMENT"]

    tab = Table(cols)
    total_warnings = 0
    total = 0
    total_valid = 0
    total_invalid = 0
    for subinst, frames in self.frameID_map.items():
        row = [subinst, len(frames)]
        valid = 0
        with_warnings = 0
        invalid = 0
        comms = []

        for frameID in frames:
            frame = self.get_frame_by_ID(frameID)

            if frame.is_valid:
                valid += 1
                if frame.has_warnings:
                    with_warnings += 1
            else:
                invalid += 1
            comms.append(frame.get_KW_value("COMMENT"))

        total_valid += valid
        total_warnings += with_warnings
        total_invalid += invalid
        total += len(frames)
        row.extend([f"{valid} [{with_warnings}]", invalid])
        if is_solar:
            if valid + invalid > 0:
                if len(set(comms)) == 1:
                    row.extend([frame.get_KW_value("COMMENT")])
                else:
                    row.extend(["Multiple OBs"])
            else:
                row.extend(["---"])
        tab.add_row(row)

    row = ["Total", total, f"{total_valid} [{total_warnings}]", total_invalid]
    tab.add_row(row)
    logger.info(tab)

    return tab

plot_quantity(xx_var: str, yy_var: str, axis=None, return_fig_and_axis: bool = False, show: bool = False, condition: None | CondModel = None)

Plot any quantity from the frames against each other.

Source code in src/ASTRA/data_objects/DataClass.py

def plot_quantity(
    self,
    xx_var: str,
    yy_var: str,
    axis=None,
    return_fig_and_axis: bool = False,
    show: bool = False,
    condition: None | CondModel = None,
):
    """Plot any quantity from the frames against each other."""

    if axis is None:
        N = len(self.get_subInstruments_with_valid_frames())
        fig, axis = plt.subplots(
            nrows=N,
            sharex=True,
        )
        if N == 1:
            axis = [axis]

    for index, inst in enumerate(self.get_subInstruments_with_valid_frames()):
        xx = self.collect_KW_observations(
            xx_var,
            [inst],
            conditions=condition,
            pad_condition_rejected=False,
        )
        yy = self.collect_KW_observations(
            yy_var,
            [inst],
            conditions=condition,
            pad_condition_rejected=False,
        )

        if len(xx) > 0:
            if isinstance(xx[0], u.Quantity):
                xx = convert_data(xx, as_value=True)
        if len(yy) > 0:
            if isinstance(yy[0], u.Quantity):
                yy = convert_data(yy, as_value=True)

        axis[index].set_title(inst)
        axis[index].scatter(xx, yy)
        axis[index].set_xlabel(xx_var)
        axis[index].set_ylabel(yy_var)

    fig.tight_layout()
    if show:
        plt.show()

    if return_fig_and_axis:
        return fig, axis

load_instrument_extra_information() -> None

See if the given instrument is one of the ones that has extra information to load.

If so, then calls the relevant function.

Source code in src/ASTRA/data_objects/DataClass.py

def load_instrument_extra_information(self) -> None:
    """See if the given instrument is one of the ones that has extra information to load.

    If so, then calls the relevant function.
    """
    logger.info("Checking if the instrument has extra data to load")

    for key, load_func in self.get_extra_loading_functions().items():
        if self.has_instrument_data(key):
            logger.info(
                f"Dataclass has {key} data. Extra loading is being triggered"
            )
            load_func(self)

    self.show_loadedData_table()

add_extra_loading_information(name: str, loader: Callable) -> None classmethod

Add a new loader function to run after the Instrument loads its data.

Args: name (str): Instrument name loader (Callable): function to run afterwards

Source code in src/ASTRA/data_objects/DataClass.py

@classmethod
def add_extra_loading_information(cls, name: str, loader: Callable) -> None:
    """Add a new loader function to run after the Instrument loads its data.

    Args:
        name (str): Instrument name
        loader (Callable): function to run afterwards

    """
    if name in cls.extra_loading_functions:
        logger.warning(f"Overloading extra-loading information from {name}")
    cls.extra_loading_functions[name] = loader

hash_path_to_int(fname: Path, algo: str = 'sha256', bits: int = 32) -> int

Deterministic integer hash of a filesystem path. - Uses hashlib (default sha256) - Returns an integer, truncated to bits if specified

Source code in src/ASTRA/data_objects/DataClass.py

def hash_path_to_int(fname: Path, algo: str = "sha256", bits: int = 32) -> int:
    """Deterministic integer hash of a filesystem path.
    - Uses hashlib (default sha256)
    - Returns an integer, truncated to `bits` if specified
    """

    h = hashlib.new(algo)
    h.update(fname.encode("utf-8"))
    digest_int = int.from_bytes(h.digest(), "big")

    if bits is not None:
        mask = (1 << bits) - 1
        return digest_int & mask
    return digest_int

MetaData

Collect metadata from the loaded Frames.

MetaData

Class to hold metadata information of the processed data.

The Frames, when loading data from disk will collect information to work as MetaData, which will be used to provide an option of avoiding the repeated computation of exactly the same dataset. This class provides an easy interface to hold such data, write and load it to disk, and compare equality between two MetaData objects.

The data stored inside this object is divided by subInstruments, with each having a unique set of values, that is not shared with other subInstruments. Comparisons of equality are supported at the subInstrument level.

Used by:

• API
  • API data_objects MetaData MetaData subInstrument_comparison
  • Data_objects MetaData MetaData subInstrument_comparison

Source code in src/ASTRA/data_objects/MetaData.py

class MetaData:
    """Class to hold metadata information of the processed data.

    The Frames, when loading data from disk will collect information to work as MetaData, which will be used to
    provide an  option of avoiding the repeated computation of exactly the same dataset. This class provides an
    easy interface to hold such data, write and load it to disk, and compare equality between two MetaData objects.

    The data stored inside this object is divided by subInstruments, with each having a unique set of values, that
    is not shared with other subInstruments. Comparisons of equality are supported at the subInstrument level.

    """

    def __init__(self, data_to_hold: dict | None = None) -> None:  # noqa: D107
        self.information = {} if data_to_hold is None else data_to_hold

    def subInstrument_comparison(self, other: MetaData, subInstrument: str) -> bool:
        """Compare the Metadata of a given subInstrument.

        Checks that must pass for equality:

        * The same keys are present in the two objects
        * A given key has the same value in the two objects

        Parameters
        ----------
        other:
            Other MetaData object
        subInstrument: str
            SubInstrument to compare

        Returns
        -------
        comparison_result: bool
            True if the MetaData matchess

        """
        equal = True

        try:
            for key, subInst_value in self.information[subInstrument].items():
                if subInst_value != other.information[subInstrument][key]:
                    equal = False
                    break
        except KeyError:
            equal = False

        return equal

    def store_json(self, path: Path) -> None:
        """Store the class as a json files.

        Args:
            path (Path):Path in which this object will be stored

        """
        logger.debug("Storing Metadata to {}", path)

        storage_path = build_filename(path, "MetaData", fmt="json", skip_version=True)

        info_to_store = {}
        for key, value in self.information.items():
            value = json_ready_converter(value)
            info_to_store[key] = value

        with open(storage_path, mode="w") as handle:
            json.dump(info_to_store, handle, indent=4)

    def add_info(self, key: str, info: str | tuple | list, subInstrument: str) -> None:
        """Add a new metric to be tracked, with the values being collected over all available frames.

        Parameters
        ----------
        key: str
            Name of the metric
        info: Union[str, Iterable]
            data to be stored
        subInstrument: str
            subInstrument to which the info belongs to


        Raises
        ------
        TypeError
            If the info is not a list nor an iterable

        """
        if not isinstance(info, (str, tuple, list)):
            raise TypeError("info must be  str or list object")
        if subInstrument not in self.information:
            self.information[subInstrument] = {}

        # remove duplicates, but keep as a list (to avoid problems when storing to json)!
        self.information[subInstrument][key] = list(set(info))

    @classmethod
    def load_from_json(cls, path: Path) -> MetaData:
        """Load from a json file."""
        storage_path = build_filename(
            path,
            "MetaData",
            fmt="json",
            ASTRA_version=None,
            skip_version=True,  # we don't want a version
        )
        try:
            with open(storage_path) as handle:
                information = json.load(handle)
        except FileNotFoundError:
            msg = f"Failed to find metadata file in {storage_path}"
            logger.warning(msg)
            raise custom_exceptions.StopComputationError(msg)

        return MetaData(information)

information = {} if data_to_hold is None else data_to_hold instance-attribute

subInstrument_comparison(other: MetaData, subInstrument: str) -> bool

Compare the Metadata of a given subInstrument.

Checks that must pass for equality:

• The same keys are present in the two objects
• A given key has the same value in the two objects

Parameters:

Name	Type	Description	Default
other	MetaData	Other MetaData object	required
subInstrument	str	SubInstrument to compare	required

Returns:

Name	Type	Description
comparison_result	bool	True if the MetaData matchess

Source code in src/ASTRA/data_objects/MetaData.py

def subInstrument_comparison(self, other: MetaData, subInstrument: str) -> bool:
    """Compare the Metadata of a given subInstrument.

    Checks that must pass for equality:

    * The same keys are present in the two objects
    * A given key has the same value in the two objects

    Parameters
    ----------
    other:
        Other MetaData object
    subInstrument: str
        SubInstrument to compare

    Returns
    -------
    comparison_result: bool
        True if the MetaData matchess

    """
    equal = True

    try:
        for key, subInst_value in self.information[subInstrument].items():
            if subInst_value != other.information[subInstrument][key]:
                equal = False
                break
    except KeyError:
        equal = False

    return equal

store_json(path: Path) -> None

Store the class as a json files.

Args: path (Path):Path in which this object will be stored

Source code in src/ASTRA/data_objects/MetaData.py

def store_json(self, path: Path) -> None:
    """Store the class as a json files.

    Args:
        path (Path):Path in which this object will be stored

    """
    logger.debug("Storing Metadata to {}", path)

    storage_path = build_filename(path, "MetaData", fmt="json", skip_version=True)

    info_to_store = {}
    for key, value in self.information.items():
        value = json_ready_converter(value)
        info_to_store[key] = value

    with open(storage_path, mode="w") as handle:
        json.dump(info_to_store, handle, indent=4)

add_info(key: str, info: str | tuple | list, subInstrument: str) -> None

Add a new metric to be tracked, with the values being collected over all available frames.

Parameters:

Name	Type	Description	Default
key	str	Name of the metric	required
info	str | tuple | list	data to be stored	required
subInstrument	str	subInstrument to which the info belongs to	required

Raises:

Type	Description
TypeError	If the info is not a list nor an iterable

Source code in src/ASTRA/data_objects/MetaData.py

def add_info(self, key: str, info: str | tuple | list, subInstrument: str) -> None:
    """Add a new metric to be tracked, with the values being collected over all available frames.

    Parameters
    ----------
    key: str
        Name of the metric
    info: Union[str, Iterable]
        data to be stored
    subInstrument: str
        subInstrument to which the info belongs to


    Raises
    ------
    TypeError
        If the info is not a list nor an iterable

    """
    if not isinstance(info, (str, tuple, list)):
        raise TypeError("info must be  str or list object")
    if subInstrument not in self.information:
        self.information[subInstrument] = {}

    # remove duplicates, but keep as a list (to avoid problems when storing to json)!
    self.information[subInstrument][key] = list(set(info))

load_from_json(path: Path) -> MetaData classmethod

Load from a json file.

Source code in src/ASTRA/data_objects/MetaData.py

@classmethod
def load_from_json(cls, path: Path) -> MetaData:
    """Load from a json file."""
    storage_path = build_filename(
        path,
        "MetaData",
        fmt="json",
        ASTRA_version=None,
        skip_version=True,  # we don't want a version
    )
    try:
        with open(storage_path) as handle:
            information = json.load(handle)
    except FileNotFoundError:
        msg = f"Failed to find metadata file in {storage_path}"
        logger.warning(msg)
        raise custom_exceptions.StopComputationError(msg)

    return MetaData(information)

Target

Representation of the star, used to query SIMBAD.

Target

Represents an observed object.

This class provides an interface to handle data sanitization and simbad lookups for the targets that we load from the S2D files.

TODO list: - [ ] Allow for the comparison of two different targets - [ ] Improve docs and interfaces of the class

Source code in src/ASTRA/data_objects/Target.py

class Target:
    """Represents an observed object.

    This class provides an interface to handle data sanitization and simbad lookups
    for the targets that we load from the S2D files.


    TODO list:
    - [ ] Allow for the comparison of two different targets
    - [ ] Improve docs and interfaces of the class
    """

    def __init__(
        self,
        target_list: list[str],
        original_name: str | None = None,
        target_dictionary_path: None | str = None,
    ) -> None:
        """Create object and run sanity checks.

        Args:
            target_list (list[str]): List of target names that have been collected across all files that have been
                loaded from disk.
            original_name (str | None): If not None, name that we will use in the place of the one in the fits header.
            target_dictionary_path (dict[str, str] | None): If not None, path to a file that maps
              'encrypted names' to SIMBAD names, separated by ,

        """
        if len(target_list) == 0:
            msg = "No valid observations. Can't create a target name"
            logger.critical(msg)
            raise custom_exceptions.NoDataError(msg)

        self.to_replace = {
            "NAME": "",
            "star": "",
        }

        self.extra_KW = {
            "KOBE-": "",
            "HD-": "HD",
            " ": "",  # TODO: do we really want to replace empty spaces in the middle of the name?
        }

        self.KOBE_alias = {}

        if target_dictionary_path is not None and Path(target_dictionary_path).exists():
            complete_removal = {**self.to_replace, **self.extra_KW}
            with open(target_dictionary_path) as dicion:
                logger.debug("Found target dictionary; Loading keys")
                for entry in dicion:
                    combination = entry.split(",")
                    KOBE_key = combination[0]
                    for key, replace in complete_removal.items():
                        KOBE_key = KOBE_key.replace(key, replace)
                    simbad_resolvable = combination[1]
                    self.KOBE_alias[KOBE_key] = simbad_resolvable
        else:
            logger.warning(f"Target dictionary not found in <{target_dictionary_path}>")
        target_list = self.clean_targ_list(target_list)
        self.validate_target_list(target_list)

        # Name from the header!
        self._original_name: str = target_list[0].strip()

        # original_nam
        if original_name is None:
            self._overriden_name: str = self._original_name
        else:
            self._overriden_name = original_name

        if self._original_name != self._overriden_name:
            msg = f"Original target name '{self._overriden_name}' "
            msg += f"validated to '{self._original_name}'"
            logger.info(msg)
        else:
            logger.info(f"Validated target to be {self._original_name}")
        self._simbad_error = False
        self._SA = np.nan

    def clean_targ_list(self, target_list: list[str]) -> list[str]:
        """Process all names to generate clean ones."""
        logger.debug("Parsing through loaded OBJECTs")
        clean_list = []

        if len(set(target_list)) == 1 and target_list[0] is None:
            # The target name took the default value!
            clean_list = ["Unknow"]
        else:
            for targ in target_list:
                clean_name = targ
                for key, replacement in self.to_replace.items():
                    clean_name = clean_name.replace(key, replacement)
                clean_list.append(clean_name.strip())
        return clean_list

    def validate_target_list(self, targets: list[str]) -> None:
        """Raise warning if we are using different targ names."""
        # ! maybe should assert np.all(np.array(targets) == targets[0])

        processed_targlist = [self.searchable_name(i) for i in targets]
        if not np.all(np.asarray(processed_targlist) == processed_targlist[0]):
            msg = f"Different targets in the input data: {np.unique(targets)}"
            logger.warning(msg)
            # raise Exception(msg)

    @property
    def secular_acceleration(self) -> RV_measurement:
        """Return the secular accelaration of the target star."""
        if self._simbad_error:
            logger.warning("\tWARNING: Failed connection to SIMBAD!!!!!! SA OF 0 BEING RETURNED")
            self._SA = 0 * meter_second

        if np.isnan(self._SA):
            try:
                logger.info(f"Querying simbad for {self.searchable_name(self.true_name)}")
                self._SA = secular_acceleration(self.searchable_name(self.true_name))
            except Exception:
                logger.opt(exception=True).critical(
                    "Could not compute the secular accelaration from {}",
                    self.true_name,
                )
                self._SA = 0 * meter_second
                self._simbad_error = True
        return self._SA

    def searchable_name(self, star: str) -> str:
        """Transform star name in one that is recognized by SIMBAD."""
        complete_removal = {**self.to_replace, **self.extra_KW}
        for key, replace in complete_removal.items():
            star = star.replace(key, replace)

        # alias list that is recognizable from SIMBAD
        alias_list = {
            "ProximaCentauri": "proxima",
            "VV645Cen": "proxima",
            "Barnards": "GJ 699",
            "VYZCet": "YZ Cet",
            "VV376Peg": " HD 209458",
            "tauCet": "tau Cet",
        }

        alias_list = {**alias_list, **self.KOBE_alias}
        return alias_list.get(star, star)

    @property
    def true_name(self) -> str:
        """Return the name of a target that is extracted from the header of the .fits files.

        This name can be overriden if the user provides a new name when instantiating this object

        """
        return self._overriden_name

    @property
    def original_name(self) -> str:
        """Name that was present in the header of the files.

        This will disregard any kind of user-provided name

        """
        return self._original_name

    @property
    def json_ready(self) -> Dict[str, Any]:
        """Data in json-compatible fmt."""
        out = {}
        out["raw_name"] = self.true_name
        return out

to_replace = {'NAME': '', 'star': ''} instance-attribute

extra_KW = {'KOBE-': '', 'HD-': 'HD', ' ': ''} instance-attribute

KOBE_alias = {} instance-attribute

secular_acceleration: RV_measurement property

Return the secular accelaration of the target star.

true_name: str property

Return the name of a target that is extracted from the header of the .fits files.

This name can be overriden if the user provides a new name when instantiating this object

original_name: str property

Name that was present in the header of the files.

This will disregard any kind of user-provided name

json_ready: Dict[str, Any] property

Data in json-compatible fmt.

clean_targ_list(target_list: list[str]) -> list[str]

Process all names to generate clean ones.

Source code in src/ASTRA/data_objects/Target.py

def clean_targ_list(self, target_list: list[str]) -> list[str]:
    """Process all names to generate clean ones."""
    logger.debug("Parsing through loaded OBJECTs")
    clean_list = []

    if len(set(target_list)) == 1 and target_list[0] is None:
        # The target name took the default value!
        clean_list = ["Unknow"]
    else:
        for targ in target_list:
            clean_name = targ
            for key, replacement in self.to_replace.items():
                clean_name = clean_name.replace(key, replacement)
            clean_list.append(clean_name.strip())
    return clean_list

validate_target_list(targets: list[str]) -> None

Raise warning if we are using different targ names.

Source code in src/ASTRA/data_objects/Target.py

def validate_target_list(self, targets: list[str]) -> None:
    """Raise warning if we are using different targ names."""
    # ! maybe should assert np.all(np.array(targets) == targets[0])

    processed_targlist = [self.searchable_name(i) for i in targets]
    if not np.all(np.asarray(processed_targlist) == processed_targlist[0]):
        msg = f"Different targets in the input data: {np.unique(targets)}"
        logger.warning(msg)

searchable_name(star: str) -> str

Transform star name in one that is recognized by SIMBAD.

Source code in src/ASTRA/data_objects/Target.py

def searchable_name(self, star: str) -> str:
    """Transform star name in one that is recognized by SIMBAD."""
    complete_removal = {**self.to_replace, **self.extra_KW}
    for key, replace in complete_removal.items():
        star = star.replace(key, replace)

    # alias list that is recognizable from SIMBAD
    alias_list = {
        "ProximaCentauri": "proxima",
        "VV645Cen": "proxima",
        "Barnards": "GJ 699",
        "VYZCet": "YZ Cet",
        "VV376Peg": " HD 209458",
        "tauCet": "tau Cet",
    }

    alias_list = {**alias_list, **self.KOBE_alias}
    return alias_list.get(star, star)