Friday, September 16, 2022

Batch Geocoding and Reverse Geocoding using HERE API

A python script for batch geocoding and reverse geocoding using HERE API

HERE API service allows you to submit batch geocoding and reverse geocoding requests.  Your submission must conform with the Input Data guidelines. The Batch Geocoder API handles the geocoding and reverse geocoding asynchronously.

As stated on the API doc page linked above, to retrieve the output of a successful batch request, you must follow the steps below:-

  1. Upload your data with a POST request to the resource jobs.
  2. Using the RequestId value contained in the response to your data upload request, check the status of the job with a GET request. You can only download the results when the job status is completed
  3. Using the RequestId value contained in the response to your data upload request, download the results by sending a GET request.


Batch Geocoding Python Script

Input Data:


Python Script:

import requests
import json
import time
import zipfile
import io
from bs4 import BeautifulSoup
import glob

import pandas as pd


mykey = 'xxxxxxxxxxxxxx' ## Register a HERE MAPS DEVELOPER API

class Batch:
    SERVICE_URL = "https://batch.geocoder.ls.hereapi.com/6.2/jobs"
    jobId = None
    
    
    def __init__(self, apikey=""): ## use a HERE MAPS DEVELOPER API
        self.apikey = apikey
        
            
    def start(self, filename='testfile.csv', indelim=",", outdelim=","):
        
        file = open(filename, 'rb')

        params = {
            "action": "run",
            "apiKey": self.apikey,
            "politicalview":"RUS",
            "gen": 9,
            "maxresults": "1",
            "header": "true",
            "indelim": indelim,
            "outdelim": outdelim,
            "outcols": "displayLatitude,displayLongitude,locationLabel,houseNumber,street,district,city,postalCode,county,state,country",
            "outputcombined": "true",
        }

        response = requests.post(self.SERVICE_URL, params=params, data=file)
        self.__stats (response)
        file.close()
    

    def status (self, jobId = None):

        if jobId is not None:
            self.jobId = jobId
        
        statusUrl = self.SERVICE_URL + "/" + self.jobId
        
        params = {
            "action": "status",
            "apiKey": self.apikey,
        }
        
        response = requests.get(statusUrl, params=params)
        self.__stats (response)
        

    def result (self, jobId = None):

        if jobId is not None:
            self.jobId = jobId
        
        print("Requesting result data ...")
        
        resultUrl = self.SERVICE_URL + "/" + self.jobId + "/result"
        
        params = {
            "apiKey": self.apikey
        }
        
        response = requests.get(resultUrl, params=params, stream=True)
        
        if (response.ok):    
            zipResult = zipfile.ZipFile(io.BytesIO(response.content))
            zipResult.extractall()
            print("File saved successfully")
        
        else:
            print("Error")
            print(response.text)
    

    
    def __stats (self, response):
        if (response.ok):
            parsedXMLResponse = BeautifulSoup(response.text, "lxml")

            self.jobId = parsedXMLResponse.find('requestid').get_text()
            
            for stat in parsedXMLResponse.find('response').findChildren():
                if(len(stat.findChildren()) == 0):
                    print("{name}: {data}".format(name=stat.name, data=stat.get_text()))
            
            # Contruct the zipfile url...
            self.zip_result = f'https://batch.geocoder.ls.hereapi.com/6.2/jobs/{self.jobId}/result?apiKey={mykey}'
            print("Zipfile URL: ", self.zip_result)
            print('-'*30)
            
            # Set delay for the zipfile to be ready for download on HERE server...
            time.sleep(50)
            
            # Download and extract zip file...
            self.r = requests.get(self.zip_result, stream=True)
            self.z = zipfile.ZipFile(io.BytesIO(self.r.content))
            self.z.extractall(r"HERE\\Output files")

        else:
            print(response.text)

