Feature/adding mdm unit profile unit categories and difference maps

examples.ipynb

@@ -2,20 +2,20 @@
  "cells": [
   {
    "cell_type": "code",
-   "execution_count": 2,
+   "execution_count": 1,
    "metadata": {},
    "outputs": [
     {
      "name": "stderr",
      "output_type": "stream",
      "text": [
-      "2024-06-14 11:18:20,153 - geosyspy.utils.oauth2_client - INFO - Authenticated\n"
+      "2024-07-31 08:29:17,754 - geosyspy.utils.oauth2_client - INFO - Authenticated\n"
      ]
     }
    ],
    "source": [
-    "from geosyspy import Geosys\n",
     "import os\n",
+    "from geosyspy import Geosys\n",
     "from dotenv import load_dotenv\n",
     "import datetime as dt\n",
     "from dateutil.relativedelta import relativedelta\n",
@@ -5516,6 +5516,31 @@
     "df_new = pd.DataFrame.from_records([data_dict])\n",
     "df_new\n"
    ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 2,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "download ok\n"
+     ]
+    }
+   ],
+   "source": [
+    "# display difference map\n",
+    "season_field_id = \"bgbrzez\"\n",
+    "image_early = \"sentinel-2-l2a%7CS2B_13SGC_20230520_0_L2A\"\n",
+    "image_late = \"sentinel-2-l2a%7CS2B_13SGC_20230530_0_L2A\"\n",
+    "image_diff = client.download_image_difference_map(season_field_id=season_field_id,image_id_earliest=image_early,image_id_latest=image_late)\n",
+    "if image_diff is not None:\n",
+    "    print(\"download ok\")\n",
+    "else:\n",
+    "    print(\"download nok\")"
+   ]
   }
  ],
  "metadata": {
@@ -5534,7 +5559,7 @@
    "name": "python",
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
-   "version": "3.11.9"
+   "version": "3.12.4"
   }
  },
  "nbformat": 4,

geosyspy/geosys.py

@@ -1,19 +1,19 @@
 """ Geosysoy class"""
 
-from enum import Enum
 import io
-import zipfile
-from typing import List, Optional
-from pathlib import Path
 import logging
+import zipfile
 from datetime import datetime
+from enum import Enum
+from pathlib import Path
+from typing import List, Optional
 
 import numpy as np
 import pandas as pd
 import rasterio
+import retrying
 import xarray as xr
 from rasterio.io import MemoryFile
-import retrying
 
 from geosyspy import image_reference
 from geosyspy.services.agriquest_service import AgriquestService
@@ -24,25 +24,25 @@
 from geosyspy.services.master_data_management_service import MasterDataManagementService
 from geosyspy.services.vegetation_time_series_service import VegetationTimeSeriesService
 from geosyspy.services.weather_service import WeatherService
-from geosyspy.utils.geosys_platform_urls import GEOSYS_API_URLS, GIS_API_URLS
 from geosyspy.utils.constants import (
-    Env,
-    Region,
-    WeatherTypeCollection,
     LR_SATELLITE_COLLECTION,
-    SatelliteImageryCollection,
     MR_SATELLITE_COLLECTION,
-    Harvest,
     AgriquestBlocks,
     AgriquestCommodityCode,
     AgriquestWeatherType,
     CropIdSeason,
     Emergence,
+    Env,
+    Harvest,
+    Region,
+    SatelliteImageryCollection,
+    WeatherTypeCollection,
     ZarcCycleType,
     ZarcSoilType,
 )
-from geosyspy.utils.http_client import HttpClient
+from geosyspy.utils.geosys_platform_urls import GEOSYS_API_URLS, GIS_API_URLS
 from geosyspy.utils.helper import Helper
+from geosyspy.utils.http_client import HttpClient
 
 
 class Geosys:
@@ -200,21 +200,30 @@ def get_satellite_image_time_series(
                 if not season_field_id:
                     # extract seasonfield id from geometry
                     season_field_id = (
-                        self.__master_data_management_service.extract_season_field_id(polygon)
+                        self.__master_data_management_service.extract_season_field_id(
+                            polygon
+                        )
+                    )
+                elif (
+                    not self.__master_data_management_service.check_season_field_exists(
+                        season_field_id
                     )
-                elif not self.__master_data_management_service.check_season_field_exists(
-                    season_field_id
                 ):
                     raise ValueError(
                         f"Cannot access {season_field_id}. It is not existing or connected user doens't have access to it."
                     )
-                
+
                 return self.__vts_service.get_time_series_by_pixel(
                     season_field_id, start_date, end_date, indicators[0]
                 )
             elif set(collections).issubset(set(MR_SATELLITE_COLLECTION)):
                 return self.__get_images_as_dataset(
-                    season_field_id, polygon, start_date, end_date, collections, indicators[0]
+                    season_field_id,
+                    polygon,
+                    start_date,
+                    end_date,
+                    collections,
+                    indicators[0],
                 )
         else:
             raise TypeError(
@@ -237,7 +246,7 @@ def get_satellite_coverage_image_references(
         ],
         polygon: Optional[str] = None,
         season_field_id: Optional[str] = None,
-        coveragePercent: Optional[int] = 80
+        coveragePercent: Optional[int] = 80,
     ) -> tuple:
         """Retrieves a list of images that covers a polygon on a specific date range.
         The return is a tuple: a dataframe with all the images covering the polygon, and
@@ -262,7 +271,13 @@ def get_satellite_coverage_image_references(
             )
 
         df = self.__map_product_service.get_satellite_coverage(
-            season_field_id, polygon, start_date, end_date, "", coveragePercent, collections
+            season_field_id,
+            polygon,
+            start_date,
+            end_date,
+            "",
+            coveragePercent,
+            collections,
         )
         images_references = {}
         if df is not None:
@@ -297,9 +312,27 @@ def download_image(self, polygon, image_id, indicator: str = "", path: str = "")
             self.logger.info("writing to %s", path)
             f.write(response_zipped_tiff.content)
 
-    def get_product(self, season_field_id, image_id, indicator, image= None):
+    def download_image_difference_map(
+        self, season_field_id, image_id_earliest, image_id_latest
+    ):
+        """Downloads a satellite image locally resulting of the difference between 2 images
+
+        Args:
+            season_field_id : season_field_id
+            image_id_earliest : the earliest image reference from the satellite coverage.
+            image_id_latest : the latest image reference from the satellite coverage.
+        """
+        response = self.__map_product_service.get_zipped_tiff_difference_map(
+            season_field_id, image_id_earliest, image_id_latest
+        )
+
+        return response
+
+    def get_product(self, season_field_id, image_id, indicator, image=None):
 
-        response = self.__map_product_service.get_product(season_field_id, image_id, indicator, image)
+        response = self.__map_product_service.get_product(
+            season_field_id, image_id, indicator, image
+        )
 
         return response
 
@@ -311,7 +344,7 @@ def __get_images_as_dataset(
         end_date: datetime,
         collections: Optional[list[SatelliteImageryCollection]],
         indicator: str,
-        coveragePercent: int = 80
+        coveragePercent: int = 80,
     ) -> "np.ndarray[np.Any , np.dtype[np.float64]]":
         """Returns all the 'sensors_list' images covering 'polygon' between
         'start_date' and 'end_date' as a xarray dataset.
@@ -349,7 +382,13 @@ def get_coordinates_by_pixel(raster):
         # and sorts them by resolution, from the highest to the lowest.
         # Keeps only the first image if two are found on the same date.
         df_coverage = self.__map_product_service.get_satellite_coverage(
-            season_field_id, polygon, start_date, end_date, indicator, coveragePercent, collections
+            season_field_id,
+            polygon,
+            start_date,
+            end_date,
+            indicator,
+            coveragePercent,
+            collections,
         )
 
         # Return empty dataset if no coverage on the polygon between start_date, end_date
@@ -606,12 +645,28 @@ def get_available_permissions(self):
             permissions: a string array containing all available permissions of the connected user
         """
         # get crop code list
-        result = self.__master_data_management_service.get_permission_codes()
+        result = self.__master_data_management_service.get_profile("permissions")
 
         # build a string array with all available permission codes for the connected user
 
         return result["permissions"]
 
+    def get_user_area_conversion_rate(self):
+        """Returns the user's defined area's unit of measurement conversion's rate to square metres."""
+
+        # get crop code list
+        result = self.__master_data_management_service.get_profile(
+            "unitProfileUnitCategories"
+        )
+
+        conversion_rate = list(
+            filter(
+                lambda x: x["unitCategory"]["id"] == "FIELD_SURFACE",
+                result["unitProfileUnitCategories"],
+            )
+        )[0]["unit"]["conversionRate"]
+        return conversion_rate
+
     def get_sfid_from_geometry(self, geometry: str):
         """Retrieves every season field ID contained within the passed geometry.
 
@@ -620,8 +675,12 @@ def get_sfid_from_geometry(self, geometry: str):
         Returns:
             ids: an array containing all the seasonfield ids
         """
-        result = self.__master_data_management_service.retrieve_season_fields_in_polygon(geometry)
-        ids = [item['id'] for item in result.json()]
+        result = (
+            self.__master_data_management_service.retrieve_season_fields_in_polygon(
+                geometry
+            )
+        )
+        ids = [item["id"] for item in result.json()]
         return ids
 
     def get_season_fields(self, season_field_ids: List[str]):
@@ -632,7 +691,9 @@ def get_season_fields(self, season_field_ids: List[str]):
         Returns:
             result: an array containing all the seasonfield
         """
-        result = self.__master_data_management_service.get_season_fields(season_field_ids)
+        result = self.__master_data_management_service.get_season_fields(
+            season_field_ids
+        )
         return result
 
     ###########################################
@@ -1218,7 +1279,7 @@ def get_zarc_analytics(
         )
         return self.check_status_and_metrics(task_id, "ZARC", sf_unique_id)
 
-    def get_farm_info_from_location(self, latitude:str, longitude:str):
+    def get_farm_info_from_location(self, latitude: str, longitude: str):
         """get farm info from CAR layer
 
         Args: