Merge pull request #146 from betonr/master

Fix landsat harmonization and update qa_mask (lc8-collection2)

Author: betonr

cube_builder_aws/cube_builder_aws/config.py

@@ -23,3 +23,4 @@
 DYNAMO_TB_ACTIVITY = os.environ.get('DYNAMO_TB_ACTIVITY', '')
 DBNAME_TB_CONTROL = os.environ.get('DBNAME_TB_CONTROL', '')
 DBNAME_TB_PROCESS = os.environ.get('DBNAME_TB_PROCESS', '')
+DBNAME_TB_HARM = os.environ.get('DBNAME_TB_HARM', '')

cube_builder_aws/cube_builder_aws/constants.py

@@ -152,4 +152,16 @@
 
 PNG_MIME_TYPE = 'image/png'
 
-SRID_ALBERS_EQUAL_AREA = 100001
+SRID_ALBERS_EQUAL_AREA = 100001
+
+HARMONIZATION = dict(
+    landsat = dict(
+        bands = dict(
+            L05 = ["SR_B1" ,"SR_B2", "SR_B3", "SR_B4", "SR_B5", "SR_B7"],
+            L07 = ["SR_B1" ,"SR_B2", "SR_B3", "SR_B4", "SR_B5", "SR_B7"],
+            L08 = ["SR_B2", "SR_B3", "SR_B4", "SR_B5", "SR_B6", "SR_B7"]
+        ),
+        bucket_src = 's3://usgs-landsat',
+        format_path = 'collection{collectionNumber}/level-2/standard/{instrument}/{acquisitionYear}/{path}/{row}/{scene_id}/{scene_id}_{band}.TIF'
+    )
+)

cube_builder_aws/cube_builder_aws/controller.py

@@ -12,34 +12,30 @@
 from datetime import datetime
 from typing import Tuple, Union
 
-import rasterio
 import sqlalchemy
 from bdc_catalog.models import (Band, BandSRC, Collection, CompositeFunction,
                                 GridRefSys, Item, MimeType, Quicklook,
                                 ResolutionUnit, SpatialRefSys, Tile)
-from bdc_catalog.models.base_sql import BaseModel, db
+from bdc_catalog.models.base_sql import db
 from geoalchemy2 import func
 from geoalchemy2.shape import from_shape
 from rasterio.crs import CRS
 from rasterio.warp import transform
 from shapely.geometry import Polygon
-from werkzeug.exceptions import BadRequest, Conflict, NotFound
+from werkzeug.exceptions import BadRequest, NotFound
 
 from .config import ITEM_PREFIX
-from .constants import (CENTER_WAVELENGTH, CLEAR_OBSERVATION_ATTRIBUTES,
-                        CLEAR_OBSERVATION_NAME, COG_MIME_TYPE,
-                        DATASOURCE_ATTRIBUTES, FULL_WIDTH_HALF_MAX,
+from .constants import (CLEAR_OBSERVATION_ATTRIBUTES, CLEAR_OBSERVATION_NAME,
+                        COG_MIME_TYPE, DATASOURCE_ATTRIBUTES,
                         PROVENANCE_ATTRIBUTES, PROVENANCE_NAME,
-                        REVISIT_BY_SATELLITE, SRID_ALBERS_EQUAL_AREA,
-                        SRID_BDC_GRID, TOTAL_OBSERVATION_ATTRIBUTES,
+                        SRID_ALBERS_EQUAL_AREA, TOTAL_OBSERVATION_ATTRIBUTES,
                         TOTAL_OBSERVATION_NAME)
 from .forms import CollectionForm
-from .maestro import (blend, merge_warped, orchestrate, posblend,
-                      prepare_merge, publish, solo)
+from .maestro import (blend, harmonization, merge_warped, orchestrate,
+                      posblend, prepare_harm, prepare_merge, publish, solo)
 from .services import CubeServices
 from .utils.image import validate_merges
-from .utils.processing import (format_version, generate_hash_md5,
-                               get_cube_name, get_cube_parts, get_date,
+from .utils.processing import (format_version, get_cube_parts, get_date,
                                get_or_create_model)
 from .utils.serializer import DecimalEncoder, Serializer
 from .utils.timeline import Timeline
@@ -59,6 +55,10 @@ def continue_process_stream(self, params_list):
         if 'channel' in params and params['channel'] == 'kinesis':
             solo(self, params_list)
 
+        # dispatch HARMONIZATION
+        elif params['action'] == 'harmonization':
+            harmonization(self, params)
+
         # dispatch MERGE
         elif params['action'] == 'merge':
             merge_warped(self, params)
@@ -246,7 +246,7 @@ def _create_cube_definition(self, cube_id: str, params: dict) -> dict:
                     collection=cube,
                     min_value=0,
                     max_value=10000 if is_not_cloud else 4,
-                    nodata=-9999 if is_not_cloud else 255,
+                    nodata=band['nodata'],
                     scale=0.0001 if is_not_cloud else 1,
                     data_type=data_type,
                     resolution_x=params['resolution'],
@@ -287,7 +287,8 @@ def _create_cube_definition(self, cube_id: str, params: dict) -> dict:
                 _ = self.get_or_create_band(cube.id, **PROVENANCE_ATTRIBUTES, resolution_unit_id=resolution_meter.id,
                                            resolution_x=params['resolution'], resolution_y=params['resolution'])
 
-        if params.get('is_combined') and function != 'MED':
+        landsat_harm = params['parameters'].get('landsat_harmonization')
+        if landsat_harm and landsat_harm.get('datasets') and function != 'MED':
             _ = self.get_or_create_band(cube.id, **DATASOURCE_ATTRIBUTES, resolution_unit_id=resolution_meter.id,
                                        resolution_x=params['resolution'], resolution_y=params['resolution'])
 
@@ -465,6 +466,15 @@ def get_cube_status(self, cube_name):
             error = 0
         ), 200
 
+
+    def start_harmonization_process(self, params):
+        _ = prepare_harm(self, params['scenes'], params['bucket_dst'], params['bucket_angles'])
+
+        return dict(
+            message='Harmonization processing started with succesfully'
+        ), 200
+
+
     def start_process(self, params):
         response = {}
         datacube_identify = f'{params["datacube"]}-{params["datacube_version"]}'

cube_builder_aws/cube_builder_aws/forms.py

@@ -56,16 +56,29 @@ class BandDefinition(Schema):
     name = fields.String(required=True, allow_none=False)
     common_name = fields.String(required=True, allow_none=False)
     data_type = fields.String(required=True, allow_none=False, validate=OneOf(SUPPORTED_DATA_TYPES))
+    nodata = fields.Integer(required=False, allow_none=False)
     metadata = fields.Dict(required=False, allow_none=False)
 
 
+class QAConfidence(Schema):
+    """Define that will discard all cloud values which has confidence greater or equal MEDIUM.
+    qa = QAConfidence(cloud='cloud >= MEDIUM', cloud_shadow=None, cirrus=None, snow=None, landsat_8=True)."""
+    
+    cloud = fields.String(required=False, allow_none=True)
+    cloud_shadow = fields.String(required=False, allow_none=True)
+    cirrus = fields.String(required=False, allow_none=True)
+    snow = fields.String(required=False, allow_none=True)
+
+
 class CustomMaskDefinition(Schema):
     """Define a custom mask."""
 
     clear_data = fields.List(fields.Integer, required=True, allow_none=False)
     not_clear_data = fields.List(fields.Integer, required=True, allow_none=False)
     saturated_data = fields.List(fields.Integer, required=True, allow_none=False)
     nodata = fields.Integer(required=True, allow_none=False)
+    bits = fields.Boolean(required=False, allow_none=False, default=False)
+    confidence = fields.Nested(QAConfidence)
 
 
 class StacDefinitionForm(Schema):
@@ -76,6 +89,15 @@ class StacDefinitionForm(Schema):
     collection = fields.String(required=True, allow_none=False)
 
 
+class LandsatHarmonization(Schema):
+    """Define parser for params of the landsat harmonization."""
+
+    apply = fields.Boolean(required=False, allow_none=False, default=False)
+    bucket_dst = fields.String(required=True, allow_none=False)
+    datasets = fields.List(fields.String(required=True, allow_none=False))
+    map_bands = fields.Dict(required=False, allow_none=False)
+
+
 class DataCubeForm(Schema):
     """Define parser for datacube creation."""
 
@@ -98,7 +120,8 @@ class DataCubeForm(Schema):
     is_combined = fields.Boolean(required=False, allow_none=False, default=False)
     parameters = fields.Dict(
         mask = fields.Nested(CustomMaskDefinition),
-        stac_list = fields.List(fields.Nested(StacDefinitionForm))
+        stac_list = fields.List(fields.Nested(StacDefinitionForm)),
+        landsat_harmonization = fields.Nested(LandsatHarmonization)
     )
 
     @pre_load
@@ -167,6 +190,15 @@ class DataCubeProcessForm(Schema):
     indexes_only_regular_cube = fields.Boolean(required=False, allow_none=True, default=False)
 
 
+class DataCubeHarmonizationForm(Schema):
+    """Define parser for harmonization generate."""
+
+    scenes = fields.List(fields.String, required=True, allow_none=False)
+    bucket_dst = fields.String(required=True, allow_none=False)
+    bucket_angles = fields.String(required=True, allow_none=False)
+    satellite = fields.String(required=True, allow_none=False, validate=OneOf(['landsat']))
+
+
 class PeriodForm(Schema):
     """Define parser for Data Cube Periods."""