Add spreadsheet factory functionality to copy_absolutes.py

Imported Enum to define a CLI parameter with a predefined set of values to choose from. Created a spreadsheets_dir option and factory option.

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