Module pyaurorax.search

Interact with the AuroraX search engine. This includes finding data sources, searching for conjunctions or ephemeris data, and uploading/managing your own data in the AuroraX platform.

Sub-modules

pyaurorax.search.api

Interface for AuroraX API requests. Primarily an under-the-hood module not needed for most use-cases.

pyaurorax.search.availability

Retrieve availability information about data in the AuroraX search engine.

pyaurorax.search.conjunctions

Use the AuroraX search engine to find conjunctions between groupings of data sources …

pyaurorax.search.data_products

Use the AuroraX search engine to search and upload data product records …

pyaurorax.search.ephemeris

Use the AuroraX search engine to search and upload ephemeris records …

pyaurorax.search.location

Class definition for AuroraX search engine location data

pyaurorax.search.metadata

Interacting with the data source metadata schemas …

pyaurorax.search.metadata_filters

Class definition for a metadata filter objects

pyaurorax.search.requests

Helper methods for retrieving data from an AuroraX search engine API request …

pyaurorax.search.sources

Manage AuroraX data sources utilized by the search engine.

pyaurorax.search.util

Utility methods. For example, converting arbitrary geographic locations to North/South B-trace geographic locations.

Classes

class AvailabilityResult (data_source: DataSource,
available_data_products: Dict | None = None,
available_ephemeris: Dict | None = None)

Expand source code

@dataclass
class AvailabilityResult:
    """
    Class definition for data availability information

    Attributes:
        data_source (pyaurorax.search.DataSource): 
            The data source that the records are associated with

        available_ephemeris (Dict): 
            The ephemeris availability information

        available_data_products (Dict): 
            The data product availability information
    """
    data_source: DataSource
    available_data_products: Optional[Dict] = None
    available_ephemeris: Optional[Dict] = None

Class definition for data availability information

Attributes

data_source : DataSource

The data source that the records are associated with

available_ephemeris : Dict

The ephemeris availability information

available_data_products : Dict

The data product availability information

Class variables

var available_data_products : Dict | None

var available_ephemeris : Dict | None

var data_source : DataSource

class Conjunction (conjunction_type: str,
start: datetime.datetime,
end: datetime.datetime,
data_sources: List[DataSource],
min_distance: float,
max_distance: float,
events: List[Dict],
closest_epoch: datetime.datetime,
farthest_epoch: datetime.datetime)

Expand source code

class Conjunction:
    """
    Conjunction object

    Attributes:
        conjunction_type (str): 
            The type of location data used when the conjunction was found (either `nbtrace`, 
            `sbtrace`, or `geographic`)
            
        start (datetime.datetime): 
            Start timestamp of the conjunction

        end (datetime.datetime): 
            End timestamp of the conjunction

        data_sources (List[DataSource]): 
            Data sources in the conjunction

        min_distance (float): 
            Minimum kilometer distance of the conjunction

        max_distance (float): 
            Maximum kilometer distance of the conjunction

        events (List[Dict]): 
            The sub-conjunctions that make up this over-arching conjunction (the conjunctions 
            between each set of two data sources)

        closest_epoch (datetime.datetime): 
            Timestamp for when data sources were closest

        farthest_epoch (datetime.datetime): 
            Timestamp for when data sources were farthest
    """

    def __init__(
        self,
        conjunction_type: str,
        start: datetime.datetime,
        end: datetime.datetime,
        data_sources: List[DataSource],
        min_distance: float,
        max_distance: float,
        events: List[Dict],
        closest_epoch: datetime.datetime,
        farthest_epoch: datetime.datetime,
    ):
        self.conjunction_type = conjunction_type
        self.start = start
        self.end = end
        self.data_sources = data_sources
        self.min_distance = min_distance
        self.max_distance = max_distance
        self.events = events
        self.closest_epoch = closest_epoch
        self.farthest_epoch = farthest_epoch

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

    def __repr__(self) -> str:
        return f"Conjunction(start={repr(self.start)}, end={repr(self.end)}, min_distance={self.min_distance:.2f}, " \
            f"max_distance={self.max_distance:.2f}, data_sources=[...], events=[...])"

Conjunction object

Attributes

conjunction_type : str

The type of location data used when the conjunction was found (either nbtrace, sbtrace, or geographic)

start : datetime.datetime

Start timestamp of the conjunction

end : datetime.datetime

End timestamp of the conjunction

data_sources : List[DataSource]

Data sources in the conjunction

min_distance : float

Minimum kilometer distance of the conjunction

max_distance : float

Maximum kilometer distance of the conjunction

events : List[Dict]

The sub-conjunctions that make up this over-arching conjunction (the conjunctions between each set of two data sources)

closest_epoch : datetime.datetime

Timestamp for when data sources were closest

farthest_epoch : datetime.datetime

Timestamp for when data sources were farthest

class ConjunctionSearch (aurorax_obj: PyAuroraX,
start: datetime.datetime,
end: datetime.datetime,
distance: Union[int, float, Dict],
ground: Sequence[Union[GroundCriteriaBlock, Dict]] = [],
space: Sequence[Union[SpaceCriteriaBlock, Dict]] = [],
events: Sequence[Union[EventsCriteriaBlock, Dict]] = [],
custom_locations: Sequence[Union[CustomLocationsCriteriaBlock, Dict]] = [],
conjunction_types: "Sequence[Union[str, Literal['nbtrace', 'sbtrace', 'geographic']]]" = ['nbtrace'],
response_format: Optional[Dict] = None)

Expand source code

class ConjunctionSearch:
    """
    Class representing a conjunction search

    Attributes:
        start (datetime.datetime): 
            Start timestamp of the search (inclusive).

        end (datetime.datetime): 
            End timestamp of the search (inclusive).

        distance (int or float or Dict): 
            The maximum distance allowed between data sources when searching for
            conjunctions. This can either be a number (int or float), or a dictionary
            modified from the output of the "get_advanced_distances_combos()" function.

        ground (List[GroundCriteriaBlock or Dict]): 
            List of ground instrument criteria blocks, defaults to [].

        space (List[SpaceCriteriaBlock or Dict]): 
            List of space instrument criteria blocks, defaults to [].

        events (List[EventsCriteriaBlock or Dict]): 
            List of event criteria blocks, defaults to [].

        custom_locations (List[CustomLocationsCriteriaBlock or Dict]): 
            List of custom location criteria blocks, defaults to [].

        conjunction_types (List[str]): 
            List of conjunction types, defaults to [] (meaning all conjunction types). Valid
            options are 'nbtrace', 'sbtrace', and 'geographic'. Defaults to 'nbtrace'.

        response_format (Dict): 
            JSON representation of desired data response format

        request (AuroraXResponse): 
            AuroraXResponse object returned when the search is executed

        request_id (str): 
            Unique ID assigned to the request by the AuroraX API

        request_url (str): 
            Unique URL assigned to the request by the AuroraX API

        executed (bool): 
            Indicates if the search has been executed/started

        completed (bool): 
            Indicates if the search has finished

        data_url (str): 
            The URL where data is accessed

        query (Dict): 
            The query for this request as JSON

        status (Dict): 
            The status of the query

        data (List[Conjunction]): 
            The conjunctions found

        logs (List[Dict]): 
            All log messages outputted by the AuroraX API for this request
    """

    __STANDARD_POLLING_SLEEP_TIME: float = 1.0

    def __init__(self,
                 aurorax_obj: PyAuroraX,
                 start: datetime.datetime,
                 end: datetime.datetime,
                 distance: Union[int, float, Dict],
                 ground: Sequence[Union[GroundCriteriaBlock, Dict]] = [],
                 space: Sequence[Union[SpaceCriteriaBlock, Dict]] = [],
                 events: Sequence[Union[EventsCriteriaBlock, Dict]] = [],
                 custom_locations: Sequence[Union[CustomLocationsCriteriaBlock, Dict]] = [],
                 conjunction_types: Sequence[Union[str, Literal["nbtrace", "sbtrace", "geographic"]]] = ["nbtrace"],
                 response_format: Optional[Dict] = None):

        # set variables using passed in args
        self.__aurorax_obj = aurorax_obj
        self.start = start
        self.end = end
        self.ground = ground
        self.space = space
        self.events = events
        self.custom_locations = custom_locations
        self.distance = distance
        self.conjunction_types = conjunction_types
        self.response_format = response_format

        # initialize additional variables
        self.request = None
        self.request_id = ""
        self.request_url = ""
        self.executed = False
        self.completed = False
        self.data_url = ""
        self.__query = {}
        self.status = {}
        self.data = []
        self.logs = []

    def __str__(self):
        return self.__repr__()

    def __repr__(self):
        return "ConjunctionSearch(executed=%s, completed=%s, request_id='%s')" % (
            self.executed,
            self.completed,
            self.request_id,
        )

    def pretty_print(self):
        """
        A special print output for this class.
        """
        # set status and query strings
        max_len = 80
        status_str = str(self.status)
        query_str = str(self.query)
        if (len(status_str) > max_len):
            status_str = "%s..." % (status_str[0:max_len])
        if (len(query_str) > max_len):
            query_str = "%s..." % (query_str[0:max_len])

        # set results string
        if (self.executed is True):
            if (len(self.data) == 0):
                data_str = "[0 conjunction results]"
            elif (len(self.data) == 1):
                data_str = "[1 conjunction result]"
            else:
                data_str = "[%d conjunction results]" % (len(self.data))
        else:
            data_str = ""

        # set logs string
        if (self.executed is True):
            if (len(self.logs) == 0):
                logs_str = "[0 log messages]"
            elif (len(self.logs) == 1):
                logs_str = "[1 log message]"
            else:
                logs_str = "[%d log messages]" % (len(self.logs))
        else:
            logs_str = ""

        # print
        print("ConjunctionSearch:")
        print("  %-13s: %s" % ("executed", self.executed))
        print("  %-13s: %s" % ("completed", self.completed))
        print("  %-13s: %s" % ("request_id", self.request_id))
        print("  %-13s: %s" % ("request", self.request))
        print("  %-13s: %s" % ("request_url", self.request_url))
        print("  %-13s: %s" % ("data_url", self.data_url))
        print("  %-13s: %s" % ("query", query_str))
        print("  %-13s: %s" % ("status", status_str))
        print("  %-13s: %s" % ("data", data_str))
        print("  %-13s: %s" % ("logs", logs_str))

    def __fill_in_missing_distances(self, curr_distances: Dict) -> Dict:
        # get all distances possible
        all_distances = self.get_advanced_distances_combos()

        # go through current distances and fill in the values
        for curr_key, curr_value in curr_distances.items():
            curr_key_split = curr_key.split('-')
            curr_key1 = curr_key_split[0].strip()
            curr_key2 = curr_key_split[1].strip()
            for all_key in all_distances.keys():
                if (curr_key1 in all_key and curr_key2 in all_key):
                    # found the matching key, replace the value
                    all_distances[all_key] = curr_value

        # return
        return all_distances

    def check_criteria_block_count_validity(self) -> None:
        """
        Check the number of of criteria blocks to see if there
        is too many. A max of 10 is allowed by the AuroraX
        conjunction search engine. An exception is raised if
        it was determined to have too many.

        Raises:
            pyaurorax.exceptions.AuroraXError: Too many criteria blocks are found
        """
        count_ground = 0
        count_space = 0
        count_events = 0
        count_custom_locations = 0
        if (self.ground is not None):
            count_ground = len(self.ground)
        if (self.space is not None):
            count_space = len(self.space)
        if (self.events is not None):
            count_events = len(self.events)
        if (self.custom_locations is not None):
            count_custom_locations = len(self.custom_locations)
        if ((count_ground + count_space + count_events + count_custom_locations) > 10):
            raise AuroraXError("Number of criteria blocks exceeds 10, please reduce the count")

    def get_advanced_distances_combos(self, default_distance: Optional[Union[int, float]] = None) -> Dict:
        """
        Get the advanced distances combinations for this search

        Args:
            default_distance (int): 
                The default distance to use, defaults to None

        Returns:
            The advanced distances combinations
        """
        # set input arrays
        options = []
        if (self.ground is not None):
            for i in range(0, len(self.ground)):
                options.append("ground%d" % (i + 1))
        if (self.space is not None):
            for i in range(0, len(self.space)):
                options.append("space%d" % (i + 1))
        if (self.events is not None):
            for i in range(0, len(self.events)):
                options.append("events%d" % (i + 1))
        if (self.custom_locations is not None):
            for i in range(0, len(self.custom_locations)):
                options.append("adhoc%d" % (i + 1))

        # derive all combinations of options of size 2
        combinations = {}
        for element in itertools.combinations(options, r=2):
            combinations["%s-%s" % (element[0], element[1])] = default_distance

        # return
        return combinations

    @property
    def distance(self) -> Union[int, float, Dict[str, Union[int, float]]]:
        """
        Property for the distance parameter

        Returns:
            The distance dictionary with all combinations
        """
        return self.__distance

    @distance.setter
    def distance(self, distance: Union[int, float, Dict[str, Union[int, float]]]) -> None:
        # set distances to a dict if it's an int or float
        if (isinstance(distance, int) or isinstance(distance, float)):
            self.__distance = self.get_advanced_distances_combos(default_distance=distance)  # type: ignore
        else:
            # is a dict, fill in any gaps
            self.__distance = self.__fill_in_missing_distances(distance)  # type: ignore

    @property
    def query(self) -> Dict:
        """
        Property for the query value

        Returns:
            The query parameter
        """
        # set ground
        ground_param = self.ground
        if (isinstance(self.ground, list) and all(isinstance(item, GroundCriteriaBlock) for item in self.ground)):
            # ground parameter is a list of GroundCriteriaBlock objects, so we
            # want to set the query to the dict version of it
            ground_param = []
            for val in self.ground:
                this_dict = deepcopy(val.__dict__)
                this_dict["ephemeris_metadata_filters"] = this_dict["metadata_filters"]
                del this_dict["metadata_filters"]
                if (this_dict["ephemeris_metadata_filters"] is not None):
                    this_dict["ephemeris_metadata_filters"] = this_dict["ephemeris_metadata_filters"].to_query_dict()
                ground_param.append(this_dict)

        # set space
        space_param = self.space
        if (isinstance(self.space, list) and all(isinstance(item, SpaceCriteriaBlock) for item in self.space)):
            # space parameter is a list of SpaceCriteriaBlock objects, so we
            # want to set the query to the dict version of it
            space_param = []
            for val in self.space:
                this_dict = deepcopy(val.__dict__)
                this_dict["ephemeris_metadata_filters"] = this_dict["metadata_filters"]
                del this_dict["metadata_filters"]
                if (this_dict["ephemeris_metadata_filters"] is not None):
                    this_dict["ephemeris_metadata_filters"] = this_dict["ephemeris_metadata_filters"].to_query_dict()
                space_param.append(this_dict)

        # set events
        events_param = self.events
        if (isinstance(self.events, list) and all(isinstance(item, EventsCriteriaBlock) for item in self.events)):
            # space parameter is a list of EventsCriteriaBlock objects, so we
            # want to set the query to the dict version of it
            events_param = []
            for val in self.events:
                this_dict = deepcopy(val.__dict__)
                this_dict["ephemeris_metadata_filters"] = this_dict["metadata_filters"]
                this_dict["programs"] = ["events"]
                del this_dict["metadata_filters"]
                if (this_dict["ephemeris_metadata_filters"] is not None):
                    this_dict["ephemeris_metadata_filters"] = this_dict["ephemeris_metadata_filters"].to_query_dict()
                events_param.append(this_dict)

        # set custom locations
        custom_param = self.custom_locations
        if (isinstance(self.custom_locations, list) and all(isinstance(item, CustomLocationsCriteriaBlock) for item in self.custom_locations)):
            # space parameter is a list of CustomLocationsCriteriaBlock objects, so we
            # want to set the query to the dict version of it
            custom_param = []
            for val in self.custom_locations:
                this_dict = val.to_search_query_dict()  # type: ignore
                custom_param.append(this_dict)

        # set query
        self.__query = {
            "start": self.start.strftime("%Y-%m-%dT%H:%M:%S"),
            "end": self.end.strftime("%Y-%m-%dT%H:%M:%S"),
            "ground": ground_param,
            "space": space_param,
            "events": events_param,
            "adhoc": custom_param,
            "conjunction_types": self.conjunction_types,
            "max_distances": self.distance,
            "epoch_search_precision": 60,
        }
        return self.__query

    @query.setter
    def query(self, query: Dict) -> None:
        self.__query = query

    def execute(self) -> None:
        """
        Initiate a conjunction search request

        Raises:
            pyaurorax.exceptions.AuroraXAPIError: An API error was encountered
        """
        # check number of criteria blocks
        self.check_criteria_block_count_validity()

        # do request
        url = "%s/%s" % (self.__aurorax_obj.api_base_url, self.__aurorax_obj.search.api.URL_SUFFIX_CONJUNCTION_SEARCH)
        req = AuroraXAPIRequest(self.__aurorax_obj, method="post", url=url, body=self.query, null_response=True)
        res = req.execute()

        # set request ID, request_url, executed
        self.executed = True
        if res.status_code == 202:
            # request successfully dispatched
            self.executed = True
            self.request_url = res.request.headers["location"]
            self.request_id = self.request_url.rsplit("/", 1)[-1]

        # set request variable
        self.request = res

    def update_status(self, status: Optional[Dict] = None) -> None:
        """
        Update the status of this conjunction search request

        Args:
            status (Dict): 
                The previously-retrieved status of this request (include
                to avoid requesting it from the API again), defaults to None

        Raises:
            pyaurorax.exceptions.AuroraXAPIError: An API error was encountered
        """
        # get the status if it isn't passed in
        if (status is None):
            status = requests_get_status(self.__aurorax_obj, self.request_url)

        # check response
        if (status is None):
            raise AuroraXAPIError("Could not retrieve status for this request")

        # update request status by checking if data URI is set
        if (status["search_result"]["data_uri"] is not None):
            self.completed = True
            self.data_url = "%s/data" % (self.request_url)

        # set class variable "status" and "logs"
        self.status = status
        self.logs = status["logs"]

    def check_for_data(self) -> bool:
        """
        Check to see if data is available for this conjunction
        search request

        Returns:
            True if data is available, else False

        Raises:
            pyaurorax.exceptions.AuroraXAPIError: An API error was encountered
        """
        self.update_status()
        return self.completed

    def get_data(self) -> None:
        """
        Retrieve the data available for this conjunction search request

        Raises:
            pyaurorax.exceptions.AuroraXAPIError: An API error was encountered
        """
        # check if completed yet
        if (self.completed is False):
            print("No data available, update status or check for data first")
            return

        # get data
        raw_data = requests_get_data(self.__aurorax_obj, self.data_url, self.response_format, False)

        # set data variable
        if (self.response_format is not None):
            self.data = raw_data
        else:
            # cast data source objects
            for i in range(0, len(raw_data)):
                for j in range(0, len(raw_data[i]["data_sources"])):
                    ds = DataSource(**raw_data[i]["data_sources"][j], format=FORMAT_BASIC_INFO)
                    raw_data[i]["data_sources"][j] = ds

            # cast conjunctions
            self.data = [Conjunction(**c) for c in raw_data]

    def wait(self, poll_interval: float = __STANDARD_POLLING_SLEEP_TIME, verbose: bool = False) -> None:
        """
        Block and wait until the request is complete and data is
        available for retrieval

        Args:
            poll_interval (float): 
                Time in seconds to wait between polling attempts, defaults to 1 second

            verbose (bool): 
                Output poll times and other progress messages, defaults to False

        Raises:
            pyaurorax.exceptions.AuroraXAPIError: An API error was encountered
        """
        url = "%s/%s" % (self.__aurorax_obj.api_base_url, self.__aurorax_obj.search.api.URL_SUFFIX_CONJUNCTION_REQUEST.format(self.request_id))
        self.update_status(requests_wait_for_data(self.__aurorax_obj, url, poll_interval, verbose))

    def cancel(self, wait: bool = False, poll_interval: float = __STANDARD_POLLING_SLEEP_TIME, verbose: bool = False) -> int:
        """
        Cancel the conjunction search request

        This method returns immediately by default since the API processes
        this request asynchronously. If you would prefer to wait for it
        to be completed, set the 'wait' parameter to True. You can adjust
        the polling time using the 'poll_interval' parameter.

        Args:
            wait (bool): 
                Wait until the cancellation request has been completed (may wait for several minutes)

            poll_interval (float): 
                Seconds to wait between polling calls, defaults to STANDARD_POLLING_SLEEP_TIME.

            verbose (bool): 
                Output poll times and other progress messages, defaults to False

        Returns:
            1 on success

        Raises:
            pyaurorax.exceptions.AuroraXAPIError: An API error was encountered
        """
        url = "%s/%s" % (self.__aurorax_obj.api_base_url, self.__aurorax_obj.search.api.URL_SUFFIX_CONJUNCTION_REQUEST.format(self.request_id))
        return requests_cancel(self.__aurorax_obj, url, wait, poll_interval, verbose)

    def describe(self):
        """
        Describe the conjunction search as an "SQL-like" string.

        Returns:
            The "SQL-like" string describing the conjunction search object
        """
        # make request
        url = "%s/%s" % (self.__aurorax_obj.api_base_url, self.__aurorax_obj.search.api.URL_SUFFIX_DESCRIBE_CONJUNCTION_QUERY)
        req = AuroraXAPIRequest(self.__aurorax_obj, method="post", url=url, body=self.query)
        res = req.execute()

        # return
        return res.data

