New FamilyProduct class to get information about l1c, l2a, cloudmask and...

New FamilyProduct class to get information about l1c, l2a, cloudmask and indice products in database from an identifier string

sen2chain/__init__.py

@@ -27,6 +27,7 @@ from .products import (
     OldCloudMaskProduct,
     NewCloudMaskProduct,
     IndiceProduct,
+    FamilyProduct,
 )
 from .library import Library
 from .data_request import DataRequest
@@ -46,6 +47,7 @@ from .utils import (
     get_Sen2Cor_version, 
     get_latest_s2c_version_path,
     set_permissions,
+    get_cm_dict, get_cm_string_from_dict, get_indice_from_identifier
 )
 from .geo_utils import (
     serialise_tiles_index,

sen2chain/products.py

@@ -20,7 +20,15 @@ from typing import (
     Union
 )
 from packaging import version
-from .utils import grouper, set_permissions, get_Sen2Cor_version, get_latest_s2c_version_path
+from .utils import (
+    grouper,
+    set_permissions,
+    get_Sen2Cor_version,
+    get_latest_s2c_version_path,
+    get_cm_dict,
+    get_cm_string_from_dict,
+    get_indice_from_identifier
+)
 from .config import Config, SHARED_DATA
 from .xmlparser import MetadataParser, Sen2ChainMetadataParser
 from .sen2cor import process_sen2cor
@@ -1733,3 +1741,100 @@ class IndiceProduct:
                 )
             else:
                 Sen2ChainMetadataParser(self._info_path).init_metadata()
+
+class FamilyProduct(dict):
+    """Family product class
+    :param identifier: cloudmask filename.
+    """
+    
+    _l1c_library_path = Path(Config().get("l1c_path"))
+    _l2a_library_path = Path(Config().get("l2a_path"))
+    _cloudmask_library_path = Path(Config().get("cloudmasks_path"))
+    _indice_library_path = Path(Config().get("indices_path"))
+    
+    def __init__(
+        self,
+        identifier: str = None,
+    ):
+        if not (identifier):
+            raise ValueError(
+                "Identifier cannot be empty"
+            )
+        else:
+            fid_tile = self.get_family_id_tile(identifier)
+            if fid_tile:
+                self.update(fid_tile)
+                self.family_id = fid_tile["family_id"]
+                self.tile = fid_tile["tile"]
+            self.l1c_id = self.get_l1c()
+            self.l2a_id = self.get_l2a()
+            self.cloudmasks = self.get_cloudmasks()
+            self.indices = self.get_indices()
+            for key, val in self.__dict__.items():
+                self[key] = val
+
+    def get_family_id_tile(self, identifier):
+        try:
+            pat = re.compile(        
+                r".*(?P<family_id>"
+                + "[0-9]{8}T[0-9]{6}"
+                + "_N[0-9]{4}_R[0-9]{3}"
+                + "_T(?P<tile>[0-9]{2}[A-Z]{3})"
+                + "_[0-9]{8}T[0-9]{6}"
+                + ").*"
+            )
+            return pat.match(identifier).groupdict()
+        except:
+            pass
+        
+    def get_l1c(self):
+        library_path = self._l1c_library_path / self.tile
+        paths = list(library_path.glob("*L1C*" + self.family_id + "*"))
+        if len(paths) > 0:
+            return paths[0].stem
+        else:
+            return None
+
+    def get_l2a(self):
+        library_path = self._l2a_library_path / self.tile
+        paths = list(library_path.glob("*L2A*" + self.family_id + "*"))
+        if len(paths) > 0:
+            return paths[0].stem
+        else:
+            return None    
+    
+    def get_cloudmasks(self):
+        cloudmasks = []
+        library_path = self._cloudmask_library_path / self.tile / self.l2a_id
+        paths = list(library_path.glob("*L2A*" + self.family_id + "_CM*.jp2"))
+        for item in paths:
+            cm_dict = get_cm_dict(item.stem)
+            cm_string = get_cm_string_from_dict(cm_dict)
+            cm_dict["cm_string"] = cm_string
+            cloudmasks.append(cm_dict)
+        return cloudmasks    
+        
+    def get_indices(self):
+        indices = []
+        library_path = self._indice_library_path
+        paths = list(
+            library_path.glob(
+                "*/"
+                + self.tile + "/"
+                + self.l2a_id + "/"
+                + "*.jp2"
+            )
+        )
+        for item in paths:
+            logger.info(item)
+            indice_dict = {}
+            indice_dict["indice"] = get_indice_from_identifier(item.stem)
+            cm_dict = get_cm_dict(item.stem)
+            if cm_dict:
+                indice_dict.update(cm_dict)
+                indice_dict["cm_string"] = get_cm_string_from_dict(cm_dict)
+                indice_dict["indice_string"] = indice_dict["indice"] + "_" + indice_dict["cm_string"]
+            else:
+                indice_dict["indice_string"] = indice_dict["indice"]
+            indices.append(indice_dict)
+        return indices    

