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, Optional
from pathlib import Path

import typer
from enum import Enum
from obspy import UTCDateTime

from ..metadata import Metadata, MetadataFactory, MetadataCategory
from ..residual import Reading, WebAbsolutesFactory, SpreadsheetAbsolutesFactory

TODAY = datetime.combine(date.today(), time(0, 0, 0))


class ResidualFactory(str, Enum):
    SPREADSHEET = "spreadsheet"
    WEB_ABSOLUTES = "webabsolutes"


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",
    ),
    directory: Optional[Path] = typer.Option(
        None,
        exists=True,
        file_okay=False,
        dir_okay=True,
        writable=False,
        readable=False,
        resolve_path=True,
        help="Residual spreadsheet directory",
    ),
    factory: ResidualFactory = ResidualFactory.WEB_ABSOLUTES,
    force: bool = typer.Option(
        default=False,
        help="Force skip the check to copy absolutes",
    ),
):
    """Copy absolutes from the web absolutes service OR residual spreadsheets into the metadata service."""
    if factory.value == ResidualFactory.WEB_ABSOLUTES:
        factory = WebAbsolutesFactory(url=web_absolutes_url)
    else:
        factory = SpreadsheetAbsolutesFactory(base_directory=directory)

    readings = get_readings(
        factory=factory,
        observatory=observatory,
        starttime=UTCDateTime(starttime),
        endtime=UTCDateTime(endtime),
    )

    # confirm whether or not to copy absolutes
    if not force:
        typer.confirm(
            f"Are you sure you want to copy {len(readings)} absolutes?", abort=True
        )
        print("Copying over 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: ResidualFactory,
    observatory: str,
    starttime: UTCDateTime,
    endtime: UTCDateTime,
) -> List[Reading]:
    """Get readings from web absolutes or residual spreadsheets.

    Parameters
    ----------
    factory
        configured WebAbsolutesFactory or SpreadsheetAbsolutesFactory
    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()