Class representing a conjunction search

Attributes

start : datetime.datetime

Start timestamp of the search (inclusive).

end : datetime.datetime

End timestamp of the search (inclusive).

distance : int or float or Dict

The maximum distance allowed between data sources when searching for conjunctions. This can either be a number (int or float), or a dictionary modified from the output of the "get_advanced_distances_combos()" function.

ground : List[GroundCriteriaBlock or Dict]

List of ground instrument criteria blocks, defaults to [].

space : List[SpaceCriteriaBlock or Dict]

List of space instrument criteria blocks, defaults to [].

events : List[EventsCriteriaBlock or Dict]

List of event criteria blocks, defaults to [].

custom_locations : List[CustomLocationsCriteriaBlock or Dict]

List of custom location criteria blocks, defaults to [].

conjunction_types : List[str]

List of conjunction types, defaults to [] (meaning all conjunction types). Valid options are 'nbtrace', 'sbtrace', and 'geographic'. Defaults to 'nbtrace'.

response_format : Dict

JSON representation of desired data response format

request : AuroraXResponse

AuroraXResponse object returned when the search is executed

request_id : str

Unique ID assigned to the request by the AuroraX API

request_url : str

Unique URL assigned to the request by the AuroraX API

executed : bool

Indicates if the search has been executed/started

completed : bool

Indicates if the search has finished

data_url : str

The URL where data is accessed

query : Dict

The query for this request as JSON

status : Dict

The status of the query

data : List[Conjunction]

The conjunctions found

logs : List[Dict]

All log messages outputted by the AuroraX API for this request

Instance variables

prop distance : Union[int, float, Dict[str, Union[int, float]]]

Expand source code

@property
def distance(self) -> Union[int, float, Dict[str, Union[int, float]]]:
    """
    Property for the distance parameter

    Returns:
        The distance dictionary with all combinations
    """
    return self.__distance

Property for the distance parameter

Returns

The distance dictionary with all combinations

prop query : Dict

Expand source code

@property
def query(self) -> Dict:
    """
    Property for the query value

    Returns:
        The query parameter
    """
    # set ground
    ground_param = self.ground
    if (isinstance(self.ground, list) and all(isinstance(item, GroundCriteriaBlock) for item in self.ground)):
        # ground parameter is a list of GroundCriteriaBlock objects, so we
        # want to set the query to the dict version of it
        ground_param = []
        for val in self.ground:
            this_dict = deepcopy(val.__dict__)
            this_dict["ephemeris_metadata_filters"] = this_dict["metadata_filters"]
            del this_dict["metadata_filters"]
            if (this_dict["ephemeris_metadata_filters"] is not None):
                this_dict["ephemeris_metadata_filters"] = this_dict["ephemeris_metadata_filters"].to_query_dict()
            ground_param.append(this_dict)

    # set space
    space_param = self.space
    if (isinstance(self.space, list) and all(isinstance(item, SpaceCriteriaBlock) for item in self.space)):
        # space parameter is a list of SpaceCriteriaBlock objects, so we
        # want to set the query to the dict version of it
        space_param = []
        for val in self.space:
            this_dict = deepcopy(val.__dict__)
            this_dict["ephemeris_metadata_filters"] = this_dict["metadata_filters"]
            del this_dict["metadata_filters"]
            if (this_dict["ephemeris_metadata_filters"] is not None):
                this_dict["ephemeris_metadata_filters"] = this_dict["ephemeris_metadata_filters"].to_query_dict()
            space_param.append(this_dict)

    # set events
    events_param = self.events
    if (isinstance(self.events, list) and all(isinstance(item, EventsCriteriaBlock) for item in self.events)):
        # space parameter is a list of EventsCriteriaBlock objects, so we
        # want to set the query to the dict version of it
        events_param = []
        for val in self.events:
            this_dict = deepcopy(val.__dict__)
            this_dict["ephemeris_metadata_filters"] = this_dict["metadata_filters"]
            this_dict["programs"] = ["events"]
            del this_dict["metadata_filters"]
            if (this_dict["ephemeris_metadata_filters"] is not None):
                this_dict["ephemeris_metadata_filters"] = this_dict["ephemeris_metadata_filters"].to_query_dict()
            events_param.append(this_dict)

    # set custom locations
    custom_param = self.custom_locations
    if (isinstance(self.custom_locations, list) and all(isinstance(item, CustomLocationsCriteriaBlock) for item in self.custom_locations)):
        # space parameter is a list of CustomLocationsCriteriaBlock objects, so we
        # want to set the query to the dict version of it
        custom_param = []
        for val in self.custom_locations:
            this_dict = val.to_search_query_dict()  # type: ignore
            custom_param.append(this_dict)

    # set query
    self.__query = {
        "start": self.start.strftime("%Y-%m-%dT%H:%M:%S"),
        "end": self.end.strftime("%Y-%m-%dT%H:%M:%S"),
        "ground": ground_param,
        "space": space_param,
        "events": events_param,
        "adhoc": custom_param,
        "conjunction_types": self.conjunction_types,
        "max_distances": self.distance,
        "epoch_search_precision": 60,
    }
    return self.__query

Property for the query value

Returns

The query parameter

Methods

def cancel(self, wait: bool = False, poll_interval: float = 1.0, verbose: bool = False) ‑> int

Expand source code

def cancel(self, wait: bool = False, poll_interval: float = __STANDARD_POLLING_SLEEP_TIME, verbose: bool = False) -> int:
    """
    Cancel the conjunction search request

    This method returns immediately by default since the API processes
    this request asynchronously. If you would prefer to wait for it
    to be completed, set the 'wait' parameter to True. You can adjust
    the polling time using the 'poll_interval' parameter.

    Args:
        wait (bool): 
            Wait until the cancellation request has been completed (may wait for several minutes)

        poll_interval (float): 
            Seconds to wait between polling calls, defaults to STANDARD_POLLING_SLEEP_TIME.

        verbose (bool): 
            Output poll times and other progress messages, defaults to False

    Returns:
        1 on success

    Raises:
        pyaurorax.exceptions.AuroraXAPIError: An API error was encountered
    """
    url = "%s/%s" % (self.__aurorax_obj.api_base_url, self.__aurorax_obj.search.api.URL_SUFFIX_CONJUNCTION_REQUEST.format(self.request_id))
    return requests_cancel(self.__aurorax_obj, url, wait, poll_interval, verbose)

Cancel the conjunction search request

This method returns immediately by default since the API processes this request asynchronously. If you would prefer to wait for it to be completed, set the 'wait' parameter to True. You can adjust the polling time using the 'poll_interval' parameter.

Args

wait : bool

Wait until the cancellation request has been completed (may wait for several minutes)

poll_interval : float

Seconds to wait between polling calls, defaults to STANDARD_POLLING_SLEEP_TIME.

verbose : bool

Output poll times and other progress messages, defaults to False

Returns

1 on success

Raises

AuroraXAPIError

An API error was encountered

def check_criteria_block_count_validity(self) ‑> None

Expand source code

def check_criteria_block_count_validity(self) -> None:
    """
    Check the number of of criteria blocks to see if there
    is too many. A max of 10 is allowed by the AuroraX
    conjunction search engine. An exception is raised if
    it was determined to have too many.

    Raises:
        pyaurorax.exceptions.AuroraXError: Too many criteria blocks are found
    """
    count_ground = 0
    count_space = 0
    count_events = 0
    count_custom_locations = 0
    if (self.ground is not None):
        count_ground = len(self.ground)
    if (self.space is not None):
        count_space = len(self.space)
    if (self.events is not None):
        count_events = len(self.events)
    if (self.custom_locations is not None):
        count_custom_locations = len(self.custom_locations)
    if ((count_ground + count_space + count_events + count_custom_locations) > 10):
        raise AuroraXError("Number of criteria blocks exceeds 10, please reduce the count")

Check the number of of criteria blocks to see if there is too many. A max of 10 is allowed by the AuroraX conjunction search engine. An exception is raised if it was determined to have too many.

Raises

AuroraXError

Too many criteria blocks are found

def check_for_data(self) ‑> bool

Expand source code

def check_for_data(self) -> bool:
    """
    Check to see if data is available for this conjunction
    search request

    Returns:
        True if data is available, else False

    Raises:
        pyaurorax.exceptions.AuroraXAPIError: An API error was encountered
    """
    self.update_status()
    return self.completed

Check to see if data is available for this conjunction search request

Returns

True if data is available, else False

Raises

AuroraXAPIError

An API error was encountered

def describe(self)

Expand source code

def describe(self):
    """
    Describe the conjunction search as an "SQL-like" string.

    Returns:
        The "SQL-like" string describing the conjunction search object
    """
    # make request
    url = "%s/%s" % (self.__aurorax_obj.api_base_url, self.__aurorax_obj.search.api.URL_SUFFIX_DESCRIBE_CONJUNCTION_QUERY)
    req = AuroraXAPIRequest(self.__aurorax_obj, method="post", url=url, body=self.query)
    res = req.execute()

    # return
    return res.data

Describe the conjunction search as an "SQL-like" string.

Returns

The "SQL-like" string describing the conjunction search object

def execute(self) ‑> None

Expand source code

def execute(self) -> None:
    """
    Initiate a conjunction search request

    Raises:
        pyaurorax.exceptions.AuroraXAPIError: An API error was encountered
    """
    # check number of criteria blocks
    self.check_criteria_block_count_validity()

    # do request
    url = "%s/%s" % (self.__aurorax_obj.api_base_url, self.__aurorax_obj.search.api.URL_SUFFIX_CONJUNCTION_SEARCH)
    req = AuroraXAPIRequest(self.__aurorax_obj, method="post", url=url, body=self.query, null_response=True)
    res = req.execute()

    # set request ID, request_url, executed
    self.executed = True
    if res.status_code == 202:
        # request successfully dispatched
        self.executed = True
        self.request_url = res.request.headers["location"]
        self.request_id = self.request_url.rsplit("/", 1)[-1]

    # set request variable
    self.request = res

Initiate a conjunction search request

Raises

AuroraXAPIError

An API error was encountered

def get_advanced_distances_combos(self, default_distance: Optional[Union[int, float]] = None) ‑> Dict

Expand source code