sen2chain/utils.py

@@ -200,3 +200,89 @@ def get_latest_s2c_version_path(l1c_identifier):
         else:
             return current_path
 
+def get_cm_dict(identifier) -> dict:
+    """Returns cloudmask version from an identifier string.
+    :param string: string from which to extract the version name.
+    can be a cloudmask or an indice identifier
+    """
+    returned_val = None
+    try:
+        pat = re.compile(r".*(?P<cm_version>CM00[1-2])")
+        returned_val = pat.match(identifier).groupdict()
+        
+    except:
+        try:
+            pat = re.compile(
+                r".*(?P<cm_version>CM003)"
+                + "-PRB(?P<probability>.*)"
+                + "-ITER(?P<iterations>.*)"
+            )
+            returned_val = pat.match(identifier).groupdict()
+            
+        except:
+            try:
+                pat = re.compile(
+                    r".*(?P<cm_version>CM004)"
+                    + "-CSH(?P<cld_shad>.*)"
+                    + "-CMP(?P<cld_med_prob>.*)"
+                    + "-CHP(?P<cld_hi_prob>.*)"
+                    + "-TCI(?P<thin_cir>.*)"
+                    + "-ITER(?P<iterations>.*)"
+                )
+                returned_val = pat.match(identifier).groupdict()
+            except:
+               pass
+    
+    if returned_val:
+        if "probability" not in returned_val:
+            returned_val["probability"] = 1
+        if "iterations" not in returned_val:
+            returned_val["iterations"] = 5
+        if "cld_shad" not in returned_val:
+            returned_val["cld_shad"] = True
+        if "cld_med_prob" not in returned_val:
+            returned_val["cld_med_prob"] = True
+        if "cld_hi_prob" not in returned_val:
+            returned_val["cld_hi_prob"] = True
+        if "thin_cir" not in returned_val:
+            returned_val["thin_cir"] = True
+        
+        returned_val["probability"] = int(returned_val["probability"])
+        returned_val["iterations"] = int(returned_val["iterations"])
+        returned_val["cld_shad"] = returned_val["cld_shad"] in ["1", 1, True, "True"] 
+        returned_val["cld_med_prob"] = returned_val["cld_med_prob"] in ["1", 1, True, "True"]
+        returned_val["cld_hi_prob"] = returned_val["cld_hi_prob"] in ["1", 1, True, "True"]
+        returned_val["thin_cir"] = returned_val["thin_cir"] in ["1", 1, True, "True"]
+    
+    return returned_val
+    
+def get_cm_string_from_dict(cm_dict) -> str:
+    if cm_dict:
+        if cm_dict["cm_version"] == "CM001":
+            cm_string = "CM001"
+        elif cm_dict["cm_version"] == "CM002":
+            cm_string = "CM002-B11"
+        elif cm_dict["cm_version"] == "CM003":
+            cm_string = (
+                "CM003" +
+                "-PRB" + str(cm_dict["probability"])  + 
+                "-ITER" + str(cm_dict["iterations"])
+            )
+        elif cm_dict["cm_version"] == "CM004":
+            cm_string = (
+                "CM004" + 
+                "-CSH" + str(1 * cm_dict["cld_shad"]) + 
+                "-CMP" + str(1 * cm_dict["cld_med_prob"]) + 
+                "-CHP" + str(1 * cm_dict["cld_hi_prob"]) + 
+                "-TCI" + str(1 * cm_dict["thin_cir"]) + 
+                "-ITER" + str(cm_dict["iterations"])
+            )
+        else:
+            cm_string = None
+    else:
+        cm_string = None
+    return cm_string
+    
+def get_indice_from_identifier(identifier) -> str:
+    indice = (identifier.replace(".", "_").split("_")[7]).upper()
+    return indice