Newer
Older
from obspy import UTCDateTime
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 SpreadsheetAbsolutesFactory, WebAbsolutesFactory
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
test_metadata = [
Metadata(
category=MetadataCategory.INSTRUMENT,
created_by="test_metadata.py",
network="NT",
station="BDT",
metadata={
"type": "FGE",
"channels": {
# each channel maps to a list of components to calculate nT
# TODO: calculate these lists based on "FGE" type
"U": [{"channel": "U_Volt", "offset": 0, "scale": 313.2}],
"V": [{"channel": "V_Volt", "offset": 0, "scale": 312.3}],
"W": [{"channel": "W_Volt", "offset": 0, "scale": 312.0}],
},
"electronics": {
"serial": "E0542",
# these scale values are used to convert voltage
"x-scale": 313.2, # V/nT
"y-scale": 312.3, # V/nT
"z-scale": 312.0, # V/nT
"temperature-scale": 0.01, # V/K
},
"sensor": {
"serial": "S0419",
# these constants combine with instrument setting for offset
"x-constant": 36958, # nT/mA
"y-constant": 36849, # nT/mA
"z-constant": 36811, # nT/mA
},
},
),
Metadata(
category=MetadataCategory.INSTRUMENT,
created_by="test_metadata.py",
network="NT",
station="NEW",
metadata={
"type": "Narod",
"channels": {
"U": [
{"channel": "U_Volt", "offset": 0, "scale": 100},
{"channel": "U_Bin", "offset": 0, "scale": 500},
],
"V": [
{"channel": "V_Volt", "offset": 0, "scale": 100},
{"channel": "V_Bin", "offset": 0, "scale": 500},
],
"W": [
{"channel": "W_Volt", "offset": 0, "scale": 100},
{"channel": "W_Bin", "offset": 0, "scale": 500},
],
},
},
),
Metadata(
category=MetadataCategory.INSTRUMENT,
created_by="test_metadata.py",
network="NT",
station="LLO",
metadata={
"type": "Narod",
"channels": {
"U": [
{"channel": "U_Volt", "offset": 0, "scale": 100},
{"channel": "U_Bin", "offset": 0, "scale": 500},
],
"V": [
{"channel": "V_Volt", "offset": 0, "scale": 100},
{"channel": "V_Bin", "offset": 0, "scale": 500},
],
"W": [
{"channel": "W_Volt", "offset": 0, "scale": 100},
{"channel": "W_Bin", "offset": 0, "scale": 500},
],
},
},
),
]
# add observatories
for observatory in OBSERVATORIES:
network = "NT"
if observatory.agency == "USGS":
network = "NT"
# rest alphabetical by agency
elif observatory.agency == "BGS":
network = "GB"
elif observatory.agency == "GSC":
network = "C2"
elif observatory.agency == "JMA":
network = "JP"
elif observatory.agency == "SANSA":
network = "AF"
test_metadata.append(
Metadata(
category=MetadataCategory.OBSERVATORY,
created_by="test_metadata.py",
network=network,
station=observatory.id,
metadata=observatory.dict(),
)
)
readings = WebAbsolutesFactory().get_readings(
observatory="BOU",
starttime=UTCDateTime("2020-01-01"),
endtime=UTCDateTime("2020-01-07"),
)
# get residual reading
reading = SpreadsheetAbsolutesFactory().parse_spreadsheet(
"etc/residual/DED-20140952332.xlsm"
)
readings.append(reading)
for reading in readings:
json_string = reading.json()
reading_dict = json.loads(json_string)
try:
reviewer = reading.metadata["reviewer"]
except KeyError:
reviewer = None
test_metadata.append(
Metadata(
category=MetadataCategory.READING,
created_by="test_metadata.py",
network="NT",
reviewed_by=reviewer,
endtime=reading.time,
station=reading.metadata["station"],
metadata=reading_dict,
metadata_valid=reading.valid,
)
)
async def load_test_metadata():
await database.connect()
for meta in test_metadata:
await metadata_table.create_metadata(meta)
await database.disconnect()
if __name__ == "__main__":
import asyncio
asyncio.run(load_test_metadata())