def get_advanced_distances_combos(self, default_distance: Optional[Union[int, float]] = None) -> Dict:
    """
    Get the advanced distances combinations for this search

    Args:
        default_distance (int): 
            The default distance to use, defaults to None

    Returns:
        The advanced distances combinations
    """
    # set input arrays
    options = []
    if (self.ground is not None):
        for i in range(0, len(self.ground)):
            options.append("ground%d" % (i + 1))
    if (self.space is not None):
        for i in range(0, len(self.space)):
            options.append("space%d" % (i + 1))
    if (self.events is not None):
        for i in range(0, len(self.events)):
            options.append("events%d" % (i + 1))
    if (self.custom_locations is not None):
        for i in range(0, len(self.custom_locations)):
            options.append("adhoc%d" % (i + 1))

    # derive all combinations of options of size 2
    combinations = {}
    for element in itertools.combinations(options, r=2):
        combinations["%s-%s" % (element[0], element[1])] = default_distance

    # return
    return combinations

Get the advanced distances combinations for this search

Args

default_distance : int

The default distance to use, defaults to None

Returns

The advanced distances combinations

def get_data(self) ‑> None

Expand source code

def get_data(self) -> None:
    """
    Retrieve the data available for this conjunction search request

    Raises:
        pyaurorax.exceptions.AuroraXAPIError: An API error was encountered
    """
    # check if completed yet
    if (self.completed is False):
        print("No data available, update status or check for data first")
        return

    # get data
    raw_data = requests_get_data(self.__aurorax_obj, self.data_url, self.response_format, False)

    # set data variable
    if (self.response_format is not None):
        self.data = raw_data
    else:
        # cast data source objects
        for i in range(0, len(raw_data)):
            for j in range(0, len(raw_data[i]["data_sources"])):
                ds = DataSource(**raw_data[i]["data_sources"][j], format=FORMAT_BASIC_INFO)
                raw_data[i]["data_sources"][j] = ds

        # cast conjunctions
        self.data = [Conjunction(**c) for c in raw_data]

Retrieve the data available for this conjunction search request

Raises

AuroraXAPIError

An API error was encountered

def pretty_print(self)

Expand source code

def pretty_print(self):
    """
    A special print output for this class.
    """
    # set status and query strings
    max_len = 80
    status_str = str(self.status)
    query_str = str(self.query)
    if (len(status_str) > max_len):
        status_str = "%s..." % (status_str[0:max_len])
    if (len(query_str) > max_len):
        query_str = "%s..." % (query_str[0:max_len])

    # set results string
    if (self.executed is True):
        if (len(self.data) == 0):
            data_str = "[0 conjunction results]"
        elif (len(self.data) == 1):
            data_str = "[1 conjunction result]"
        else:
            data_str = "[%d conjunction results]" % (len(self.data))
    else:
        data_str = ""

    # set logs string
    if (self.executed is True):
        if (len(self.logs) == 0):
            logs_str = "[0 log messages]"
        elif (len(self.logs) == 1):
            logs_str = "[1 log message]"
        else:
            logs_str = "[%d log messages]" % (len(self.logs))
    else:
        logs_str = ""

    # print
    print("ConjunctionSearch:")
    print("  %-13s: %s" % ("executed", self.executed))
    print("  %-13s: %s" % ("completed", self.completed))
    print("  %-13s: %s" % ("request_id", self.request_id))
    print("  %-13s: %s" % ("request", self.request))
    print("  %-13s: %s" % ("request_url", self.request_url))
    print("  %-13s: %s" % ("data_url", self.data_url))
    print("  %-13s: %s" % ("query", query_str))
    print("  %-13s: %s" % ("status", status_str))
    print("  %-13s: %s" % ("data", data_str))
    print("  %-13s: %s" % ("logs", logs_str))

A special print output for this class.

def update_status(self, status: Optional[Dict] = None) ‑> None

Expand source code

def update_status(self, status: Optional[Dict] = None) -> None:
    """
    Update the status of this conjunction search request

    Args:
        status (Dict): 
            The previously-retrieved status of this request (include
            to avoid requesting it from the API again), defaults to None

    Raises:
        pyaurorax.exceptions.AuroraXAPIError: An API error was encountered
    """
    # get the status if it isn't passed in
    if (status is None):
        status = requests_get_status(self.__aurorax_obj, self.request_url)

    # check response
    if (status is None):
        raise AuroraXAPIError("Could not retrieve status for this request")

    # update request status by checking if data URI is set
    if (status["search_result"]["data_uri"] is not None):
        self.completed = True
        self.data_url = "%s/data" % (self.request_url)

    # set class variable "status" and "logs"
    self.status = status
    self.logs = status["logs"]

Update the status of this conjunction search request

Args

status : Dict

The previously-retrieved status of this request (include to avoid requesting it from the API again), defaults to None

Raises

AuroraXAPIError

An API error was encountered

def wait(self, poll_interval: float = 1.0, verbose: bool = False) ‑> None

Expand source code

def wait(self, poll_interval: float = __STANDARD_POLLING_SLEEP_TIME, verbose: bool = False) -> None:
    """
    Block and wait until the request is complete and data is
    available for retrieval

    Args:
        poll_interval (float): 
            Time in seconds to wait between polling attempts, defaults to 1 second

        verbose (bool): 
            Output poll times and other progress messages, defaults to False

    Raises:
        pyaurorax.exceptions.AuroraXAPIError: An API error was encountered
    """
    url = "%s/%s" % (self.__aurorax_obj.api_base_url, self.__aurorax_obj.search.api.URL_SUFFIX_CONJUNCTION_REQUEST.format(self.request_id))
    self.update_status(requests_wait_for_data(self.__aurorax_obj, url, poll_interval, verbose))

Block and wait until the request is complete and data is available for retrieval

Args

poll_interval : float

Time in seconds to wait between polling attempts, defaults to 1 second

verbose : bool

Output poll times and other progress messages, defaults to False

Raises

AuroraXAPIError

An API error was encountered

class CustomLocationsCriteriaBlock (locations: List[Tuple[float, float]] = [])

Expand source code

class CustomLocationsCriteriaBlock:
    """
    Representation of a custom location criteria block used for conjunction searches. 

    Attributes:
        locations (List[Tuple[float, float]]): 
            List of locations to use in this criteria block. Location are 2-element tuples
            of (latitude, longitude). Latitude values must be between -90 and 90, and longitude
            values must be between -180 and 180. Optional, default is `[]`.
    """

    def __init__(self, locations: List[Tuple[float, float]] = []):
        self.locations = locations

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

    def __repr__(self) -> str:
        return "CustomLocationsCriteriaBlock(locations=%s)" % (self.locations)

    def pretty_print(self):
        """
        A special print output for this class.
        """
        print("CustomLocationsCriteriaBlock:")
        print("  %-11s: %s" % ("locations", self.locations))

    def to_search_query_dict(self):
        query_dict = {}
        if (len(self.locations) > 0):
            query_dict["locations"] = []
            for location in self.locations:
                query_dict["locations"].append({
                    "lat": location[0],
                    "lon": location[1],
                })
        return query_dict

Representation of a custom location criteria block used for conjunction searches.

Attributes

locations : List[Tuple[float, float]]

List of locations to use in this criteria block. Location are 2-element tuples of (latitude, longitude). Latitude values must be between -90 and 90, and longitude values must be between -180 and 180. Optional, default is [].

Methods

def pretty_print(self)

Expand source code

def pretty_print(self):
    """
    A special print output for this class.
    """
    print("CustomLocationsCriteriaBlock:")
    print("  %-11s: %s" % ("locations", self.locations))

A special print output for this class.

def to_search_query_dict(self)

Expand source code

def to_search_query_dict(self):
    query_dict = {}
    if (len(self.locations) > 0):
        query_dict["locations"] = []
        for location in self.locations:
            query_dict["locations"].append({
                "lat": location[0],
                "lon": location[1],
            })
    return query_dict

class DataProductData (data_source: DataSource,
data_product_type: str,
start: datetime.datetime,
end: datetime.datetime,
url: str,
metadata: Dict | None = None)

Expand source code

class DataProductData:
    """
    Data product object

    Attributes:
        data_source (DataSource): 
            Data source that the ephemeris record is associated with

        data_product_type (str): 
            Data product type ("keogram", "movie", "summary_plot", etc.)

        start (datetime.datetime): 
            Starting timestamp for the record (assumed it is in UTC), inclusive

        end (datetime.datetime): 
            Ending timestamp for the record (assumed it is in UTC), inclusive

        url (str): 
            The URL of data product

        metadata (Dict): 
            Metadata for this record (arbitrary keys and values)
    """

    def __init__(self,
                 data_source: DataSource,
                 data_product_type: str,
                 start: datetime.datetime,
                 end: datetime.datetime,
                 url: str,
                 metadata: Optional[Dict] = None):
        self.data_source = data_source
        self.data_product_type = data_product_type
        self.start = start
        self.end = end
        self.url = url
        self.metadata = metadata

    def to_json_serializable(self) -> Dict:
        """
        Convert object to a JSON-serializable object (ie. translate
        datetime objects to strings)

        Returns:
            A dictionary object that is JSON-serializable
        """
        # init
        d = self.__dict__

        # format epoch as str
        if (type(d["start"]) is datetime.datetime):
            d["start"] = d["start"].strftime("%Y-%m-%dT%H:%M:00.000")
        if (type(d["end"]) is datetime.datetime):
            d["end"] = d["end"].strftime("%Y-%m-%dT%H:%M:00.000")

        # format metadata
        if (self.metadata is not None):
            for key, value in self.metadata.items():
                if (isinstance(value, datetime.datetime) is True or isinstance(value, datetime.date) is True):
                    self.metadata[key] = self.metadata[key].strftime("%Y-%m-%dT%H:%M:%S.%f")
        # if (isinstance(self.metadata, list) is True):
        #     self.metadata = {}

        # format data source fields for query
        d["program"] = self.data_source.program
        d["platform"] = self.data_source.platform
        d["instrument_type"] = self.data_source.instrument_type
        del d["data_source"]

        # return
        return d

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

    def __repr__(self) -> str:
        # shorten the metadata and url
        max_len = 20
        attr_metadata = f"{self.metadata}"
        if (len(attr_metadata) > max_len):
            attr_metadata = attr_metadata[0:max_len] + "...}"
        attr_url = f"{self.url}"
        if (len(attr_url) > max_len):
            attr_url = attr_url[0:max_len] + "..."

        # return formatted representation
        return f"DataProductData(start={repr(self.start)}, end={repr(self.end)}, data_product_type='{self.data_product_type}', " \
            f"url='{attr_url}', metadata={attr_metadata}, data_source=DataSource(...))"

Data product object

Attributes

data_source : DataSource

Data source that the ephemeris record is associated with

data_product_type : str

Data product type ("keogram", "movie", "summary_plot", etc.)

start : datetime.datetime

Starting timestamp for the record (assumed it is in UTC), inclusive

end : datetime.datetime

Ending timestamp for the record (assumed it is in UTC), inclusive

url : str

The URL of data product

metadata : Dict

Metadata for this record (arbitrary keys and values)

Methods

def to_json_serializable(self) ‑> Dict

Expand source code

def to_json_serializable(self) -> Dict:
    """
    Convert object to a JSON-serializable object (ie. translate
    datetime objects to strings)

    Returns:
        A dictionary object that is JSON-serializable
    """
    # init
    d = self.__dict__

    # format epoch as str
    if (type(d["start"]) is datetime.datetime):
        d["start"] = d["start"].strftime("%Y-%m-%dT%H:%M:00.000")
    if (type(d["end"]) is datetime.datetime):
        d["end"] = d["end"].strftime("%Y-%m-%dT%H:%M:00.000")

    # format metadata
    if (self.metadata is not None):
        for key, value in self.metadata.items():
            if (isinstance(value, datetime.datetime) is True or isinstance(value, datetime.date) is True):
                self.metadata[key] = self.metadata[key].strftime("%Y-%m-%dT%H:%M:%S.%f")
    # if (isinstance(self.metadata, list) is True):
    #     self.metadata = {}

    # format data source fields for query
    d["program"] = self.data_source.program
    d["platform"] = self.data_source.platform
    d["instrument_type"] = self.data_source.instrument_type
    del d["data_source"]

    # return
    return d

Convert object to a JSON-serializable object (ie. translate datetime objects to strings)

Returns

A dictionary object that is JSON-serializable

class DataProductSearch (aurorax_obj: PyAuroraX,
start: datetime.datetime,
end: datetime.datetime,
programs: Optional[List[str]] = None,
platforms: Optional[List[str]] = None,
instrument_types: Optional[List[str]] = None,
data_product_types: "Optional[Literal['keogram', 'montage', 'movie', 'summary_plot', 'data_availability']]" = None,
metadata_filters: Optional[Union[MetadataFilter, List[Dict]]] = None,
metadata_filters_logical_operator: "Optional[Literal['and', 'or', 'AND', 'OR']]" = None,
response_format: Optional[Dict] = None)

Expand source code

