google_geocoding_api.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d39ba66e",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import requests\n",
    "import logging\n",
    "import time\n",
    "import re\n",
    "import string\n",
    "import pickle\n",
    "import os"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "14c547a3",
   "metadata": {},
   "outputs": [],
   "source": [
    "API_KEY = 'YOUR_API_KEY'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9fca0027",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Backoff time sets how many minutes to wait between google pings when your API limit is hit\n",
    "BACKOFF_TIME = 30\n",
    "\n",
    "# Return Full Google Results? If True, full JSON results from Google are included in output\n",
    "RETURN_FULL_RESULTS = True\n",
    "# Include country bias to increase accuracy\n",
    "BIAS = False"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c2e63113",
   "metadata": {},
   "outputs": [],
   "source": [
    "logger = logging.getLogger(\"root\")\n",
    "logger.setLevel(logging.DEBUG)\n",
    "\n",
    "# create console handler\n",
    "ch = logging.StreamHandler()\n",
    "ch.setLevel(logging.DEBUG)\n",
    "logger.addHandler(ch)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5e5133e8",
   "metadata": {},
   "outputs": [],
   "source": [
    "input_list = ['lille', 'los angeles', 'jakarta']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2ad05806",
   "metadata": {},
   "outputs": [],
   "source": [
    "locations = pd.DataFrame(data=input_list)\n",
    "locations.columns = ['ADDRESS']\n",
    "locations.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6f43fed8",
   "metadata": {},
   "outputs": [],
   "source": [
    "output_folder  = 'YOUR_OUTPUT_FOLDER'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ce7417c5",
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_google_results(address, api_key=None, return_full_response=False):\n",
    "    \"\"\"\n",
    "    Get geocode results from Google Maps Geocoding API.\n",
    "    \n",
    "    Note, that in the case of multiple google geocode reuslts, this function returns details of the FIRST result.\n",
    "    \n",
    "    @param address: String address as accurate as possible. For Example \"18 Grafton Street, Dublin, Ireland\"\n",
    "    @param api_key: String API key if present from google. \n",
    "                    If supplied, requests will use your allowance from the Google API. If not, you\n",
    "                    will be limited to the free usage of 2500 requests per day.\n",
    "    @param return_full_response: Boolean to indicate if you'd like to return the full response from google. This\n",
    "                    is useful if you'd like additional location details for storage or parsing later.\n",
    "    \"\"\"\n",
    "    # Set up your Geocoding url\n",
    "    geocode_url = \"https://maps.googleapis.com/maps/api/geocode/json?address={}\".format(address)\n",
    "    if api_key is not None:\n",
    "        geocode_url = geocode_url + \"&key={}\".format(api_key)\n",
    "        \n",
    "    # Ping google for the reuslts:\n",
    "    results = requests.get(geocode_url)\n",
    "    # Results will be in JSON format - convert to dict using requests functionality\n",
    "    results = results.json()\n",
    "    \n",
    "    # if there's no results or an error, return empty results.\n",
    "    if len(results['results']) == 0:\n",
    "        output = {\n",
    "            \"formatted_address\" : None,\n",
    "            \"latitude\": None,\n",
    "            \"longitude\": None,\n",
    "            \"accuracy\": None,\n",
    "            \"google_place_id\": None,\n",
    "            \"type\": None,\n",
    "            \"postcode\": None\n",
    "        }\n",
    "    else:    \n",
    "        answer = results['results'][0]\n",
    "        output = {\n",
    "            \"formatted_address\" : answer.get('formatted_address'),\n",
    "            \"latitude\": answer.get('geometry').get('location').get('lat'),\n",
    "            \"longitude\": answer.get('geometry').get('location').get('lng'),\n",
    "            \"accuracy\": answer.get('geometry').get('location_type'),\n",
    "            \"google_place_id\": answer.get(\"place_id\"),\n",
    "            \"type\": \",\".join(answer.get('types')),\n",
    "            \"postcode\": \",\".join([x['long_name'] for x in answer.get('address_components') \n",
    "                                  if 'postal_code' in x.get('types')])\n",
    "        }\n",
    "        \n",
    "    # Append some other details:    \n",
    "    output['input_string'] = address\n",
    "    output['number_of_results'] = len(results['results'])\n",
    "    output['status'] = results.get('status')\n",
    "    if return_full_response is True:\n",
    "        output['response'] = results\n",
    "    \n",
    "    return output"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "03031f5b",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Ensure, before we start, that the API key is ok/valid, and internet access is ok\n",
    "test_result = get_google_results(\"London, England\", API_KEY, RETURN_FULL_RESULTS)\n",
    "if (test_result['status'] != 'OK') or (test_result['formatted_address'] != 'London, UK'):\n",
    "    logger.warning(\"There was an error when testing the Google Geocoder.\")\n",
    "    raise ConnectionError('Problem with test results from Google Geocode - check your API key and internet connection.')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "31e049ce",
   "metadata": {},
   "outputs": [],
   "source": [
    "test_result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d4dc284d",
   "metadata": {},
   "outputs": [],
   "source": [
    "geocoded_addresses = list()\n",
    "# Add existing addresses from the output text file to the set\n",
    "if os.path.exists('YOUR_OUTPUT_FOLDER/geocoding_results.txt'):\n",
    "    with open('YOUR_OUTPUT_FOLDER/geocoding_results.txt', 'r') as file:\n",
    "        for line in file:\n",
    "            try:\n",
    "                address, _ = line.strip().split(';-;')\n",
    "                geocoded_addresses.append(address)\n",
    "            except Exception as e:\n",
    "                print(line)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "698609ed",
   "metadata": {},
   "outputs": [],
   "source": [
    "len(geocoded_addresses)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7ed685f2",
   "metadata": {},
   "outputs": [],
   "source": [
    "geocoded_addresses"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "dae8d2ba",
   "metadata": {},
   "outputs": [],
   "source": [
    "locations.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cd576b50",
   "metadata": {},
   "outputs": [],
   "source": [
    "locations.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c4fe826a",
   "metadata": {},
   "outputs": [],
   "source": [
    "geocoded_addresses"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "823783c4",
   "metadata": {},
   "outputs": [],
   "source": [
    "locations['ADDRESS'][0] in geocoded_addresses"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "08dfb394",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# Create a list to hold results\n",
    "results = []\n",
    "\n",
    "# Go through each address in turn\n",
    "for idx,location in locations.iterrows():\n",
    "    if location['ADDRESS'] not in geocoded_addresses:\n",
    "        # While the address geocoding is not finished:\n",
    "        geocoded = False\n",
    "\n",
    "        while geocoded is not True:\n",
    "\n",
    "            # Geocode the address with google\n",
    "            try:\n",
    "\n",
    "                geocode_result = get_google_results(location['ADDRESS'], \n",
    "                                                    API_KEY, \n",
    "                                                    return_full_response=RETURN_FULL_RESULTS)\n",
    "\n",
    "#                 geocode_result['raw_string'] = location['LOCATION']\n",
    "\n",
    "            except Exception as e:\n",
    "\n",
    "                logger.exception(e)\n",
    "\n",
    "                logger.error(\"Major error with {}\".format(location['ADDRESS']))\n",
    "\n",
    "                logger.error(\"Skipping!\")\n",
    "\n",
    "                geocoded = True\n",
    "\n",
    "            # If we're over the API limit, backoff for a while and try again later.\n",
    "            if geocode_result['status'] == 'OVER_QUERY_LIMIT':\n",
    "\n",
    "                logger.info(\"Hit Query Limit! Backing off for a bit.\")\n",
    "\n",
    "                time.sleep(BACKOFF_TIME * 60) # sleep for 30 minutes\n",
    "\n",
    "                geocoded = False\n",
    "\n",
    "            else:\n",
    "\n",
    "                # If we're ok with API use, save the results\n",
    "                # Note that the results might be empty / non-ok - log this\n",
    "                if geocode_result['status'] != 'OK':\n",
    "\n",
    "                    logger.warning(\"Error geocoding {}: {}\".format(location['ADDRESS'], geocode_result['status']))\n",
    "\n",
    "                logger.debug(\"Geocoded: {}: {}\".format(location['ADDRESS'], geocode_result['status']))\n",
    "\n",
    "                results.append(geocode_result)           \n",
    "\n",
    "                geocoded = True\n",
    "        # Save the geocoding result to the text file\n",
    "        with open('YOUR_OUTPUT_FOLDER/geocoding_results.txt', 'a') as file:\n",
    "            file.write(f\"{location['ADDRESS']};-;{geocode_result}\\n\")\n",
    "        # Print status every 100 addresses\n",
    "        if len(results) % 100 == 0:\n",
    "\n",
    "            logger.info(\"Completed {} of {} address\".format(len(results), len(locations)))\n",
    "\n",
    "        # Every 100 addresses, save progress to file(in case of a failure so you have something!)\n",
    "    #     if len(results) % 1000 == 0:\n",
    "\n",
    "    #         pd.DataFrame(results).to_pickle(os.path.join(output_folder, ))\n",
    "\n",
    "# All done\n",
    "logger.info(\"Finished geocoding all addresses\")"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}