Merge branch 'add_absolutes' into 'master'

Add absolutes

See merge request !189

geomagio/processing/copy_absolutes.py

+# -*- coding: utf-8 -*-
+"""
+Created on Thu Jul  7 11:05:19 2022
+
+@author: awernle
+"""
+import json
+import os
+from datetime import date, datetime, time, timedelta
+from typing import List
+
+import typer
+from obspy import UTCDateTime
+
+from ..metadata import Metadata, MetadataCategory, MetadataFactory
+from ..residual import Reading, WebAbsolutesFactory
+
+TODAY = datetime.combine(date.today(), time(0, 0, 0))
+
+
+def copy_absolutes(
+    observatory: str = typer.Option(..., help="Observatory code"),
+    starttime: datetime = typer.Option(
+        default=TODAY - timedelta(days=1), help="Search start time, default "
+    ),
+    endtime: datetime = typer.Option(default=TODAY),
+    metadata_token: str = typer.Option(
+        default=os.getenv("GITLAB_API_TOKEN"),
+        help="Token for metadata web service.",
+        metavar="TOKEN",
+        show_default="environment variable GITLAB_API_TOKEN",
+    ),
+    metadata_url: str = typer.Option(
+        default="https://geomag.usgs.gov/ws/secure",
+        help="URL to metadata web service",
+        metavar="URL",
+    ),
+    web_absolutes_url: str = typer.Option(
+        default="https://geomag.usgs.gov/baselines/observation.json.php",
+        help="URL to web absolutes service",
+        metavar="URL",
+    ),
+):
+    """Copy absolutes from the web absolutes service into the metadata service."""
+    # read readings from web absolutes
+    web_absolutes_factory = WebAbsolutesFactory(url=web_absolutes_url)
+    readings = get_readings(
+        factory=web_absolutes_factory,
+        observatory=observatory,
+        starttime=UTCDateTime(starttime),
+        endtime=UTCDateTime(endtime),
+    )
+    print(f"Found {len(readings)} absolutes")
+    # write readings to metadata service
+    metadata_factory = MetadataFactory(token=metadata_token, url=metadata_url)
+    with typer.progressbar(
+        iterable=readings, label="Uploading to metadata service"
+    ) as progressbar:
+        for reading in progressbar:
+            upload_reading(factory=metadata_factory, reading=reading)
+
+
+def create_reading_metadata(reading: Reading) -> Metadata:
+    """Create reading metadata object.
+
+    Parameters
+    ----------
+    reading:
+        reading to convert to metadata.
+
+    Returns
+    -------
+    metadata object for reading
+    """
+    measurement_times = [m.time for m in reading.measurements if m.time]
+    metadata = Metadata(
+        category=MetadataCategory.READING,
+        created_by=reading.metadata.get("observer", "copy_absolutes"),
+        endtime=max(measurement_times),
+        metadata=json.loads(reading.json()),
+        network="NT",
+        starttime=min(measurement_times),
+        station=reading.metadata["station"],
+        status="reviewed" if reading.metadata.get("reviewed") else "new",
+        updated_by=reading.metadata.get("reviewer"),
+    )
+    return metadata
+
+
+def get_readings(
+    factory: WebAbsolutesFactory,
+    observatory: str,
+    starttime: UTCDateTime,
+    endtime: UTCDateTime,
+) -> List[Reading]:
+    """Get readings from web absolutes.
+
+    Parameters
+    ----------
+    factory
+        configured WebAbsolutesFactory
+    observatory
+        search observatory
+    starttime
+        search start time
+    endtime
+        search end time
+
+    Returns
+    -------
+    list of found readings
+    """
+    readings = factory.get_readings(
+        observatory=observatory,
+        starttime=UTCDateTime(starttime),
+        endtime=UTCDateTime(endtime),
+        include_measurements=True,
+    )
+    return readings
+
+
+def main() -> None:
+    """Entrypoint for copy absolutes method."""
+    # for one command, can use typer.run
+    typer.run(copy_absolutes)
+
+
+def upload_reading(factory: MetadataFactory, reading: Reading) -> Metadata:
+    """Upload reading to metadata service
+
+    Parameters
+    ----------
+    factory
+        factory configured for metadata service
+    reading
+        reading to upload
+
+    Returns
+    -------
+    created metadata object
+    """
+    metadata = create_reading_metadata(reading=reading)
+    # TODO: should this check if metadata was already uploaded?
+    # TODO: should that check occur before calling this method?
+    return factory.create_metadata(metadata=metadata)
+
+
+if __name__ == "__main__":
+    main()

pyproject.toml

@@ -71,3 +71,4 @@ geomag-py = "geomagio.Controller:main"
 magproc-prepfiles = "geomagio.processing.magproc:main"
 make-cal = "geomagio.processing.make_cal:main"
 geomag-filter = "geomagio.processing.filters:main"
+copy-absolutes = "geomagio.processing.copy_absolutes:main"
\ No newline at end of file