Converts latitude and longitude columns in a CSV file from EPSG:3347 NAD83 / Statistics Canada Lambert to EPSG:4326 WGS 84 coordinate system

convert_coordinates_NAD83_to_WGS84.py

import csv
import os # For splitting filename and extension
import argparse # For command-line arguments
from pyproj import Transformer, CRS
from pyproj.exceptions import ProjError

def convert_coords_write_newfile_replace_values(
    input_csv_path,
    output_csv_path,
    lon_column_name, # Column to read EPSG:3347 from AND write EPSG:4326 to
    lat_column_name  # Column to read EPSG:3347 from AND write EPSG:4326 to
):
    """
    Reads a CSV file, converts coordinates in specified columns from
    EPSG:3347 to EPSG:4326, and writes to a new file, replacing
    the values in the original-named columns.

    Args:
        input_csv_path (str): Path to the input CSV file.
        output_csv_path (str): Path to the new output CSV file.
        lon_column_name (str): Name of the longitude column.
        lat_column_name (str): Name of the latitude column.
    """
    try:
        crs_3347 = CRS.from_epsg(3347)
        crs_4326 = CRS.from_epsg(4326)
        transformer = Transformer.from_crs(crs_3347, crs_4326, always_xy=True)
    except ProjError as e:
        print(f"ERROR: Initializing CRS or Transformer failed: {e}")
        print("Please ensure pyproj is installed correctly and EPSG codes are valid.")
        return False # Indicate failure

    try:
        with open(input_csv_path, 'r', newline='', encoding='utf-8') as infile, \
             open(output_csv_path, 'w', newline='', encoding='utf-8') as outfile:

            reader = csv.DictReader(infile)
            if not reader.fieldnames:
                print(f"ERROR: Could not read header from {input_csv_path}. Is the file empty or malformed?")
                return False

            if lon_column_name not in reader.fieldnames:
                print(f"ERROR: Longitude column '{lon_column_name}' not found in the input CSV.")
                print(f"Available columns: {reader.fieldnames}")
                return False
            if lat_column_name not in reader.fieldnames:
                print(f"ERROR: Latitude column '{lat_column_name}' not found in the input CSV.")
                print(f"Available columns: {reader.fieldnames}")
                return False

            fieldnames = list(reader.fieldnames)
            writer = csv.DictWriter(outfile, fieldnames=fieldnames)
            writer.writeheader()

            print(f"Processing file: {input_csv_path}")
            print(f"Output will be saved to: {output_csv_path}")
            processed_rows_count = 0
            conversion_errors_count = 0

            for row_num, row in enumerate(reader, 1):
                processed_rows_count += 1
                try:
                    x_3347_str = row.get(lon_column_name)
                    y_3347_str = row.get(lat_column_name)

                    if x_3347_str is None or y_3347_str is None or \
                       x_3347_str.strip() == '' or y_3347_str.strip() == '':
                        row[lon_column_name] = 'EMPTY_COORD_INPUT'
                        row[lat_column_name] = 'EMPTY_COORD_INPUT'
                        conversion_errors_count += 1
                        writer.writerow(row)
                        continue

                    x_3347 = float(x_3347_str)
                    y_3347 = float(y_3347_str)

                    lon_4326, lat_4326 = transformer.transform(x_3347, y_3347)

                    row[lon_column_name] = lon_4326
                    row[lat_column_name] = lat_4326

                except ValueError:
                    row[lon_column_name] = 'CONVERSION_ERROR_FLOAT'
                    row[lat_column_name] = 'CONVERSION_ERROR_FLOAT'
                    conversion_errors_count +=1
                except ProjError:
                    row[lon_column_name] = 'CONVERSION_ERROR_PROJ'
                    row[lat_column_name] = 'CONVERSION_ERROR_PROJ'
                    conversion_errors_count +=1
                except Exception:
                    row[lon_column_name] = 'CONVERSION_ERROR_UNEXPECTED'
                    row[lat_column_name] = 'CONVERSION_ERROR_UNEXPECTED'
                    conversion_errors_count +=1
                
                writer.writerow(row)
            
            print(f"Finished processing. Total rows processed: {processed_rows_count}.")
            if conversion_errors_count > 0:
                print(f"Number of rows with conversion issues (marked in output file): {conversion_errors_count}")
            print(f"Output successfully saved to: {output_csv_path}")
            return True # Indicate success

    except FileNotFoundError:
        print(f"ERROR: Input file not found at {input_csv_path}")
        return False
    except IOError as e:
        print(f"ERROR: Could not read from '{input_csv_path}' or write to '{output_csv_path}'. Error: {e}")
        return False
    except Exception as e:
        print(f"ERROR: An unexpected error occurred: {e}")
        return False

if __name__ == '__main__':
    parser = argparse.ArgumentParser(
        description="Convert coordinates in a CSV file from EPSG:3347 to EPSG:4326. "
                    "The script reads specified longitude and latitude columns, converts their values, "
                    "and writes them back into the same columns in a new file named "
                    "'<original_filename>_wgs84.<original_extension>'."
    )
    parser.add_argument(
        "input_file", 
        help="Path to the input CSV file."
    )
    
    args = parser.parse_args()
    
    # --- Configuration for Column Names (remains internal to the script) ---
    # These are the names of the columns in your INPUT CSV that hold the EPSG:3347 coordinates.
    # In the OUTPUT CSV, these same columns will contain the converted EPSG:4326 (WGS 84) values.
    # For EPSG:3347, "longitude" typically refers to Easting (X) and "latitude" to Northing (Y).
    LONGITUDE_COLUMN_NAME = 'Longitude' # <--- CHANGE THIS if your column name is different
    LATITUDE_COLUMN_NAME = 'Latitude'  # <--- CHANGE THIS if your column name is different
    # --- End Configuration for Column Names ---

    # Generate the output file name from the provided input_file argument
    base, ext = os.path.splitext(args.input_file)
    output_file_path = base + "_wgs84" + ext

    print("Starting coordinate conversion process...")
    success = convert_coords_write_newfile_replace_values(
        args.input_file,
        output_file_path,
        LONGITUDE_COLUMN_NAME,
        LATITUDE_COLUMN_NAME
    )
    
    if success:
        print("Script finished successfully.")
    else:
        print("Script finished with errors.")