From a9d3ea16bfdc8fb411358a05231465cf46a89bfa Mon Sep 17 00:00:00 2001
From: Jeremy Fee <jmfee@usgs.gov>
Date: Fri, 30 Sep 2022 12:16:26 -0600
Subject: [PATCH] Clean up copy_absolutes
---
geomagio/processing/copy_absolutes.py | 24 +++++++++---------------
1 file changed, 9 insertions(+), 15 deletions(-)
diff --git a/geomagio/processing/copy_absolutes.py b/geomagio/processing/copy_absolutes.py
index bd3efb5b3..f60c214f4 100644
--- a/geomagio/processing/copy_absolutes.py
+++ b/geomagio/processing/copy_absolutes.py
@@ -4,16 +4,15 @@ Created on Thu Jul 7 11:05:19 2022
@author: awernle
"""
-import os
import json
+import os
from datetime import date, datetime, time, timedelta
-
from typing import List
import typer
from obspy import UTCDateTime
-from ..metadata import Metadata, MetadataFactory, MetadataCategory
+from ..metadata import Metadata, MetadataCategory, MetadataFactory
from ..residual import Reading, WebAbsolutesFactory
TODAY = datetime.combine(date.today(), time(0, 0, 0))
@@ -51,7 +50,7 @@ def copy_absolutes(
starttime=UTCDateTime(starttime),
endtime=UTCDateTime(endtime),
)
- typer.echo(f"Found {len(readings)} absolutes")
+ print(f"Found {len(readings)} absolutes")
# write readings to metadata service
metadata_factory = MetadataFactory(token=metadata_token, url=metadata_url)
with typer.progressbar(
@@ -73,25 +72,20 @@ def create_reading_metadata(reading: Reading) -> Metadata:
-------
metadata object for reading
"""
- json_string = reading.json()
- reading_dict = json.loads(json_string)
- reviewer = reading.metadata.get["reviewer"]
+ measurement_times = [m.time for m in reading.measurements if m.time]
metadata = Metadata(
category=MetadataCategory.READING,
created_by=reading.metadata.get("observer", "copy_absolutes"),
+ endtime=max(measurement_times),
+ metadata=json.loads(reading.json()),
network="NT",
- updated_by=reviewer,
- status="reviewed" if reading.metadata.get("reviewed") else "new",
- starttime=min([m.time for m in reading.measurements if m.time]),
- endtime=max([m.time for m in reading.measurements if m.time]),
+ starttime=min(measurement_times),
station=reading.metadata["station"],
- metadata=reading_dict,
+ status="reviewed" if reading.metadata.get("reviewed") else "new",
+ updated_by=reading.metadata.get("reviewer"),
)
-
return metadata
- pass
-
def get_readings(
factory: WebAbsolutesFactory,
--
GitLab