class DataProductSearch:
    """
    Class representing a data product search

    Attributes:
        start (datetime.datetime): 
            Start timestamp of the search (inclusive)

        end (datetime.datetime): 
            End timestamp of the search (inclusive)

        programs (List[str]): 
            List of program names to search

        platforms (List[str]): 
            List of platform names to search

        instrument_types (List[str]): 
            List of instrument types to search

        data_product_types (List[str]): 
            List of strings describing data product types to filter on e.g. "keogram", defaults 
            to None. Valid options are: `keogram`, `montage`, `movie`, `summary_plot`, and 
            `data_availability`.
    
        metadata_filters (MetadataFilter or List[Dict]): 
            The metadata filters to use when searching, defaults to None

        metadata_filters_logical_operator (str): 
            The logical operator to use when evaluating metadata filters (either `and` or `or`), 
            defaults to `and`. This parameter is deprecated in exchange for passing a 
            MetadataFilter object into the metadata_filters parameter. 

        response_format (Dict): 
            JSON representation of desired data response format

        request (AuroraXResponse): 
            AuroraXResponse object returned when the search is executed

        request_id (str): 
            Unique ID assigned to the request by the AuroraX API

        request_url (str): 
            Unique URL assigned to the request by the AuroraX API

        executed (bool): 
            Indicates if the search has been executed/started

        completed (bool): 
            Indicates if the search has finished

        data_url (str): 
            The URL where data is accessed

        query (Dict): 
            The query for this request as JSON

        status (Dict): 
            The status of the query

        data (List[DataProductData]): 
            The data product records found

        logs (List[Dict]): 
            All log messages outputted by the AuroraX API for this request
    """

    __STANDARD_POLLING_SLEEP_TIME: float = 1.0

    def __init__(self,
                 aurorax_obj: PyAuroraX,
                 start: datetime.datetime,
                 end: datetime.datetime,
                 programs: Optional[List[str]] = None,
                 platforms: Optional[List[str]] = None,
                 instrument_types: Optional[List[str]] = None,
                 data_product_types: Optional[Literal["keogram", "montage", "movie", "summary_plot", "data_availability"]] = None,
                 metadata_filters: Optional[Union[MetadataFilter, List[Dict]]] = None,
                 metadata_filters_logical_operator: Optional[Literal["and", "or", "AND", "OR"]] = None,
                 response_format: Optional[Dict] = None) -> None:

        # set variables using passed in args
        self.__aurorax_obj = aurorax_obj
        self.start = start
        self.end = end
        self.programs = programs
        self.platforms = platforms
        self.instrument_types = instrument_types
        self.data_product_types = data_product_types
        self.metadata_filters = metadata_filters
        self.metadata_filters_logical_operator = "AND" if metadata_filters_logical_operator is None else metadata_filters_logical_operator
        self.response_format = response_format

        # initialize additional variables
        self.request = None
        self.request_id = ""
        self.request_url = ""
        self.executed = False
        self.completed = False
        self.data_url = ""
        self.__query = {}
        self.status = {}
        self.data = []
        self.logs = []

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

    def __repr__(self) -> str:
        return "DataProductSearch(executed=%s, completed=%s, request_id='%s')" % (
            self.executed,
            self.completed,
            self.request_id,
        )

    def pretty_print(self):
        """
        A special print output for this class.
        """
        # set status and query strings
        max_len = 80
        status_str = str(self.status)
        query_str = str(self.query)
        if (len(status_str) > max_len):
            status_str = "%s..." % (status_str[0:max_len])
        if (len(query_str) > max_len):
            query_str = "%s..." % (query_str[0:max_len])

        # set results string
        if (self.executed is True):
            if (len(self.data) == 0):
                data_str = "[0 data product results]"
            elif (len(self.data) == 1):
                data_str = "[1 data product result]"
            else:
                data_str = "[%d data product results]" % (len(self.data))
        else:
            data_str = ""

        # set logs string
        if (self.executed is True):
            if (len(self.logs) == 0):
                logs_str = "[0 log messages]"
            elif (len(self.logs) == 1):
                logs_str = "[1 log message]"
            else:
                logs_str = "[%d log messages]" % (len(self.logs))
        else:
            logs_str = ""

        # print
        print("DataProductSearch:")
        print("  %-13s: %s" % ("executed", self.executed))
        print("  %-13s: %s" % ("completed", self.completed))
        print("  %-13s: %s" % ("request_id", self.request_id))
        print("  %-13s: %s" % ("request", self.request))
        print("  %-13s: %s" % ("request_url", self.request_url))
        print("  %-13s: %s" % ("data_url", self.data_url))
        print("  %-13s: %s" % ("query", query_str))
        print("  %-13s: %s" % ("status", status_str))
        print("  %-13s: %s" % ("data", data_str))
        print("  %-13s: %s" % ("logs", logs_str))

    @property
    def query(self):
        """
        Property for the query value
        """
        # set metadata filter value
        if (self.metadata_filters is None):
            metadata_filters_dict = {}
        elif (isinstance(self.metadata_filters, MetadataFilter) is True):
            # metadata filter is a class
            metadata_filters_dict = self.metadata_filters.to_query_dict()  # type: ignore
        else:
            # metadata filter is a dictionary
            metadata_filters_dict = {
                "expressions": self.metadata_filters,
                "logical_operator": self.metadata_filters_logical_operator,
            }

        # set query
        self.__query = {
            "data_sources": {
                "programs": [] if not self.programs else self.programs,
                "platforms": [] if not self.platforms else self.platforms,
                "instrument_types": [] if not self.instrument_types else self.instrument_types,
                "data_product_metadata_filters": metadata_filters_dict,
            },
            "start": self.start.strftime("%Y-%m-%dT%H:%M:%S"),
            "end": self.end.strftime("%Y-%m-%dT%H:%M:%S"),
            "data_product_type_filters": [] if not self.data_product_types else self.data_product_types,
        }

        # return
        return self.__query

    @query.setter
    def query(self, query):
        self.__query = query

    def execute(self) -> None:
        """
        Initiate a data product search request
        """
        # do request
        url = "%s/%s" % (self.__aurorax_obj.api_base_url, self.__aurorax_obj.search.api.URL_SUFFIX_DATA_PRODUCTS_SEARCH)
        req = AuroraXAPIRequest(self.__aurorax_obj, method="post", url=url, body=self.query, null_response=True)
        res = req.execute()

        # set request ID, request_url, executed
        self.executed = True
        if (res.status_code == 202):
            # request successfully dispatched
            self.executed = True
            self.request_url = res.request.headers["location"]
            self.request_id = self.request_url.rsplit("/", 1)[-1]

        # set request variable
        self.request = res

    def update_status(self, status: Optional[Dict] = None) -> None:
        """
        Update the status of this data product search request

        Args:
            status (Dict): 
                The previously-retrieved status of this request (include
                to avoid requesting it from the API again), defaults to None
        """
        # get the status if it isn't passed in
        if (status is None):
            status = requests_get_status(self.__aurorax_obj, self.request_url)

        # check response
        if (status is None):
            raise AuroraXAPIError("Could not retrieve status for this request")

        # update request status by checking if data URI is set
        if (status["search_result"]["data_uri"] is not None):
            self.completed = True
            self.data_url = "%s/data" % (self.request_url)

        # set class variable "status" and "logs"
        self.status = status
        self.logs = status["logs"]

    def check_for_data(self) -> bool:
        """
        Check to see if data is available for this data product
        search request

        Returns:
            True if data is available, else False
        """
        self.update_status()
        return self.completed

    def get_data(self) -> None:
        """
        Retrieve the data available for this data product search request
        """
        # check if completed yet
        if (self.completed is False):
            print("No data available, update status or check for data first")
            return

        # get data
        raw_data = requests_get_data(self.__aurorax_obj, self.data_url, self.response_format, False)

        # set data variable
        if (self.response_format is not None):
            self.data = raw_data
        else:
            # cast data source objects
            for i in range(0, len(raw_data)):
                ds = DataSource(**raw_data[i]["data_source"], format=FORMAT_BASIC_INFO)
                raw_data[i]["data_source"] = ds

            # cast data product objects
            self.data = [DataProductData(**dp) for dp in raw_data]

    def wait(self, poll_interval: float = __STANDARD_POLLING_SLEEP_TIME, verbose: bool = False) -> None:
        """
        Block and wait for the request to complete and data is available
        for retrieval

        Args:
            poll_interval (float): 
                Time in seconds to wait between polling attempts, defaults to 1 second

            verbose (bool): 
                Output poll times and other progress messages, defaults to False
        """
        url = "%s/%s" % (self.__aurorax_obj.api_base_url, self.__aurorax_obj.search.api.URL_SUFFIX_DATA_PRODUCTS_REQUEST.format(self.request_id))
        self.update_status(requests_wait_for_data(self.__aurorax_obj, url, poll_interval, verbose))

    def cancel(self, wait: bool = False, poll_interval: float = __STANDARD_POLLING_SLEEP_TIME, verbose: bool = False) -> int:
        """
        Cancel the data product search request

        This method returns immediately by default since the API processes
        this request asynchronously. If you would prefer to wait for it
        to be completed, set the 'wait' parameter to True. You can adjust
        the polling time using the 'poll_interval' parameter.

        Args:
            wait (bool): 
                Wait until the cancellation request has been completed (may wait 
                for several minutes)
            
            poll_interval (float): 
                Seconds to wait between polling calls, defaults to 1 second.
            
            verbose (bool): 
                Output poll times and other progress messages, defaults to False

        Returns:
            1 on success

        Raises:
            pyaurorax.exceptions.AuroraXAPIError: An API error was encountered
            pyaurorax.exceptions.AuroraXUnauthorizedError: Invalid API key for this operation
        """
        url = "%s/%s" % (self.__aurorax_obj.api_base_url, self.__aurorax_obj.search.api.URL_SUFFIX_DATA_PRODUCTS_REQUEST.format(self.request_id))
        return requests_cancel(self.__aurorax_obj, url, wait, poll_interval, verbose)

    def describe(self):
        """
        Describe the data products search as an "SQL-like" string.

        Returns:
            The "SQL-like" string describing the data products search object
        """
        # make request
        url = "%s/%s" % (self.__aurorax_obj.api_base_url, self.__aurorax_obj.search.api.URL_SUFFIX_DESCRIBE_DATA_PRODUCTS_QUERY)
        req = AuroraXAPIRequest(self.__aurorax_obj, method="post", url=url, body=self.query)
        res = req.execute()

        # return
        return res.data

Class representing a data product search

Attributes

start : datetime.datetime

Start timestamp of the search (inclusive)

end : datetime.datetime

End timestamp of the search (inclusive)

programs : List[str]

List of program names to search

platforms : List[str]

List of platform names to search

instrument_types : List[str]

List of instrument types to search

data_product_types : List[str]

List of strings describing data product types to filter on e.g. "keogram", defaults to None. Valid options are: keogram, montage, movie, summary_plot, and data_availability.

metadata_filters : MetadataFilter or List[Dict]

The metadata filters to use when searching, defaults to None

metadata_filters_logical_operator : str

The logical operator to use when evaluating metadata filters (either and or or), defaults to and. This parameter is deprecated in exchange for passing a MetadataFilter object into the metadata_filters parameter.

response_format : Dict

JSON representation of desired data response format

request : AuroraXResponse

AuroraXResponse object returned when the search is executed

request_id : str

Unique ID assigned to the request by the AuroraX API

request_url : str

Unique URL assigned to the request by the AuroraX API

executed : bool

Indicates if the search has been executed/started

completed : bool

Indicates if the search has finished

data_url : str

The URL where data is accessed

query : Dict

The query for this request as JSON

status : Dict

The status of the query

data : List[DataProductData]

The data product records found

logs : List[Dict]

All log messages outputted by the AuroraX API for this request

Instance variables

prop query

Expand source code

@property
def query(self):
    """
    Property for the query value
    """
    # set metadata filter value
    if (self.metadata_filters is None):
        metadata_filters_dict = {}
    elif (isinstance(self.metadata_filters, MetadataFilter) is True):
        # metadata filter is a class
        metadata_filters_dict = self.metadata_filters.to_query_dict()  # type: ignore
    else:
        # metadata filter is a dictionary
        metadata_filters_dict = {
            "expressions": self.metadata_filters,
            "logical_operator": self.metadata_filters_logical_operator,
        }

    # set query
    self.__query = {
        "data_sources": {
            "programs": [] if not self.programs else self.programs,
            "platforms": [] if not self.platforms else self.platforms,
            "instrument_types": [] if not self.instrument_types else self.instrument_types,
            "data_product_metadata_filters": metadata_filters_dict,
        },
        "start": self.start.strftime("%Y-%m-%dT%H:%M:%S"),
        "end": self.end.strftime("%Y-%m-%dT%H:%M:%S"),
        "data_product_type_filters": [] if not self.data_product_types else self.data_product_types,
    }

    # return
    return self.__query

Property for the query value

Methods

def cancel(self, wait: bool = False, poll_interval: float = 1.0, verbose: bool = False) ‑> int

Expand source code

def cancel(self, wait: bool = False, poll_interval: float = __STANDARD_POLLING_SLEEP_TIME, verbose: bool = False) -> int:
    """
    Cancel the data product search request

    This method returns immediately by default since the API processes
    this request asynchronously. If you would prefer to wait for it
    to be completed, set the 'wait' parameter to True. You can adjust
    the polling time using the 'poll_interval' parameter.

    Args:
        wait (bool): 
            Wait until the cancellation request has been completed (may wait 
            for several minutes)
        
        poll_interval (float): 
            Seconds to wait between polling calls, defaults to 1 second.
        
        verbose (bool): 
            Output poll times and other progress messages, defaults to False

    Returns:
        1 on success

    Raises:
        pyaurorax.exceptions.AuroraXAPIError: An API error was encountered
        pyaurorax.exceptions.AuroraXUnauthorizedError: Invalid API key for this operation
    """
    url = "%s/%s" % (self.__aurorax_obj.api_base_url, self.__aurorax_obj.search.api.URL_SUFFIX_DATA_PRODUCTS_REQUEST.format(self.request_id))
    return requests_cancel(self.__aurorax_obj, url, wait, poll_interval, verbose)

Cancel the data product search request

This method returns immediately by default since the API processes this request asynchronously. If you would prefer to wait for it to be completed, set the 'wait' parameter to True. You can adjust the polling time using the 'poll_interval' parameter.

Args

wait : bool

Wait until the cancellation request has been completed (may wait for several minutes)

poll_interval : float

Seconds to wait between polling calls, defaults to 1 second.

verbose : bool

Output poll times and other progress messages, defaults to False

Returns

1 on success

Raises

AuroraXAPIError

An API error was encountered

AuroraXUnauthorizedError

Invalid API key for this operation

def check_for_data(self) ‑> bool

Expand source code

def check_for_data(self) -> bool:
    """
    Check to see if data is available for this data product
    search request

    Returns:
        True if data is available, else False
    """
    self.update_status()
    return self.completed

Check to see if data is available for this data product search request

Returns

True if data is available, else False

def describe(self)

Expand source code

def describe(self):
    """
    Describe the data products search as an "SQL-like" string.

    Returns:
        The "SQL-like" string describing the data products search object
    """
    # make request
    url = "%s/%s" % (self.__aurorax_obj.api_base_url, self.__aurorax_obj.search.api.URL_SUFFIX_DESCRIBE_DATA_PRODUCTS_QUERY)
    req = AuroraXAPIRequest(self.__aurorax_obj, method="post", url=url, body=self.query)
    res = req.execute()

    # return
    return res.data

Describe the data products search as an "SQL-like" string.

Returns

The "SQL-like" string describing the data products search object

def execute(self) ‑> None

Expand source code

def execute(self) -> None:
    """
    Initiate a data product search request
    """
    # do request
    url = "%s/%s" % (self.__aurorax_obj.api_base_url, self.__aurorax_obj.search.api.URL_SUFFIX_DATA_PRODUCTS_SEARCH)
    req = AuroraXAPIRequest(self.__aurorax_obj, method="post", url=url, body=self.query, null_response=True)
    res = req.execute()

    # set request ID, request_url, executed
    self.executed = True
    if (res.status_code == 202):
        # request successfully dispatched
        self.executed = True
        self.request_url = res.request.headers["location"]
        self.request_id = self.request_url.rsplit("/", 1)[-1]

    # set request variable
    self.request = res

Initiate a data product search request

def get_data(self) ‑> None

Expand source code

def get_data(self) -> None:
    """
    Retrieve the data available for this data product search request
    """
    # check if completed yet
    if (self.completed is False):
        print("No data available, update status or check for data first")
        return

    # get data
    raw_data = requests_get_data(self.__aurorax_obj, self.data_url, self.response_format, False)

    # set data variable
    if (self.response_format is not None):
        self.data = raw_data
    else:
        # cast data source objects
        for i in range(0, len(raw_data)):
            ds = DataSource(**raw_data[i]["data_source"], format=FORMAT_BASIC_INFO)
            raw_data[i]["data_source"] = ds

        # cast data product objects
        self.data = [DataProductData(**dp) for dp in raw_data]

Retrieve the data available for this data product search request

def pretty_print(self)

Expand source code

def pretty_print(self):
    """
    A special print output for this class.
    """
    # set status and query strings
    max_len = 80
    status_str = str(self.status)
    query_str = str(self.query)
    if (len(status_str) > max_len):
        status_str = "%s..." % (status_str[0:max_len])
    if (len(query_str) > max_len):
        query_str = "%s..." % (query_str[0:max_len])

    # set results string
    if (self.executed is True):
        if (len(self.data) == 0):
            data_str = "[0 data product results]"
        elif (len(self.data) == 1):
            data_str = "[1 data product result]"
        else:
            data_str = "[%d data product results]" % (len(self.data))
    else:
        data_str = ""

    # set logs string
    if (self.executed is True):
        if (len(self.logs) == 0):
            logs_str = "[0 log messages]"
        elif (len(self.logs) == 1):
            logs_str = "[1 log message]"
        else:
            logs_str = "[%d log messages]" % (len(self.logs))
    else:
        logs_str = ""

    # print
    print("DataProductSearch:")
    print("  %-13s: %s" % ("executed", self.executed))
    print("  %-13s: %s" % ("completed", self.completed))
    print("  %-13s: %s" % ("request_id", self.request_id))
    print("  %-13s: %s" % ("request", self.request))
    print("  %-13s: %s" % ("request_url", self.request_url))
    print("  %-13s: %s" % ("data_url", self.data_url))
    print("  %-13s: %s" % ("query", query_str))
    print("  %-13s: %s" % ("status", status_str))
    print("  %-13s: %s" % ("data", data_str))
    print("  %-13s: %s" % ("logs", logs_str))

A special print output for this class.

def update_status(self, status: Optional[Dict] = None) ‑> None

Expand source code

