clean scripts

geosyspy/image_reference.py

@@ -1,4 +1,3 @@
-
 class ImageReference:
 
     def __init__(self, image_id, image_date, image_sensor, season_field_id):

geosyspy/services/analytics_fabric_service.py

@@ -36,7 +36,7 @@ def build_timestamp_query_parameters(start_date: Optional[datetime] = None,
             return f'&Timestamp=$between:{start_date}|{end_date}'
 
     def create_schema_id(self, schema_id: str,
-                         schema: dict):
+                        schema: dict):
         """Create a schema in Analytics Fabrics
 
         Args:

geosyspy/services/vegetation_time_series_service.py

@@ -64,7 +64,7 @@ def get_time_series_by_pixel(self, season_field_id: str,
                                    start_date: datetime,
                                    end_date: datetime,
                                    indicator: str) -> pd.DataFrame:
-            """Returns a pandas DataFrame.
+        """Returns a pandas DataFrame.
 
             This method returns a time series of 'indicator' by pixel within the range 'start_date' -> 'end_date'
             as well as the pixel's coordinates X,Y in the MODIS's sinusoidal projection as a pandas DataFrame :
@@ -93,39 +93,43 @@ def get_time_series_by_pixel(self, season_field_id: str,
 
             """
 
-            self.logger.info("Calling APIs for time series by the pixel")
-            start_date: str = start_date.strftime("%Y-%m-%d")
-            end_date: str = end_date.strftime("%Y-%m-%d")
-            parameters: str = f"/values?$offset=0&$limit=None&$count=false&SeasonField.Id={season_field_id}&index={indicator}&$filter=Date >= '{start_date}' and Date <= '{end_date}'"
-            vts_url: str = urljoin(self.base_url, GeosysApiEndpoints.VTS_BY_PIXEL_ENDPOINT.value + parameters)
-            # PSX/PSY : size in meters of one pixel
-            # MODIS_GRID_LENGTH : theoretical length of the modis grid in meters
-            # MODIS_GRID_HEIGHT : theoretical height of the modis grid in meters
-            PSX = 231.65635826
-            PSY = -231.65635826
-            MODIS_GRID_LENGTH = 4800 * PSX * 36
-            MODIS_GRID_HEIGHT = 4800 * PSY * 18
-
-            response = self.http_client.get(vts_url)
-
-            if response.status_code == 200:
-                df = pd.json_normalize(response.json())
-                df.set_index("date", inplace=True)
-
-                # Extracts h, v, i and j from the pixel dataframe
-                self.logger.info("Computing X and Y coordinates per pixel... ")
-                df["h"] = df["pixel.id"].str.extract(r"h(.*)v").astype(int)
-                df["v"] = df["pixel.id"].str.extract(r"v(.*)i").astype(int)
-                df["i"] = df["pixel.id"].str.extract(r"i(.*)j").astype(int)
-                df["j"] = df["pixel.id"].str.extract(r"j(.*)$").astype(int)
-
-                # XUL/YUL : The coordinates of the top left corner of the tile h,v's top left pixel
-                #  X/Y : the coordinates of the top left corner of the i,j pixel
-                df["XUL"] = (df["h"] + 1) * 4800 * PSX - MODIS_GRID_LENGTH / 2
-                df["YUL"] = (df["v"] + 1) * 4800 * PSY + MODIS_GRID_HEIGHT / 2
-                df["X"] = df["i"] * PSX + df["XUL"]
-                df["Y"] = df["j"] * PSY + df["YUL"]
-                self.logger.info("Done ! ")
-                return df[["index", "value", "pixel.id", "X", "Y"]]
-            else:
-                self.logger.info(response.status_code)
+        self.logger.info("Calling APIs for time series by the pixel")
+        start_date: str = start_date.strftime("%Y-%m-%d")
+        end_date: str = end_date.strftime("%Y-%m-%d")
+        parameters: str = f"/values?$offset=0&$limit=None&$count=false&SeasonField.Id={season_field_id}&index={indicator}&$filter=Date >= '{start_date}' and Date <= '{end_date}'"
+        vts_url: str = urljoin(self.base_url, GeosysApiEndpoints.VTS_BY_PIXEL_ENDPOINT.value + parameters)
+        response = self.http_client.get(vts_url)
+
+        if response.status_code == 200:
+            return self._extracted_from_get_time_series_by_pixel_50(response)
+        else:
+            self.logger.info(response.status_code)
+
+    # TODO Rename this here and in `get_time_series_by_pixel`
+    def _extracted_from_get_time_series_by_pixel_50(self, response):
+        df = pd.json_normalize(response.json())
+        df.set_index("date", inplace=True)
+
+        # Extracts h, v, i and j from the pixel dataframe
+        self.logger.info("Computing X and Y coordinates per pixel... ")
+        df["h"] = df["pixel.id"].str.extract(r"h(.*)v").astype(int)
+        df["v"] = df["pixel.id"].str.extract(r"v(.*)i").astype(int)
+        df["i"] = df["pixel.id"].str.extract(r"i(.*)j").astype(int)
+        df["j"] = df["pixel.id"].str.extract(r"j(.*)$").astype(int)
+
+        # PSX/PSY : size in meters of one pixel
+        # MODIS_GRID_LENGTH : theoretical length of the modis grid in meters
+        # MODIS_GRID_HEIGHT : theoretical height of the modis grid in meters
+        PSX = 231.65635826
+        MODIS_GRID_LENGTH = 4800 * PSX * 36
+        # XUL/YUL : The coordinates of the top left corner of the tile h,v's top left pixel
+        #  X/Y : the coordinates of the top left corner of the i,j pixel
+        df["XUL"] = (df["h"] + 1) * 4800 * PSX - MODIS_GRID_LENGTH / 2
+        PSY = -231.65635826
+        MODIS_GRID_HEIGHT = 4800 * PSY * 18
+
+        df["YUL"] = (df["v"] + 1) * 4800 * PSY + MODIS_GRID_HEIGHT / 2
+        df["X"] = df["i"] * PSX + df["XUL"]
+        df["Y"] = df["j"] * PSY + df["YUL"]
+        self.logger.info("Done ! ")
+        return df[["index", "value", "pixel.id", "X", "Y"]]

