add new Tile function to clean residual processing error files (l2a, cloud_masks, indices)

sen2chain/products.py

@@ -482,6 +482,7 @@ class L2aProduct(Product):
 
         return self
 
+
     # def has_indice(self, indice):
     # """
     # Returns if the indice file exists on disk.

sen2chain/tiles.py

@@ -7,6 +7,7 @@ import pathlib
 import logging
 import re
 import fiona
+import shutil
 
 from pathlib import Path
 from collections import namedtuple
@@ -320,7 +321,7 @@ class Tile:
         return prods_list
 
     def missing_indices(self, indice: str) -> "ProductsList":
-        """Returns tile's L2A products that don't have a cloud mask as a ProductsList."""
+        """Returns tile's L2A products that don't have indices as a ProductsList."""
         prods_list = ProductsList()
         try:
             missings_indice_set = set(self.l2a.products) - {identifier.replace("_" + indice.upper() + ".jp2", ".SAFE") \
@@ -331,3 +332,64 @@ class Tile:
                 prods_list[prod] = {"date": self._products["l2a"][prod].date,
                                 "cloud_cover": self._products["l2a"][prod].cloud_cover}            
         return prods_list
+
+    def clean_lib(self,
+                  remove: bool = False):
+        """
+        Clean processing errors
+        - unmoved error l2a products from l1c folder
+        - moved  error l2a products from l2a folder
+        - cloud masks error (0kb)
+        - indices error (0kb)        
+        """
+        logger.info("Cleaning {} library".format(self.name))
+        # identify residual l2a from l1c folder
+        for f in self._paths["l1c"].glob("*L2A*.SAFE"):
+            logger.info("Identified {} from L1C folder".format(f.name))
+            if remove:
+                try:
+                    shutil.rmtree(str(f))
+                    logger.info("Removing {} from L1C folder".format(f.name))
+                except:
+                    logger.error("Can't remove {} from L1C folder".format(f.name))
+        # identify residual l2a from l1c folder
+        for f in self._paths["l2a"].glob("*L2A*.SAFE"):
+            if len(list(f.glob("GRANULE/*/IMG_DATA/R10m/*.jp2"))) == 0:
+                logger.info("Corrupted L2A {} in L2A folder".format(f.name))
+                if remove:
+                    try:
+                        shutil.rmtree(str(f))
+                        logger.info("Removing corrupted L2A {} from L2A folder".format(f.name))
+                    except:
+                        logger.error("Can't remove {} from L2A folder".format(f.name))
+        # identify 0B cloud masks
+        for f in self._paths["l2a"].glob("*L2A*_CLOUD_MASK.jp2"):
+            if f.stat().st_size == 0:
+                logger.info("Corrupted cloud mask {} in L2A folder".format(f.name))
+                if remove:
+                    logger.info("Removing corrupted cloud mask {} from L2A folder".format(f.name))
+                    f.unlink()
+        # identify 0B or absent indices QL
+        for f in self._paths["indices"]:
+            #~ print(f, self._paths["indices"][f])
+            for p in self._paths["indices"][f].glob("*_MSIL2A_*/"):
+                if p.is_file():
+                    logger.info("Identified old indice format {}".format(p.name))
+                    if remove:
+                        logger.info("Removing old indice format {}".format(p.name))
+                        p.unlink()
+                else:
+                    for q in p.glob("*_NDVI_QUICKLOOK.tif"):
+                        if q.stat().st_size == 0:
+                            logger.info("Identified indice QL error {}".format(q.name))
+                            if remove:
+                                logger.info("Removing indice QL error {}".format(q.name))
+                                q.unlink()
+    
+                
+        
+        
+        
+
+
+