Skip to content
Snippets Groups Projects

Add spreadsheet factory functionality to copy_absolutes.py

1 file
+ 21
13
Compare changes
  • Side-by-side
  • Inline
@@ -7,7 +7,8 @@ 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
@@ -19,9 +20,9 @@ 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"
class ResidualFactory(str, Enum):
SPREADSHEET = "spreadsheet"
WEB_ABSOLUTES = "webabsolutes"
def copy_absolutes(
@@ -46,19 +47,23 @@ def copy_absolutes(
help="URL to web absolutes service",
metavar="URL",
),
spreadsheets_dir: str = typer.Option(
default="---", help="Directory to residual spreadsheets", metavar="DIR"
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",
),
# 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,
factory: ResidualFactory = ResidualFactory.WEB_ABSOLUTES,
):
"""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:
if factory.value == ResidualFactory.WEB_ABSOLUTES:
factory = WebAbsolutesFactory(url=web_absolutes_url)
else:
factory = SpreadsheetAbsolutesFactory(base_directory=spreadsheets_dir)
factory = SpreadsheetAbsolutesFactory(base_directory=directory)
readings = get_readings(
factory=factory,
@@ -104,7 +109,10 @@ def create_reading_metadata(reading: Reading) -> Metadata:
def get_readings(
factory: factories, observatory: str, starttime: UTCDateTime, endtime: UTCDateTime
factory: ResidualFactory,
observatory: str,
starttime: UTCDateTime,
endtime: UTCDateTime,
) -> List[Reading]:
"""Get readings from web absolutes or residual spreadsheets.
Loading