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()