Module pyucrio.data

Instrument data downloading and reading module. This module presently has support for data provided by the University of Calgary, such as NORSTAR Riometers, and SWAN Hyper Spectral Riometers (HSR).

Sub-modules

pyucrio.data.ucalgary

Data downloading and reading routines for data provided by the University of Calgary.

Classes

class DataManager (rio_obj)

The DataManager object is initialized within every PyUCRio object. It acts as a way to access the submodules and carry over configuration information in the super class.

Expand source code

class DataManager:
    """
    The DataManager object is initialized within every PyUCRio object. It acts as a way to access 
    the submodules and carry over configuration information in the super class.
    """

    def __init__(self, rio_obj):
        self.__rio_obj = rio_obj

        # initialize sub-modules
        self.__ucalgary = UCalgaryManager(self.__rio_obj)

    # ------------------------------------------
    # properties for submodule managers
    # ------------------------------------------
    @property
    def ucalgary(self):
        """
        Access to the `ucalgary` submodule from within a PyUCRio object.
        """
        return self.__ucalgary

    def list_datasets(self, name: Optional[str] = None, timeout: Optional[int] = None) -> List[Dataset]:
        """
        List available datasets from all providers

        Args:
            name (str): 
                Supply a name used for filtering. If that name is found in the available dataset 
                names received from the API, it will be included in the results. This parameter is
                optional.
            
            timeout (int): 
                Represents how many seconds to wait for the API to send data before giving up. The 
                default is 10 seconds, or the `api_timeout` value in the super class' `pyucrio.PyUCRio`
                object. This parameter is optional.
            
        Returns:
            A list of [`Dataset`](https://docs-pyucalgarysrs.phys.ucalgary.ca/data/classes.html#pyucalgarysrs.data.classes.Dataset)
            objects.
        
        Raises:
            pyucrio.exceptions.PyUCRioAPIError: An API error was encountered.
        """
        # init
        datasets = []

        # get ucalgary datasets
        ucalgary_datasets = self.__ucalgary.list_datasets(name=name, timeout=timeout)

        # merge
        datasets = datasets + ucalgary_datasets

        # sort by name
        datasets = sorted(datasets, key=lambda x: x.name)

        # return
        return datasets

    def get_dataset(self, name: str, timeout: Optional[int] = None) -> Dataset:
        """
        Get a specific dataset

        Args:
            name (str): 
                The dataset name to get. Case is insensitive.

            timeout (int): 
                Represents how many seconds to wait for the API to send data before giving up. The 
                default is 10 seconds, or the `api_timeout` value in the super class' `pyucrio.PyUCRio`
                object. This parameter is optional.
            
        Returns:
            The found [`Dataset`](https://docs-pyucalgarysrs.phys.ucalgary.ca/data/classes.html#pyucalgarysrs.data.classes.Dataset)
            object. Raises an exception if not found.
        
        Raises:
            pyucrio.exceptions.PyUCRioAPIError: An API error was encountered.
        """
        # init
        dataset = None

        # search ucalgary datasets
        try:
            ucalgary_dataset = self.__ucalgary.get_dataset(name, timeout=timeout)
            dataset = ucalgary_dataset
        except Exception:  # nosec
            pass

        # return
        if (dataset is None):
            # never could find it
            raise PyUCRioAPIError("Dataset not found")
        else:
            return dataset

    def list_datasets_in_table(self, name: Optional[str] = None, max_width: int = 200, timeout: Optional[int] = None) -> None:
        """
        Print available datasets from all providers in a table

        Args:
            name (str): 
                Supply a name used for filtering. If that name is found in the available dataset 
                names received from the API, it will be included in the results. This parameter is
                optional.
            
            max_width (int): 
                Maximum width of the table. Default is `200`. This parameter is optional.

            timeout (int): 
                Represents how many seconds to wait for the API to send data before giving up. The 
                default is 10 seconds, or the `api_timeout` value in the super class' `pyucrio.PyUCRio`
                object. This parameter is optional.
            
        Returns:
            Printed table.

        Raises:
            pyucrio.exceptions.PyUCRioAPIError: An API error was encountered.
        """
        # get datasets
        datasets = self.list_datasets(name=name, timeout=timeout)

        # set table lists
        table_names = []
        table_providers = []
        table_levels = []
        table_doi_details = []
        table_short_descriptions = []
        for d in datasets:
            table_names.append(d.name)
            table_providers.append(d.provider)
            table_levels.append(d.level)
            table_doi_details.append(d.doi_details)
            table_short_descriptions.append(d.short_description)

        # set header values
        table_headers = [
            "Name",
            "Provider",
            "Level",
            "DOI Details",
            "Short Description",
        ]

        # print as table
        table = Texttable(max_width=max_width)
        table.set_deco(Texttable.HEADER)
        table.set_cols_dtype(["t"] * len(table_headers))
        table.set_header_align(["l"] * len(table_headers))
        table.set_cols_align(["l"] * len(table_headers))
        table.header(table_headers)
        for i in range(0, len(table_names)):
            table.add_row([
                table_names[i],
                table_providers[i],
                table_levels[i],
                table_doi_details[i],
                table_short_descriptions[i],
            ])
        print(table.draw())

    def list_observatories(self,
                           instrument_array: Literal["norstar_riometer", "swan_hsr"],
                           uid: Optional[str] = None,
                           timeout: Optional[int] = None) -> List[Observatory]:
        """
        List information about observatories utilized by all providers.

        Args:
            instrument_array (str): 
                The instrument array to list observatories for. Valid values are: norstar_riometer, and swan_hsr.

            uid (str): 
                Supply a observatory unique identifier used for filtering (usually 4-letter site code). If that UID 
                is found in the available observatories received from the API, it will be included in the results. This 
                parameter is optional.
            
            timeout (int): 
                Represents how many seconds to wait for the API to send data before giving up. The 
                default is 10 seconds, or the `api_timeout` value in the super class' `pyucrio.PyUCRio`
                object. This parameter is optional.
            
        Returns:
            A list of [`Observatory`](https://docs-pyucalgarysrs.phys.ucalgary.ca/data/classes.html#pyucalgarysrs.data.classes.Observatory)
            objects.
        
        Raises:
            pyucrio.exceptions.PyUCRioAPIError: An API error was encountered.
        """
        # init
        observatories = []

        # get ucalgary datasets
        ucalgary_observatories = self.__ucalgary.list_observatories(instrument_array, uid=uid, timeout=timeout)

        # merge
        observatories = observatories + ucalgary_observatories

        # sort by name
        observatories = sorted(observatories, key=lambda x: x.uid)

        # return
        return observatories

    def list_observatories_in_table(self,
                                    instrument_array: Literal["norstar_riometer", "swan_hsr"],
                                    uid: Optional[str] = None,
                                    max_width: int = 200,
                                    timeout: Optional[int] = None) -> None:
        """
        Print available observatories for a given instrument array in a table

        Args:
            instrument_array (str): 
                The instrument array to list observatories for. Valid values are: norstar_riometer, and swan_hsr.

            uid (str): 
                Supply a observatory unique identifier used for filtering (usually 4-letter site code). If that UID 
                is found in the available observatories received from the API, it will be included in the results. This 
                parameter is optional.
            
            max_width (int): 
                Maximum width of the table. Default is `200`. This parameter is optional.

            timeout (int): 
                Represents how many seconds to wait for the API to send data before giving up. The 
                default is 10 seconds, or the `api_timeout` value in the super class' `pyucrio.PyUCRio`
                object. This parameter is optional.
            
        Returns:
            Printed table.
        
        Raises:
            pyucrio.exceptions.PyUCRioAPIError: An API error was encountered.
        """
        # get observatories
        observatories = self.list_observatories(instrument_array, uid=uid, timeout=timeout)

        # set table lists
        table_uids = []
        table_full_names = []
        table_geo_lats = []
        table_geo_lons = []
        table_providers = []
        for o in observatories:
            table_uids.append(o.uid)
            table_full_names.append(o.full_name)
            table_geo_lats.append(o.geodetic_latitude)
            table_geo_lons.append(o.geodetic_longitude)
            table_providers.append(o.provider)

        # set header values
        table_headers = [
            "UID",
            "Full Name",
            "Geodetic Latitude",
            "Geodetic Longitude",
            "Provider",
        ]

        # print as table
        table = Texttable(max_width=max_width)
        table.set_deco(Texttable.HEADER)
        table.set_cols_dtype(["t"] * len(table_headers))
        table.set_header_align(["l"] * len(table_headers))
        table.set_cols_align(["l"] * len(table_headers))
        table.header(table_headers)
        for i in range(0, len(table_uids)):
            table.add_row([
                table_uids[i],
                table_full_names[i],
                table_geo_lats[i],
                table_geo_lons[i],
                table_providers[i],
            ])
        print(table.draw())

