Where do you live? May be a simple question, but when companies are asking where someone lives, the answer isn’t always so easy. The physical address and the mailing address of a person can be two completely separate things. And when companies want to send direct mail, they need to ensure the message is appropriate. <\/p>\n\n\n\n
Today I am going to show a Python program that I wrote that can process millions of addresses and verify if the address is a valid mailing address that matches the physical address. And how the majority of online mapping services get it wrong.<\/p>\n\n\n\n
<\/p>\n\n\n\n\n\n\n\n
One the main challenges about physical addresses is that you can’t rely on online mapping services for 100% verification. Many of the top mapping services, i.e. Google Maps, will approximate an address if it isn’t in its database. An example of this is 1500 West Pine Street, Anywhere, AK. If Google Maps knows the street and the city, it will mark the location as the closest street number to the requested street number. <\/p>\n\n\n\n
The result may be close enough for driving directions or to help give you an estimate of where the location is, but for the postal system, this doesn’t work. If the building does exist, it is ok; however if it doesn’t exist, then your mail will be returned. And a marketing opportunity will have been lost. <\/p>\n\n\n\n
The United States Postal Service does have an address verification tool. However it doesn’t allow for large datasets to be verified. Many of the other online API tools, paid or unpaid are the same. I resolved the issue by splitting the source file into manageable block sizes for requests. Next when the response is received I combine the responses and re-arrange the responses to match the order the requests were made. <\/p>\n\n\n\n
The program uses concurrent futures package to allow for multi-threaded operations and allow for numerous requests to be sent and received at the same time. <\/p>\n\n\n\n
<\/p>\n\n\n\n
Program Overview<\/h2>\n\n\n\n
The program uses the tkinter UI package. I won’t go in-depth on how the tkinter package works, that would be a separate posting all on its own. This post will be long enough as it is. <\/p>\n\n\n\n
<\/figure>\n\n\n\n
Program Files<\/h2>\n\n\n\n
File Name<\/th>
Description<\/th><\/tr><\/thead>
LocusApiSearch.py<\/td>
The main file imports the packages and contains the main program loop <\/td><\/tr>
cls\/apiHandler.py<\/td>
class to handle connecting to the API<\/td><\/tr>
cls\/csvFileHandler.py<\/td>
creates interim .csv files and <\/td><\/tr>
cls\/outputFile.py<\/td>
defines the definition, i.e. colums of the output file<\/td><\/tr>
cls\/pyUtilities.py<\/td>
helper function to merge interim .csv files into the final file<\/td><\/tr>
cls\/runtimeParameters.py<\/td>
class to store user parameters from the UI<\/td><\/tr>
cls\/ui.py<\/td>
main class for handling the UI components<\/td><\/tr>
cls\/uiConfig.py<\/td>
reading configuration options from the UI and saving to local file<\/td><\/tr>
cls\/userAddresses.py<\/td>
definition of the input data file and methods to clean data<\/td><\/tr>
<\/td>
<\/td><\/tr><\/tbody><\/table><\/figure>\n\n\n\n
<\/p>\n\n\n\n
<\/p>\n\n\n\n
Packages<\/h2>\n\n\n\n
datetime<\/li>
time<\/li>
os<\/li>
pathlib<\/li>
numpy<\/li>
pandas<\/li>
json<\/li>
requests<\/li>
tkinter<\/li>
concurrent.futures<\/li><\/ul>\n\n\n\n
<\/p>\n\n\n\n
<\/p>\n\n\n\n
Program Flow<\/h2>\n\n\n\n
LocusApiSearch\n Initiate the UI\n Read the configuration settings, populate the UI\n Configuration settings file, doesn't exist use default\n\n Start the Main Loop\n Load the runtime parameters from the UI\n Initiate the log file\n Read the input file, store into a data frame\n Start new threads, each handling a block of data\n Process the csv file\n prep the data, based on parser and UI options\n save json file\n Make API call\n Process returned json file\n If file not processed, sleep, then retry\n Save file as .csv file\n Momentary pause, don't overload remote api\n Merge all .csv files<\/pre><\/div>\n\n\n\n
<\/p>\n\n\n\n
<\/p>\n\n\n\n
Main<\/h2>\n\n\n\n
LocusApiSearch.py<\/p>\n\n\n\n
# main entry point to start the gui\nif __name__ == '__main__':\n # load the gui\n root = tk.Tk()\n app = Application(app_root=root)\n # start the gui\n app.mainloop()\n print(\"complete\")<\/pre><\/div>\n\n\n\n
The first code block starts the UI, I assigned the UI to root. and assigned the application to app. Next I tell the UI to start the main loop. <\/p>\n\n\n\n
<\/p>\n\n\n\n
UI<\/h2>\n\n\n\n<\/figure>\n\n\n\n
I am going to skip over the UI startup and drawing. There are multiple tutorials on how to use the tkinter UI package for Python. In summary cls\/ui.py and cls\/uiConfig.py handle drawing the screen windows and controls in the UI.<\/p>\n\n\n\n
<\/p>\n\n\n\n
<\/p>\n\n\n\n
Application Start<\/h2>\n\n\n\n
<\/p>\n\n\n\n
LocusApiSearch.py<\/h3>\n\n\n\n
start_job()<\/h4>\n\n\n\n
<\/p>\n\n\n\n
def start_job(user_args: dict):\n \"\"\"f(x) to initiate & control the API request threads\"\"\"\n try:\n global logfile\n\n # execution\n app.update_progress(\"Starting Job\")\n\n # --------------------------------------------------------------------\n # set the run time parameters\n # --------------------------------------------------------------------\n args = runtimeParameters(user_args, batch=True)\n\n # --------------------------------------------------------------------\n # check and set logfile locale, write log file to results directory\n # --------------------------------------------------------------------\n log_path = Path(args.usr_args['output_dir'] + '\/logs\/')\n if not log_path.is_dir():\n os.mkdir(log_path)\n nm = logfile.name\n logfile = Path(log_path \/ nm)\n\n with open(logfile, 'a+') as file_handle:\n file_handle.write(\"starting job: \" + datetime.now().strftime(\"%Y-%m-%d.%H%M%S\") + \"\\n\")\n file_handle.close()<\/pre><\/div>\n\n\n\n
start_job(user_args: dict)<\/p>\n\n\n\n
Pass in the dictionary that was generated from reading contents of the UI. <\/p>\n\n\n\n
Next, use the runtimeParameters class to organize the parameters. Because we are using multi-threaded processing need to have a local copy of the parameters for the current batch of data that is being processed. <\/p>\n\n\n\n
The next thing is store into the log file that starting the job, and release the lock on the file.<\/p>\n\n\n\n
\n # initiate file handler to read the input file\n cfh = csvFileHandler(args.usr_args)\n\n # load the .csv file into a panda data frame\n # rm # cfh.get_file(args.csv_file_name, args.test_mode)\n cfh.get_file(csv_file_name=args.usr_args['input_file'])\n app.update_progress(f\"File Length: {len(cfh.csv_df)}\")\n\n # set the run time argument for file size\n # and starting rows, ending rows, and rows to process in batch\n args.set_file_length(len(cfh.csv_df))<\/pre><\/div>\n\n\n\n
<\/p>\n\n\n\n
Will use the csvFileHandler class to read the data file. Write a message to the UI to the let user know where we are in the processing. Update the args.set_file_length property with the file length. <\/p>\n\n\n\n
<\/p>\n\n\n\n
Initiate the Thread Pool<\/h3>\n\n\n\n
# multi thread the process_csv_file f(x)\n # to initiate multiple threads\n # loop through input source document in groups specified by step size\n with cf.ThreadPoolExecutor(max_workers=8) as executor:\n call_locus_api = {executor.submit(process_csv_file, args, cfh, curr_idx):\n curr_idx for curr_idx in np.arange(args.start_row, args.end_row, args.step_size)}\n for worker in cf.as_completed(call_locus_api):\n # stop new threads from being processed,\n # won't prevent existing results from being processed\n if app.continue_processing == False:\n executor.shutdown(wait=False)\n worker.cancel()\n break\n \n fx_result = call_locus_api[worker]\n try:\n worker_result = worker.result()\n except Exception as e:\n print(f\"Exception {e}\")\n app.update_progress(f\"{fx_result} raised an exception {e}\")\n else:\n app.update_result(f\"{fx_result} result: {worker_result}\")<\/pre><\/div>\n\n\n\n
Here we will start processing the blocks of data. Each block of data will be assigned its own thread. I capped the maximum number of threads at 8. The number of threads could be set in the UI, but I opted against it.<\/p>\n\n\n\n
with cf.ThreadPoolExecutor(max_workers=8) as executor:<\/pre>\n\n\n\n
cf is from the concurrent.futures package and the method ThreadPoolExecutor is what is used to manage the pool of threads.<\/p>\n\n\n\n
call_locus_api = {executor.submit(process_csv_file, args, cfh, curr_idx):\n curr_idx for curr_idx in np.arange(args.start_row, args.end_row, args.step_size)}<\/pre>\n\n\n\n
call_locus_api is definition of the function I want to execute with all of the applicable parameters. <\/p>\n\n\n\n
for worker in cf.as_completed(call_locus_api):\n # stop new threads from being processed,\n # won't prevent existing results from being processed\n if app.continue_processing == False:\n executor.shutdown(wait=False)\n worker.cancel()\n break<\/pre>\n\n\n\n
Use a for loop to process each data block. I do a quick check to see if the cancel button was pushed to prevent the data block from processing. Threads that are currently processing will finish. <\/p>\n\n\n\n
![\"\"](\"https:\/\/eipsoftware.com\/musings\/wp-content\/uploads\/2021\/10\/Process_1-1024x488.png\")
<\/figure>\n\n\n\n![\"\"](\"https:\/\/eipsoftware.com\/musings\/wp-content\/uploads\/2021\/10\/options_edited-1024x488.png\")
<\/figure>\n\n\n\n![\"\"](\"https:\/\/eipsoftware.com\/musings\/wp-content\/uploads\/2021\/10\/Messages_2-1024x488.png\")
<\/figure>\n\n\n\n