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.
Expand source code
# Copyright 2024 University of Calgary
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
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.
"""
# pull in classes
from .location import Location
from .sources.classes.data_source import DataSource
from .availability.classes.availability_result import AvailabilityResult
from .ephemeris.classes.ephemeris import EphemerisData
from .ephemeris.classes.search import EphemerisSearch
from .data_products.classes.data_product import DataProductData
from .data_products.classes.search import DataProductSearch
from .conjunctions.classes.conjunction import Conjunction
from .conjunctions.classes.search import ConjunctionSearch
# pull in constants
from .sources.classes.data_source import (
FORMAT_BASIC_INFO,
FORMAT_BASIC_INFO_WITH_METADATA,
FORMAT_FULL_RECORD,
FORMAT_IDENTIFIER_ONLY,
FORMAT_DEFAULT,
SOURCE_TYPE_EVENT_LIST,
SOURCE_TYPE_GROUND,
SOURCE_TYPE_HEO,
SOURCE_TYPE_LEO,
SOURCE_TYPE_LUNAR,
)
from .data_products.classes.data_product import (
DATA_PRODUCT_TYPE_KEOGRAM,
DATA_PRODUCT_TYPE_MONTAGE,
DATA_PRODUCT_TYPE_MOVIE,
DATA_PRODUCT_TYPE_SUMMARY_PLOT,
DATA_PRODUCT_TYPE_DATA_AVAILABILITY,
)
from .conjunctions.classes.conjunction import CONJUNCTION_TYPE_NBTRACE, CONJUNCTION_TYPE_SBTRACE
# imports for this file
from . import api as module_api
from .util import UtilManager
from .sources import SourcesManager
from .availability import AvailabilityManager
from .metadata import MetadataManager
from .requests import RequestsManager
from .ephemeris import EphemerisManager
from .data_products import DataProductsManager
from .conjunctions import ConjunctionsManager
__all__ = [
"SearchManager",
"FORMAT_BASIC_INFO",
"FORMAT_BASIC_INFO_WITH_METADATA",
"FORMAT_FULL_RECORD",
"FORMAT_IDENTIFIER_ONLY",
"FORMAT_DEFAULT",
"SOURCE_TYPE_EVENT_LIST",
"SOURCE_TYPE_GROUND",
"SOURCE_TYPE_HEO",
"SOURCE_TYPE_LEO",
"SOURCE_TYPE_LUNAR",
"DATA_PRODUCT_TYPE_KEOGRAM",
"DATA_PRODUCT_TYPE_MONTAGE",
"DATA_PRODUCT_TYPE_MOVIE",
"DATA_PRODUCT_TYPE_SUMMARY_PLOT",
"DATA_PRODUCT_TYPE_DATA_AVAILABILITY",
"CONJUNCTION_TYPE_NBTRACE",
"CONJUNCTION_TYPE_SBTRACE",
"DataSource",
"Location",
"AvailabilityResult",
"EphemerisData",
"EphemerisSearch",
"DataProductData",
"DataProductSearch",
"Conjunction",
"ConjunctionSearch",
]
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)
# ------------------------------------------
# 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.__conjunctionsSub-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-
AuroraX
Locationclass definition pyaurorax.search.metadata-
Interacting with the data source metadata schemas …
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: 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
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] = NoneClass variables
var available_data_products : Optional[Dict]var available_ephemeris : Optional[Dict]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)-
Conjunction object
Attributes
conjunction_type- the type of location data used when the conjunction was found (either 'nbtrace', 'sbtrace', or 'geographic')
start- start timestamp of the conjunction
end- end timestamp of the conjunction
data_sources- data sources in the conjunction
min_distance- minimum kilometer distance of the conjunction
max_distance- maximum kilometer distance of the conjunction
events- the sub-conjunctions that make up this over-arching conjunction (the conjunctions between each set of two data sources)
closest_epoch- timestamp for when data sources were closest
farthest_epoch- timestamp for when data sources were farthest
Expand source code
class Conjunction: """ Conjunction object Attributes: conjunction_type: the type of location data used when the conjunction was found (either 'nbtrace', 'sbtrace', or 'geographic') start: start timestamp of the conjunction end: end timestamp of the conjunction data_sources: data sources in the conjunction min_distance: minimum kilometer distance of the conjunction max_distance: maximum kilometer distance of the conjunction events: the sub-conjunctions that make up this over-arching conjunction (the conjunctions between each set of two data sources) closest_epoch: timestamp for when data sources were closest farthest_epoch: 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=[...])" class ConjunctionSearch (aurorax_obj: PyAuroraX, start: datetime.datetime, end: datetime.datetime, distance: Union[int, float, Dict], ground: Optional[List[Dict]] = None, space: Optional[List[Dict]] = None, events: Optional[List[Dict]] = None, conjunction_types: Optional[List[str]] = None, epoch_search_precision: Optional[int] = None, response_format: Optional[Dict] = None)-
Class representing a conjunction search
Attributes
start- start timestamp of the search (inclusive)
end- end timestamp of the search (inclusive)
distance- 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 of ground instrument search parameters, defaults to []
Example:
[{ "programs": ["themis-asi"], "platforms": ["gillam", "rabbit lake"], "instrument_types": ["RGB"], "ephemeris_metadata_filters": { "logical_operator": "AND", "expressions": [ { "key": "calgary_apa_ml_v1", "operator": "in", "values": [ "classified as APA" ] } ] } }] space-
list of one or more space instrument search parameters, defaults to []
Example:
[{ "programs": ["themis-asi", "swarm"], "platforms": ["themisa", "swarma"], "instrument_types": ["footprint"], "ephemeris_metadata_filters": { "logical_operator": "AND", "expressions": [ { "key": "nbtrace_region", "operator": "in", "values": [ "north auroral oval" ] } ] }, "hemisphere": [ "northern" ] }] events-
list of one or more events search parameters, defaults to []
Example:
[{ "programs": [ "events" ], "instrument_types": [ "substorm onsets" ] }] conjunction_types- list of conjunction types, defaults to ["nbtrace"]. Options are in the pyaurorax.conjunctions module, or at the top level using the pyaurorax.CONJUNCTION_TYPE_* variables.
epoch_search_precision- the time precision to which conjunctions are calculated. Can be 30 or 60 seconds. Defaults to 60 seconds. Note - this parameter is under active development and still considered "alpha".
response_format- JSON representation of desired data response format
request- AuroraXResponse object returned when the search is executed
request_id- unique ID assigned to the request by the AuroraX API
request_url- unique URL assigned to the request by the AuroraX API
executed- indicates if the search has been executed/started
completed- indicates if the search has finished
data_url- the URL where data is accessed
query- the query for this request as JSON
status- the status of the query
data- the conjunctions found
logs- all log messages outputted by the AuroraX API for this request
Expand source code
class ConjunctionSearch: """ Class representing a conjunction search Attributes: start: start timestamp of the search (inclusive) end: end timestamp of the search (inclusive) distance: 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 of ground instrument search parameters, defaults to [] Example: [{ "programs": ["themis-asi"], "platforms": ["gillam", "rabbit lake"], "instrument_types": ["RGB"], "ephemeris_metadata_filters": { "logical_operator": "AND", "expressions": [ { "key": "calgary_apa_ml_v1", "operator": "in", "values": [ "classified as APA" ] } ] } }] space: list of one or more space instrument search parameters, defaults to [] Example: [{ "programs": ["themis-asi", "swarm"], "platforms": ["themisa", "swarma"], "instrument_types": ["footprint"], "ephemeris_metadata_filters": { "logical_operator": "AND", "expressions": [ { "key": "nbtrace_region", "operator": "in", "values": [ "north auroral oval" ] } ] }, "hemisphere": [ "northern" ] }] events: list of one or more events search parameters, defaults to [] Example: [{ "programs": [ "events" ], "instrument_types": [ "substorm onsets" ] }] conjunction_types: list of conjunction types, defaults to ["nbtrace"]. Options are in the pyaurorax.conjunctions module, or at the top level using the pyaurorax.CONJUNCTION_TYPE_* variables. epoch_search_precision: the time precision to which conjunctions are calculated. Can be 30 or 60 seconds. Defaults to 60 seconds. Note - this parameter is under active development and still considered "alpha". response_format: JSON representation of desired data response format request: AuroraXResponse object returned when the search is executed request_id: unique ID assigned to the request by the AuroraX API request_url: unique URL assigned to the request by the AuroraX API executed: indicates if the search has been executed/started completed: indicates if the search has finished data_url: the URL where data is accessed query: the query for this request as JSON status: the status of the query data: the conjunctions found logs: 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: Optional[List[Dict]] = None, space: Optional[List[Dict]] = None, events: Optional[List[Dict]] = None, conjunction_types: Optional[List[str]] = None, epoch_search_precision: Optional[int] = None, response_format: Optional[Dict] = None): # set variables using passed in args self.aurorax_obj = aurorax_obj self.start = start self.end = end self.ground = [] if ground is None else ground self.space = [] if space is None else space self.events = [] if events is None else events self.distance = distance self.conjunction_types = [CONJUNCTION_TYPE_NBTRACE] if conjunction_types is None else conjunction_types self.epoch_search_precision = 60 if epoch_search_precision is None else epoch_search_precision 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 f"ConjunctionSearch(executed={self.executed}, completed={self.completed}, request_id='{self.request_id}')" 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 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 ((count_ground + count_space + count_events) > 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: 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)) # 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 """ self._query = { "start": self.start.strftime("%Y-%m-%dT%H:%M:%S"), "end": self.end.strftime("%Y-%m-%dT%H:%M:%S"), "ground": self.ground, "space": self.space, "events": self.events, "conjunction_types": self.conjunction_types, "max_distances": self.distance, "epoch_search_precision": self.epoch_search_precision if self.epoch_search_precision in [30, 60] else 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: 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: time in seconds to wait between polling attempts, defaults to pyaurorax.requests.STANDARD_POLLING_SLEEP_TIME verbose: 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: wait until the cancellation request has been completed (may wait for several minutes) poll_interval: seconds to wait between polling calls, defaults to STANDARD_POLLING_SLEEP_TIME. verbose: 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)Instance variables
var distance : Union[int, float, Dict[str, Union[int, float]]]-
Property for the distance parameter
Returns
the distance dictionary with all combinations
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 var query : Dict-
Property for the query value
Returns
the query parameter
Expand source code
@property def query(self) -> Dict: """ Property for the query value Returns: the query parameter """ self._query = { "start": self.start.strftime("%Y-%m-%dT%H:%M:%S"), "end": self.end.strftime("%Y-%m-%dT%H:%M:%S"), "ground": self.ground, "space": self.space, "events": self.events, "conjunction_types": self.conjunction_types, "max_distances": self.distance, "epoch_search_precision": self.epoch_search_precision if self.epoch_search_precision in [30, 60] else 60, } return self._query
Methods
def cancel(self, wait: bool = False, poll_interval: float = 1.0, 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- wait until the cancellation request has been completed (may wait for several minutes)
poll_interval- seconds to wait between polling calls, defaults to STANDARD_POLLING_SLEEP_TIME.
verbose- output poll times and other progress messages, defaults to False
Returns
1 on success
Raises
AuroraXAPIError- An API error was encountered
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: wait until the cancellation request has been completed (may wait for several minutes) poll_interval: seconds to wait between polling calls, defaults to STANDARD_POLLING_SLEEP_TIME. verbose: 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 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
AuroraXError- too many criteria blocks are found
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 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 ((count_ground + count_space + count_events) > 10): raise AuroraXError("Number of criteria blocks exceeds 10, please reduce the count") 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
AuroraXAPIError- An API error was encountered
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 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 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- the default distance to use, defaults to None
Returns
the advanced distances combinations
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: 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)) # 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 def get_data(self) ‑> None-
Retrieve the data available for this conjunction search request
Raises
AuroraXAPIError- An API error was encountered
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] def update_status(self, status: Optional[Dict] = None) ‑> None-
Update the status of this conjunction search request
Args
status- 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
Expand source code
def update_status(self, status: Optional[Dict] = None) -> None: """ Update the status of this conjunction search request Args: status: 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 wait(self, poll_interval: float = 1.0, verbose: bool = False) ‑> None-
Block and wait until the request is complete and data is available for retrieval
Args
poll_interval- time in seconds to wait between polling attempts, defaults to pyaurorax.requests.STANDARD_POLLING_SLEEP_TIME
verbose- output poll times and other progress messages, defaults to False
Raises
AuroraXAPIError- An API error was encountered
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: time in seconds to wait between polling attempts, defaults to pyaurorax.requests.STANDARD_POLLING_SLEEP_TIME verbose: 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))
class DataProductData (data_source: DataSource, data_product_type: str, start: datetime.datetime, end: datetime.datetime, url: str, metadata: Optional[Dict] = None)-
Data product object
Attributes
data_source- data source that the ephemeris record is associated with
data_product_type- data product type ("keogram", "movie", "summary_plot")
start- starting timestamp for the record (assumed it is in UTC), inclusive
end- ending timestamp for the record (assumed it is in UTC), inclusive
url- the URL of data product
metadata- metadata for this record (arbitrary keys and values)
Expand source code
class DataProductData: """ Data product object Attributes: data_source: data source that the ephemeris record is associated with data_product_type: data product type ("keogram", "movie", "summary_plot") start: starting timestamp for the record (assumed it is in UTC), inclusive end: ending timestamp for the record (assumed it is in UTC), inclusive url: the URL of data product metadata: 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(...))"Methods
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
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
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[List[str]] = None, metadata_filters: Optional[List[Dict]] = None, metadata_filters_logical_operator: Optional[str] = None, response_format: Optional[Dict] = None)-
Class representing a data product search
Attributes
start- start timestamp of the search (inclusive)
end- end timestamp of the search (inclusive)
programs- list of program names to search
platforms- list of platform names to search
instrument_types- list of instrument types to search
data_product_types- list of dictionaries describing data product types to filter on e.g. "keogram", defaults to None. Options are in the pyaurorax.data_products module, or at the top level using the pyaurorax.DATA_PRODUCT_TYPE* variables.
metadata_filters-
list of dictionaries describing metadata keys and values to filter on, defaults to None
Example:
[{ "key": "nbtrace_region", "operator": "in", "values": ["north polar cap"] }] metadata_filters_logical_operator- the logical operator to use when evaluating metadata filters (either 'AND' or 'OR'), defaults to "AND"
response_format- JSON representation of desired data response format
request- AuroraXResponse object returned when the search is executed
request_id- unique ID assigned to the request by the AuroraX API
request_url- unique URL assigned to the request by the AuroraX API
executed- indicates if the search has been executed/started
completed- indicates if the search has finished
data_url- the URL where data is accessed
query- the query for this request as JSON
status- the status of the query
data- the data product records found
logs- all log messages outputted by the AuroraX API for this request
Expand source code
class DataProductSearch: """ Class representing a data product search Attributes: start: start timestamp of the search (inclusive) end: end timestamp of the search (inclusive) programs: list of program names to search platforms: list of platform names to search instrument_types: list of instrument types to search data_product_types: list of dictionaries describing data product types to filter on e.g. "keogram", defaults to None. Options are in the pyaurorax.data_products module, or at the top level using the pyaurorax.DATA_PRODUCT_TYPE* variables. metadata_filters: list of dictionaries describing metadata keys and values to filter on, defaults to None Example: [{ "key": "nbtrace_region", "operator": "in", "values": ["north polar cap"] }] metadata_filters_logical_operator: the logical operator to use when evaluating metadata filters (either 'AND' or 'OR'), defaults to "AND" response_format: JSON representation of desired data response format request: AuroraXResponse object returned when the search is executed request_id: unique ID assigned to the request by the AuroraX API request_url: unique URL assigned to the request by the AuroraX API executed: indicates if the search has been executed/started completed: indicates if the search has finished data_url: the URL where data is accessed query: the query for this request as JSON status: the status of the query data: the data product records found logs: 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[List[str]] = None, metadata_filters: Optional[List[Dict]] = None, metadata_filters_logical_operator: Optional[str] = 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 f"DataProductsSearch(executed={self.executed}, completed={self.completed}, request_id='{self.request_id}')" @property def query(self): """ Property for the query value """ 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": {} if not self.metadata_filters else { "logical_operator": self.metadata_filters_logical_operator, "expressions": self.metadata_filters }, }, "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 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: 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: time in seconds to wait between polling attempts, defaults to pyaurorax.requests.STANDARD_POLLING_SLEEP_TIME verbose: 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: wait until the cancellation request has been completed (may wait for several minutes) poll_interval: seconds to wait between polling calls, defaults to STANDARD_POLLING_SLEEP_TIME. verbose: 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)Instance variables
var query-
Property for the query value
Expand source code
@property def query(self): """ Property for the query value """ 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": {} if not self.metadata_filters else { "logical_operator": self.metadata_filters_logical_operator, "expressions": self.metadata_filters }, }, "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 self._query
Methods
def cancel(self, wait: bool = False, poll_interval: float = 1.0, 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- wait until the cancellation request has been completed (may wait for several minutes)
poll_interval- seconds to wait between polling calls, defaults to STANDARD_POLLING_SLEEP_TIME.
verbose- 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
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: wait until the cancellation request has been completed (may wait for several minutes) poll_interval: seconds to wait between polling calls, defaults to STANDARD_POLLING_SLEEP_TIME. verbose: 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 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
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 def execute(self) ‑> None-
Initiate a data product search request
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 def get_data(self) ‑> None-
Retrieve the data available for this data product search request
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] def update_status(self, status: Optional[Dict] = None) ‑> None-
Update the status of this data product search request
Args
status- the previously-retrieved status of this request (include to avoid requesting it from the API again), defaults to None
Expand source code
def update_status(self, status: Optional[Dict] = None) -> None: """ Update the status of this data product search request Args: status: 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 wait(self, poll_interval: float = 1.0, verbose: bool = False) ‑> None-
Block and wait for the request to complete and data is available for retrieval
Args
poll_interval- time in seconds to wait between polling attempts, defaults to pyaurorax.requests.STANDARD_POLLING_SLEEP_TIME
verbose- output poll times and other progress messages, defaults to False
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: time in seconds to wait between polling attempts, defaults to pyaurorax.requests.STANDARD_POLLING_SLEEP_TIME verbose: 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))
class DataSource (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 = 'full_record')-
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
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.
Expand source code
@dataclass 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 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. """ 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_RECORDClass variables
var data_product_metadata_schema : Optional[List[Dict]]var display_name : Optional[str]var ephemeris_metadata_schema : Optional[List[Dict]]var format : strvar identifier : Optional[int]var instrument_type : Optional[str]var maintainers : Optional[List[str]]var metadata : Optional[Dict]var owner : Optional[str]var platform : Optional[str]var program : Optional[str]var source_type : Optional[str]var stats : Optional[DataSourceStatistics]
class EphemerisData (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)-
Ephemeris object
Attributes
data_source- data source that the ephemeris record is associated with
epoch- timestamp for the record (assumed it is in UTC)
location_geo- Location object containing geographic latitude and longitude
location_gsm- Location object containing GSM latitude and longitude (leave empty for data sources with a type of 'ground')
nbtrace- Location object with north B-trace geographic latitude and longitude
sbtrace- Location object with south B-trace geographic latitude and longitude
metadata- metadata for this record (arbitrary keys and values)
Expand source code
class EphemerisData: """ Ephemeris object Attributes: data_source: data source that the ephemeris record is associated with epoch: timestamp for the record (assumed it is in UTC) location_geo: Location object containing geographic latitude and longitude location_gsm: Location object containing GSM latitude and longitude (leave empty for data sources with a type of 'ground') nbtrace: Location object with north B-trace geographic latitude and longitude sbtrace: Location object with south B-trace geographic latitude and longitude metadata: 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(...))"Methods
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
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
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[List[Dict]] = None, metadata_filters_logical_operator: "Optional[Literal['AND', 'OR']]" = None, response_format: Optional[Dict] = None)-
Class representing an ephemeris search
Note: At least one search criteria from programs, platforms, or instrument_types must be specified.
Args
start- start timestamp of the search (inclusive)
end- end timestamp of the search (inclusive)
programs- list of programs to search through, defaults to None
platforms- list of platforms to search through, defaults to None
instrument_types- list of instrument types to search through, defaults to None
metadata_filters-
list of dictionaries describing metadata keys and values to filter on, defaults to None
e.g. { "key": "string", "operator": "=", "values": [ "string" ] }
metadata_filters_logical_operator- the logical operator to use when evaluating metadata filters (either 'AND' or 'OR'), defaults to "AND"
response_format- JSON representation of desired data response format
request- AuroraXResponse object returned when the search is executed
request_id- unique ID assigned to the request by the AuroraX API
request_url- unique URL assigned to the request by the AuroraX API
executed- indicates if the search has been executed/started
completed- indicates if the search has finished
data_url- the URL where data is accessed
query- the query for this request as JSON
status- the status of the query
data- the ephemeris records found
logs- all log messages outputted by the AuroraX API for this request
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. Args: start: start timestamp of the search (inclusive) end: end timestamp of the search (inclusive) programs: list of programs to search through, defaults to None platforms: list of platforms to search through, defaults to None instrument_types: list of instrument types to search through, defaults to None metadata_filters: list of dictionaries describing metadata keys and values to filter on, defaults to None e.g. { "key": "string", "operator": "=", "values": [ "string" ] } metadata_filters_logical_operator: the logical operator to use when evaluating metadata filters (either 'AND' or 'OR'), defaults to "AND" response_format: JSON representation of desired data response format request: AuroraXResponse object returned when the search is executed request_id: unique ID assigned to the request by the AuroraX API request_url: unique URL assigned to the request by the AuroraX API executed: indicates if the search has been executed/started completed: indicates if the search has finished data_url: the URL where data is accessed query: the query for this request as JSON status: the status of the query data: the ephemeris records found logs: 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[List[Dict]] = None, metadata_filters_logical_operator: Optional[Literal["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 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 f"EphemerisSearch(executed={self.executed}, completed={self.completed}, request_id='{self.request_id}')" @property def query(self): """ Property for the query value """ 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": {} if not self.metadata_filters else { "logical_operator": self.metadata_filters_logical_operator, "expressions": self.metadata_filters }, }, "start": self.start.strftime("%Y-%m-%dT%H:%M:%S"), "end": self.end.strftime("%Y-%m-%dT%H:%M:%S"), } 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: 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: time in seconds to wait between polling attempts, defaults to pyaurorax.requests.STANDARD_POLLING_SLEEP_TIME verbose: 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: wait until the cancellation request has been completed (may wait for several minutes) poll_interval: seconds to wait between polling calls, defaults to STANDARD_POLLING_SLEEP_TIME. verbose: 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)Instance variables
var query-
Property for the query value
Expand source code
@property def query(self): """ Property for the query value """ 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": {} if not self.metadata_filters else { "logical_operator": self.metadata_filters_logical_operator, "expressions": self.metadata_filters }, }, "start": self.start.strftime("%Y-%m-%dT%H:%M:%S"), "end": self.end.strftime("%Y-%m-%dT%H:%M:%S"), } return self._query
Methods
def cancel(self, wait: bool = False, poll_interval: float = 1.0, 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- wait until the cancellation request has been completed (may wait for several minutes)
poll_interval- seconds to wait between polling calls, defaults to STANDARD_POLLING_SLEEP_TIME.
verbose- 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
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: wait until the cancellation request has been completed (may wait for several minutes) poll_interval: seconds to wait between polling calls, defaults to STANDARD_POLLING_SLEEP_TIME. verbose: 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 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
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 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 def get_data(self) ‑> None-
Retrieve the data available for this ephemeris search request
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] def update_status(self, status: Optional[Dict] = None) ‑> None-
Update the status of this ephemeris search request
Args
status- the previously-retrieved status of this request (include to avoid requesting it from the API again), defaults to None
Expand source code
def update_status(self, status: Optional[Dict] = None) -> None: """ Update the status of this ephemeris search request Args: status: 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 wait(self, poll_interval: float = 1.0, verbose: bool = False) ‑> None-
Block and wait for the request to complete and data is available for retrieval
Args
poll_interval- time in seconds to wait between polling attempts, defaults to pyaurorax.requests.STANDARD_POLLING_SLEEP_TIME
verbose- output poll times and other progress messages, defaults to False
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: time in seconds to wait between polling attempts, defaults to pyaurorax.requests.STANDARD_POLLING_SLEEP_TIME verbose: 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))
class Location (lat: Optional[float] = None, lon: Optional[float] = None)-
Representation for an AuroraX location, such as geographic coordinates, GSM coordinates, or 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.
Expand source code
class Location: """ Representation for an AuroraX location, such as geographic coordinates, GSM coordinates, or 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))Instance variables
var lat-
Expand source code
@property def lat(self): return self.__lat var lon-
Expand source code
@property def lon(self): return self.__lon
Methods
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
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}
class SearchManager (aurorax_obj)-
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.
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) # ------------------------------------------ # 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.__conjunctionsInstance variables
var api-
Access to the
pyaurorax.search.apisubmodule from within a PyAuroraX object.Expand source code
@property def api(self): """ Access to the `api` submodule from within a PyAuroraX object. """ return self.__api var availability-
Access to the
pyaurorax.search.availabilitysubmodule from within a PyAuroraX object.Expand source code
@property def availability(self): """ Access to the `availability` submodule from within a PyAuroraX object. """ return self.__availability var conjunctions-
Access to the
pyaurorax.search.conjunctionssubmodule from within a PyAuroraX object.Expand source code
@property def conjunctions(self): """ Access to the `conjunctions` submodule from within a PyAuroraX object. """ return self.__conjunctions var data_products-
Access to the
pyaurorax.search.data_productssubmodule from within a PyAuroraX object.Expand source code
@property def data_products(self): """ Access to the `data_products` submodule from within a PyAuroraX object. """ return self.__data_products var ephemeris-
Access to the
pyaurorax.search.ephemerissubmodule from within a PyAuroraX object.Expand source code
@property def ephemeris(self): """ Access to the `ephemeris` submodule from within a PyAuroraX object. """ return self.__ephemeris var metadata-
Access to the
pyaurorax.search.metadatasubmodule from within a PyAuroraX object.Expand source code
@property def metadata(self): """ Access to the `metadata` submodule from within a PyAuroraX object. """ return self.__metadata var requests-
Access to the
pyaurorax.search.requestssubmodule from within a PyAuroraX object.Expand source code
@property def requests(self): """ Access to the `requests` submodule from within a PyAuroraX object. """ return self.__requests var sources-
Access to the
pyaurorax.search.sourcessubmodule from within a PyAuroraX object.Expand source code
@property def sources(self): """ Access to the `sources` submodule from within a PyAuroraX object. """ return self.__sources var util-
Access to the
pyaurorax.search.utilsubmodule from within a PyAuroraX object.Expand source code
@property def util(self): """ Access to the `util` submodule from within a PyAuroraX object. """ return self.__util