Merge branch 'master' of https://framagit.org/espace-dev/sen2chain

sen2chain/data/job_ini.cfg

@@ -5,23 +5,30 @@
 # tries: the number of times the download should loop before stopping, to download OFFLINE products
 # sleep: the time in min to wait between loops
 # nb_proc: the number of cpu cores to use for this job, default 8
+# copy_l2a_side_products: to duplicate msk_cldprb_20m and scl_20m from l2a folder to cloudmask folder after l2a production. 
+#   Interesting if you plan to remove l2a to save disk space, but want to keep these 2 files for cloudmask generation and better extraction
+#   Possible values: True | False
 
 # tile: tile identifier, format ##XXX, comment line using ! before tile name
 # date_min the start date for this task, possible values:
-#    empty (2015-01-01 will be used) | any date | now-xx (xx nb of days before now to consider)
+#    empty (2015-01-01 will be used) | any date | today-xx (xx nb of days before today to consider)
 # date_max the last date for this task, possible values:
-#    empty (9999-12-31 will be used) | any date | now
-#
+#    empty (9999-12-31 will be used) | any date | today
+# max_clouds: max cloud cover to consider for downloading images, and computing l2a products 
 # l1c: download l1c: True|False
 # l2a: compute l2a with sen2chain: True | False
-# cloudmasks: False | CM001 | CM002 | CM003-PRB1-ITER5 | CM004-CSH1-CMP1-CHP1-TCI1-ITER0
+# cloudmasks: the cloudmask(s) to compute and use to mask indice(s). Possible values range from none (False) to multiple cloudmasks:
+#    False | CM001/CM002/CM003-PRB1-ITER5/CM004-CSH1-CMP1-CHP1-TCI1-ITER0/etc.
 # indices: False | All | NDVI/NDWIGAO/etc.
-# remove: to remove downloaded L1C and/or produced L2A, possible values: False | l1c | l2a | l1c/l2a
+# remove: used to remove downloaded L1C and/or produced L2A, possible values: False | l1c | l2a | l1c/l2a
 # comments: free user comments, ie tile name, etc.
 
 logs = False
 timing = 0 0 * * *
 provider = peps
+tries = 1
+sleep = 0
 nb_proc = 8
+copy_l2a_sideproducts = False
 
-tile;date_min;date_max;max_clouds;l1c;l2a;cloudmask;indices;remove
+tile;date_min;date_max;max_clouds;l1c;l2a;cloudmask;indices;remove;comments

sen2chain/download_eodag.py

@@ -116,7 +116,11 @@ class S2cEodag:
                 self.products.remove(p)
                 logger.info("{} - local l1c {} - filtering (wrong Tile)".format(p.properties["title"], l1c_presence))
             else:
-                # logger.info(p.properties["cloudCover"])
+                try:
+                    prop = p.properties["storageStatus"]
+                except:
+                    if p.properties["storage"]["mode"] == "tier2":
+                        p.properties["storageStatus"] = "ONLINE"
                 if not(min_cloudcover <= float(p.properties["cloudCover"] or 0) <= max_cloudcover):
                     self.products.remove(p)
                     logger.info("{} - local l1c {} - filtering (CC = {}%)".format(p.properties["title"], l1c_presence, int(p.properties["cloudCover"])))
@@ -127,6 +131,7 @@ class S2cEodag:
                             self.products.remove(p)
                         else:
                             logger.info("{} - local l1c {} - remote {}".format(p.properties["title"], l1c_presence, p.properties["storageStatus"]))
+                            
                     elif ref == "l2a":
                         if (Path(Config().get("l2a_path")) / self.name / (p.properties["title"].replace("L1C_", "L2A_").replace("__OPER__", "_USER_") + ".SAFE")).exists():
                             logger.info("{} - local l1c {} - local l2a {} - filtering".format(p.properties["title"], l1c_presence, l2a_presence, p.properties["storageStatus"]))

sen2chain/multi_processing.py

@@ -2,7 +2,7 @@
 
 import multiprocessing, subprocess
 import os
-from time import sleep
+import time
 import logging
 from functools import partial
 