def update_status(self, status: Optional[Dict] = None) -> None:
    """
    Update the status of this data product search request

    Args:
        status (Dict): 
            The previously-retrieved status of this request (include
            to avoid requesting it from the API again), defaults to None
    """
    # get the status if it isn't passed in
    if (status is None):
        status = requests_get_status(self.__aurorax_obj, self.request_url)

    # check response
    if (status is None):
        raise AuroraXAPIError("Could not retrieve status for this request")

    # update request status by checking if data URI is set
    if (status["search_result"]["data_uri"] is not None):
        self.completed = True
        self.data_url = "%s/data" % (self.request_url)

    # set class variable "status" and "logs"
    self.status = status
    self.logs = status["logs"]

Update the status of this data product search request

Args

status : Dict

The previously-retrieved status of this request (include to avoid requesting it from the API again), defaults to None

def wait(self, poll_interval: float = 1.0, verbose: bool = False) ‑> None

Expand source code

def wait(self, poll_interval: float = __STANDARD_POLLING_SLEEP_TIME, verbose: bool = False) -> None:
    """
    Block and wait for the request to complete and data is available
    for retrieval

    Args:
        poll_interval (float): 
            Time in seconds to wait between polling attempts, defaults to 1 second

        verbose (bool): 
            Output poll times and other progress messages, defaults to False
    """
    url = "%s/%s" % (self.__aurorax_obj.api_base_url, self.__aurorax_obj.search.api.URL_SUFFIX_DATA_PRODUCTS_REQUEST.format(self.request_id))
    self.update_status(requests_wait_for_data(self.__aurorax_obj, url, poll_interval, verbose))

Block and wait for the request to complete and data is available for retrieval

Args

poll_interval : float

Time in seconds to wait between polling attempts, defaults to 1 second

verbose : bool

Output poll times and other progress messages, defaults to False

class DataSource (identifier: int | None = None,
program: str | None = None,
platform: str | None = None,
instrument_type: str | None = None,
source_type: str | None = None,
display_name: str | None = None,
metadata: Dict | None = None,
owner: str | None = None,
maintainers: List[str] | None = None,
ephemeris_metadata_schema: List[Dict] | None = None,
data_product_metadata_schema: List[Dict] | None = None,
stats: DataSourceStatistics | None = None,
format: str = 'full_record')

Expand source code

class DataSource:
    """
    AuroraX data source record

    Attributes:
        identifier (int): 
            The unique AuroraX data source identifier

        program (str): 
            The program for this data source

        platform (str): 
            The platform for this data source

        instrument_type (str): 
            The instrument type for this data source

        source_type (str): 
            The data source type for this data source. Options are in the 
            pyaurorax.search.sources module, or at the top level using the
            pyaurorax.search.SOURCE_TYPE_* variables.

        display_name (str): 
            The display name for this data source

        metadata (Dict): 
            Metadata for this data source (arbitrary keys and values)

        owner (str): 
            The owner's email address of this data source

        maintainers (List[str]): 
            The email addresses of AuroraX accounts that can alter this data source and 
            its associated records

        ephemeris_metadata_schema (Dict): 
            A list of dictionaries capturing the metadata keys and values that can appear 
            in ephemeris records associated with this data source

        data_product_metadata_schema (Dict): 
            A list of dictionaries capturing the metadata keys and values that can appear 
            in data product records associated with this data source

        stats (DataSourceStatistics): 
            Data source statistics information
            
        format (str): 
            The format used when printing the data source, defaults to `full_record`. Other 
            options are in the pyaurorax.search.sources module, or at the top level using 
            the pyaurorax.search.FORMAT_* variables.
    """

    def __init__(self,
                 identifier: Optional[int] = None,
                 program: Optional[str] = None,
                 platform: Optional[str] = None,
                 instrument_type: Optional[str] = None,
                 source_type: Optional[str] = None,
                 display_name: Optional[str] = None,
                 metadata: Optional[Dict] = None,
                 owner: Optional[str] = None,
                 maintainers: Optional[List[str]] = None,
                 ephemeris_metadata_schema: Optional[List[Dict]] = None,
                 data_product_metadata_schema: Optional[List[Dict]] = None,
                 stats: Optional[DataSourceStatistics] = None,
                 format: str = FORMAT_FULL_RECORD):
        self.identifier = identifier
        self.program = program
        self.platform = platform
        self.instrument_type = instrument_type
        self.source_type = source_type
        self.display_name = display_name
        self.metadata = metadata
        self.owner = owner
        self.maintainers = maintainers
        self.ephemeris_metadata_schema = ephemeris_metadata_schema
        self.data_product_metadata_schema = data_product_metadata_schema
        self.stats = stats
        self.format = format

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

    def __repr__(self) -> str:
        return "DataSource(identifier=%s, program='%s', platform='%s', instrument_type='%s', source_type='%s', display_name='%s', ...)" % (
            self.identifier,
            self.program,
            self.platform,
            self.instrument_type,
            self.source_type,
            self.display_name,
        )

    def pretty_print(self):
        """
        A special print output for this class.
        """
        max_len = 80
        print("DataSource:")
        print("  %-30s: %d" % ("identifier", self.identifier))
        print("  %-30s: %s" % ("program", self.program))
        print("  %-30s: %s" % ("platform", self.platform))
        print("  %-30s: %s" % ("instrument_type", self.instrument_type))
        print("  %-30s: %s" % ("source_type", self.source_type))
        print("  %-30s: %s" % ("display_name", self.display_name))
        print("  %-30s: %s" % ("metadata", self.metadata))
        print("  %-30s: %s" % ("owner", self.owner))
        print("  %-30s: %s" % ("maintainers", self.maintainers))
        if (self.ephemeris_metadata_schema is not None and len(str(self.ephemeris_metadata_schema)) > max_len):
            ephemeris_metadata_schema_str = "%s..." % (str(self.ephemeris_metadata_schema)[0:max_len])
        else:
            ephemeris_metadata_schema_str = self.ephemeris_metadata_schema
        print("  %-30s: %s" % ("ephemeris_metadata_schema", ephemeris_metadata_schema_str))
        if (self.data_product_metadata_schema is not None and len(str(self.data_product_metadata_schema)) > max_len):
            data_product_metadata_schema_str = "%s..." % (str(self.data_product_metadata_schema)[0:max_len])
        else:
            data_product_metadata_schema_str = self.data_product_metadata_schema
        print("  %-30s: %s" % ("data_product_metadata_schema", data_product_metadata_schema_str))
        if (self.stats is not None and len(str(self.stats)) > max_len):
            stats_str = "%s..." % (str(self.stats)[0:max_len])
        else:
            stats_str = self.data_product_metadata_schema
        print("  %-30s: %s" % ("stats", stats_str))
        print("  %-30s: %s" % ("format", self.format))

AuroraX data source record

Attributes

identifier : int

The unique AuroraX data source identifier

program : str

The program for this data source

platform : str

The platform for this data source

instrument_type : str

The instrument type for this data source

source_type : str

The data source type for this data source. Options are in the pyaurorax.search.sources module, or at the top level using the pyaurorax.search.SOURCE_TYPE_* variables.

display_name : str

The display name for this data source

metadata : Dict

Metadata for this data source (arbitrary keys and values)

owner : str

The owner's email address of this data source

maintainers : List[str]

The email addresses of AuroraX accounts that can alter this data source and its associated records

ephemeris_metadata_schema : Dict

A list of dictionaries capturing the metadata keys and values that can appear in ephemeris records associated with this data source

data_product_metadata_schema : Dict

A list of dictionaries capturing the metadata keys and values that can appear in data product records associated with this data source

stats : DataSourceStatistics

Data source statistics information

format : str

The format used when printing the data source, defaults to full_record. Other options are in the pyaurorax.search.sources module, or at the top level using the pyaurorax.search.FORMAT_* variables.

Methods

def pretty_print(self)

Expand source code

def pretty_print(self):
    """
    A special print output for this class.
    """
    max_len = 80
    print("DataSource:")
    print("  %-30s: %d" % ("identifier", self.identifier))
    print("  %-30s: %s" % ("program", self.program))
    print("  %-30s: %s" % ("platform", self.platform))
    print("  %-30s: %s" % ("instrument_type", self.instrument_type))
    print("  %-30s: %s" % ("source_type", self.source_type))
    print("  %-30s: %s" % ("display_name", self.display_name))
    print("  %-30s: %s" % ("metadata", self.metadata))
    print("  %-30s: %s" % ("owner", self.owner))
    print("  %-30s: %s" % ("maintainers", self.maintainers))
    if (self.ephemeris_metadata_schema is not None and len(str(self.ephemeris_metadata_schema)) > max_len):
        ephemeris_metadata_schema_str = "%s..." % (str(self.ephemeris_metadata_schema)[0:max_len])
    else:
        ephemeris_metadata_schema_str = self.ephemeris_metadata_schema
    print("  %-30s: %s" % ("ephemeris_metadata_schema", ephemeris_metadata_schema_str))
    if (self.data_product_metadata_schema is not None and len(str(self.data_product_metadata_schema)) > max_len):
        data_product_metadata_schema_str = "%s..." % (str(self.data_product_metadata_schema)[0:max_len])
    else:
        data_product_metadata_schema_str = self.data_product_metadata_schema
    print("  %-30s: %s" % ("data_product_metadata_schema", data_product_metadata_schema_str))
    if (self.stats is not None and len(str(self.stats)) > max_len):
        stats_str = "%s..." % (str(self.stats)[0:max_len])
    else:
        stats_str = self.data_product_metadata_schema
    print("  %-30s: %s" % ("stats", stats_str))
    print("  %-30s: %s" % ("format", self.format))

A special print output for this class.

class EphemerisData (data_source: DataSource,
epoch: datetime.datetime,
location_geo: Location | None = None,
nbtrace: Location | None = None,
sbtrace: Location | None = None,
location_gsm: Location | None = None,
metadata: Dict | None = None)

Expand source code

class EphemerisData:
    """
    Ephemeris object

    Attributes:
        data_source (DataSource): 
            Data source that the ephemeris record is associated with

        epoch (datetime.datetime): 
            Timestamp for the record (assumed it is in UTC)

        location_geo (Location): 
            Location object containing geographic latitude and longitude

        location_gsm (Location): 
            Location object containing GSM latitude and longitude (leave empty for data 
            sources with a type of 'ground')

        nbtrace (Location): 
            Location object with north B-trace geographic latitude and longitude

        sbtrace (Location): 
            Location object with south B-trace geographic latitude and longitude

        metadata (Dict): 
            Metadata for this record (arbitrary keys and values)
    """

    def __init__(self,
                 data_source: DataSource,
                 epoch: datetime.datetime,
                 location_geo: Optional[Location] = None,
                 nbtrace: Optional[Location] = None,
                 sbtrace: Optional[Location] = None,
                 location_gsm: Optional[Location] = None,
                 metadata: Optional[Dict] = None):
        self.data_source = data_source
        self.epoch = epoch
        self.location_geo = Location(lat=None, lon=None) if location_geo is None else location_geo
        self.nbtrace = Location(lat=None, lon=None) if nbtrace is None else nbtrace
        self.sbtrace = Location(lat=None, lon=None) if sbtrace is None else sbtrace
        self.location_gsm = Location(lat=None, lon=None) if location_gsm is None else location_gsm
        self.metadata = metadata

    def to_json_serializable(self) -> Dict:
        """
        Convert object to a JSON-serializable object (ie. translate
        datetime objects to strings)

        Returns:
            A dictionary object that is JSON-serializable
        """
        # init
        d = self.__dict__

        # format epoch as str
        if (isinstance(d["epoch"], datetime.datetime) is True):
            d["epoch"] = d["epoch"].strftime("%Y-%m-%dT%H:%M:00.000Z")

        # format location
        if (isinstance(d["location_geo"], Location) is True):
            d["location_geo"] = d["location_geo"].to_json_serializable()
        if (isinstance(d["location_gsm"], Location) is True):
            d["location_gsm"] = d["location_gsm"].to_json_serializable()
        if (isinstance(d["nbtrace"], Location) is True):
            d["nbtrace"] = d["nbtrace"].to_json_serializable()
        if (isinstance(d["sbtrace"], Location) is True):
            d["sbtrace"] = d["sbtrace"].to_json_serializable()

        # format metadata
        if (self.metadata is not None):
            for key, value in self.metadata.items():
                if (isinstance(value, datetime.datetime) is True or isinstance(value, datetime.date) is True):
                    self.metadata[key] = self.metadata[key].strftime("%Y-%m-%dT%H:%M:%S.%f")
        # if (isinstance(self.metadata, list) is True):
        #     self.metadata = {}

        # format data source fields for query
        d["program"] = self.data_source.program
        d["platform"] = self.data_source.platform
        d["instrument_type"] = self.data_source.instrument_type
        del d["data_source"]

        # return
        return d

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

    def __repr__(self) -> str:
        # shorten the metadata
        max_len = 20
        attr_metadata = f"{self.metadata}"
        if (len(attr_metadata) > max_len):
            attr_metadata = attr_metadata[0:max_len] + "...}"

        # return formatted representation
        return f"EphemerisData(epoch={repr(self.epoch)}, location_geo={repr(self.location_geo)}, " \
            f"location_gsm={repr(self.location_gsm)}, nbtrace={repr(self.nbtrace)}, sbtrace={repr(self.sbtrace)}, " \
            f"metadata={attr_metadata}, data_source=DataSource(...))"

Ephemeris object

Attributes

data_source : DataSource

Data source that the ephemeris record is associated with

epoch : datetime.datetime

Timestamp for the record (assumed it is in UTC)

location_geo : Location

Location object containing geographic latitude and longitude

location_gsm : Location

Location object containing GSM latitude and longitude (leave empty for data sources with a type of 'ground')

nbtrace : Location

Location object with north B-trace geographic latitude and longitude

sbtrace : Location

Location object with south B-trace geographic latitude and longitude

metadata : Dict

Metadata for this record (arbitrary keys and values)

Methods

def to_json_serializable(self) ‑> Dict

Expand source code

def to_json_serializable(self) -> Dict:
    """
    Convert object to a JSON-serializable object (ie. translate
    datetime objects to strings)

    Returns:
        A dictionary object that is JSON-serializable
    """
    # init
    d = self.__dict__

    # format epoch as str
    if (isinstance(d["epoch"], datetime.datetime) is True):
        d["epoch"] = d["epoch"].strftime("%Y-%m-%dT%H:%M:00.000Z")

    # format location
    if (isinstance(d["location_geo"], Location) is True):
        d["location_geo"] = d["location_geo"].to_json_serializable()
    if (isinstance(d["location_gsm"], Location) is True):
        d["location_gsm"] = d["location_gsm"].to_json_serializable()
    if (isinstance(d["nbtrace"], Location) is True):
        d["nbtrace"] = d["nbtrace"].to_json_serializable()
    if (isinstance(d["sbtrace"], Location) is True):
        d["sbtrace"] = d["sbtrace"].to_json_serializable()

    # format metadata
    if (self.metadata is not None):
        for key, value in self.metadata.items():
            if (isinstance(value, datetime.datetime) is True or isinstance(value, datetime.date) is True):
                self.metadata[key] = self.metadata[key].strftime("%Y-%m-%dT%H:%M:%S.%f")
    # if (isinstance(self.metadata, list) is True):
    #     self.metadata = {}

    # format data source fields for query
    d["program"] = self.data_source.program
    d["platform"] = self.data_source.platform
    d["instrument_type"] = self.data_source.instrument_type
    del d["data_source"]

    # return
    return d

Convert object to a JSON-serializable object (ie. translate datetime objects to strings)

Returns

A dictionary object that is JSON-serializable

class EphemerisSearch (aurorax_obj: PyAuroraX,
start: datetime.datetime,
end: datetime.datetime,
programs: Optional[List[str]] = None,
platforms: Optional[List[str]] = None,
instrument_types: Optional[List[str]] = None,
metadata_filters: Optional[Union[MetadataFilter, List[Dict]]] = None,
metadata_filters_logical_operator: "Optional[Literal['and', 'or', 'AND', 'OR']]" = None,
response_format: Optional[Dict] = None)

Expand source code