geosyspy/utils/__init__.py

@@ -1 +1,8 @@
-__all__ = ["constants", "geosys_platform_urls", "oauth2_client", "http_client", "helper", "jwt_validator"]
+__all__ = [
+    "constants",
+    "geosys_platform_urls",
+    "oauth2_client",
+    "http_client",
+    "helper",
+    "jwt_validator"
+    ]

geosyspy/utils/constants.py

@@ -14,7 +14,7 @@ class SatelliteImageryCollection(Enum):
     RESOURCESAT2 = "RESOURCESAT2"
     CBERS_4 = "CBERS_4"
     KAZSTSAT = "KAZSTSAT"
-    GAOPFEN = "GAOPFEN"
+    GAOFEN = "GAOFEN"
 
 
 class WeatherTypeCollection(Enum):
@@ -171,11 +171,13 @@ class GeosysApiEndpoints(Enum):
 
 
 LR_SATELLITE_COLLECTION = [SatelliteImageryCollection.MODIS]
-MR_SATELLITE_COLLECTION = [SatelliteImageryCollection.LANDSAT_8, SatelliteImageryCollection.LANDSAT_9,
-                           SatelliteImageryCollection.SENTINEL_2, SatelliteImageryCollection.ALSAT_1B,
-                           SatelliteImageryCollection.CBERS_4, SatelliteImageryCollection.DEIMOS_1, 
-                           SatelliteImageryCollection.GAOPFEN, SatelliteImageryCollection.KAZSTSAT, 
-                           SatelliteImageryCollection.RESOURCESAT2]
+MR_SATELLITE_COLLECTION = [
+    SatelliteImageryCollection.LANDSAT_8,SatelliteImageryCollection.LANDSAT_9,
+    SatelliteImageryCollection.SENTINEL_2, SatelliteImageryCollection.ALSAT_1B,
+    SatelliteImageryCollection.CBERS_4, SatelliteImageryCollection.DEIMOS_1,
+    SatelliteImageryCollection.GAOFEN, SatelliteImageryCollection.KAZSTSAT,
+    SatelliteImageryCollection.RESOURCESAT2
+    ]
 
 PRIORITY_HEADERS = {"bulk": "Geosys_API_Bulk", "realtime": ""}
 SEASON_FIELD_ID_REGEX = r"\sId:\s(\w+),"

