diff --git a/geomagio/processing/copy_instrument.py b/geomagio/processing/copy_instrument.py new file mode 100644 index 0000000000000000000000000000000000000000..dab5240e483c3501d6a4c8d503e4c662aa337ba1 --- /dev/null +++ b/geomagio/processing/copy_instrument.py @@ -0,0 +1,170 @@ +# -*- coding: utf-8 -*- +""" +Created on Thu Jul 7 11:05:19 2022 + +@author: awernle +""" +import os +from datetime import date, datetime, time, timedelta +import json + +import typer +from obspy import UTCDateTime +from typing import Dict, Optional + +from ..metadata.Instrument.Instrument import Instrument, Offsets, Sensor, Electronics +from ..metadata import Metadata, MetadataFactory, MetadataCategory + +TODAY = datetime.combine(date.today(), time(0, 0, 0)) + + +# TODO: edit the metadata categories? +# TODO: add function to check if metadata already exists with the option to update values +def copy_instrument( + observatory: str = typer.Option(..., help="Observatory code"), + starttime: datetime = typer.Option( + default=TODAY - timedelta(days=1), help="Default is 1 day from now" + ), + endtime: datetime = typer.Option(default=None), + offsets_json: Optional[str] = typer.Option( + default=None, + help='JSON string representing Offsets metadata (refer to Instrument.py for detailed attributes). CLI example usage: --offsets-json \'{"u_offset": 10, "v_offset": 20, "w_offset": 30}\'', + ), + sensor_json: Optional[str] = typer.Option( + default=None, + help='JSON string representing Sensor metadata (refer to Instrument.py for detailed attributes). CLI example usage: --offsets-json \'{"type":"FGE", "u_constant": 10, "v_constant": 20, "w_constant": 30}\'', + ), + electronics_json: Optional[str] = typer.Option( + default=None, + help='JSON string representing Electronics metadata (refer to Instrument.py for detailed attributes). CLI example usage: --offsets-json \'{"type": str, "resistance": float, "temperature_scale": float, "u_bin": 10, "v_bin": 20, "w_bin": 30}\'', + ), + 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://staging-geomag.cr.usgs.gov/ws/secure/metadata", + help="URL to metadata web service", + metavar="URL", + ), +): + """Copy instrument metadata into the metadata webservice.""" + + created_by_label = "copy_instrument.py" + + instrument_metadata = [] + + # deserialize JSON strings and create metadata if it exists + for metadata_type, json_str in { + "offsets": offsets_json, + "sensor": sensor_json, + "electronics": electronics_json, + }.items(): + if json_str: + metadata_dict = json.loads(json_str) + metadata_obj = create_metadata( + observatory=observatory, + metadata_type=metadata_type, + metadata=metadata_dict, + starttime=starttime, + endtime=endtime, + created_by_label=created_by_label, + ) + instrument_metadata.append(metadata_obj) + + print(instrument_metadata) + + # confirm whether or not to copy instrument metadata + typer.confirm(f"Are you sure you want to copy instrument metadata?", abort=True) + print("Copying over instrument metadata") + + # write instrument metadata to metadata service + metadata_factory = MetadataFactory(token=metadata_token, url=metadata_url) + with typer.progressbar( + iterable=instrument_metadata, label="Uploading to metadata service" + ) as progressbar: + for instrument in progressbar: + upload_instrument_metadata(factory=metadata_factory, instrument=instrument) + + +def create_metadata( + observatory: str, + metadata_type: str, + metadata: dict, + starttime: UTCDateTime, + endtime: UTCDateTime, + created_by_label: str, +) -> Metadata: + """Create instrument metadata object. + + Parameters + ---------- + observatory: + observatory that metadata applies to. + metadata_type: + type of metadata to create. + metadata: + metadata information to convert to metadata. + starttime: + when the metadata should start. + endtime: + when the metadata should end. + created_by_label: + The script that created the metadata. + + Returns + ------- + Instrument metadata object. + """ + metadata_class = None + if metadata_type == "offsets": + metadata_class = Offsets + elif metadata_type == "sensor": + metadata_class = Sensor + elif metadata_type == "electronics": + metadata_class = Electronics + + if metadata_class: + metadata_obj = Metadata( + category=MetadataCategory.INSTRUMENT, + created_by=created_by_label, + endtime=endtime, + metadata=metadata_class(**metadata), + network="NT", + starttime=starttime, + station=observatory, + ) + return metadata_obj + else: + raise ValueError(f"Invalid metadata type: {metadata_type}") + + +def main() -> None: + """Entrypoint for copy observatory method.""" + # for one command, can use typer.run + typer.run(copy_instrument) + + +def upload_instrument_metadata( + factory: MetadataFactory, instrument: Metadata +) -> Metadata: + """Upload instrument metadata to metadata service. + + Parameters + ---------- + factory + factory configured for metadata service. + instrument + instrument metadata to be uploaded. + + Returns + ------- + created metadata object. + """ + return factory.create_metadata(instrument=instrument) + + +if __name__ == "__main__": + main()