@@ -60,6 +60,7 @@ def __init__(self, *args, **kwargs):
60
60
61
61
if self .debug :
62
62
import http .client as http_client
63
+
63
64
http_client .HTTPConnection .debuglevel = 1
64
65
65
66
# You must initialize logging,
@@ -70,7 +71,9 @@ def __init__(self, *args, **kwargs):
70
71
requests_log .setLevel (logging .DEBUG )
71
72
requests_log .propagate = True
72
73
73
- def get (self , query , responseformat = "geojson" , verbosity = "body" , build = True , date = '' ):
74
+ def get (
75
+ self , query , responseformat = "geojson" , verbosity = "body" , build = True , date = ""
76
+ ):
74
77
"""Pass in an Overpass query in Overpass QL.
75
78
76
79
:param query: the Overpass QL query to send to the endpoint
@@ -89,7 +92,7 @@ def get(self, query, responseformat="geojson", verbosity="body", build=True, dat
89
92
date = datetime .fromisoformat (date )
90
93
except ValueError :
91
94
# The 'Z' in a standard overpass date will throw fromisoformat() off
92
- date = datetime .strptime (date , ' %Y-%m-%dT%H:%M:%SZ' )
95
+ date = datetime .strptime (date , " %Y-%m-%dT%H:%M:%SZ" )
93
96
# Construct full Overpass query
94
97
if build :
95
98
full_query = self ._construct_ql_query (
@@ -142,34 +145,26 @@ def _api_status() -> dict:
142
145
r = requests .get (endpoint )
143
146
lines = tuple (r .text .splitlines ())
144
147
145
- available_re = re .compile (r' \d(?= slots? available)' )
148
+ available_re = re .compile (r" \d(?= slots? available)" )
146
149
available_slots = int (
147
- next (
148
- (
149
- m .group ()
150
- for line in lines
151
- if (m := available_re .search (line ))
152
- ), 0
153
- )
150
+ next ((m .group () for line in lines if (m := available_re .search (line ))), 0 )
154
151
)
155
152
156
- waiting_re = re .compile (r' (?<=Slot available after: )[\d\-TZ:]{20}' )
153
+ waiting_re = re .compile (r" (?<=Slot available after: )[\d\-TZ:]{20}" )
157
154
waiting_slots = tuple (
158
155
datetime .strptime (m .group (), "%Y-%m-%dT%H:%M:%S%z" )
159
156
for line in lines
160
157
if (m := waiting_re .search (line ))
161
158
)
162
159
163
160
current_idx = next (
164
- i for i , word in enumerate (lines )
165
- if word .startswith ('Currently running queries' )
161
+ i
162
+ for i , word in enumerate (lines )
163
+ if word .startswith ("Currently running queries" )
166
164
)
167
- running_slots = tuple (tuple (line .split ()) for line in lines [current_idx + 1 :])
165
+ running_slots = tuple (tuple (line .split ()) for line in lines [current_idx + 1 :])
168
166
running_slots_datetimes = tuple (
169
- datetime .strptime (
170
- slot [3 ], "%Y-%m-%dT%H:%M:%S%z"
171
- )
172
- for slot in running_slots
167
+ datetime .strptime (slot [3 ], "%Y-%m-%dT%H:%M:%S%z" ) for slot in running_slots
173
168
)
174
169
175
170
return {
@@ -217,11 +212,10 @@ def slot_available_countdown(self) -> int:
217
212
return max (
218
213
ceil (
219
214
(
220
- self .slot_available_datetime -
221
- datetime .now (timezone .utc )
215
+ self .slot_available_datetime - datetime .now (timezone .utc )
222
216
).total_seconds ()
223
217
),
224
- 0
218
+ 0 ,
225
219
)
226
220
except TypeError :
227
221
# Can't subtract from None, which means slot is available now
@@ -246,7 +240,8 @@ def _construct_ql_query(self, userquery, responseformat, verbosity, date):
246
240
if responseformat == "geojson" :
247
241
template = self ._GEOJSON_QUERY_TEMPLATE
248
242
complete_query = template .format (
249
- query = raw_query , verbosity = verbosity , date = date )
243
+ query = raw_query , verbosity = verbosity , date = date
244
+ )
250
245
else :
251
246
template = self ._QUERY_TEMPLATE
252
247
complete_query = template .format (
0 commit comments