geosyspy/utils/helper.py

@@ -5,10 +5,18 @@
 
 
 class Helper:
+    """A class providing helper functions for working with geometries.
 
+    Methods:
+        get_matched_str_from_pattern(pattern, text): 
+            Returns the first occurrence of the matched pattern in text.
+        convert_to_wkt(geometry): Convert a geometry (WKT or geoJson) to WKT.
+        is_valid_wkt(geometry): Check if the geometry is a valid WKT.
+
+    """
     @staticmethod
     def get_matched_str_from_pattern(pattern: str,
-                                     text: str) -> str:
+                                    text: str) -> str:
         """Returns the first occurence of the matched pattern in text.
 
         Args:
@@ -35,16 +43,13 @@ def convert_to_wkt(geometry):
 
         try:
             # check if the geometry is a valid WKT
-            if Helper.is_valid_wkt(geometry):
-                # return the wkt
-                return geometry
-        except:
+            return geometry if Helper.is_valid_wkt(geometry) else None
+        except Exception:
             try:
                 # check if the geometry is a valid geoJson
                 geojson_data = json.loads(geometry)
                 geom = shape(geojson_data)
                 geometry = geom.wkt
-
                 return geometry
 
             except ValueError:
@@ -65,4 +70,5 @@ def is_valid_wkt(geometry):
             wkt.loads(geometry)
             return True
         except ValueError:
-            return False
+            return False
+

geosyspy/utils/http_client.py

@@ -1,7 +1,7 @@
-from . import oauth2_client
 from oauthlib.oauth2 import TokenExpiredError
 from requests_oauthlib import OAuth2Session
 
+from . import oauth2_client
 
 def renew_access_token(func):
     """Decorator used to wrap the Geosys class's http methods.
@@ -21,9 +21,16 @@ def wrapper(self, *args, **kwargs):
 
     return wrapper
 
-
 class HttpClient:
