-
Notifications
You must be signed in to change notification settings - Fork 5
/
bulk_submitter.py
160 lines (124 loc) · 4.24 KB
/
bulk_submitter.py
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
#!/usr/bin/python
# The following Python code is an example of how to submit geocode
# requests to the synchronous bulk geocoding API using multiple
# python threads <MAX_WORKERS> to make simultaneous requests, with
# each request containing <GEOCODES_PER_REQUEST> addresses to geocode.
import concurrent.futures
import copy
import csv
import requests
import statistics
import sys
import threading
import time
import io
# Uncomment the appropriate SERVICE_URL for the target environment
# Dev
SERVICE_URL = 'https://ssl.refractions.net/ols/pub/geocoder/'
# Test
# SERVICE_URL = 'https://geocodertst.api.gov.bc.ca/'
# Production
#SERVICE_URL = 'https://geocoder.api.gov.bc.ca/'
REQUEST_URL = SERVICE_URL + 'addresses/bulk.csv'
# Setup parameters for job submission
headers = {'Accept': 'text/csv'}
global_params = {'outputSrs': '4326',
'interpolation': 'adaptive',
'locationDescriptor': 'any'}
GEOCODES_PER_REQUEST = 10
MAX_WORKERS = 4
fileHeader = 'yourId,addressString\n'
seq_num = 0
request_times = []
def log(msg):
print((time.strftime("%Y/%m/%d|%H:%M:%S|") + msg))
# reads GEOCODES_PER_REQUEST rows of input, sends them in a geocode request, and writes the results
def geocode(dictReader, readLock, outFile, writeLock):
global seq_num, request_times
#log('running geocode...')
buffer = io.StringIO()
buffer.write(fileHeader)
moreToRead = True
num_read = 0
#log('acquiring readLock...')
with readLock:
#log('acquired readLock...')
start_seq = seq_num
for i in range(GEOCODES_PER_REQUEST):
#log('i = ' + str(i))
try:
row = next(dictReader)
seq_num += 1
num_read += 1
if row['yourId']:
buffer.write(','.join([row['yourId'], row['addressString']]) + '\n')
else:
buffer.write(','.join(['', row['addressString']]) + '\n')
except StopIteration:
moreToRead = False
break
# if there were no rows to read at all, there is no request to make, return now
if len(buffer.getvalue()) == len(fileHeader):
return False
buffer.seek(0)
files = {'file': buffer}
request_start = time.time()
log(f'Sending request for {start_seq}-{start_seq+num_read-1}')
params = global_params.copy()
params['startSeqNum'] = start_seq
req = requests.post(REQUEST_URL, files=files, headers=headers, data=params)
resultBuffer = io.StringIO(req.text)
resultBuffer.seek(0)
request_time = time.time() - request_start
with writeLock:
request_times.append(request_time)
# if we aren't writing the first line, skip the header line
if outFile.tell() != 0:
resultBuffer.readline()
outFile.write(resultBuffer.read())
return moreToRead
# keeps geocoding until it runs out of input
def geocodeLoop(dictReader, readLock, outFile, writeLock):
keepGoing = True
while keepGoing:
keepGoing = geocode(dictReader, readLock, outFile, writeLock)
def main(args):
# Get parameters from the command-line
if len(args) != 3:
print(('Usage: <python> bulk_submitter.py ' +
'<infile> <outfile>'))
sys.exit(1)
inFileName = args[1]
outFileName = args[2]
t0 = time.time()
# open the input file
csvFile = open(inFileName, mode='r', newline='')
reader = csv.DictReader(csvFile)
if 'addressString' not in reader.fieldnames:
print('Input CSV must have an \'addressString\' column; columns found:')
print(reader.fieldnames)
sys.exit(1)
readLock = threading.Lock()
# open the output file
outFile = open(outFileName, mode='w')
writeLock = threading.Lock()
print('Bulk Geocoding Submission')
print(f'Input File name: {inFileName}')
print(f'Worker Threads: {MAX_WORKERS}')
print(f'Geocodes per Request: {GEOCODES_PER_REQUEST}')
print(f'API URL: {REQUEST_URL}')
print(f'Output File name: {outFileName}')
# run MAX_WORKERS geocodeLoops in MAX_WORKERS threads until they all complete
with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
for i in range(MAX_WORKERS):
executor.submit(geocodeLoop, reader, readLock, outFile, writeLock)
# clean up
csvFile.close()
outFile.close()
t1 = time.time()
print(f'Total Time taken: {t1-t0} seconds')
print(f'Average Time per request {round(statistics.mean(request_times),3)} seconds')
print(f'Max Time per request {round(max(request_times),3)} seconds')
print(f'Min Time per request {round(min(request_times),3)} seconds')
if __name__ == '__main__':
main(sys.argv)