geocoding a source CVS with google maps and then feed into a table in PostGIS

geocode_csv_to_postgis_gmaps.py

# geocoding a source CVS with google maps
# and then feed into a table in PostGIS
# 
# interestingly some parts of the code were created with ChatGPT4
# but the output, of course, had to be revised a little
# 
# license CC-Attribution
3

import csv
import requests
import psycopg2
from psycopg2.extras import execute_values
from pyproj import Transformer

# Read the CSV file
def read_csv(file_path):
    addresses = []
    with open(file_path, newline='', encoding='utf-8') as csvfile:
        reader = csv.DictReader(csvfile)
        for row in reader:
            addresses.append(row)
    return addresses

# Geocode an address using Google Maps API
def geocode(address, city, country, api_key):
    url = f'https://maps.googleapis.com/maps/api/geocode/json?address={address},{city},{country}&key={api_key}'
    response = requests.get(url)
    data = response.json()
    if data['status'] == 'OK':
        location = data['results'][0]['geometry']['location']
        return location['lat'], location['lng']
    else:
        return None

# Transform coordinates from EPSG:4326 to EPSG:4258
def transform_coords(lat, lng):
    transformer = Transformer.from_crs("EPSG:4326", "EPSG:7801", always_xy=True)
    x, y = transformer.transform(lng, lat)
    return f"POINT({x} {y})"

# Connect to the PostGIS database
def connect_to_db(database, user, password, host, port):
    connection = psycopg2.connect(database=database, user=user, password=password, host=host, port=port)
    return connection

# Insert data into PostGIS database
def insert_into_postgis(connection, data):
    cursor = connection.cursor()

    # Create table if not exists
    create_table_query = """
        CREATE TABLE IF NOT EXISTS geocoded_addresses (
            id SERIAL PRIMARY KEY,
            address VARCHAR(255),
            city VARCHAR(255),
            country VARCHAR(255),
            location GEOMETRY(Point, 7801)
        );
    """
    cursor.execute(create_table_query)

    # Insert data into the table
    query = """
        INSERT INTO geocoded_addresses (address, city, country, location)
        VALUES %s;
    """
    execute_values(cursor, query, data)
    connection.commit()
    cursor.close()


# Main function
def main():
    # Replace these values with your own
    csv_file_path = 'yourdata/input.csv'
    google_api_key = 'yourkey'
    postgis_config = {
        'database': 'yourdb',
        'user': 'youruser',
        'password': 'yourpass',
        'host': 'your.host.com',
        'port': '5432',
    }

    addresses = read_csv(csv_file_path)
    geocoded_data = []

    for address in addresses:
        coords = geocode(address['address'], address['city'], address['countryCode'], google_api_key)
        if coords:
            wktcoords = transform_coords(*coords)
            geocoded_data.append((address['address'], address['city'], address['countryCode'], wktcoords))

    connection = connect_to_db(**postgis_config)
    insert_into_postgis(connection, geocoded_data)
    connection.close()

if __name__ == '__main__':
    main()