FilesExpand file tree

 master

/

batch.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

150 lines (118 loc) · 4.54 KB

 master

/

batch.py

Copy path

Top

File metadata and controls

• Code
• Blame

150 lines (118 loc) · 4.54 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

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

#!/usr/bin/env python3

# Background tutorial on async programming with Python

# https://realpython.com/async-io-python/

# Requires Python 3.7 or newer. Tested with 3.8 and 3.9.

# Installation:

# pip3 install opencage asyncio aiohttp backoff

import sys, random, time

import csv

import backoff

import asyncio

from opencage.geocoder import OpenCageGeocode, AioHttpError

api_key = ''

infile = 'file_to_geocode.csv'

outfile = 'file_geocoded.csv'

max_items = 100 # Set to 0 for unlimited

num_workers = 3 # For 10 requests per second try 2-5

timeout = 5 # For individual HTTP requests. In seconds, default is 1

retry_max_tries = 10 # How often to retry if a HTTP request times out

retry_max_time = 60 # Limit in seconds for retries

csv_writer = csv.writer(open(outfile, 'w', newline=''))

async def write_one_geocoding_result(geocoding_result, address, address_id):

if geocoding_result != None:

geocoding_result = geocoding_result[0]

row = [

address_id,

geocoding_result['geometry']['lat'],

geocoding_result['geometry']['lng'],

# Any of these components might be empty :

geocoding_result['components'].get('country', ''),

geocoding_result['components'].get('county', ''),

geocoding_result['components'].get('city', ''),

geocoding_result['components'].get('postcode', ''),

geocoding_result['components'].get('road', ''),

geocoding_result['components'].get('house_number', ''),

geocoding_result['confidence'],

geocoding_result['formatted']

]

else:

row = [

address_id,

0, # not to be confused with https://en.wikipedia.org/wiki/Null_Island

0,

'',

'',

'',

'',

'',

'',

-1, # confidence values are 1-10 (lowest to highest), use -1 for unknown

''

]

sys.stderr.write("not found, writing empty result: %s\n" % address)

csv_writer.writerow(row)

# Backing off 0.4 seconds afters 1 tries calling function <function geocode_one_address

# at 0x10dbf5e50> with args ('14464 3RD ST # 4, 91423, CA, USA', '1780245') and kwargs {}

def backoff_hdlr(details):

sys.stderr.write("Backing off {wait:0.1f} seconds afters {tries} tries "

"calling function {target} with args {args} and kwargs "

"{kwargs}\n".format(**details))

# https://pypi.org/project/backoff/

@backoff.on_exception(backoff.expo,

(asyncio.TimeoutError),

max_time=retry_max_time, # seconds

max_tries=retry_max_tries,

on_backoff=backoff_hdlr)

async def geocode_one_address(address, address_id):

async with OpenCageGeocode(api_key) as geocoder:

geocoding_result = await geocoder.geocode_async(address)

try:

await write_one_geocoding_result(geocoding_result, address, address_id)

except Exception as e:

sys.stderr.write(e)

async def run_worker(worker_name, queue):

sys.stderr.write("Worker %s starts...\n" % worker_name)

while True:

work_item = await queue.get()

address_id = work_item['id']

address = work_item['address']

await geocode_one_address(address, address_id)

queue.task_done()

async def main():

assert sys.version_info >= (3, 7), "Script requires Python 3.7+."

## 1. Read CSV into a Queue

## Each work_item is an address an id. The id will be part of the output,

## easy to add more settings. Named 'work_item' to avoid the words

## 'address' or 'task' which are used elsewhere

##

## https://docs.python.org/3/library/asyncio-queue.html

##

queue = asyncio.Queue(maxsize=max_items)

csv_reader = csv.reader(open(infile, 'r'))

for row in csv_reader:

work_item = {'id': row[0], 'address': row[1]}

await queue.put(work_item)

if queue.full():

break

sys.stderr.write("%d work_items in queue\n" % queue.qsize())

## 2. Create tasks workers. That is coroutines, each taks take work_items

## from the queue until it's empty. Tasks run in parallel

##

## https://docs.python.org/3/library/asyncio-task.html#creating-tasks

## https://docs.python.org/3/library/asyncio-task.html#coroutine

##

sys.stderr.write("Creating %d task workers...\n" % num_workers)

tasks = []

for i in range(num_workers):

task = asyncio.create_task(run_worker(f'worker {i}', queue))

tasks.append(task)

## 3. Now workers do the geocoding

##

sys.stderr.write("Now waiting for workers to finish processing queue...\n")

await queue.join()

## 4. Cleanup

##

for task in tasks:

task.cancel()

sys.stderr.write("All done.\n")

asyncio.run(main())