diff --git a/geomagio/processing/copy_absolutes.py b/geomagio/processing/copy_absolutes.py index f60c214f462b239206dafd2670fdd14eebd8ac05..86d0748c591c375656d630f674806ac1bd2f1752 100644 --- a/geomagio/processing/copy_absolutes.py +++ b/geomagio/processing/copy_absolutes.py @@ -7,17 +7,24 @@ Created on Thu Jul 7 11:05:19 2022 import json import os from datetime import date, datetime, time, timedelta -from typing import List +from typing import List, Optional +from pathlib import Path import typer +from enum import Enum from obspy import UTCDateTime -from ..metadata import Metadata, MetadataCategory, MetadataFactory -from ..residual import Reading, WebAbsolutesFactory +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( @@ -40,17 +47,42 @@ def copy_absolutes( 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 into the metadata service.""" - # read readings from web absolutes - web_absolutes_factory = WebAbsolutesFactory(url=web_absolutes_url) + """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=web_absolutes_factory, + factory=factory, observatory=observatory, starttime=UTCDateTime(starttime), endtime=UTCDateTime(endtime), ) - print(f"Found {len(readings)} absolutes") + + # 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( @@ -88,17 +120,17 @@ def create_reading_metadata(reading: Reading) -> Metadata: def get_readings( - factory: WebAbsolutesFactory, + factory: ResidualFactory, observatory: str, starttime: UTCDateTime, endtime: UTCDateTime, ) -> List[Reading]: - """Get readings from web absolutes. + """Get readings from web absolutes or residual spreadsheets. Parameters ---------- factory - configured WebAbsolutesFactory + configured WebAbsolutesFactory or SpreadsheetAbsolutesFactory observatory search observatory starttime