diff --git a/geomagio/metadata/MetadataFactory.py b/geomagio/metadata/MetadataFactory.py
index 92b449752f6739dab281b9280dfa6ffe86414972..fdd134e602a76857c71f85f1379d8883674825c9 100644
--- a/geomagio/metadata/MetadataFactory.py
+++ b/geomagio/metadata/MetadataFactory.py
@@ -9,7 +9,7 @@ from pydantic import parse_obj_as
from .Metadata import Metadata
from .MetadataQuery import MetadataQuery
-
+# Set the API host and URL for the Geomagnetic Metadata service
GEOMAG_API_HOST = os.getenv("GEOMAG_API_HOST", "geomag.usgs.gov")
GEOMAG_API_URL = f"https://{GEOMAG_API_HOST}/ws/secure/metadata"
if "127.0.0.1" in GEOMAG_API_URL:
@@ -17,22 +17,70 @@ if "127.0.0.1" in GEOMAG_API_URL:
class MetadataFactory(object):
+ """
+ A factory class for interacting with metadata servers.
+
+ Attributes:
+ ----------
+ url: str
+ The base URL for the metadata server.
+ token: str
+ The authorization token for accessing the API.
+ """
+
def __init__(
self,
url: str = GEOMAG_API_URL,
token: str = os.getenv("GITLAB_API_TOKEN"),
):
+ """
+ Initializes the MetadataFactory with the specified URL and token.
+
+ Parameters:
+ -----------
+ url: str
+ The base URL for the metadata server (default is GEOMAG_API_URL).
+ token: str
+ The authorization token (default is fetched from environment).
+ """
self.url = url
self.token = token
def _get_headers(self):
+ """
+ Constructs the headers for the API request.
+
+ Returns:
+ --------
+ dict:
+ A dictionary containing the authorization token and content type.
+ """
return (
{"Authorization": self.token, "content-type": "application/json"}
if self.token
else None
)
- def get_metadata(self, query: MetadataQuery) -> List[Metadata]:
+ def get_metadata(
+ self, query: MetadataQuery, timeout: float | tuple = None
+ ) -> List[Metadata]:
+ """
+ Retrieves metadata based on the provided query.
+
+ Parameters:
+ -----------
+ query: MetadataQuery
+ The query object containing search parameters.
+ timeout: float or tuple, optional
+ The timeout for the API request.
+ Can be a single number (for both connection and read timeouts)
+ or a tuple (connection timeout, read timeout). Defaults to None.
+
+ Returns:
+ --------
+ metadata: List[Metadata]
+ A list of Metadata objects matching the query.
+ """
if query.id:
metadata = [self.get_metadata_by_id(id=query.id)]
else:
@@ -40,10 +88,11 @@ class MetadataFactory(object):
url=self.url,
headers=self._get_headers(),
params=parse_params(query=query),
+ timeout=timeout,
)
metadata = parse_obj_as(List[Metadata], response.json())
- # if metadata dict() provided in query, only include Metadata objects with matching key-value pairs
+ # Filter metadata based on additional criteria if provided
if query.metadata is not None:
query_metadata_items = query.metadata.items()
metadata_list = metadata
@@ -59,42 +108,110 @@ class MetadataFactory(object):
return metadata
- def get_metadata_by_id(self, id: int) -> Metadata:
+ def get_metadata_by_id(self, id: int, timeout: float | tuple = None) -> Metadata:
+ """
+ Retrieves metadata for a specific ID.
+
+ Parameters:
+ -----------
+ id: int
+ The ID of the metadata to retrieve.
+ timeout: float or tuple, optional
+ The timeout for the API request.
+ Can be a single number (for both connection and read timeouts)
+ or a tuple (connection timeout, read timeout). Defaults to None.
+
+ Returns:
+ --------
+ Metadata: object
+ The Metadata object corresponding to the given ID.
+ """
response = requests.get(
- url=f"{self.url}/{id}",
- headers=self._get_headers(),
+ url=f"{self.url}/{id}", headers=self._get_headers(), timeout=timeout
)
return Metadata(**response.json())
- def create_metadata(self, metadata: Metadata) -> Metadata:
+ def create_metadata(
+ self, metadata: Metadata, timeout: float | tuple = None
+ ) -> Metadata:
+ """
+ Creates new metadata objects in the database.
+
+ Parameters:
+ -----------
+ metadata: Metadata
+ The Metadata object to create.
+ timeout: float or tuple, optional
+ The timeout for the API request.
+ Can be a floating-point number (for both connection and read timeouts)
+ or a tuple (connection timeout, read timeout). Defaults to None.
+
+ Returns:
+ --------
+ Metadata: The created Metadata object.
+ """
response = requests.post(
url=self.url,
data=metadata.json(),
headers=self._get_headers(),
+ timeout=timeout,
)
return Metadata(**response.json())
- def update_metadata(self, metadata: Metadata) -> Metadata:
+ def update_metadata(
+ self, metadata: Metadata, timeout: float | tuple = None
+ ) -> Metadata:
+ """
+ Updates existing metadata in the system.
+
+ Parameters:
+ -----------
+ metadata: Metadata
+ The Metadata object to update.
+ timeout: float or tuple, optional
+ The timeout for the API request.
+ Can be a floating-point number (for both connection and read timeouts)
+ or a tuple (connection timeout, read timeout). Defaults to None.
+
+ Returns:
+ -----------
+ Metadata:
+ The updated Metadata object.
+ """
response = requests.put(
url=f"{self.url}/{metadata.id}",
data=metadata.json(),
headers=self._get_headers(),
+ timeout=timeout,
)
return Metadata(**response.json())
def parse_params(query: MetadataQuery) -> str:
+ """
+ Parses the query object into a dictionary of parameters for the API request.
+
+ Parameters:
+ -----------
+ query: MetadataQuery
+ The query object containing search parameters.
+
+ Returns:
+ --------
+ dict:
+ A dictionary of parameters formatted for the API request.
+ """
query = query.dict(exclude_none=True)
args = {}
for key in query.keys():
element = query[key]
- # convert times to strings
+ # Convert times to ISO format strings
if isinstance(element, UTCDateTime):
element = element.isoformat()
- # serialize the metadata dictionary to a JSON string
+ # Serialize the metadata dictionary to a JSON string
elif key == "metadata" and isinstance(element, dict):
args[key] = json.dumps(element)
- # get string value of metadata category
+ # Get string value of metadata category
if key == "category":
element = element.value
args[key] = element
diff --git a/geomagio/metadata/instrument/InstrumentCalibrations.py b/geomagio/metadata/instrument/InstrumentCalibrations.py
index 1bdc33d792afe18c9e1f2afb2ed9789738cfff02..8dfcef480e4d070e767bbb982e8868a677d55422 100644
--- a/geomagio/metadata/instrument/InstrumentCalibrations.py
+++ b/geomagio/metadata/instrument/InstrumentCalibrations.py
@@ -14,33 +14,33 @@ class InstrumentCalibrations:
Attributes
----------
- metadata_list : list
- a list of metadata objects to be processed
- previous_calibration : dict
- a dictionary to store the previous calibration values for each axis and key
- station : str
- the station code, checked for consistency across all metadata objects
- location : str
- the location code, checked for consistency across all metadata objects
- network : str
- the network code, checked for consistency across all metadata objects
+ metadata_list : list
+ a list of metadata objects to be processed
+ previous_calibration : dict
+ a dictionary to store the previous calibration values for each axis and key
+ station : str
+ the station code, checked for consistency across all metadata objects
+ location : str
+ the location code, checked for consistency across all metadata objects
+ network : str
+ the network code, checked for consistency across all metadata objects
Methods
-------
- get_calibrations():
- Main method to gather applicable calibrations from the metadata list.
- create_overlap_interval(current_metadata):
- Creates an overlap interval from the current metadata.
- update_overlap_interval(overlap_interval, current_metadata):
- Updates the overlap interval with the current metadata.
- update_overlap_interval_with_same_starttime(i, sorted_metadata_list, overlap_interval):
- Updates the overlap interval with metadata that have the same starttime.
- set_endtime(i, sorted_metadata_list, overlap_interval, current_metadata):
- Sets the endtime for the overlap interval.
- convert_to_calibration(overlap_metadata):
- Converts overlapping metadata information to an applicable calibration.
- get_channels(overlap_metadata):
- Gets the channels for the applicable calibration.
+ get_calibrations():
+ Main method to gather applicable calibrations from the metadata list.
+ create_overlap_interval(current_metadata):
+ Creates an overlap interval from the current metadata.
+ update_overlap_interval(overlap_interval, current_metadata):
+ Updates the overlap interval with the current metadata.
+ update_overlap_interval_with_same_starttime(i, sorted_metadata_list, overlap_interval):
+ Updates the overlap interval with metadata that have the same starttime.
+ set_endtime(i, sorted_metadata_list, overlap_interval, current_metadata):
+ Sets the endtime for the overlap interval.
+ convert_to_calibration(overlap_metadata):
+ Converts overlapping metadata information to an applicable calibration.
+ get_channels(overlap_metadata):
+ Gets the channels for the applicable calibration.
"""
def __init__(self, metadata_list):
@@ -81,14 +81,14 @@ class InstrumentCalibrations:
Main method to compile applicable calibrations from the metadata list.
Parameters
- ----------
- query_starttime : UTCDateTime
- the query starttime
+ ----------
+ query_starttime : UTCDateTime
+ the query starttime
Returns
-------
- list
- a list of dictionaries, each representing an applicable calibration
+ list
+ a list of dictionaries, each representing an applicable calibration
"""
sorted_metadata_list = sorted(
self.metadata_list,
@@ -120,13 +120,13 @@ class InstrumentCalibrations:
Parameters
----------
- current_metadata : Metadata
- the current metadata object
+ current_metadata : Metadata
+ the current metadata object
Returns
-------
- dict
- a dictionary representing an overlap interval
+ dict
+ a dictionary representing an overlap interval
"""
overlap_interval = {"starttime": current_metadata.starttime}
self.update_overlap_interval(overlap_interval, current_metadata)
@@ -138,10 +138,10 @@ class InstrumentCalibrations:
Parameters
----------
- overlap_interval : dict
- the overlap interval to be updated
- current_metadata : Metadata
- the current metadata object
+ overlap_interval : dict
+ the overlap interval to be updated
+ current_metadata : Metadata
+ the current metadata object
"""
for axis in ["u", "v", "w"]:
for key in ["constant", "bin", "offset"]:
@@ -172,17 +172,17 @@ class InstrumentCalibrations:
Parameters
----------
- i : int
- the current index in the sorted metadata list
- sorted_metadata_list : list
- the sorted list of metadata objects
- overlap_interval : dict
- the overlap interval to be updated
+ i : int
+ the current index in the sorted metadata list
+ sorted_metadata_list : list
+ the sorted list of metadata objects
+ overlap_interval : dict
+ the overlap interval to be updated
Returns
-------
- int
- the updated index in the sorted metadata list
+ int
+ the updated index in the sorted metadata list
"""
while (
i + 1 < len(sorted_metadata_list)
@@ -200,14 +200,14 @@ class InstrumentCalibrations:
Parameters
----------
- i : int
- the current index in the sorted metadata list
- sorted_metadata_list : list
- the sorted list of metadata objects
- overlap_interval : dict
- the overlap interval to be updated
- current_metadata : Metadata
- the current metadata object
+ i : int
+ the current index in the sorted metadata list
+ sorted_metadata_list : list
+ the sorted list of metadata objects
+ overlap_interval : dict
+ the overlap interval to be updated
+ current_metadata : Metadata
+ the current metadata object
"""
if (
i + 1 < len(sorted_metadata_list)
@@ -223,13 +223,13 @@ class InstrumentCalibrations:
Parameters
----------
- overlap_metadata : dict
- the metadata overlap data to be converted
+ overlap_metadata : dict
+ the metadata overlap data to be converted
Returns
-------
- dict
- a dictionary representing an applicable calibration
+ dict
+ a dictionary representing an applicable calibration
"""
calibration = {
"network": self.network,
@@ -250,13 +250,13 @@ class InstrumentCalibrations:
Parameters
----------
- overlap_metadata : dict
- the metadata overlap data from which to get the channels
+ overlap_metadata : dict
+ the metadata overlap data from which to get the channels
Returns
-------
- dict
- a dictionary representing the channels for the applicable calibration
+ dict
+ a dictionary representing the channels for the applicable calibration
"""
channels = {}
for axis in ["U", "V", "W"]:
@@ -304,14 +304,17 @@ def get_instrument_calibrations(
):
"""Get instrument metadata
- Args:
- observatory: observatory code
- start_time: start time to match, or None to match any.
- end_time: end time to match, or None to match any.
- calibrations: use custom list, defaults to pulling and converting instrument metadata
- metadata_url: metadata database url
+ Parameters:
+ -----------
+ observatory: observatory code
+ start_time: start time to match, or None to match any.
+ end_time: end time to match, or None to match any.
+ calibrations: use custom list, defaults to pulling and converting instrument metadata
+ metadata_url: metadata database url
+
Returns:
- list of applicable instrument calibrations
+ --------
+ list of applicable instrument calibrations
"""
if not calibrations:
@@ -329,7 +332,7 @@ def get_instrument_calibrations(
data_valid=True,
)
try:
- metadata = factory.get_metadata(query=query)
+ metadata = factory.get_metadata(query=query, timeout=28)
except:
print(
"Warning: An error occurred while trying to pull metadata from the metadata server!"