@@ -15,7 +15,10 @@ logging.basicConfig(level=logging.INFO)
 logger.setLevel(logging.INFO)
 
 
-def multi(product):
+def multi(product_copyl2asideproducts):
+    product = product_copyl2asideproducts[0]
+    copy_l2a_sideproducts = product_copyl2asideproducts[1]
+    proc = None
     try:
         fwd = os.path.dirname(os.path.realpath(__file__))
         logger.info("Processing {}".format(product))
@@ -26,6 +29,7 @@ def multi(product):
                 "/usr/bin/python3",
                 fwd + "/multiprocess_l2a.py",
                 product,
+                copy_l2a_sideproducts,
             ]
             proc = subprocess.Popen(cmd)
 
@@ -34,14 +38,21 @@ def multi(product):
                 "_OPER_", "_USER_"
             )
             l2a_prod = L2aProduct(l2a_identifier)
+            timeout = time.time() + 2*60*60
             while not (l2a_prod.in_library):
-                sleep(5)
+                if time.time() > timeout:
+                    logger.info("Timeout (2h) reached for Sen2Cor processing, killing process {}".format(proc.pid))
+                    os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
+                    break
+                time.sleep(5)
             logger.info("End {}".format(product))
     except:
         logger.info("Plante {}".format(product))
+        if proc:
+            logger.info("Killing process {}".format(proc.pid))
+            os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
         pass
 
-
 def l2a_multiprocessing(process_list, nb_proc=4):
     """ """
     nb_proc = max(min(len(os.sched_getaffinity(0)) - 1, nb_proc), 1)
@@ -95,7 +106,7 @@ def multi_cld_ver_pro_iter_repro(l2a_ver_pro_iter_repro):
     reprocess = l2a_ver_pro_iter_repro[8]
     try:
         l2a.compute_cloud_mask(
-            cm_version=cm_version.lower(),
+            cm_version=cm_version,
             probability=probability,
             iterations=iterations,
             cld_shad=cld_shad,
@@ -109,7 +120,8 @@ def multi_cld_ver_pro_iter_repro(l2a_ver_pro_iter_repro):
 
 
 def cld_version_probability_iterations_reprocessing_multiprocessing(
-    process_list, nb_proc=4
+    process_list, 
+    nb_proc=4
 ):
     """ """
     nb_proc = max(min(len(os.sched_getaffinity(0)) - 1, nb_proc), 1)
@@ -140,7 +152,7 @@ def multi_idx(l2a_id_idx):
             reprocess=reprocess,
             nodata_clouds=nodata_clouds,
             quicklook=quicklook,
-            cm_version=cm_version.lower(),
+            cm_version=cm_version,
             probability=probability,
             iterations=iterations,
             cld_shad=cld_shad,
@@ -152,7 +164,10 @@ def multi_idx(l2a_id_idx):
         pass
 
 
-def idx_multiprocessing(process_list, nb_proc=4):
+def idx_multiprocessing(
+    process_list, 
+    nb_proc=4,
+):
     """ """
     nb_proc = max(min(len(os.sched_getaffinity(0)) - 1, nb_proc), 1)
     pool = multiprocessing.Pool(nb_proc)

sen2chain/multiprocess_l2a.py

@@ -10,6 +10,7 @@ import sys
 from sen2chain import L1cProduct
 
 identifier = sys.argv[1]
+copy_l2a_sideproducts = sys.argv[2]
 
 l1c = L1cProduct(identifier)
-l1c.process_l2a()
+l1c.process_l2a(copy_l2a_sideproducts = copy_l2a_sideproducts)

sen2chain/products.py

@@ -225,6 +225,7 @@ class L1cProduct(Product):
         self, 
         reprocess: bool = False,
         s2c_path: Union[str, pathlib.PosixPath] = None,
+        copy_l2a_sideproducts: bool = False,
     ) -> "L1cProduct":
         """ process with sen2cor """
         logger.info("{}: processing L2A".format(self.identifier))
@@ -245,6 +246,7 @@ class L1cProduct(Product):
         else:
             if not reprocess:
                 logger.info("{} already exists.".format(l2a_identifier))
