Newer
Older
Wernle, Alexandra Nicole
committed
1
2
3
4
5
6
7
8
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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
# -*- coding: utf-8 -*-
"""
Created on Thu Jul 7 11:05:19 2022
@author: awernle
"""
import os
import json
from datetime import date, datetime, time, timedelta
# from types import NoneType
from typing import List
import typer
from obspy import UTCDateTime
"""
from ..metadata import Metadata, MetadataFactory, MetadataCategory
from ..residual import Reading, WebAbsolutesFactory
"""
from geomagio.metadata import Metadata, MetadataFactory, MetadataCategory
from geomagio.residual import Reading, WebAbsolutesFactory
TODAY = datetime.combine(date.today(), time(0, 0, 0))
NoneType = type(None)
def copy_absolutes(
observatory: str = typer.Option(..., help="Observatory code"),
starttime: datetime = typer.Option(
default=TODAY - timedelta(days=1), help="Search start time, default "
),
endtime: datetime = typer.Option(default=TODAY),
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://geomag.usgs.gov/ws/secure", # insert staging
help="URL to metadata web service",
metavar="URL",
),
web_absolutes_url: str = typer.Option(
default="https://geomag.usgs.gov/baselines/observation.json.php",
help="URL to web absolutes service",
metavar="URL",
),
) -> List[Metadata]: # make this nonetype?
"""Copy absolutes from the web absolutes service into the metadata service."""
# read readings from web absolutes
web_absolutes_factory = WebAbsolutesFactory(url=web_absolutes_url)
readings = get_readings(
factory=web_absolutes_factory,
observatory=observatory,
starttime=UTCDateTime(starttime),
endtime=UTCDateTime(endtime),
)
typer.echo(f"Found {len(readings)} absolutes")
# write readings to metadata service
metadata_factory = MetadataFactory(token=metadata_token, url=metadata_url)
with typer.progressbar(
iterable=readings, label="Uploading to metadata service"
) as progressbar:
for reading in progressbar:
upload_reading(factory=metadata_factory, reading=reading)
def create_reading_metadata(reading: Reading) -> Metadata:
"""Create reading metadata object.
Parameters
----------
reading:
reading to convert to metadata.
Returns
-------
metadata object for reading
"""
json_string = reading.json()
reading_dict = json.loads(json_string)
metadata = []
try:
reviewer = reading.metadata["reviewer"]
except KeyError:
reviewer = None
metadata.append(
Metadata(
category=MetadataCategory.READING,
created_by="absolutes_test.py",
network="NT",
updated_by=reviewer,
starttime=reading.time,
endtime=reading.time,
station=reading.metadata["station"],
metadata=reading_dict,
)
)
return metadata
# TODO: should reading starttime/endtime be for range of measurement times?
pass
def get_readings(
factory: WebAbsolutesFactory,
observatory: str,
starttime: UTCDateTime,
endtime: UTCDateTime,
) -> List[Reading]:
"""Get readings from web absolutes.
Parameters
----------
factory
configured WebAbsolutesFactory
observatory
search observatory
starttime
search start time
endtime
search end time
Returns
-------
list of found readings
"""
readings = factory.get_readings(
observatory=observatory,
starttime=UTCDateTime(starttime),
endtime=UTCDateTime(endtime),
include_measurements=True,
)
return readings
def upload_reading(factory: MetadataFactory, reading: Reading) -> Metadata:
"""Upload reading to metadata service
Parameters
----------
factory
factory configured for metadata service
reading
reading to upload
Returns
-------
created metadata object
"""
metadata = create_reading_metadata(reading=reading)
# TODO: should this check if metadata was already uploaded?
# TODO: should that check occur before calling this method?
for meta in metadata:
return factory.create_metadata(metadata=meta)
def main() -> None:
"""Entrypoint for copy absolutes method."""
# for one command, can use typer.run
typer.run(copy_absolutes)
if __name__ == "__main__":
main()