class EphemerisSearch:
    """
    Class representing an ephemeris search

    Note: At least one search criteria from programs, platforms, or instrument_types
    must be specified.

    Attributes:
        start (datetime.datetime): 
            Start timestamp of the search (inclusive)

        end (datetime.datetime): 
            End timestamp of the search (inclusive)

        programs (List[str]): 
            List of programs to search through, defaults to None

        platforms (List[str]): 
            List of platforms to search through, defaults to None

        instrument_types (List[str]): 
            List of instrument types to search through, defaults to None

        metadata_filters (MetadataFilter or List[Dict]): 
            List of dictionaries describing metadata keys and values to filter on, defaults 
            to None

        metadata_filters_logical_operator (str): 
            The logical operator to use when evaluating metadata filters (either `AND` or `OR`), 
            defaults to `AND`

        response_format (Dict): 
            JSON representation of desired data response format
        
        request (AuroraXResponse): 
            AuroraXResponse object returned when the search is executed
        
        request_id (str): 
            Unique ID assigned to the request by the AuroraX API
       
        request_url (str): 
            Unique URL assigned to the request by the AuroraX API
     
        executed (bool): 
            Indicates if the search has been executed/started
     
        completed (bool): 
            Indicates if the search has finished
       
        data_url (str): 
            The URL where data is accessed
      
        query (Dict): 
            The query for this request as JSON
      
        status (Dict): 
            The status of the query
      
        data (List[EphemerisData]): 
            The ephemeris records found
      
        logs (List[Dict]): 
            All log messages outputted by the AuroraX API for this request
    """

    __STANDARD_POLLING_SLEEP_TIME: float = 1.0

    def __init__(self,
                 aurorax_obj: PyAuroraX,
                 start: datetime.datetime,
                 end: datetime.datetime,
                 programs: Optional[List[str]] = None,
                 platforms: Optional[List[str]] = None,
                 instrument_types: Optional[List[str]] = None,
                 metadata_filters: Optional[Union[MetadataFilter, List[Dict]]] = None,
                 metadata_filters_logical_operator: Optional[Literal["and", "or", "AND", "OR"]] = None,
                 response_format: Optional[Dict] = None) -> None:

        # set variables using passed in args
        self.__aurorax_obj = aurorax_obj
        self.start = start
        self.end = end
        self.programs = programs
        self.platforms = platforms
        self.instrument_types = instrument_types
        self.metadata_filters = metadata_filters
        self.metadata_filters_logical_operator = "AND" if metadata_filters_logical_operator is None else metadata_filters_logical_operator.upper()
        self.response_format = response_format

        # initialize additional variables
        self.request = None
        self.request_id = ""
        self.request_url = ""
        self.executed = False
        self.completed = False
        self.data_url = ""
        self.__query = {}
        self.status = {}
        self.data = []
        self.logs = []

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

    def __repr__(self) -> str:
        return "EphemerisSearch(executed=%s, completed=%s, request_id='%s')" % (
            self.executed,
            self.completed,
            self.request_id,
        )

    def pretty_print(self):
        """
        A special print output for this class.
        """
        # set status and query strings
        max_len = 80
        status_str = str(self.status)
        query_str = str(self.query)
        if (len(status_str) > max_len):
            status_str = "%s..." % (status_str[0:max_len])
        if (len(query_str) > max_len):
            query_str = "%s..." % (query_str[0:max_len])

        # set results string
        if (self.executed is True):
            if (len(self.data) == 0):
                data_str = "[0 ephemeris results]"
            elif (len(self.data) == 1):
                data_str = "[1 ephemeris result]"
            else:
                data_str = "[%d ephemeris results]" % (len(self.data))
        else:
            data_str = ""

        # set logs string
        if (self.executed is True):
            if (len(self.logs) == 0):
                logs_str = "[0 log messages]"
            elif (len(self.logs) == 1):
                logs_str = "[1 log message]"
            else:
                logs_str = "[%d log messages]" % (len(self.logs))
        else:
            logs_str = ""

        # print
        print("EphemerisSearch:")
        print("  %-13s: %s" % ("executed", self.executed))
        print("  %-13s: %s" % ("completed", self.completed))
        print("  %-13s: %s" % ("request_id", self.request_id))
        print("  %-13s: %s" % ("request", self.request))
        print("  %-13s: %s" % ("request_url", self.request_url))
        print("  %-13s: %s" % ("data_url", self.data_url))
        print("  %-13s: %s" % ("query", query_str))
        print("  %-13s: %s" % ("status", status_str))
        print("  %-13s: %s" % ("data", data_str))
        print("  %-13s: %s" % ("logs", logs_str))

    @property
    def query(self):
        """
        Property for the query value
        """
        # set metadata filter value
        if (self.metadata_filters is None):
            metadata_filters_dict = {}
        elif (isinstance(self.metadata_filters, MetadataFilter) is True):
            # metadata filter is a class
            metadata_filters_dict = self.metadata_filters.to_query_dict()  # type: ignore
        else:
            # metadata filter is a dictionary
            metadata_filters_dict = {
                "expressions": self.metadata_filters,
                "logical_operator": self.metadata_filters_logical_operator,
            }

        # set query
        self.__query = {
            "data_sources": {
                "programs": [] if not self.programs else self.programs,
                "platforms": [] if not self.platforms else self.platforms,
                "instrument_types": [] if not self.instrument_types else self.instrument_types,
                "ephemeris_metadata_filters": metadata_filters_dict,
            },
            "start": self.start.strftime("%Y-%m-%dT%H:%M:%S"),
            "end": self.end.strftime("%Y-%m-%dT%H:%M:%S"),
        }

        # return
        return self.__query

    @query.setter
    def query(self, query):
        self.__query = query

    def execute(self) -> None:
        """
        Initiate ephemeris search request

        Raises:
            pyaurorax.exceptions.AuroraXError: Invalid request parameters are set
        """
        # check for at least one filter criteria
        if not (self.programs or self.platforms or self.instrument_types or self.metadata_filters):
            raise AuroraXError("At least one filter criteria parameter besides 'start' and 'end' must be specified")

        # do request
        url = "%s/%s" % (self.__aurorax_obj.api_base_url, self.__aurorax_obj.search.api.URL_SUFFIX_EPHEMERIS_SEARCH)
        req = AuroraXAPIRequest(self.__aurorax_obj, method="post", url=url, body=self.query, null_response=True)
        res = req.execute()

        # set request ID, request_url, executed
        self.executed = True
        if (res.status_code == 202):
            # request successfully dispatched
            self.executed = True
            self.request_url = res.request.headers["location"]
            self.request_id = self.request_url.rsplit("/", 1)[-1]

        # set the request variable
        self.request = res

    def update_status(self, status: Optional[Dict] = None) -> None:
        """
        Update the status of this ephemeris search request

        Args:
            status (Dict): 
                The previously-retrieved status of this request (include to avoid requesting it 
                from the API again), defaults to None
        """
        # get the status if it isn't passed in
        if (status is None):
            status = requests_get_status(self.__aurorax_obj, self.request_url)

        # check response
        if (status is None):
            raise AuroraXAPIError("Could not retrieve status for this request")

        # update request status by checking if data URI is set
        if (status["search_result"]["data_uri"] is not None):
            self.completed = True
            self.data_url = "%s/data" % (self.request_url)

        # set class variable "status" and "logs"
        self.status = status
        self.logs = status["logs"]

    def check_for_data(self) -> bool:
        """
        Check to see if data is available for this ephemeris search request

        Returns:
            True if data is available, else False
        """
        self.update_status()
        return self.completed

    def get_data(self) -> None:
        """
        Retrieve the data available for this ephemeris search request
        """
        # check if completed yet
        if (self.completed is False):
            print("No data available, update status or check for data first")
            return

        # get data
        raw_data = requests_get_data(self.__aurorax_obj, self.data_url, self.response_format, False)

        # set data variable
        if (self.response_format is not None):
            self.data = raw_data
        else:
            # cast data source objects
            for i in range(0, len(raw_data)):
                ds = DataSource(**raw_data[i]["data_source"], format=FORMAT_BASIC_INFO)
                raw_data[i]["data_source"] = ds

            # cast ephemeris objects
            self.data = [EphemerisData(**e) for e in raw_data]

    def wait(self, poll_interval: float = __STANDARD_POLLING_SLEEP_TIME, verbose: bool = False) -> None:
        """
        Block and wait for the request to complete and data is available for retrieval

        Args:
            poll_interval (float): 
                Time in seconds to wait between polling attempts, defaults to 1 second

            verbose (bool): 
                Output poll times and other progress messages, defaults to False
        """
        url = "%s/%s" % (self.__aurorax_obj.api_base_url, self.__aurorax_obj.search.api.URL_SUFFIX_EPHEMERIS_REQUEST.format(self.request_id))
        self.update_status(requests_wait_for_data(self.__aurorax_obj, url, poll_interval, verbose))

    def cancel(self, wait: bool = False, poll_interval: float = __STANDARD_POLLING_SLEEP_TIME, verbose: bool = False) -> int:
        """
        Cancel the ephemeris search request

        This method returns immediately by default since the API processes
        this request asynchronously. If you would prefer to wait for it
        to be completed, set the 'wait' parameter to True. You can adjust
        the polling time using the 'poll_interval' parameter.

        Args:
            wait (bool): 
                Wait until the cancellation request has been completed (may wait for 
                several minutes)

            poll_interval (float): 
                Seconds to wait between polling calls, defaults to 1 second

            verbose (bool): 
                Output poll times and other progress messages, defaults to False

        Returns:
            1 on success

        Raises:
            pyaurorax.exceptions.AuroraXUnauthorizedError: Invalid API key for this operation
            pyaurorax.exceptions.AuroraXAPIError: An API error was encountered
        """
        url = "%s/%s" % (self.__aurorax_obj.api_base_url, self.__aurorax_obj.search.api.URL_SUFFIX_EPHEMERIS_REQUEST.format(self.request_id))
        return requests_cancel(self.__aurorax_obj, url, wait, poll_interval, verbose)

    def describe(self):
        """
        Describe the ephemeris search as an "SQL-like" string.

        Returns:
            The "SQL-like" string describing the ephemeris search object
        """
        # make request
        url = "%s/%s" % (self.__aurorax_obj.api_base_url, self.__aurorax_obj.search.api.URL_SUFFIX_DESCRIBE_EPHEMERIS_QUERY)
        req = AuroraXAPIRequest(self.__aurorax_obj, method="post", url=url, body=self.query)
        res = req.execute()

        # return
        return res.data

Class representing an ephemeris search

Note: At least one search criteria from programs, platforms, or instrument_types must be specified.

Attributes

start : datetime.datetime

Start timestamp of the search (inclusive)

end : datetime.datetime

End timestamp of the search (inclusive)

programs : List[str]

List of programs to search through, defaults to None

platforms : List[str]

List of platforms to search through, defaults to None

instrument_types : List[str]

List of instrument types to search through, defaults to None

metadata_filters : MetadataFilter or List[Dict]

List of dictionaries describing metadata keys and values to filter on, defaults to None

metadata_filters_logical_operator : str

The logical operator to use when evaluating metadata filters (either AND or OR), defaults to AND

response_format : Dict

JSON representation of desired data response format

request : AuroraXResponse

AuroraXResponse object returned when the search is executed

request_id : str

Unique ID assigned to the request by the AuroraX API

request_url : str

Unique URL assigned to the request by the AuroraX API

executed (bool): Indicates if the search has been executed/started

completed (bool): Indicates if the search has finished

data_url (str): The URL where data is accessed

query (Dict): The query for this request as JSON

status (Dict): The status of the query

data (List[EphemerisData]): The ephemeris records found

logs (List[Dict]): All log messages outputted by the AuroraX API for this request

Instance variables

prop query

Expand source code

@property
def query(self):
    """
    Property for the query value
    """
    # set metadata filter value
    if (self.metadata_filters is None):
        metadata_filters_dict = {}
    elif (isinstance(self.metadata_filters, MetadataFilter) is True):
        # metadata filter is a class
        metadata_filters_dict = self.metadata_filters.to_query_dict()  # type: ignore
    else:
        # metadata filter is a dictionary
        metadata_filters_dict = {
            "expressions": self.metadata_filters,
            "logical_operator": self.metadata_filters_logical_operator,
        }

    # set query
    self.__query = {
        "data_sources": {
            "programs": [] if not self.programs else self.programs,
            "platforms": [] if not self.platforms else self.platforms,
            "instrument_types": [] if not self.instrument_types else self.instrument_types,
            "ephemeris_metadata_filters": metadata_filters_dict,
        },
        "start": self.start.strftime("%Y-%m-%dT%H:%M:%S"),
        "end": self.end.strftime("%Y-%m-%dT%H:%M:%S"),
    }

    # return
    return self.__query

Property for the query value

Methods

def cancel(self, wait: bool = False, poll_interval: float = 1.0, verbose: bool = False) ‑> int

Expand source code

def cancel(self, wait: bool = False, poll_interval: float = __STANDARD_POLLING_SLEEP_TIME, verbose: bool = False) -> int:
    """
    Cancel the ephemeris search request

    This method returns immediately by default since the API processes
    this request asynchronously. If you would prefer to wait for it
    to be completed, set the 'wait' parameter to True. You can adjust
    the polling time using the 'poll_interval' parameter.

    Args:
        wait (bool): 
            Wait until the cancellation request has been completed (may wait for 
            several minutes)

        poll_interval (float): 
            Seconds to wait between polling calls, defaults to 1 second

        verbose (bool): 
            Output poll times and other progress messages, defaults to False

    Returns:
        1 on success

    Raises:
        pyaurorax.exceptions.AuroraXUnauthorizedError: Invalid API key for this operation
        pyaurorax.exceptions.AuroraXAPIError: An API error was encountered
    """
    url = "%s/%s" % (self.__aurorax_obj.api_base_url, self.__aurorax_obj.search.api.URL_SUFFIX_EPHEMERIS_REQUEST.format(self.request_id))
    return requests_cancel(self.__aurorax_obj, url, wait, poll_interval, verbose)

Cancel the ephemeris search request

This method returns immediately by default since the API processes this request asynchronously. If you would prefer to wait for it to be completed, set the 'wait' parameter to True. You can adjust the polling time using the 'poll_interval' parameter.

Args

wait : bool

Wait until the cancellation request has been completed (may wait for several minutes)

poll_interval : float

Seconds to wait between polling calls, defaults to 1 second

verbose : bool

Output poll times and other progress messages, defaults to False

Returns

1 on success

Raises

AuroraXUnauthorizedError

Invalid API key for this operation

AuroraXAPIError

An API error was encountered

def check_for_data(self) ‑> bool

Expand source code

def check_for_data(self) -> bool:
    """
    Check to see if data is available for this ephemeris search request

    Returns:
        True if data is available, else False
    """
    self.update_status()
    return self.completed

Check to see if data is available for this ephemeris search request

Returns

True if data is available, else False

def describe(self)

Expand source code

def describe(self):
    """
    Describe the ephemeris search as an "SQL-like" string.

    Returns:
        The "SQL-like" string describing the ephemeris search object
    """
    # make request
    url = "%s/%s" % (self.__aurorax_obj.api_base_url, self.__aurorax_obj.search.api.URL_SUFFIX_DESCRIBE_EPHEMERIS_QUERY)
    req = AuroraXAPIRequest(self.__aurorax_obj, method="post", url=url, body=self.query)
    res = req.execute()

    # return
    return res.data

Describe the ephemeris search as an "SQL-like" string.

Returns

The "SQL-like" string describing the ephemeris search object

def execute(self) ‑> None

Expand source code

