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

@@ -10,14 +10,20 @@ from datetime import date, datetime, time, timedelta
 from typing import List
 
 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 factories(str, Enum):
+    web_absolutes_factory = "web_absolutes_factory"
+    spreadsheet_factory = "spreadsheet_factory"
+
+
 def copy_absolutes(
     observatory: str = typer.Option(..., help="Observatory code"),
     starttime: datetime = typer.Option(
@@ -40,12 +46,22 @@ def copy_absolutes(
         help="URL to web absolutes service",
         metavar="URL",
     ),
+    spreadsheets_dir: str = typer.Option(
+        default="---", help="Directory to residual spreadsheets", metavar="DIR"
+    ),
+    # I am not sure what to specify as the default directory, I was using a local directory as a test
+    factory: factories = factories.web_absolutes_factory,
 ):
-    """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."""
+    print(f"Using factory: {factory.value}")
+
+    if factory.value == factories.web_absolutes_factory:
+        factory = WebAbsolutesFactory(url=web_absolutes_url)
+    else:
+        factory = SpreadsheetAbsolutesFactory(base_directory=spreadsheets_dir)
+
     readings = get_readings(
-        factory=web_absolutes_factory,
+        factory=factory,
         observatory=observatory,
         starttime=UTCDateTime(starttime),
         endtime=UTCDateTime(endtime),
@@ -88,17 +104,14 @@ def create_reading_metadata(reading: Reading) -> Metadata:
 
 
 def get_readings(
-    factory: WebAbsolutesFactory,
-    observatory: str,
-    starttime: UTCDateTime,
-    endtime: UTCDateTime,
+    factory: factories, 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