Instance variables

prop ucalgary

Access to the pyucrio.data.ucalgary submodule from within a PyUCRio object.

Expand source code

@property
def ucalgary(self):
    """
    Access to the `ucalgary` submodule from within a PyUCRio object.
    """
    return self.__ucalgary

Methods

def get_dataset(self, name: str, timeout: int | None = None) ‑> pyucalgarysrs.data.classes.Dataset

Get a specific dataset

Args

name : str

The dataset name to get. Case is insensitive.

timeout : int

Represents how many seconds to wait for the API to send data before giving up. The default is 10 seconds, or the api_timeout value in the super class' PyUCRio object. This parameter is optional.

Returns

The found Dataset object. Raises an exception if not found.

Raises

PyUCRioAPIError

An API error was encountered.

def list_datasets(self, name: str | None = None, timeout: int | None = None) ‑> List[pyucalgarysrs.data.classes.Dataset]

List available datasets from all providers

Args

name : str

Supply a name used for filtering. If that name is found in the available dataset names received from the API, it will be included in the results. This parameter is optional.

timeout : int

Represents how many seconds to wait for the API to send data before giving up. The default is 10 seconds, or the api_timeout value in the super class' PyUCRio object. This parameter is optional.

Returns

A list of Dataset objects.

Raises

PyUCRioAPIError