+            
             else:
                 if l2a_prod.in_library:
                     shutil.rmtree(str(l2a_prod.path))
@@ -274,6 +276,17 @@ class L1cProduct(Product):
             l2a_prod = L2aProduct(l2a_identifier)
             l2a_prod.set_permissions()
             l2a_prod.update_md(sen2cor_version = get_Sen2Cor_version(s2c_path))
+            
+            if copy_l2a_sideproducts:
+                if l2a_prod.path.exists():
+                    cloudmask = NewCloudMaskProduct(l2a_identifier = l2a_identifier)
+                    cloudmask.path.parent.mkdir(parents = True, exist_ok = True)
+                    if not (cloudmask.path.parent / Path(cloudmask.msk_cldprb_20m).name).exists():
+                        logger.info("Copying msk_cldprb_20m to cloumask folder")
+                        shutil.copy2(cloudmask.msk_cldprb_20m, cloudmask.path.parent)
+                    if not (cloudmask.path.parent / Path(cloudmask.scl_20m).name).exists():
+                        shutil.copy2(cloudmask.scl_20m, cloudmask.path.parent)
+                        logger.info("Copying scl_20m to cloumask folder")
         return self
 
     def process_ql(
@@ -455,7 +468,10 @@ class L2aProduct(Product):
     _tiled_metadata = "MTD_MSIL2A.xml"
 
     def __init__(
-        self, identifier: str = None, tile: str = None, path: str = None
+        self, 
+        identifier: str = None, 
+        tile: str = None, 
+        path: str = None
     ) -> None:
         super().__init__(identifier=identifier, tile=tile, path=path)
         if not re.match(r".*L2A_.*", identifier):
@@ -636,6 +652,7 @@ class L2aProduct(Product):
         # out_path_mask_b11 = None
     ) -> "L2aProduct":
         """ """
+        cm_version = cm_version.upper()
         if cm_version == "CM003":
             logger.info(
                 "Computing cloudmask version {}, probability {}%, iteration(s) {}: {}".format(
@@ -721,7 +738,7 @@ class L2aProduct(Product):
                             "No cloudmask version cm001 found, please compute this one first"
                         )
                 else:
-                    logger.info("Skipping: no L2A")
+                    logger.info("Skipping: no B11 from L2A")
             elif cm_version == "CM003":
                 if cloudmask.msk_cldprb_20m:
                     cloudmask.path.unlink(missing_ok = True)
@@ -990,14 +1007,15 @@ class L2aProduct(Product):
             sen2cor_version=sen2cor_version,
         )
 
-    def remove(self):
+    def remove(
+        self,
+        copy_l2a_sideproducts: bool = False,
+    ):
+        
+        if copy_l2a_sideproducts:
+            self.copy_l2a_sideproducts()
+
         if self.path.exists():
-            cloudmask = NewCloudMaskProduct(l2a_identifier = self.identifier)
-            if cloudmask.path.parent.exists():
-                if not (cloudmask.path.parent / Path(cloudmask.msk_cldprb_20m).name).exists():
-                    shutil.copy2(cloudmask.msk_cldprb_20m, cloudmask.path.parent)
-                if not (cloudmask.path.parent / Path(cloudmask.scl_20m).name).exists():
-                    shutil.copy2(cloudmask.scl_20m, cloudmask.path.parent)
             if self.path.is_symlink():
                 l2a_path = os.readlink(str(self.path))
                 logger.info("Removing: {}".format(l2a_path))
@@ -1009,7 +1027,22 @@ class L2aProduct(Product):
                 logger.info("Removing: {}".format(self.path))
         else:
             logger.info("L2A product not on disk")
