geocode.py

   1 #!/usr/bin/env python3
   2
   3 # © Copyright 2022, Scott Gasch
   4
   5 """Wrapper around US Census address geocoder API described here:
   6 https://www2.census.gov/geo/pdfs/maps-data/data/Census_Geocoder_User_Guide.pdf
   7 https://geocoding.geo.census.gov/geocoder/Geocoding_Services_API.pdf
   8
   9 Also try:
  10
  11     $ curl --form [email protected] \
  12            --form benchmark=2020 \
  13            https://geocoding.geo.census.gov/geocoder/locations/addressbatch \
  14            --output geocoderesult.csv
  15 """
  16
  17 import json
  18 import logging
  19 from typing import Any, Dict, List, Optional
  20
  21 import requests
  22 from requests.utils import requote_uri
  23
  24 import list_utils
  25
  26 logger = logging.getLogger(__name__)
  27
  28
  29 def geocode_address(address: str) -> Optional[Dict[str, Any]]:
  30     """Send a single address to the US Census geocoding API.  The response
  31     is a parsed JSON chunk of data with N addressMatches in the result
  32     section and the details of each match within it.  Returns None on error.
  33
  34     >>> json = geocode_address('4600 Silver Hill Rd,, 20233')
  35     >>> json['result']['addressMatches'][0]['matchedAddress']
  36     '4600 SILVER HILL RD, WASHINGTON, DC, 20233'
  37
  38     >>> json['result']['addressMatches'][0]['coordinates']
  39     {'x': -76.92743, 'y': 38.84599}
  40
  41     """
  42     url = 'https://geocoding.geo.census.gov/geocoder/geographies/onelineaddress'
  43     url += f'?address={address}'
  44     url += '&returntype=geographies&layers=all&benchmark=4&vintage=4&format=json'
  45     url = requote_uri(url)
  46     logger.debug('GET: %s', url)
  47     try:
  48         r = requests.get(url)
  49     except Exception as e:
  50         logger.exception(e)
  51         return None
  52
  53     if r.status_code != 200:
  54         logger.debug(r.text)
  55         logger.error('Unexpected response code %d, wanted 200.  Fail.', r.status_code)
  56         return None
  57     logger.debug('Response: %s', json.dumps(r.json(), indent=4, sort_keys=True))
  58     return r.json()
  59
  60
  61 def batch_geocode_addresses(addresses: List[str]):
  62     """Send up to addresses for batch geocoding.  Each line of the input
  63     list should be a single address of the form: STREET ADDRESS, CITY,
  64     STATE, ZIP.  Components may be omitted but the commas may not be.
  65     Result is an array of the same size as the input array with one
  66     answer record per line.  Returns None on error.
  67
  68     This code will deal with requests >10k addresses by chunking them
  69     internally because the census website disallows requests > 10k lines.
  70
  71     >>> batch_geocode_addresses(
  72     ...     [
  73     ...         '4600 Silver Hill Rd, Washington, DC, 20233',
  74     ...         '935 Pennsylvania Avenue, NW, Washington, DC, 20535-0001',
  75     ...         '1600 Pennsylvania Avenue NW, Washington, DC, 20500',
  76     ...         '700 Pennsylvania Avenue NW, Washington, DC, 20408',
  77     ...     ]
  78     ... )
  79     ['"1"," 4600 Silver Hill Rd,  Washington,  DC,  20233","Match","Exact","4600 SILVER HILL RD, WASHINGTON, DC, 20233","-76.92743,38.84599","76355984","L","24","033","802405","2004"', '"2"," 935 Pennsylvania Avenue,  NW,  Washington,  DC","No_Match"', '"3"," 1600 Pennsylvania Avenue NW,  Washington,  DC,  20500","Match","Exact","1600 PENNSYLVANIA AVE NW, WASHINGTON, DC, 20500","-77.03534,38.898754","76225813","L","11","001","980000","1034"', '"4"," 700 Pennsylvania Avenue NW,  Washington,  DC,  20408","Match","Exact","700 PENNSYLVANIA AVE NW, WASHINGTON, DC, 20408","-77.02304,38.89362","76226346","L","11","001","980000","1025"']
  80     """
  81
  82     n = 1
  83     url = 'https://geocoding.geo.census.gov/geocoder/geographies/addressbatch'
  84     payload = {'benchmark': '4', 'vintage': '4'}
  85     out = []
  86     for chunk in list_utils.shard(addresses, 9999):
  87         raw_file = ''
  88         for address in chunk:
  89             raw_file += f'{n}, {address}\n'
  90             n += 1
  91         files = {'addressFile': ('input.csv', raw_file)}
  92         logger.debug('POST: %s', url)
  93         try:
  94             r = requests.post(url, files=files, data=payload)
  95         except Exception as e:
  96             logger.exception(e)
  97             return None
  98
  99         if r.status_code != 200:
 100             logger.debug(r.text)
 101             logger.error('Unexpected response code %d, wanted 200.  Fail.', r.status_code)
 102             return None
 103         logger.debug('Response: %s', r.text)
 104         for line in r.text.split('\n'):
 105             line = line.strip()
 106             if len(line) > 0:
 107                 out.append(line)
 108     return out
 109
 110
 111 if __name__ == '__main__':
 112     import doctest
 113
 114     doctest.testmod()