if __name__=="__main__":
    
    service = Batch(apikey=mykey) ## use a HERE MAPS DEVELOPER API
    
    spreadsheet_folder = glob.glob(r'HERE\\Input files\\csv_file\\df_newPU\\*.csv')
    for cvs_file in spreadsheet_folder:
        print('Geocoding file...', cvs_file)
        service.start (cvs_file, indelim = ",", outdelim = ",")



Batch Reverse Geocoding Python Script

Input Data:


Python Script:

l# Reverse Geocode....
class Batch:
    SERVICE_URL = "https://batch.geocoder.ls.hereapi.com/6.2/jobs"
    jobId = None
    
    
    def __init__(self, apikey=""): ## use a HERE MAPS DEVELOPER API
        self.apikey = apikey
        
            
    def start(self, filename='togeocode619.csv', indelim=",", outdelim=","):
        
        file = open(filename, 'rb')

        params = {
            "action": "run",
            "apiKey": self.apikey,
            "politicalview":"RUS",
            "gen": 9,
            "maxresults": "1",
            "header": "true",
            "indelim": indelim,
            "outdelim": outdelim,
            "outCols": "recId,latitude,longitude,locationLabel",
            "outputcombined": "true",
            "mode":"retrieveAddresses",
        }

        
        response = requests.post(self.SERVICE_URL, params=params, data=file)
        self.__stats (response)
        file.close()
    

    def status (self, jobId = None):

        if jobId is not None:
            self.jobId = jobId
        
        statusUrl = self.SERVICE_URL + "/" + self.jobId
        
        params = {
            "action": "status",
            "apiKey": self.apikey,
        }
        
        response = requests.get(statusUrl, params=params)
        self.__stats (response)
        

    def result (self, jobId = None):

        if jobId is not None:
            self.jobId = jobId
        
        print("Requesting result data ...")
        
        resultUrl = self.SERVICE_URL + "/" + self.jobId + "/result"
        
        params = {
            "apiKey": self.apikey
        }
        
        response = requests.get(resultUrl, params=params, stream=True)
        
        if (response.ok):    
            zipResult = zipfile.ZipFile(io.BytesIO(response.content))
            zipResult.extractall()
            print("File saved successfully")
        
        else:
            print("Error")
            print(response.text)
    

    
    def __stats (self, response):
        if (response.ok):
            parsedXMLResponse = BeautifulSoup(response.text, "lxml")

            self.jobId = parsedXMLResponse.find('requestid').get_text()
            
            for stat in parsedXMLResponse.find('response').findChildren():
                if(len(stat.findChildren()) == 0):
                    print("{name}: {data}".format(name=stat.name, data=stat.get_text()))
            
            # Contruct the zipfile url...
            self.zip_result = f'https://batch.geocoder.ls.hereapi.com/6.2/jobs/{self.jobId}/result?apiKey={mykey}'
            print("Zipfile URL: ", self.zip_result)
            print('-'*30)
            
            # Set delay for the zipfile to be ready for download on HERE server...
            time.sleep(50)
            
            # Download and extract zip file...
            self.r = requests.get(self.zip_result, stream=True)
            self.z = zipfile.ZipFile(io.BytesIO(self.r.content))
            self.z.extractall(r"HERE\\Output files")

        else:
            print(response.text)

if __name__=="__main__":
    
    service = Batch(apikey=mykey) ## use a HERE MAPS DEVELOPER API
    cvs_file = r"C:\Users\Yusuf_08039508010\Documents\Jupyter_Notebook\2022\Geocode Addresses\19k LatLong in Fr\reverse_geocode_input.csv"
    
    print('Geocoding file...', cvs_file)
    service.start (cvs_file, indelim = ",", outdelim = ",")
    
#     spreadsheet_folder = glob.glob(r'HERE\\Input files\\csv_file\\df_newPU\\*.csv')
#     for cvs_file in spreadsheet_folder:
#         print('Geocoding file...', cvs_file)
#         service.start (cvs_file, indelim = ",", outdelim = ",")


Happy geocoding!

No comments:

Post a Comment