02_asf_static.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# ASF Static RTC Files\n",
    "\n",
    "> Module for accessing static files hosted by ASF. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#| default_exp asf_static"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Module Imports"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "import logging\n",
    "import os\n",
    "import re\n",
    "import tempfile\n",
    "\n",
    "import asf_search\n",
    "import ee\n",
    "from fastcore.basics import patch\n",
    "import gcsfs\n",
    "from IPython.display import JSON\n",
    "import pandas as pd\n",
    "\n",
    "from sar_asf_to_gee.core import (\n",
    "    FORMAT_GEE_DATETIME_STRING,\n",
    "    create_gee_image_collection\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Setup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "logger = logging.getLogger()\n",
    "logger.setLevel(logging.INFO)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# ASF Static OPERA Files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "class SearchOpera():   \n",
    "\n",
    "    LOCAL_PROPNAME = 'local_paths'\n",
    "    GCS_PATH_PROPNAME = 'gcs_path'\n",
    "    GEE_ASSET_PROPNAME = 'gee_asset'\n",
    "    \n",
    "    def __init__(\n",
    "        self,\n",
    "        search_opts,\n",
    "        gcs_bucket,  # GCS bucket\n",
    "        gee_gcp_project, # GCP project used by Earth Engine\n",
    "        gee_image_collection=None,  # Name of the Earth Engine ImageCollection (optional)\n",
    "        local_storage=None,\n",
    "    ):\n",
    "        self.search_opts = search_opts\n",
    "        self.gcs_bucket = gcs_bucket\n",
    "        self.gee_gcp_project = gee_gcp_project\n",
    "        self.gee_image_collection = gee_image_collection\n",
    "        if local_storage:\n",
    "            self.tempdir = None\n",
    "            self.local_storage = local_storage\n",
    "        else:\n",
    "            self.tempdir = tempfile.TemporaryDirectory() \n",
    "            self.local_storage = self.tempdir.name\n",
    "            logging.debug(f'created temporary directory: {self.tempdir.name}')\n",
    "        self._search_results=None\n",
    "        # self._extracted={}\n",
    "        # self._gcs_path={}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "search_opts = {\n",
    "    'dataset': 'OPERA-S1',\n",
    "    'processingLevel': 'RTC',\n",
    "    'intersectsWith': 'Point (-122.299 37.702)',  # San Francisco\n",
    "    'start': '2023-12-14 00:00',\n",
    "    'end': '2024-01-01 00:00',\n",
    "    'maxResults': 50 \n",
    "}\n",
    "obj = SearchOpera(\n",
    "    search_opts,\n",
    "    gcs_bucket='asf-static-data-staging',\n",
    "    local_storage='temp_downloads',\n",
    "    gee_gcp_project='sar-asf-to-gee',\n",
    "    gee_image_collection=f'example-opera-s1-rtc',\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create a method to return the count of search results."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "@patch\n",
    "def search_count(\n",
    "    self:SearchOpera,\n",
    "):\n",
    "    \"Returns a count of records (w/ duplicates)\"\n",
    "    return asf_search.search_count(**self.search_opts)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "obj.search_count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create a method to return the search results."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "@patch\n",
    "def search(\n",
    "    self:SearchOpera,\n",
    "):\n",
    "    if not self._search_results:\n",
    "        self._search_results = asf_search.search(**self.search_opts)\n",
    "    return self._search_results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "obj.search()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Return search results as a dataframe."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "@patch\n",
    "def as_dataframe(\n",
    "    self:SearchOpera,\n",
    "):\n",
    "    \"Returns results as a pandas dataframe (w/o duplicates)\"\n",
    "    df = pd.DataFrame.from_records([r.properties for r in self.search()])\n",
    "    # For datasets that have been processed more than once, retain the last result.\n",
    "    df = (df.sort_values(by=['processingDate'])\n",
    "            .drop_duplicates(subset=['groupID',\n",
    "                                     'beamMode',\n",
    "                                     'processingLevel',\n",
    "                                     'startTime',\n",
    "                                     'stopTime'], keep='last')\n",
    "    )\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "obj.as_dataframe()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "@patch\n",
    "def scene_list(\n",
    "    self:SearchOpera,\n",
    "):\n",
    "    return self.as_dataframe()['sceneName'].to_list()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "obj.scene_list()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Transfer files locally."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_urls(r):\n",
    "    # Construct a dictionary of URLs for the polarization bands.\n",
    "    pols = r.properties['polarization']\n",
    "    pattern = re.compile(f\"({'|'.join(pols) + '|mask'})\\.tif$\")\n",
    "    tif_dict = {}\n",
    "    for url in [r.properties['url']] + r.properties['additionalUrls']:\n",
    "        m = pattern.search(url)\n",
    "        if m:\n",
    "            tif_dict[m.group(1)] = url\n",
    "    return tif_dict"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "@patch\n",
    "def to_local(\n",
    "    self:SearchOpera,\n",
    "):\n",
    "    \"Transfer static ASF results to local system, unzip, and update the job dictionary.\"    \n",
    "    logging.info(f'Starting asf_static.to_local()')\n",
    "\n",
    "    for r in self.search():\n",
    "        logging.info(f'  Processing {r.properties[\"fileID\"]}')\n",
    "        r.properties['url_set'] = get_urls(r)\n",
    "        r.properties[self.LOCAL_PROPNAME] = {}\n",
    "        for url_key, url_value in r.properties['url_set'].items():\n",
    "            filename = f'{r.properties[\"fileID\"]}_{url_key}.tif'\n",
    "            asf_search.download_url(\n",
    "                url=url_value,\n",
    "                path=self.local_storage,\n",
    "                filename=filename\n",
    "            )\n",
    "            r.properties[self.LOCAL_PROPNAME][url_key] = os.path.join(self.local_storage, filename)\n",
    "        # display(JSON(r.properties))\n",
    "    logging.info(f'Finished asf_static.to_local()')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pprint import pprint\n",
    "obj.to_local()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Display the extracted local files."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for r in obj.search():\n",
    "    display(JSON(r.properties['local_paths']))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Transfer to Google Cloud Storage\n",
    "\n",
    "Create an instance method for transferring results from a local computer to Google Cloud Storage."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "@patch\n",
    "def to_gcs(\n",
    "    self:SearchOpera,\n",
    "):\n",
    "    logging.info('Starting to_gcs()')\n",
    "\n",
    "    fs = gcsfs.GCSFileSystem(token='google_default')\n",
    "\n",
    "    if not fs.exists(self.gcs_bucket):\n",
    "        print('Bucket does not exist!!!')\n",
    "        fs.mkdir(self.gcs_bucket)\n",
    "\n",
    "    for r in self.search():\n",
    "        logging.info(f'  Transferring {r.properties[\"fileID\"]}')\n",
    "        r.properties[self.GCS_PATH_PROPNAME] = {}\n",
    "        \n",
    "        for key, local_filepath in r.properties[self.LOCAL_PROPNAME].items():\n",
    "            path_split = os.path.split(local_filepath)\n",
    "            filename = path_split[-1]\n",
    "            gcs_path = f'{self.gcs_bucket}/{filename}'\n",
    "            \n",
    "            if fs.exists(gcs_path):\n",
    "                logging.info(f'  GCS file already exists:\\n    {gcs_path}')\n",
    "            else:\n",
    "                logging.info(f'  Starting to transfer file to GCS:\\n    {gcs_path}')\n",
    "                # Transfer the local file to GCS.\n",
    "                print('filename', filename)\n",
    "                print('gcs_path', gcs_path)\n",
    "                fs.put_file(\n",
    "                    lpath=filepath,\n",
    "                    rpath=gcs_path\n",
    "                )    \n",
    "                logging.info(f'  Transferred file to GCS: {gcs_path}')\n",
    "            r.properties[self.GCS_PATH_PROPNAME][key] = gcs_path"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "obj.to_gcs()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Display the GCS paths."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for r in obj.search():\n",
    "    display(JSON(r.properties[obj.GCS_PATH_PROPNAME]))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create a GEE Asset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "@patch\n",
    "def create_gee_asset(\n",
    "    self:SearchOpera,\n",
    "):\n",
    "    \"Create an Earth Engine asset.\"\n",
    "    logging.info(f'Starting create_gee_asset()')\n",
    "    \n",
    "    ee.Initialize(project=self.gee_gcp_project)\n",
    "    \n",
    "    create_gee_image_collection(self.gee_gcp_project, self.gee_image_collection)\n",
    "\n",
    "    for r in self.search():\n",
    "        logging.info(f'  Creating GEE assets for {r.properties[\"fileID\"]}')\n",
    "        r.properties[self.GEE_ASSET_PROPNAME] = {}\n",
    "\n",
    "        display(JSON({'r.meta': r.meta, 'r.properties': r.properties}))\n",
    "        start_time = r.properties['startTime']\n",
    "        end_time = r.properties['stopTime']\n",
    "        description = (f\"{r.properties['platform']}\"\n",
    "                       f\" - {r.properties['processingLevel']}\"\n",
    "                       f\" - {r.properties['beamModeType']}\")\n",
    "        # id = f\"{self.job_dict['job_id']}\"\n",
    "        id = r.properties[\"fileID\"]\n",
    "        \n",
    "        for band, gcs_path in r.properties[self.GCS_PATH_PROPNAME].items():\n",
    "            print('band', band)\n",
    "            print('gcs_path', gcs_path)\n",
    "\n",
    "            request = {\n",
    "                'type': 'IMAGE',\n",
    "                'bands': {  # TODO: Update this once multi-band COG assets are supported\n",
    "                    'id': band\n",
    "                },\n",
    "                'gcs_location': {\n",
    "                    'uris': [f'gs://{gcs_path}']\n",
    "                },\n",
    "                'properties': {\n",
    "                    'source':  r.properties['url'],\n",
    "                    'band': band  # TODO: Remove this once multi-band COG assets are supported\n",
    "                },\n",
    "                'startTime': start_time,  #.strftime(FORMAT_GEE_DATETIME_STRING),\n",
    "                'endTime': end_time,  #.strftime(FORMAT_GEE_DATETIME_STRING),\n",
    "                'description': description\n",
    "            }\n",
    "\n",
    "            path_parts = [\n",
    "                'projects',\n",
    "                self.gee_gcp_project,\n",
    "                'assets',\n",
    "                self.gee_image_collection,\n",
    "                # TODO: Remove the band suffix once multi-band COG assets are supported\n",
    "                f'{id}_{band}'.replace(\".\", \"_\") \n",
    "            ]\n",
    "            assetname = os.path.join(*[x for x in path_parts if x is not None])\n",
    "\n",
    "\n",
    "            logging.debug(f'request = {request}')\n",
    "            logging.debug(f'assetname = {assetname}')\n",
    "            try:\n",
    "                ee.data.createAsset(\n",
    "                    value=request,\n",
    "                    path=assetname\n",
    "                )  \n",
    "                logging.info(f'Finished creating a GEE asset:\\n    {assetname}.')\n",
    "            except ee.EEException as e:\n",
    "                print(f'e = {e}')\n",
    "                if \"does not exist or doesn't allow this operation\" in str(e):\n",
    "                    raise(e)\n",
    "                else:\n",
    "                    raise(e)  # TODO: Add logic to parse the EEException message.\n",
    "                    logging.info('GEE asset already exists. Skipping.')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "obj.create_gee_asset()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#| hide\n",
    "import nbdev; nbdev.nbdev_export()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}