Insert sample readings into test metadata

geomagio/residual/Reading.py

@@ -2,7 +2,7 @@ import collections
 from typing import Dict, List, Optional
 from typing_extensions import Literal
 
-from obspy import Stream
+from obspy import Stream, UTCDateTime
 from pydantic import BaseModel
 
 from .. import TimeseriesUtility
@@ -97,3 +97,34 @@ class Reading(BaseModel):
                 time=measurement.time,
                 default=default_existing and measurement.f or None,
             )
+
+    @property
+    def time(self) -> Optional[UTCDateTime]:
+        h = self.get_absolute("H")
+        if h:
+            return h.endtime
+        return None
+
+    @property
+    def valid(self) -> bool:
+        """ensures that readings used in calculations have been marked as valid
+
+        Attributes
+        ----------
+        readings: list containing valid and invalid readings
+        """
+        if (
+            self.get_absolute("D").valid == True
+            and self.get_absolute("H").valid == True
+            and self.get_absolute("Z").valid == True
+        ):
+            return True
+
+    def get_absolute(
+        self,
+        element: str,
+    ) -> Optional[Absolute]:
+        for absolute in self.absolutes:
+            if absolute.element == element:
+                return absolute
+        return None

test_metadata.py

+import json
 from obspy import UTCDateTime
 
+from geomagio.residual.SpreadsheetAbsolutesFactory import SpreadsheetAbsolutesFactory
+from geomagio import metadata
 from geomagio.api.db import database, metadata_table
 from geomagio.api.ws.Observatory import OBSERVATORIES
 from geomagio.metadata import Metadata, MetadataCategory
+from geomagio.residual import WebAbsolutesFactory, MeasurementType
 
 
 test_metadata = [
@@ -109,6 +113,49 @@ for observatory in OBSERVATORIES:
         )
     )
 
+# add null readings
+starttime = UTCDateTime("2020-01-01")
+endtime = UTCDateTime("2020-01-07")
+
+readings = WebAbsolutesFactory().get_readings(
+    observatory="BOU",
+    starttime=UTCDateTime("2020-01-01"),
+    endtime=UTCDateTime("2020-01-07"),
+)
+
+for reading in readings:
+    json_string = reading.json()
+    reading_dict = json.loads(json_string)
+    test_metadata.append(
+        Metadata(
+            category=MetadataCategory.READING,
+            created_by="test_metadata.py",
+            reviewed_by=reading.metadata["reviewer"],
+            starttime=reading.time,
+            station="BOU",
+            metadata=reading_dict,
+            metadata_valid=reading.valid,
+        )
+    )
+
+reading = SpreadsheetAbsolutesFactory().parse_spreadsheet(
+    "etc/residual/DED-20140952332.xlsm"
+)
+
+json_string = reading.json()
+reading_dict = json.loads(json_string)
+
+test_metadata.append(
+    Metadata(
+        category=MetadataCategory.READING,
+        created_by="test_metadata.py",
+        starttime=reading.time,
+        station="DED",
+        metadata=reading_dict,
+        metadata_valid=reading.valid,
+    )
+)
+
 
 async def load_test_metadata():
     await database.connect()