1_Tiff_Matcher_Osgeo.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "from osgeo import gdal\n",
    "from typing import Tuple"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# get environment variables\n",
    "cep_tiles_directory = os.environ['CEP_TILES_DIRECTORY']\n",
    "water_tiles_directory = os.environ['WATER_TILES_DIRECTORY']\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Matching Tiles from JRC by extent"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "class TiffMatcher:\n",
    "\n",
    "    def __init__(self, water_tiff_path:str, cep_tiff_path:str):\n",
    "        self.water_tiff = water_tiff_path\n",
    "        self.cep_tiff = cep_tiff_path\n",
    "        self.matches = None\n",
    "\n",
    "    def write_matches(self, output_file):\n",
    "        # Write the matches to a file\n",
    "        with open(output_file, \"w\") as file:\n",
    "            for water_file, cep_file in self.matches.items():\n",
    "                file.write(f\"{water_file},{cep_file}\\n\")\n",
    "\n",
    "    def get_extent(self, tiff_file , rounding:bool=False) -> Tuple[float, float, float, float]:\n",
    "        # Open the TIFF file\n",
    "        dataset = gdal.Open(tiff_file)\n",
    "\n",
    "        # Get raster dimensions\n",
    "        width = dataset.RasterXSize\n",
    "        height = dataset.RasterYSize\n",
    "\n",
    "        # Get geotransform information\n",
    "        geotransform = dataset.GetGeoTransform()\n",
    "\n",
    "        # Calculate the extent\n",
    "        xmin = geotransform[0]\n",
    "        ymin = geotransform[3] + width*geotransform[4] + height*geotransform[5]\n",
    "        xmax = geotransform[0] + width*geotransform[1] + height*geotransform[2]\n",
    "        ymax = geotransform[3]\n",
    "\n",
    "        if not rounding:\n",
    "            return (xmin, ymin, xmax, ymax)\n",
    "        else:\n",
    "            # return and round to nearest 10 to make extents comparable\n",
    "            return (round(xmin, -1), round(ymin, -1), round(xmax, -1), round(ymax, -1))\n",
    "\n",
    "    def get_all_tiff_extents(self,directory, rounding:bool) -> dict[Tuple[float, float, float, float], str]:\n",
    "        # Get all the files in the directory\n",
    "        files = os.listdir(directory)\n",
    "\n",
    "        # Get the extents of all the tiff files\n",
    "        extents = {}\n",
    "        for file in files:\n",
    "            if file.endswith(\".tif\") or file.endswith(\".tiff\"):\n",
    "                extents[self.get_extent(os.path.join(directory, file), rounding)] = file\n",
    "        return extents\n",
    "\n",
    "    def match_tiffs_by_extent(self,water_extents, cep_extents, append_file_path=False) -> dict[str, str]:\n",
    "        # For each water extent, find the CEP extent that matches\n",
    "        matches = {}\n",
    "        for water_extent in water_extents:\n",
    "            for cep_extent in cep_extents:\n",
    "                if water_extent == cep_extent:\n",
    "                    if append_file_path:\n",
    "                        water_path = os.path.join(self.water_tiff, water_extents[water_extent])\n",
    "                        cep_path = os.path.join(self.cep_tiff, cep_extents[cep_extent])\n",
    "                        matches[water_path] = cep_path \n",
    "                    else:\n",
    "                        matches[water_extents[water_extent]] = cep_extents[cep_extent]\n",
    "        return matches\n",
    "\n",
    "    def match_tiffs_by_nearest_extent(self,water_extents, cep_extents,append_file_path) -> dict[str, str]:\n",
    "        # For each water extent, find the nearest CEP extent\n",
    "        matches = {}\n",
    "        for water_extent in water_extents:\n",
    "            nearest_cep_extent = None\n",
    "            nearest_cep_extent_distance = None\n",
    "            for cep_extent in cep_extents:\n",
    "                # Calculate the distance between the water and CEP extents using the Euclidean distance\n",
    "                # distance of first point from second point = sqrt((x1-x2)^2 + (y1-y2)^2)\n",
    "                distance = ((cep_extent[0] - water_extent[0])**2 + (cep_extent[1] - water_extent[1])**2)**0.5\n",
    "                if nearest_cep_extent is None or distance < nearest_cep_extent_distance:\n",
    "                    nearest_cep_extent = cep_extent\n",
    "                    nearest_cep_extent_distance = distance\n",
    "            if append_file_path:\n",
    "                    water_path = os.path.join(self.water_tiff, water_extents[water_extent])\n",
    "                    cep_path = os.path.join(self.cep_tiff, cep_extents[cep_extent])\n",
    "                    matches[water_path] = cep_path \n",
    "            else:\n",
    "                matches[water_extents[water_extent]] = cep_extents[cep_extent]\n",
    "        return matches \n",
    "    \n",
    "    def match_tiffs(self, rounding:bool=False, append_file_path:bool=False) -> dict[str, str]:\n",
    "        # Get the extents of all the tiff files in the water and CEP directories\n",
    "        water_extents = self.get_all_tiff_extents(self.water_tiff, rounding)\n",
    "        cep_extents = self.get_all_tiff_extents(self.cep_tiff, rounding)\n",
    "        \n",
    "        # Match the tiff files by extent\n",
    "        self.matches = self.match_tiffs_by_extent(water_extents, cep_extents, append_file_path)\n",
    "        return self.matches\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "tiffmatcher = TiffMatcher(water_tiles_directory, cep_tiles_directory)\n",
    "tiffmatcher.match_tiffs(rounding=True, append_file_path=True)\n",
    "tiffmatcher.write_matches(\"matches.txt\")\n",
    "\n",
    "#check first match pair\n",
    "for key, value in tiffmatcher.matches.items():\n",
    "    print(key, value)\n",
    "    break"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "ad4gd",
   "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.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}