-
+    
+    def copy_l2a_sideproducts(self):
+        if self.path.exists():
+            cloudmask = NewCloudMaskProduct(l2a_identifier = self.identifier)
+            cloudmask.path.parent.mkdir(parents = True, exist_ok = True)
+            if not (cloudmask.path.parent / Path(cloudmask.msk_cldprb_20m).name).exists():
+                logger.info("{} - Copying msk_cldprb_20m to cloumask folder".format(self.identifier))
+                shutil.copy2(cloudmask.msk_cldprb_20m, cloudmask.path.parent)
+            else:
+                logger.info("{} - msk_cldprb_20m already copied to cloumask folder".format(self.identifier))
+            if not (cloudmask.path.parent / Path(cloudmask.scl_20m).name).exists():
+                logger.info("{} - Copying scl_20m to cloumask folder".format(self.identifier))
+                shutil.copy2(cloudmask.scl_20m, cloudmask.path.parent)
+            else:
+                logger.info("{} - scl_20m already copied to cloumask folder".format(self.identifier))
+    
     @property
     def sen2chain_version(self):
         return Sen2ChainMetadataParser(

sen2chain/tiles.py

@@ -227,7 +227,7 @@ class CloudMaskList(ProductsList):
     """Class for managing mask product list"""
 
     @property
-    def cm001(self) -> "CloudMaskList":
+    def CM001(self) -> "CloudMaskList":
         filtered = CloudMaskList()
         for k, v in self._dict.items():
             if "_CM001" in k:
@@ -238,7 +238,7 @@ class CloudMaskList(ProductsList):
         return filtered
 
     @property
-    def cm002(self) -> "CloudMaskList":
+    def CM002(self) -> "CloudMaskList":
         filtered = CloudMaskList()
         for k, v in self._dict.items():
             if "_CM002" in k:
@@ -249,7 +249,7 @@ class CloudMaskList(ProductsList):
         return filtered
 
     @property
-    def cm003(self) -> "CloudMaskList":
+    def CM003(self) -> "CloudMaskList":
         filtered = CloudMaskList()
         for k, v in self._dict.items():
             if "_CM003" in k:
@@ -260,7 +260,7 @@ class CloudMaskList(ProductsList):
         return filtered
 
     @property
-    def cm004(self) -> "CloudMaskList":
+    def CM004(self) -> "CloudMaskList":
         filtered = CloudMaskList()
         for k, v in self._dict.items():
             if "_CM004" in k:
@@ -675,7 +675,7 @@ class Tile:
         return prod_list
         
     # def cloudmasks_missing(self,
-                           # cm_version: str = "cm001",
+                           # cm_version: str = "CM001",
                            # probability: int = 1,
                            # iterations: int = 5,
                            # cld_shad: bool = True,
@@ -687,7 +687,7 @@ class Tile:
 
     def cloudmasks_missing(
         self,
-        cm_version: str = "cm001",
+        cm_version: str = "CM001",
         probability: int = 1,
         iterations: int = 5,
         cld_shad: bool = True,
@@ -769,7 +769,7 @@ class Tile:
         self,
         indice: str,
         nodata_clouds: bool = False,
-        cm_version: list = "cm001",
+        cm_version: list = "CM001",
         probability: int = 1,
         iterations: int = 5,
         cld_shad: bool = True,
@@ -799,7 +799,14 @@ class Tile:
                         getattr(getattr(self, indice.lower()), "masks"),
                         cm_version,
                     )
-                    .params(probability=probability, iterations=iterations)
+                    .params(
+                        probability = probability, 
+                        iterations = iterations,
+                        cld_shad = cld_shad,
+                        cld_med_prob = cld_med_prob,
+                        cld_hi_prob = cld_hi_prob,
+                        thin_cir = thin_cir,
+                    )
                     .products
                 }
         except:
@@ -946,6 +953,7 @@ class Tile:
         cover_min: int = 0,
         cover_max: int = 100,        
         nb_proc: int = 4,
+        copy_l2a_sideproducts: bool = False,
     ):
         """
         Compute all missing l2a for l1c products between date_min and date_max
@@ -979,9 +987,13 @@ class Tile:
         l1c_process_list = []
         l1c_process_list.append(
             list(
-                p.identifier
+                [
+                    p.identifier,
+                    copy_l2a_sideproducts,
+                ]
                 for p in self.l2a_missings.filter_dates(
-                    date_min=date_min, date_max=date_max
+                    date_min=date_min, 
+                    date_max=date_max,
                 ).filter_clouds(
                     cover_min = cover_min, 
                     cover_max = cover_max
@@ -999,10 +1011,10 @@ class Tile:
         l2a_res = False
         if l1c_process_list:
             l2a_res = l2a_multiprocessing(l1c_process_list, nb_proc=nb_proc)
-
+            
     def compute_cloudmasks(
         self,
-        cm_version: str = "cm001",
+        cm_version: str = "CM001",
         probability: int = 1,
         iterations: int = 5,
         cld_shad: bool = True,
@@ -1016,11 +1028,11 @@ class Tile:
     ):
         """Compute all (missing) cloud masks for l2a products.
 
