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
|
'''Test Script/CLI Interface to pyairnow'''
import argparse
import asyncio
from pyairnow import WebServiceAPI
def optionalArgs(args):
'''Load optional keyword arguments (date and dict)'''
kwargs = dict()
if args.date is not None:
kwargs['date'] = args.date
if args.distance is not None:
kwargs['distance'] = args.distance
return kwargs
def llArgs(args):
'''Load latitude and longitude from the location argument'''
try:
lat, long = args.split(',')
except ValueError:
raise ValueError(
'Invalid location string, expected <latitude>,<longitude>'
)
return [float(lat.strip()), float(long.strip())]
async def get_forecast_zip(args: argparse.Namespace) -> None:
'''Print forecast information for a zip code'''
client = WebServiceAPI(args.api_key)
data = await client.forecast.zipCode(args.zipcode, **optionalArgs(args))
print(data)
async def get_forecast_ll(args: argparse.Namespace) -> None:
'''Print forecast information for a latitude and longitude'''
client = WebServiceAPI(args.api_key)
data = await client.forecast.latLong(*llArgs(args), **optionalArgs(args))
print(data)
async def get_current_zip(args: argparse.Namespace) -> None:
'''Print current observation information for a zip code'''
client = WebServiceAPI(args.api_key)
data = await client.observations.zipCode(
args.zipcode,
**optionalArgs(args),
)
print(data)
async def get_current_ll(args: argparse.Namespace) -> None:
'''Print current observation information for a latitude and longitude'''
client = WebServiceAPI(args.api_key)
data = await client.observations.latLong(
*llArgs(args),
**optionalArgs(args),
)
print(data)
if __name__ == '__main__':
parser = argparse.ArgumentParser(
description='CLI Client for pyAirNow, which obtains current and '
'forecasted air quality information from AirNow. The '
'returned JSON data is printed to stdout.',
epilog='To obtain an API key, visit https://docs.airnowapi.org/login',
)
parser.add_argument(
'-k', '--api-key',
required=True,
help='AirNow API Key (required)',
)
location_group = parser.add_mutually_exclusive_group(required=True)
location_group.add_argument(
'-z', '--zipcode',
help='Zip Code for Observations or Forecast. May not be used with '
'--location.',
)
location_group.add_argument(
'-l', '--location',
help='Latitude and Longitude for Observations or Forecast (must be '
'formatted as "<latitude>,<longitude>" with an optional space '
'before the longitude value. May not be used with --zipcode.'
)
parser.add_argument(
'-r', '--distance',
help='If no reporting area is associated with the zip code or '
'latitude and longitude, return data from a station within this '
'distance (in miles)'
)
parser.add_argument(
'-d', '--date',
help='Desired forecast date. If omitted, the current day forecast is '
'returned. Format as yyyy-mm-dd'
)
parser.add_argument(
'type', choices=['forecast', 'observations'],
help='Look up either the forecast or the current observations for the '
'given location'
)
args = parser.parse_args()
loop = asyncio.get_event_loop()
if args.type == 'forecast':
if args.zipcode is not None:
loop.run_until_complete(get_forecast_zip(args))
else:
loop.run_until_complete(get_forecast_ll(args))
else:
if args.zipcode is not None:
loop.run_until_complete(get_current_zip(args))
else:
loop.run_until_complete(get_current_ll(args))
|