+    """A class for making HTTP requests with OAuth2 authentication.
+
+    Methods:
+        get(url_endpoint): Gets the url_endpoint.
+        post(url_endpoint, payload): Posts payload to the url_endpoint.
+        patch(url_endpoint, payload): Patches payload to the url_endpoint.
+        get_access_token(): Returns the access token.
 
+    """
     def __init__(
             self,
             client_id: str,
@@ -45,10 +52,11 @@ def __init__(
         )
         self.access_token = self.__client_oauth.token
 
-        self.__client = OAuth2Session(self.__client_oauth.client_id, token=self.__client_oauth.token)
+        self.__client = OAuth2Session(self.__client_oauth.client_id,
+                                    token=self.__client_oauth.token)
 
     @renew_access_token
-    def get(self, url_endpoint: str, headers={}):
+    def get(self, url_endpoint: str, headers=None):
         """Gets the url_endpopint.
 
         Args:
@@ -57,10 +65,12 @@ def get(self, url_endpoint: str, headers={}):
         Returns:
             A response object.
         """
+        if headers is None:
+            headers = {}
         return self.__client.get(url_endpoint, headers=headers)
 
     @renew_access_token
-    def post(self, url_endpoint: str, payload: dict, headers={}):
+    def post(self, url_endpoint: str, payload: dict, headers=None):
         """Posts payload to the url_endpoint.
 
         Args:
@@ -70,6 +80,8 @@ def post(self, url_endpoint: str, payload: dict, headers={}):
         Returns:
             A response object.
         """
+        if headers is None:
+            headers = {}
         return self.__client.post(url_endpoint, json=payload, headers=headers)
 
     @renew_access_token
@@ -86,4 +98,9 @@ def patch(self, url_endpoint: str, payload: dict):
         return self.__client.patch(url_endpoint, json=payload)
 
     def get_access_token(self):
+        """Returns the access token.
+
+        Returns:
+            The access token.
+        """
         return self.access_token

geosyspy/utils/jwt_validator.py

@@ -1,17 +1,18 @@
-import jwt
 import base64
 import json
 from datetime import datetime, timezone
+import logging
+
+import jwt
 from cryptography import x509
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import serialization
-import logging
 
 
 logger = logging.getLogger(__name__)
 
 
-def check_token_validity(token: str, certificate_key: str, algorithms=['RS256']) -> bool:
+def check_token_validity(token: str, certificate_key: str, algorithms=None) -> bool:
     """
     Check the validity of a JWT token.
 
@@ -24,36 +25,42 @@ def check_token_validity(token: str, certificate_key: str, algorithms=['RS256'])
     Returns:
         bool: True if the token is valid, False otherwise.
     """
+    if algorithms is None:
+        algorithms = ['RS256']
     try:
-        cert_bytes = certificate_key.encode()
-        cert = x509.load_pem_x509_certificate(cert_bytes, default_backend())
-        public_key = cert.public_key()
-
-        # extract public key in PEM format
-        public_key = public_key.public_bytes(
-            encoding=serialization.Encoding.PEM,
-            format=serialization.PublicFormat.SubjectPublicKeyInfo
-        ).decode()
-        aud = __get_audience_from_token(token)
-        decoded_token = jwt.decode(token, public_key, algorithms=algorithms, audience=aud)
-        expiration_timestamp = decoded_token['exp']
-        expiration_datetime = datetime.fromtimestamp(expiration_timestamp, tz=timezone.utc)
-
-        if expiration_datetime > datetime.now(timezone.utc):
-            return True
-        else:
-            return False
+        return _extracted_from_check_token_validity_17(
+            certificate_key, token, algorithms
+        )
     except jwt.ExpiredSignatureError:
         logger.error("Expired Token")
         return False
     except jwt.InvalidTokenError as e:
-        logger.error("Invalid Token." + str(e))
+        logger.error("Invalid Token.%s",str(e))
         return False
     except Exception as e:
-        logger.error("Invalid Token." + str(e))
+        logger.error("Invalid Token.%s",str(e))
         return False
 
 
+# TODO Rename this here and in `check_token_validity`
+def _extracted_from_check_token_validity_17(certificate_key, token, algorithms):
+    cert_bytes = certificate_key.encode()
+    cert = x509.load_pem_x509_certificate(cert_bytes, default_backend())
+    public_key = cert.public_key()
+
+    # extract public key in PEM format
+    public_key = public_key.public_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PublicFormat.SubjectPublicKeyInfo
+    ).decode()
+    aud = __get_audience_from_token(token)
+    decoded_token = jwt.decode(token, public_key, algorithms=algorithms, audience=aud)
+    expiration_timestamp = decoded_token['exp']
+    expiration_datetime = datetime.fromtimestamp(expiration_timestamp, tz=timezone.utc)
+
+    return expiration_datetime > datetime.now(timezone.utc)
+
+
 def __get_audience_from_token(token: str) -> str:
     """
     Get the audience from a JWT token.
@@ -76,5 +83,4 @@ def __get_audience_from_token(token: str) -> str:
     payload_json = decoded_payload.decode('utf-8')
 
     payload_dict = json.loads(payload_json)
-    audience = payload_dict.get('aud')
-    return audience
+    return payload_dict.get('aud')