-        :param cm_version: version of cloudmask to compute. Can be either cm001,
-            cm002, cm003, or cm004.
-        :param probability: only used by cm003: threshold probability of clouds
+        :param cm_version: version of cloudmask to compute. Can be either CM001,
+            CM002, CM003, or CM004.
+        :param probability: only used by CM003: threshold probability of clouds
             to be considered.
-        :param iterations: only used by cm003: number of iterations for
+        :param iterations: only used by CM003: number of iterations for
             dilatation process while computing cloudmask.
         :param reprocess: if False (default), only missing cloudmasks will be
             computed. if True already processed cloudmask will be computed again.
@@ -1031,7 +1043,8 @@ class Tile:
         :param nb_proc: number of parallel process, limited to the number of
             proc of your PC (default 4).
         """
-
+        
+        cm_version = cm_version.upper()
         if not reprocess:
             cld_l2a_process_list = list(
                 [
@@ -1074,7 +1087,7 @@ class Tile:
             )
         if cld_l2a_process_list:
             logger.info(
-                "{} l2a products to process:".format(len(cld_l2a_process_list))
+                "{} l2a product(s) to process".format(len(cld_l2a_process_list))
             )
             # logger.info("{}".format(cld_l2a_process_list))
             cld_version_probability_iterations_reprocessing_multiprocessing(
@@ -1090,7 +1103,7 @@ class Tile:
         reprocess: bool = False,
         nodata_clouds: bool = True,
         quicklook: bool = False,
-        cm_version: list = "cm001",
+        cm_version: list = "CM001",
         probability: int = 1,
         iterations: int = 5,
         cld_shad: bool = True,
@@ -1107,6 +1120,8 @@ class Tile:
         - if indices not provided, will compute missing dates of already existing indices for this tile (no new indice computed)
         - indices won't be masked if no cloud masks are present, you have to compute cloudmasks first
         """
+        
+        cm_version = cm_version.upper()
         if not indices:
             indices = list(self._paths["indices"].keys())
         # else:
@@ -1615,7 +1630,7 @@ class Tile:
         """
         Rename old indices to match new cloudmask nomenclature
         """
-        # Rename old indices to default cm_version cm001
+        # Rename old indices to default cm_version CM001
         logger.info("Moving and renaming old indices")
 
         for indice, path in self._paths["indices"].items():
@@ -1675,6 +1690,7 @@ class Tile:
         self,
         product_list: list = [],
         entire: bool = False,
+        copy_l2a_sideproducts: bool = False,
     ):
         """
         Remove l2a files
@@ -1683,5 +1699,26 @@ class Tile:
             product_list = [product.identifier for product in self.l2a]
         for identifier in product_list:
             l2a = L2aProduct(identifier)
-            l2a.remove()
+            l2a.remove(copy_l2a_sideproducts = copy_l2a_sideproducts)
         logger.info("Removed: {} products".format(len(product_list)))
+    
+    def copy_sp(self, l2a_product):
+            l2a = L2aProduct(l2a_product)
+            l2a.copy_l2a_sideproducts() 
+    
+    def copy_l2a_sideproducts(
+        self,
+        product_list: list = [],
+        nb_proc: int = 4,
+    ):
+        if not product_list:
+            product_list = [product.identifier for product in self.l2a]
+        
+        nb_proc = max(min(len(os.sched_getaffinity(0)) - 1, nb_proc), 1)
+        pool = multiprocessing.Pool(nb_proc)
+        results = [pool.map(self.copy_sp, product_list)]
+        pool.close()
+        pool.join()
+        return True
+        
+