geo_spatial_analysis.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Data Analysis and Reporting Notebook"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Setup of Analysis\n",
    "\n",
    "**Imports**\n",
    "\n",
    "Libraries needed to run analysis and create report "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import geopandas as gpd\n",
    "import shapely\n",
    "from shapely.errors import TopologicalError\n",
    "import pandas as pd\n",
    "import os\n",
    "from docx import Document\n",
    "from docx.shared import Pt"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Dataset Folder for Analysis**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dataset = \"\"\n",
    "\n",
    "dataset_folder = f\"datasets/{dataset}\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Filepath Setup**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def find_geospatial_files(folder_path):\n",
    "    \"\"\"\n",
    "    Returns a list of all geospatial files (.shp, .gpkg, or .csv) found in the given folder path.\n",
    "    If no geospatial files are found, returns an empty list.\n",
    "    \"\"\"\n",
    "    geospatial_files = [\n",
    "        file_name for file_name in os.listdir(folder_path)\n",
    "        if file_name.endswith('.shp') or file_name.endswith('.gpkg') or file_name.endswith('.csv')\n",
    "    ]\n",
    "    return geospatial_files\n",
    "\n",
    "# Define folder path containing the files\n",
    "folder_path = f\"{dataset_folder}/files\"  # The folder containing the geospatial files\n",
    "print(f\"Looking for geospatial files in folder: {folder_path}\")\n",
    "\n",
    "# Find geospatial files\n",
    "geospatial_files = find_geospatial_files(folder_path)\n",
    "\n",
    "if geospatial_files:\n",
    "    print(\"Geospatial files found:\")\n",
    "    for file in geospatial_files:\n",
    "        print(f\"- {file}\")\n",
    "else:\n",
    "    print(\"No geospatial files found in the folder.\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Select shapefile to analyse\n",
    "geospatial_file = geospatial_files[0]\n",
    "print(geospatial_file)\n",
    "\n",
    "# Define paths for output files\n",
    "geojson_output_file = f\"{dataset_folder}/{dataset}_{geospatial_file}_sample.geojson\"\n",
    "print(f\"GeoJSON output file path: {geojson_output_file}\")\n",
    "\n",
    "features_plot_file = f\"{dataset_folder}/{dataset}_plot.png\"\n",
    "print(f\"Features plot file path: {features_plot_file}\")\n",
    "\n",
    "report_output_file = f\"{dataset_folder}/{dataset}_analysis_report\"\n",
    "print(f\"Analysis report file path: {report_output_file}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Read in shapefile and initially inspect first 5 rows (and headers)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def load_geospatial_file(folder_path, file_name):\n",
    "    \"\"\"\n",
    "    Loads a geospatial file (either .shp, .gpkg, or .csv) into a GeoDataFrame.\n",
    "    Attempts to handle invalid geometries using pyogrio's on_invalid option.\n",
    "    For CSV files, expects columns 'latitude' and 'longitude' to convert to geometries.\n",
    "    \n",
    "    Args:\n",
    "        folder_path (str): Path to the folder containing the file.\n",
    "        file_name (str): Name of the file to load.\n",
    "    \n",
    "    Returns:\n",
    "        GeoDataFrame: A GeoDataFrame containing the data.\n",
    "    \"\"\"\n",
    "    file_path = os.path.join(folder_path, file_name)\n",
    "    \n",
    "    try:\n",
    "        if file_name.endswith('.shp') or file_name.endswith('.gpkg'):\n",
    "            # Load shapefile or GeoPackage using pyogrio\n",
    "            gdf = gpd.read_file(file_path, engine=\"pyogrio\", on_invalid=\"ignore\")\n",
    "        \n",
    "        elif file_name.endswith('.csv'):\n",
    "            # Load CSV file and attempt to create geometries if lat/lon columns are present\n",
    "            gdf = pd.read_csv(file_path)\n",
    "                  \n",
    "        else:\n",
    "            raise ValueError(\"Unsupported file format. Only .shp, .gpkg, and .csv files are supported.\")\n",
    "    \n",
    "    except TopologicalError as e:\n",
    "        print(f\"TopologicalError encountered: {e}\")\n",
    "        raise\n",
    "    except Exception as e:\n",
    "        print(f\"An error occurred: {e}\")\n",
    "        raise\n",
    "    \n",
    "    return gdf\n",
    "\n",
    "file_path = f\"{geospatial_file}\"\n",
    "\n",
    "# Load the geospatial file (shapefile or GeoPackage)\n",
    "gdf = load_geospatial_file(folder_path, file_path)\n",
    "print(f\"Geospatial data loaded with {len(gdf)} records.\")\n",
    "gdf.head()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Analyse folder and shapefile"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Analyse the contents of the folder\n",
    "def analyze_folder(folder_path):\n",
    "    \"\"\"Returns a detailed description of files contained in the folder.\"\"\"\n",
    "    folder_contents = os.listdir(folder_path)\n",
    "    content_details = {}\n",
    "    total_size = 0\n",
    "    for file in folder_contents:\n",
    "        file_path = os.path.join(folder_path, file)\n",
    "        file_size = os.path.getsize(file_path)\n",
    "        content_details[file] = f\"Size: {file_size} bytes\"\n",
    "        total_size += file_size\n",
    "    content_details[\"Total Folder Size\"] = f\"{total_size} bytes\"\n",
    "    return content_details\n",
    "\n",
    "folder_description = analyze_folder(folder_path)\n",
    "folder_description"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def analyze_geospatial_file(gdf, geojson_output_file, file_path):\n",
    "    \"\"\"\n",
    "    Analyzes a geospatial file and returns important information for the report.\n",
    "    Converts Timestamp columns to string to handle JSON serialization.\n",
    "    \n",
    "    Args:\n",
    "        gdf (GeoDataFrame): The GeoDataFrame to analyze.\n",
    "        geojson_output_file (str): The path to save the sample GeoJSON file.\n",
    "        file_path (str): The path to the original geospatial file.\n",
    "        \n",
    "    Returns:\n",
    "        dict: Dictionary containing analysis information.\n",
    "    \"\"\"\n",
    "    \n",
    "    # Convert any Timestamp columns to strings for JSON serialization\n",
    "    for column in gdf.select_dtypes(include=['datetime']):\n",
    "        gdf[column] = gdf[column].astype(str)\n",
    "\n",
    "    # Save a sample as GeoJSON (first 50 records)\n",
    "    try:\n",
    "        gdf.head(50).to_file(geojson_output_file, driver=\"GeoJSON\")\n",
    "        print(f\"Sample GeoJSON saved as {geojson_output_file}\")\n",
    "    except Exception as e:\n",
    "        return None, {\"error\": f\"Error saving GeoJSON: {e}\"}\n",
    "    \n",
    "    # Analyze the geospatial data\n",
    "    geometry_type = gdf.geom_type.unique()\n",
    "    record_count = len(gdf)\n",
    "    features = gdf.columns.tolist()\n",
    "    crs = str(gdf.crs) if gdf.crs is not None else \"Unknown\"\n",
    "    \n",
    "    # Convert a sample of the data to GeoJSON for preview\n",
    "    try:\n",
    "        geojson_sample = gdf.head(50).to_json()\n",
    "    except Exception as e:\n",
    "        return None, {\"error\": f\"Error converting to GeoJSON: {e}\"}\n",
    "    \n",
    "    # Get the file size\n",
    "    file_size = os.path.getsize(file_path)\n",
    "    \n",
    "    # Compile the analysis into a dictionary\n",
    "    geospatial_data = {\n",
    "        \"geometry_type\": ', '.join(geometry_type),\n",
    "        \"record_count\": record_count,\n",
    "        \"features\": ', '.join(features),\n",
    "        \"crs\": crs,\n",
    "        \"geojson_sample\": geojson_sample,\n",
    "        \"file_size\": file_size\n",
    "    }\n",
    "\n",
    "    return geospatial_data\n",
    "\n",
    "\n",
    "path = f\"{folder_path}/{file_path}\"\n",
    "\n",
    "geospatial_data = analyze_geospatial_file(gdf, geojson_output_file, path)\n",
    "geospatial_data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create Report"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def generate_report(folder_data, geospatial_data, output_file):\n",
    "    \"\"\"\n",
    "    Generates a DOCX report summarising the dataset\n",
    "    \n",
    "    Args:\n",
    "        folder_data (dict): Information about folder contents.\n",
    "        geospatial_data (dict): Information about the shapefile.\n",
    "        output_file (str): The file path to save the report.\n",
    "    \"\"\"\n",
    "    doc = Document()\n",
    "    \n",
    "    # Add title\n",
    "    doc.add_heading(f\"{dataset} Analysis Report\", 0)\n",
    "\n",
    "    # Add section for folder analysis\n",
    "    doc.add_heading('Folder Contents:', level=1)\n",
    "    doc.add_paragraph('The following files were found in the folder:')\n",
    "    for file, details in folder_data.items():\n",
    "        doc.add_paragraph(f\"{file}: {details}\", style='List Bullet')\n",
    "\n",
    "    # Add file analysis section\n",
    "    custom_heading = doc.add_heading(f\"{geospatial_file} File Analysis:\", level=0)\n",
    "    for run in custom_heading.runs:\n",
    "        run.font.size = Pt(16)\n",
    "        run.bold = True\n",
    "    doc.add_paragraph(f\"Geometry Type: {geospatial_data['geometry_type']}\", style='List Bullet')\n",
    "    doc.add_paragraph(f\"Number of Records: {geospatial_data['record_count']}\", style='List Bullet')\n",
    "    doc.add_paragraph(f\"Features/Attributes: {geospatial_data['features']}\", style='List Bullet')\n",
    "    doc.add_paragraph(f\"Coordinate Reference System (CRS): {geospatial_data['crs']}\", style='List Bullet')\n",
    "    doc.add_paragraph(f\"File Size: {geospatial_data['file_size']} bytes\", style='List Bullet')\n",
    "\n",
    "    doc.add_heading(\"Plot of Dataset\")\n",
    "    doc.add_paragraph(f\"Plot of first row of data:\")\n",
    "        \n",
    "    doc.add_heading(\"GeoJSON Sample Data\", level=1)\n",
    "    try:\n",
    "        with open(geojson_output_file, 'r') as geojson_file:\n",
    "            geojson_content = geojson_file.read(2000)  # Read the first 2000 characters\n",
    "            doc.add_paragraph(\"Sample GeoJSON content:\")\n",
    "            doc.add_paragraph(geojson_content)\n",
    "    except Exception as e:\n",
    "        doc.add_paragraph(f\"Error loading GeoJSON file: {e}\")\n",
    "\n",
    "    # Save the report to a DOCX file\n",
    "    output = f\"{output_file}_({geospatial_file}).docx\"\n",
    "\n",
    "    doc.save(output)\n",
    "    print(f\"Report saved as {output}\")\n",
    "\n",
    "\n",
    "image_files = [features_plot_file]\n",
    "\n",
    "generate_report(folder_description, geospatial_data, report_output_file)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## View data in interactive map"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# simple interactive\n",
    "# check subset of rows\n",
    "gdf.iloc[:1].explore()\n",
    "\n",
    "# Whole table\n",
    "#gdf.explore()\n",
    "\n",
    "## clip to a subset\n",
    "\n",
    "#row = 1\n",
    "#col = 'geometry'\n",
    "\n",
    "# draw a bounding box using https://wktmap.com/\n",
    "#bbox_wkt = f\"{gdf.loc[row, col]}\"\n",
    "#bbox_poly = shapely.wkt.loads(bbox_wkt)\n",
    "\n",
    "#gdf_clipped = gdf.clip(bbox_poly)\n",
    "\n",
    "#print(bbox_wkt)\n",
    "#print(len(gdf_clipped))\n",
    "\n",
    "#gdf_clipped.explore()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv",
   "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.11.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}