def execute(self) -> None:
    """
    Initiate ephemeris search request

    Raises:
        pyaurorax.exceptions.AuroraXError: Invalid request parameters are set
    """
    # check for at least one filter criteria
    if not (self.programs or self.platforms or self.instrument_types or self.metadata_filters):
        raise AuroraXError("At least one filter criteria parameter besides 'start' and 'end' must be specified")

    # do request
    url = "%s/%s" % (self.__aurorax_obj.api_base_url, self.__aurorax_obj.search.api.URL_SUFFIX_EPHEMERIS_SEARCH)
    req = AuroraXAPIRequest(self.__aurorax_obj, method="post", url=url, body=self.query, null_response=True)
    res = req.execute()

    # set request ID, request_url, executed
    self.executed = True
    if (res.status_code == 202):
        # request successfully dispatched
        self.executed = True
        self.request_url = res.request.headers["location"]
        self.request_id = self.request_url.rsplit("/", 1)[-1]

    # set the request variable
    self.request = res

Initiate ephemeris search request

Raises

AuroraXError

Invalid request parameters are set

def get_data(self) ‑> None

Expand source code

def get_data(self) -> None:
    """
    Retrieve the data available for this ephemeris search request
    """
    # check if completed yet
    if (self.completed is False):
        print("No data available, update status or check for data first")
        return

    # get data
    raw_data = requests_get_data(self.__aurorax_obj, self.data_url, self.response_format, False)

    # set data variable
    if (self.response_format is not None):
        self.data = raw_data
    else:
        # cast data source objects
        for i in range(0, len(raw_data)):
            ds = DataSource(**raw_data[i]["data_source"], format=FORMAT_BASIC_INFO)
            raw_data[i]["data_source"] = ds

        # cast ephemeris objects
        self.data = [EphemerisData(**e) for e in raw_data]

Retrieve the data available for this ephemeris search request

def pretty_print(self)

Expand source code

def pretty_print(self):
    """
    A special print output for this class.
    """
    # set status and query strings
    max_len = 80
    status_str = str(self.status)
    query_str = str(self.query)
    if (len(status_str) > max_len):
        status_str = "%s..." % (status_str[0:max_len])
    if (len(query_str) > max_len):
        query_str = "%s..." % (query_str[0:max_len])

    # set results string
    if (self.executed is True):
        if (len(self.data) == 0):
            data_str = "[0 ephemeris results]"
        elif (len(self.data) == 1):
            data_str = "[1 ephemeris result]"
        else:
            data_str = "[%d ephemeris results]" % (len(self.data))
    else:
        data_str = ""

    # set logs string
    if (self.executed is True):
        if (len(self.logs) == 0):
            logs_str = "[0 log messages]"
        elif (len(self.logs) == 1):
            logs_str = "[1 log message]"
        else:
            logs_str = "[%d log messages]" % (len(self.logs))
    else:
        logs_str = ""

    # print
    print("EphemerisSearch:")
    print("  %-13s: %s" % ("executed", self.executed))
    print("  %-13s: %s" % ("completed", self.completed))
    print("  %-13s: %s" % ("request_id", self.request_id))
    print("  %-13s: %s" % ("request", self.request))
    print("  %-13s: %s" % ("request_url", self.request_url))
    print("  %-13s: %s" % ("data_url", self.data_url))
    print("  %-13s: %s" % ("query", query_str))
    print("  %-13s: %s" % ("status", status_str))
    print("  %-13s: %s" % ("data", data_str))
    print("  %-13s: %s" % ("logs", logs_str))

A special print output for this class.

def update_status(self, status: Optional[Dict] = None) ‑> None

Expand source code

def update_status(self, status: Optional[Dict] = None) -> None:
    """
    Update the status of this ephemeris search request

    Args:
        status (Dict): 
            The previously-retrieved status of this request (include to avoid requesting it 
            from the API again), defaults to None
    """
    # get the status if it isn't passed in
    if (status is None):
        status = requests_get_status(self.__aurorax_obj, self.request_url)

    # check response
    if (status is None):
        raise AuroraXAPIError("Could not retrieve status for this request")

    # update request status by checking if data URI is set
    if (status["search_result"]["data_uri"] is not None):
        self.completed = True
        self.data_url = "%s/data" % (self.request_url)

    # set class variable "status" and "logs"
    self.status = status
    self.logs = status["logs"]

Update the status of this ephemeris search request

Args

status : Dict

The previously-retrieved status of this request (include to avoid requesting it from the API again), defaults to None

def wait(self, poll_interval: float = 1.0, verbose: bool = False) ‑> None

Expand source code

def wait(self, poll_interval: float = __STANDARD_POLLING_SLEEP_TIME, verbose: bool = False) -> None:
    """
    Block and wait for the request to complete and data is available for retrieval

    Args:
        poll_interval (float): 
            Time in seconds to wait between polling attempts, defaults to 1 second

        verbose (bool): 
            Output poll times and other progress messages, defaults to False
    """
    url = "%s/%s" % (self.__aurorax_obj.api_base_url, self.__aurorax_obj.search.api.URL_SUFFIX_EPHEMERIS_REQUEST.format(self.request_id))
    self.update_status(requests_wait_for_data(self.__aurorax_obj, url, poll_interval, verbose))

Block and wait for the request to complete and data is available for retrieval

Args

poll_interval : float

Time in seconds to wait between polling attempts, defaults to 1 second

verbose : bool

Output poll times and other progress messages, defaults to False

class EventsCriteriaBlock (platforms: List[str] = [],
instrument_types: List[str] = [],
metadata_filters: MetadataFilter | None = None)

Expand source code

class EventsCriteriaBlock:
    """
    Representation of a event criteria block used for conjunction searches. 

    Attributes:
        platforms (List[str]): 
            List of platform strings to use in this criteria block. Optional, default is `[]`.

        instrument_types (List[str]): 
            List of instrument type strings to use in this criteria block. Optional, default is `[]`.

        metadata_filters (MetadataFilter): 
            The metadata filters to use in this criteria block. Optional, default is None.
    """

    def __init__(self, platforms: List[str] = [], instrument_types: List[str] = [], metadata_filters: Optional[MetadataFilter] = None):
        self.platforms = platforms
        self.instrument_types = instrument_types
        self.metadata_filters = metadata_filters

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

    def __repr__(self) -> str:
        return "EventsCriteriaBlock(platforms=%s, instrument_types=%s, metadata_filters=%s)" % (
            self.platforms,
            self.instrument_types,
            self.metadata_filters,
        )

    def pretty_print(self):
        """
        A special print output for this class.
        """
        # set special strings
        max_len = 80
        metadata_filters_str = str(self.metadata_filters)
        if (len(metadata_filters_str) > max_len):
            metadata_filters_str = "%s...)" % (metadata_filters_str[0:max_len])

        # print
        print("EventsCriteriaBlock:")
        print("  %-18s: %s" % ("platforms", self.platforms))
        print("  %-18s: %s" % ("instrument_types", self.instrument_types))
        print("  %-18s: %s" % ("metadata_filters", metadata_filters_str))

Representation of a event criteria block used for conjunction searches.

Attributes

platforms : List[str]

List of platform strings to use in this criteria block. Optional, default is [].

instrument_types : List[str]

List of instrument type strings to use in this criteria block. Optional, default is [].

metadata_filters : MetadataFilter

The metadata filters to use in this criteria block. Optional, default is None.

Methods

def pretty_print(self)

Expand source code

def pretty_print(self):
    """
    A special print output for this class.
    """
    # set special strings
    max_len = 80
    metadata_filters_str = str(self.metadata_filters)
    if (len(metadata_filters_str) > max_len):
        metadata_filters_str = "%s...)" % (metadata_filters_str[0:max_len])

    # print
    print("EventsCriteriaBlock:")
    print("  %-18s: %s" % ("platforms", self.platforms))
    print("  %-18s: %s" % ("instrument_types", self.instrument_types))
    print("  %-18s: %s" % ("metadata_filters", metadata_filters_str))

A special print output for this class.

class GroundCriteriaBlock (programs: List[str] = [],
platforms: List[str] = [],
instrument_types: List[str] = [],
metadata_filters: MetadataFilter | None = None)

Expand source code

class GroundCriteriaBlock:
    """
    Representation of a ground criteria block used for conjunction searches. 

    Attributes:
        programs (List[str]): 
            List of program strings to use in this criteria block. Optional, default is `[]`.
        
        platforms (List[str]): 
            List of platform strings to use in this criteria block. Optional, default is `[]`.

        instrument_types (List[str]): 
            List of instrument type strings to use in this criteria block. Optional, default is `[]`.

        metadata_filters (MetadataFilter): 
            The metadata filters to use in this criteria block. Optional, default is None.
    """

    def __init__(self,
                 programs: List[str] = [],
                 platforms: List[str] = [],
                 instrument_types: List[str] = [],
                 metadata_filters: Optional[MetadataFilter] = None):
        self.programs = programs
        self.platforms = platforms
        self.instrument_types = instrument_types
        self.metadata_filters = metadata_filters

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

    def __repr__(self) -> str:
        return "GroundCriteriaBlock(programs=%s, platforms=%s, instrument_types=%s, metadata_filters=%s)" % (
            self.programs,
            self.platforms,
            self.instrument_types,
            self.metadata_filters,
        )

    def pretty_print(self):
        """
        A special print output for this class.
        """
        # set special strings
        max_len = 80
        metadata_filters_str = str(self.metadata_filters)
        if (len(metadata_filters_str) > max_len):
            metadata_filters_str = "%s...)" % (metadata_filters_str[0:max_len])

        # print
        print("GroundCriteriaBlock:")
        print("  %-18s: %s" % ("programs", self.programs))
        print("  %-18s: %s" % ("platforms", self.platforms))
        print("  %-18s: %s" % ("instrument_types", self.instrument_types))
        print("  %-18s: %s" % ("metadata_filters", metadata_filters_str))

Representation of a ground criteria block used for conjunction searches.

Attributes

programs : List[str]

List of program strings to use in this criteria block. Optional, default is [].

platforms : List[str]

List of platform strings to use in this criteria block. Optional, default is [].

instrument_types : List[str]

List of instrument type strings to use in this criteria block. Optional, default is [].

metadata_filters : MetadataFilter

The metadata filters to use in this criteria block. Optional, default is None.

Methods

def pretty_print(self)

Expand source code

def pretty_print(self):
    """
    A special print output for this class.
    """
    # set special strings
    max_len = 80
    metadata_filters_str = str(self.metadata_filters)
    if (len(metadata_filters_str) > max_len):
        metadata_filters_str = "%s...)" % (metadata_filters_str[0:max_len])

    # print
    print("GroundCriteriaBlock:")
    print("  %-18s: %s" % ("programs", self.programs))
    print("  %-18s: %s" % ("platforms", self.platforms))
    print("  %-18s: %s" % ("instrument_types", self.instrument_types))
    print("  %-18s: %s" % ("metadata_filters", metadata_filters_str))

A special print output for this class.

class Location (lat: float | None = None, lon: float | None = None)

Expand source code

class Location:
    """
    Representation for an AuroraX search engine location. This data can be in geodetic 
    coordinates, GSM coordinates, or geodetic northern/southern B-trace magnetic footprints.

    Latitude and longitude values are in decimal degrees format, ranging from -90 to 90
    for latitude and -180 to 180 for longitude.

    Note that latitude and longitude must both be numbers, or both be None.

    Attributes:
        lat (float): latitude value
        lon (float): longitude value
    
    Raises:
        ValueError: if both latitude and longitude are not real numbers, or not both None.
    """

    def __init__(self, lat: Optional[float] = None, lon: Optional[float] = None):
        if (lat is None and lon is not None) or (lat is not None and lon is None):
            # one of them is None, not allowed
            raise ValueError("Latitude and longitude must both be numbers, or both be None")
        self.__lat = lat
        self.__lon = lon

    @property
    def lat(self):
        return self.__lat

    @lat.setter
    def lat(self, value: float):
        if (self.__lon is None and value is not None) or (self.__lon is not None and value is None):
            # one of them is None, not allowed
            raise ValueError("Latitude and longitude must both be numbers, or both be None")
        self.__lat = value

    @property
    def lon(self):
        return self.__lon

    @lon.setter
    def lon(self, value: float):
        if (self.__lat is None and value is not None) or (self.__lat is not None and value is None):
            # one of them is None, not allowed
            raise ValueError("Latitude and longitude must both be numbers, or both be None")
        self.__lon = value

    def to_json_serializable(self) -> Dict:
        """
        Convert object to a JSON-serializable object (ie. translate
        datetime objects to strings)

        Returns:
            a dictionary object that is JSON-serializable
        """
        return {"lat": self.lat, "lon": self.lon}

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

    def __repr__(self) -> str:
        return "%s(lat=%s, lon=%s)" % (self.__class__.__name__, str(self.lat), str(self.lon))

Representation for an AuroraX search engine location. This data can be in geodetic coordinates, GSM coordinates, or geodetic northern/southern B-trace magnetic footprints.

Latitude and longitude values are in decimal degrees format, ranging from -90 to 90 for latitude and -180 to 180 for longitude.

Note that latitude and longitude must both be numbers, or both be None.

Attributes

lat : float

latitude value

lon : float

longitude value

Raises

ValueError

if both latitude and longitude are not real numbers, or not both None.

Instance variables

prop lat

Expand source code

@property
def lat(self):
    return self.__lat

prop lon

Expand source code

@property
def lon(self):
    return self.__lon

Methods

def to_json_serializable(self) ‑> Dict

Expand source code

def to_json_serializable(self) -> Dict:
    """
    Convert object to a JSON-serializable object (ie. translate
    datetime objects to strings)

    Returns:
        a dictionary object that is JSON-serializable
    """
    return {"lat": self.lat, "lon": self.lon}

Convert object to a JSON-serializable object (ie. translate datetime objects to strings)

Returns

a dictionary object that is JSON-serializable

class MetadataFilter (expressions: List[MetadataFilterExpression],
operator: Literal['and', 'or', 'AND', 'OR'] = 'and')

Expand source code

class MetadataFilter:
    """
    Representation for an AuroraX search engine metadata filter. These are used 
    as part of conjunction, ephemeris, and data product searching.

    Attributes:
        expressions (List[MetadataFilterExpression]): 
            The list of metadata filter expressions for use with conjunction, ephemeris, and 
            data product searches.
        
        operator (str): 
            The logical operator to use when the search engine will evaluate multiple expressions. If 
            not supplied, the search engine will perform a logical 'AND' between each expression. Possible
            choices are 'and' or 'or'.

    Raises:
        pyaurorax.exceptions.AuroraXError: if invalid operator was specified.
    """

    def __init__(
        self,
        expressions: List[MetadataFilterExpression],
        operator: Literal["and", "or", "AND", "OR"] = "and",
    ):
        self.expressions = expressions
        self.__operator = operator

    @property
    def operator(self):
        return self.__operator

    @operator.setter
    def operator(self, val: Literal["and", "or", "AND", "OR"] = "and"):
        if (val.lower() not in ["and", "or"]):
            AuroraXError("Operator '%s' not allowed. You must use one of the following: ['and', 'or']" % (val))
        self.__operator = val.lower()

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

    def __repr__(self) -> str:
        # set special strings
        if (len(self.expressions) == 0):
            expressions_str = "[0 expressions]"
        elif (len(self.expressions) == 1):
            expressions_str = "[1 expression]"
        else:
            expressions_str = "[%d expressions]" % (len(self.expressions))

        # return
        return "MetadataFilter(expressions=%s, operator='%s')" % (expressions_str, self.operator)

    def pretty_print(self):
        """
        A special print output for this class.
        """
        # set special strings
        if (len(self.expressions) == 0):
            expressions_str = "[0 expressions]"
        elif (len(self.expressions) == 1):
            expressions_str = "[1 expression]"
        else:
            expressions_str = "[%d expressions]" % (len(self.expressions))

        # print
        print("MetadataFilter:")
        print("  %-13s: %s" % ("expressions", expressions_str))
        print("  %-13s: %s" % ("operator", self.operator))

    def to_query_dict(self):
        """
        Convert the expression object to a dictionary that will be used when executing a search.
        """
        return {
            "expressions": [x.to_query_dict() for x in self.expressions],
            "logical_operator": self.operator.upper(),
        }