An API error was encountered.

def list_datasets_in_table(self, name: str | None = None, max_width: int = 200, timeout: int | None = None) ‑> None

Print available datasets from all providers in a table

Args

name : str

Supply a name used for filtering. If that name is found in the available dataset names received from the API, it will be included in the results. This parameter is optional.

max_width : int

Maximum width of the table. Default is 200. This parameter is optional.

timeout : int

Represents how many seconds to wait for the API to send data before giving up. The default is 10 seconds, or the api_timeout value in the super class' PyUCRio object. This parameter is optional.

Returns

Printed table.

Raises

PyUCRioAPIError

An API error was encountered.

def list_observatories(self,
instrument_array: Literal['norstar_riometer', 'swan_hsr'],
uid: str | None = None,
timeout: int | None = None) ‑> List[pyucalgarysrs.data.classes.Observatory]

List information about observatories utilized by all providers.

Args

instrument_array : str

The instrument array to list observatories for. Valid values are: norstar_riometer, and swan_hsr.

uid : str

Supply a observatory unique identifier used for filtering (usually 4-letter site code). If that UID is found in the available observatories received from the API, it will be included in the results. This parameter is optional.

timeout : int

Represents how many seconds to wait for the API to send data before giving up. The default is 10 seconds, or the api_timeout value in the super class' PyUCRio object. This parameter is optional.

Returns

A list of Observatory objects.

Raises

PyUCRioAPIError

An API error was encountered.

def list_observatories_in_table(self,
instrument_array: Literal['norstar_riometer', 'swan_hsr'],
uid: str | None = None,
max_width: int = 200,
timeout: int | None = None) ‑> None

Print available observatories for a given instrument array in a table

Args

instrument_array : str

The instrument array to list observatories for. Valid values are: norstar_riometer, and swan_hsr.

uid : str

Supply a observatory unique identifier used for filtering (usually 4-letter site code). If that UID is found in the available observatories received from the API, it will be included in the results. This parameter is optional.

max_width : int

Maximum width of the table. Default is 200. This parameter is optional.

timeout : int

Represents how many seconds to wait for the API to send data before giving up. The default is 10 seconds, or the api_timeout value in the super class' PyUCRio object. This parameter is optional.

Returns

Printed table.

Raises

PyUCRioAPIError

An API error was encountered.