Representation for an AuroraX search engine metadata filter. These are used as part of conjunction, ephemeris, and data product searching.

Attributes

expressions : List[MetadataFilterExpression]

The list of metadata filter expressions for use with conjunction, ephemeris, and data product searches.

operator : str

The logical operator to use when the search engine will evaluate multiple expressions. If not supplied, the search engine will perform a logical 'AND' between each expression. Possible choices are 'and' or 'or'.

Raises

AuroraXError

if invalid operator was specified.

Instance variables

prop operator

Expand source code

@property
def operator(self):
    return self.__operator

Methods

def pretty_print(self)

Expand source code

def pretty_print(self):
    """
    A special print output for this class.
    """
    # set special strings
    if (len(self.expressions) == 0):
        expressions_str = "[0 expressions]"
    elif (len(self.expressions) == 1):
        expressions_str = "[1 expression]"
    else:
        expressions_str = "[%d expressions]" % (len(self.expressions))

    # print
    print("MetadataFilter:")
    print("  %-13s: %s" % ("expressions", expressions_str))
    print("  %-13s: %s" % ("operator", self.operator))

A special print output for this class.

def to_query_dict(self)

Expand source code

def to_query_dict(self):
    """
    Convert the expression object to a dictionary that will be used when executing a search.
    """
    return {
        "expressions": [x.to_query_dict() for x in self.expressions],
        "logical_operator": self.operator.upper(),
    }

Convert the expression object to a dictionary that will be used when executing a search.

class MetadataFilterExpression (key: str,
values: Any | List[Any],
operator: Literal['=', '!=', '>', '<', '>=', '<=', 'between', 'in', 'not in'] = 'in')

Expand source code

class MetadataFilterExpression:
    """
    Representation for an AuroraX search engine metadata filter expression. These are used 
    as part of conjunction, ephemeris, and data product searching.

    Attributes:
        key (str): 
            The special key for the metadata filter. For example, 'nbtrace_region'.
        
        values (Any or List[Any]): 
            The value(s) that the search will use when filtering. This can either be a single value, 
            or a list of values.

        operator (str): 
            The operator to use when the search engine evaluates the expression. Valid choices
            are: "=", "!=", ">", "<", ">=", "<=", "between", "in", "not in".

            The "in" and "not in" operators are meant exclusively for expressions where there 
            are multiple values (ie. the values parameter is a list of strings).
        
    Raises:
        pyaurorax.exceptions.AuroraXError: if invalid operator was specified.
    """

    def __init__(
        self,
        key: str,
        values: Union[Any, List[Any]],
        operator: Literal["=", "!=", ">", "<", ">=", "<=", "between", "in", "not in"] = "in",
    ):
        # set required parameters
        self.key = key
        self.values = values
        self.__operator = operator

    @property
    def operator(self) -> str:
        return self.__operator

    @operator.setter
    def operator(self, val: Literal["=", "!=", ">", "<", ">=", "<=", "between", "in", "not in"]):
        if (val not in ["=", "!=", ">", "<", ">=", "<=", "between", "in", "not in"]):
            AuroraXError("Operator '%s' not allowed. You must use one of the following: ['=', '!=', '>', '<', " +
                         "'>=', '<=', 'between', 'in', 'not in']" % (val))
        self.__operator = val

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

    def __repr__(self) -> str:
        # set special strings
        values_str = "'%s'" % (self.values)
        if (type(self.values) is list):
            values_str = str(self.values)

        # return
        return "MetadataFilterExpression(key='%s', values=%s, operator='%s')" % (self.key, values_str, self.operator)

    def pretty_print(self):
        """
        A special print output for this class.
        """

        # print
        print("MetadataFilterExpression:")
        print("  %-10s: %s" % ("key", self.key))
        print("  %-10s: %s" % ("values", self.values))
        print("  %-10s: %s" % ("operator", self.operator))

    def to_query_dict(self):
        """
        Convert the expression object to a dictionary that will be used when executing a search.
        """
        return {
            "key": str(self.key),
            "values": [str(self.values)] if type(self.values) is not list else self.values,
            "operator": str(self.operator),
        }

Representation for an AuroraX search engine metadata filter expression. These are used as part of conjunction, ephemeris, and data product searching.

Attributes

key : str

The special key for the metadata filter. For example, 'nbtrace_region'.

values : Any or List[Any]

The value(s) that the search will use when filtering. This can either be a single value, or a list of values.

operator : str

The operator to use when the search engine evaluates the expression. Valid choices are: "=", "!=", ">", "<", ">=", "<=", "between", "in", "not in".

The "in" and "not in" operators are meant exclusively for expressions where there are multiple values (ie. the values parameter is a list of strings).

Raises

AuroraXError

if invalid operator was specified.

Instance variables

prop operator : str

Expand source code

@property
def operator(self) -> str:
    return self.__operator

Methods

def pretty_print(self)

Expand source code

def pretty_print(self):
    """
    A special print output for this class.
    """

    # print
    print("MetadataFilterExpression:")
    print("  %-10s: %s" % ("key", self.key))
    print("  %-10s: %s" % ("values", self.values))
    print("  %-10s: %s" % ("operator", self.operator))

A special print output for this class.

def to_query_dict(self)

Expand source code

def to_query_dict(self):
    """
    Convert the expression object to a dictionary that will be used when executing a search.
    """
    return {
        "key": str(self.key),
        "values": [str(self.values)] if type(self.values) is not list else self.values,
        "operator": str(self.operator),
    }

Convert the expression object to a dictionary that will be used when executing a search.

class SearchManager (aurorax_obj)

Expand source code

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

    def __init__(self, aurorax_obj):
        self.__aurorax_obj = aurorax_obj

        # initialize sub-modules
        self.__util = UtilManager()
        self.__api = module_api
        self.__sources = SourcesManager(self.__aurorax_obj)
        self.__availability = AvailabilityManager(self.__aurorax_obj)
        self.__metadata = MetadataManager(self.__aurorax_obj)
        self.__requests = RequestsManager(self.__aurorax_obj)
        self.__ephemeris = EphemerisManager(self.__aurorax_obj)
        self.__data_products = DataProductsManager(self.__aurorax_obj)
        self.__conjunctions = ConjunctionsManager(self.__aurorax_obj)

        # initialize class vars
        self.DataSource = DataSource
        self.Location = Location
        self.MetadataFilter = MetadataFilter
        self.MetadataFilterExpression = MetadataFilterExpression
        self.GroundCriteriaBlock = GroundCriteriaBlock
        self.SpaceCriteriaBlock = SpaceCriteriaBlock
        self.EventsCriteriaBlock = EventsCriteriaBlock
        self.CustomLocationsCriteriaBlock = CustomLocationsCriteriaBlock
        self.ConjunctionSearch = ConjunctionSearch
        self.EphemerisSearch = EphemerisSearch
        self.DataProductSearch = DataProductSearch
        self.EphemerisData = EphemerisData
        self.DataProductData = DataProductData

        # initialize static vars
        self.FORMAT_BASIC_INFO = FORMAT_BASIC_INFO
        self.FORMAT_BASIC_INFO_WITH_METADATA = FORMAT_BASIC_INFO_WITH_METADATA
        self.FORMAT_FULL_RECORD = FORMAT_FULL_RECORD
        self.FORMAT_IDENTIFIER_ONLY = FORMAT_IDENTIFIER_ONLY
        self.FORMAT_DEFAULT = FORMAT_DEFAULT
        self.SOURCE_TYPE_EVENT_LIST = SOURCE_TYPE_EVENT_LIST
        self.SOURCE_TYPE_GROUND = SOURCE_TYPE_GROUND
        self.SOURCE_TYPE_HEO = SOURCE_TYPE_HEO
        self.SOURCE_TYPE_LEO = SOURCE_TYPE_LEO
        self.SOURCE_TYPE_LUNAR = SOURCE_TYPE_LUNAR
        self.DATA_PRODUCT_TYPE_KEOGRAM = DATA_PRODUCT_TYPE_KEOGRAM
        self.DATA_PRODUCT_TYPE_MONTAGE = DATA_PRODUCT_TYPE_MONTAGE
        self.DATA_PRODUCT_TYPE_MOVIE = DATA_PRODUCT_TYPE_MOVIE
        self.DATA_PRODUCT_TYPE_SUMMARY_PLOT = DATA_PRODUCT_TYPE_SUMMARY_PLOT
        self.DATA_PRODUCT_TYPE_DATA_AVAILABILITY = DATA_PRODUCT_TYPE_DATA_AVAILABILITY
        self.CONJUNCTION_TYPE_NBTRACE = CONJUNCTION_TYPE_NBTRACE
        self.CONJUNCTION_TYPE_SBTRACE = CONJUNCTION_TYPE_SBTRACE

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

    @property
    def api(self):
        """
        Access to the `api` submodule from within a PyAuroraX object.
        """
        return self.__api

    @property
    def sources(self):
        """
        Access to the `sources` submodule from within a PyAuroraX object.
        """
        return self.__sources

    @property
    def availability(self):
        """
        Access to the `availability` submodule from within a PyAuroraX object.
        """
        return self.__availability

    @property
    def metadata(self):
        """
        Access to the `metadata` submodule from within a PyAuroraX object.
        """
        return self.__metadata

    @property
    def requests(self):
        """
        Access to the `requests` submodule from within a PyAuroraX object.
        """
        return self.__requests

    @property
    def ephemeris(self):
        """
        Access to the `ephemeris` submodule from within a PyAuroraX object.
        """
        return self.__ephemeris

    @property
    def data_products(self):
        """
        Access to the `data_products` submodule from within a PyAuroraX object.
        """
        return self.__data_products

    @property
    def conjunctions(self):
        """
        Access to the `conjunctions` submodule from within a PyAuroraX object.
        """
        return self.__conjunctions

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

Instance variables

prop api

Expand source code

@property
def api(self):
    """
    Access to the `api` submodule from within a PyAuroraX object.
    """
    return self.__api

Access to the pyaurorax.search.api submodule from within a PyAuroraX object.

prop availability

Expand source code

@property
def availability(self):
    """
    Access to the `availability` submodule from within a PyAuroraX object.
    """
    return self.__availability

Access to the pyaurorax.search.availability submodule from within a PyAuroraX object.

prop conjunctions

Expand source code

@property
def conjunctions(self):
    """
    Access to the `conjunctions` submodule from within a PyAuroraX object.
    """
    return self.__conjunctions

Access to the pyaurorax.search.conjunctions submodule from within a PyAuroraX object.

prop data_products

Expand source code

@property
def data_products(self):
    """
    Access to the `data_products` submodule from within a PyAuroraX object.
    """
    return self.__data_products

Access to the pyaurorax.search.data_products submodule from within a PyAuroraX object.

prop ephemeris

Expand source code

@property
def ephemeris(self):
    """
    Access to the `ephemeris` submodule from within a PyAuroraX object.
    """
    return self.__ephemeris

Access to the pyaurorax.search.ephemeris submodule from within a PyAuroraX object.

prop metadata

Expand source code

@property
def metadata(self):
    """
    Access to the `metadata` submodule from within a PyAuroraX object.
    """
    return self.__metadata

Access to the pyaurorax.search.metadata submodule from within a PyAuroraX object.

prop requests

Expand source code

@property
def requests(self):
    """
    Access to the `requests` submodule from within a PyAuroraX object.
    """
    return self.__requests

Access to the pyaurorax.search.requests submodule from within a PyAuroraX object.

prop sources

Expand source code

@property
def sources(self):
    """
    Access to the `sources` submodule from within a PyAuroraX object.
    """
    return self.__sources

Access to the pyaurorax.search.sources submodule from within a PyAuroraX object.

prop util

Expand source code

@property
def util(self):
    """
    Access to the `util` submodule from within a PyAuroraX object.
    """
    return self.__util

Access to the pyaurorax.search.util submodule from within a PyAuroraX object.

class SpaceCriteriaBlock (programs: List[str] = [],
platforms: List[str] = [],
instrument_types: List[str] = [],
hemisphere: List[Literal['northern', 'southern']] = [],
metadata_filters: MetadataFilter | None = None)

Expand source code

class SpaceCriteriaBlock:
    """
    Representation of a space criteria block used for conjunction searches. 

    Attributes:
        programs (List[str]): 
            List of program strings to use in this criteria block. Optional, default is `[]`.
        
        platforms (List[str]): 
            List of platform strings to use in this criteria block. Optional, default is `[]`.

        instrument_types (List[str]): 
            List of instrument type strings to use in this criteria block. Optional, default is `[]`.

        hemisphere (List[str]): 
            List of hemisphere strings to use in this criteria block. Valid values are 'northern' 
            or 'southern'. Optional, default is `[]`.

        metadata_filters (MetadataFilter): 
            The metadata filters to use in this criteria block. Optional, default is None.
    """

    def __init__(self,
                 programs: List[str] = [],
                 platforms: List[str] = [],
                 instrument_types: List[str] = [],
                 hemisphere: List[Literal["northern", "southern"]] = [],
                 metadata_filters: Optional[MetadataFilter] = None):
        self.programs = programs
        self.platforms = platforms
        self.instrument_types = instrument_types
        self.hemisphere = hemisphere
        self.metadata_filters = metadata_filters

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

    def __repr__(self) -> str:
        return "SpaceCriteriaBlock(programs=%s, platforms=%s, instrument_types=%s, hemisphere=%s, metadata_filters=%s)" % (
            self.programs,
            self.platforms,
            self.instrument_types,
            self.hemisphere,
            self.metadata_filters,
        )

    def pretty_print(self):
        """
        A special print output for this class.
        """
        # set special strings
        max_len = 80
        metadata_filters_str = str(self.metadata_filters)
        if (len(metadata_filters_str) > max_len):
            metadata_filters_str = "%s...)" % (metadata_filters_str[0:max_len])

        # print
        print("SpaceCriteriaBlock:")
        print("  %-18s: %s" % ("programs", self.programs))
        print("  %-18s: %s" % ("platforms", self.platforms))
        print("  %-18s: %s" % ("instrument_types", self.instrument_types))
        print("  %-18s: %s" % ("hemisphere", self.hemisphere))
        print("  %-18s: %s" % ("metadata_filters", metadata_filters_str))

    def to_search_query_dict(self):
        query_dict = self.__dict__
        query_dict["ephemeris_metadata_filters"] = None if self.metadata_filters is None else self.metadata_filters.__dict__
        del query_dict["metadata_filters"]
        return query_dict

Representation of a space criteria block used for conjunction searches.

Attributes

programs : List[str]

List of program strings to use in this criteria block. Optional, default is [].

platforms : List[str]

List of platform strings to use in this criteria block. Optional, default is [].

instrument_types : List[str]

List of instrument type strings to use in this criteria block. Optional, default is [].

hemisphere : List[str]

List of hemisphere strings to use in this criteria block. Valid values are 'northern' or 'southern'. Optional, default is [].

metadata_filters : MetadataFilter

The metadata filters to use in this criteria block. Optional, default is None.

Methods

def pretty_print(self)

Expand source code

def pretty_print(self):
    """
    A special print output for this class.
    """
    # set special strings
    max_len = 80
    metadata_filters_str = str(self.metadata_filters)
    if (len(metadata_filters_str) > max_len):
        metadata_filters_str = "%s...)" % (metadata_filters_str[0:max_len])

    # print
    print("SpaceCriteriaBlock:")
    print("  %-18s: %s" % ("programs", self.programs))
    print("  %-18s: %s" % ("platforms", self.platforms))
    print("  %-18s: %s" % ("instrument_types", self.instrument_types))
    print("  %-18s: %s" % ("hemisphere", self.hemisphere))
    print("  %-18s: %s" % ("metadata_filters", metadata_filters_str))

A special print output for this class.

def to_search_query_dict(self)

Expand source code

def to_search_query_dict(self):
    query_dict = self.__dict__
    query_dict["ephemeris_metadata_filters"] = None if self.metadata_filters is None else self.metadata_filters.__dict__
    del query_dict["metadata_filters"]
    return query_dict