Merge pull request #173 from raphaelrpl/b-0.8

Add minimal support to deal with bitwise mask for Landsat Collection 2 products

Author: raphaelrpl

CHANGES.rst

@@ -21,6 +21,14 @@ Changes
 =======
 
 
+Version 0.8.4
+-------------
+
+- Add support to generate data cube from Landsat Collection 2 (`#172 <https://github.com/brazil-data-cube/cube-builder/issues/172>`_)
+- Add support to combine Landsat Collection 2 sensors (L5/L7, L7/L8, L7/L8/L9) using single collection (`#172 <https://github.com/brazil-data-cube/cube-builder/issues/172>`_)
+- Review API error when no parameter is set.
+
+
 Version 0.8.3 (2022-10-03)
 --------------------------
 

Dockerfile

@@ -15,18 +15,29 @@
 # You should have received a copy of the GNU General Public License
 # along with this program. If not, see <https://www.gnu.org/licenses/gpl-3.0.html>.
 #
-
-FROM python:3.8
+ARG GIT_COMMIT
+ARG BASE_IMAGE=python:3.8
+FROM ${BASE_IMAGE}
 
 ARG GIT_COMMIT
 
 LABEL "org.brazildatacube.maintainer"="Brazil Data Cube <brazildatacube@inpe.br>"
 LABEL "org.brazildatacube.title"="Docker image for Data Cube Builder Service"
+LABEL "org.brazildatacube.description"="Docker image for Data Cube Builder application."
 LABEL "org.brazildatacube.git_commit"="${GIT_COMMIT}"
 
-ADD . /app
+# Build arguments
+ARG CUBE_BUILDER_VERSION="0.8.4"
+ARG CUBE_BUILDER_INSTALL_PATH="/opt/cube-builder/${CUBE_BUILDER_VERSION}"
+
+ADD . ${CUBE_BUILDER_INSTALL_PATH}
 
-WORKDIR /app
+WORKDIR ${CUBE_BUILDER_INSTALL_PATH}
 
 RUN python3 -m pip install pip --upgrade setuptools wheel && \
-    python3 -m pip install -e .[rabbitmq]
+    python3 -m pip install -e .[rabbitmq] && \
+    python3 -m pip install gunicorn
+
+EXPOSE 5000
+
+CMD ["gunicorn", "-w4", "--bind=0.0.0.0:5000", "cube_builder:create_app()"]

cube_builder/celery/tasks.py

@@ -89,7 +89,9 @@ def warp_merge(activity, band_map, mask, force=False, data_dir=None, **kwargs):
     Returns:
         Validated activity
     """
-    logging.warning('Executing merge {} - {}'.format(activity.get('warped_collection_id'), activity['band']))
+    logging.warning('Executing merge {} - {} - {}'.format(activity.get('warped_collection_id'),
+                                                          activity['band'],
+                                                          activity['date']))
 
     record = create_execution(activity)
 
@@ -117,7 +119,7 @@ def warp_merge(activity, band_map, mask, force=False, data_dir=None, **kwargs):
                                           tile_id, version=version, band=record.band,
                                           prefix=data_dir, composed=False, **kwargs)
 
-        if activity['band'] == quality_band and len(activity['args']['datasets']):
+        if activity['band'] == quality_band and (len(activity['args']['datasets']) or kwargs.get('combined')):
             kwargs['build_provenance'] = True
 
     reused = False
@@ -304,6 +306,8 @@ def _is_not_stk(merge):
         """
         return merge['band'] == quality_band and composite_function not in ('STK', 'LCF')
 
+    platforms = []
+
     for _merge in merges:
         # Skip quality generation for MEDIAN, AVG
         if _merge['band'] in activities and _merge['args']['date'] in activities[_merge['band']]['scenes'] or \
@@ -341,6 +345,9 @@ def _is_not_stk(merge):
 
         if _merge['args'].get(DATASOURCE_NAME):
             activity['scenes'][_merge['args']['date']]['ARDfiles'][DATASOURCE_NAME] = _merge['args'][DATASOURCE_NAME]
+            if _merge['args'].get('platforms'):
+                activity.setdefault('platforms', _merge['args'].get('platforms'))
+                platforms = _merge['args'].get('platforms')
 
         activities[_merge['band']] = activity
 
@@ -386,9 +393,6 @@ def _is_not_stk(merge):
             for idx, date in enumerate(ordered_dates):
                 activity['scenes'][date]['efficacy'] = weights[idx]
 
-    # Prepare list of activities to dispatch
-    activity_list = list(activities.values())
-
     # For IDENTITY data cube trigger, just publish
     if composite_function == 'IDT':
         task = publish.s(list(activities.values()), reuse_data_cube=reuse_data_cube, band_map=band_map, **kwargs)
@@ -401,10 +405,10 @@ def _is_not_stk(merge):
     # We must keep track of last activity to run
     # Since the Clear Observation must only be execute by single process. It is important
     # to avoid concurrent processes to write same data set in disk
-    last_activity = activity_list[-1]
+    last_activity = activities.pop(quality_band)
 
     # Trigger all except the last
-    for activity in activity_list[:-1]:
+    for activity in list(activities.values()):
         # TODO: Persist
         blends.append(blend.s(activity, band_map, reuse_data_cube=reuse_data_cube, **kwargs))
 

cube_builder/cli.py

@@ -110,7 +110,7 @@ def worker(ctx: click.Context):
 @click.option('--export-files', type=click.Path(writable=True), help='Export Identity Merges in file')
 @with_appcontext
 def build(datacube: str, collections: str, tiles: str, start: str, end: str, bands: str = None,
-          stac_url: str = None, force=False, with_rgb=False, shape=None, export_files=None, **kwargs):
+          stac_url: str = None, force=False, shape=None, export_files=None, **kwargs):
     """Build data cube through command line.
 
     Args:
@@ -138,7 +138,6 @@ def build(datacube: str, collections: str, tiles: str, start: str, end: str, ban
         end_date=end,
         tiles=tiles.split(','),
         force=force,
-        with_rgb=with_rgb,
         stac_url=stac_url,
         **kwargs
     )
@@ -175,6 +174,7 @@ def build(datacube: str, collections: str, tiles: str, start: str, end: str, ban
 @click.option('--mask',  type=click.STRING, help='Custom mask values for data cube.')
 @click.option('--quality-band', type=click.STRING, help='Quality band name')
 @click.option('--cloud-cover', type=click.FLOAT, help='Cloud Cover Factor. Default is 100 to use all.', default=100)
+@click.option('--band-map', type=click.STRING, help='Band mapping when combine sensors. Default is None')
 @with_appcontext
 def _configure_parameters(datacube: str, **kwargs):
     """Configure the default parameters for data cube.

cube_builder/forms.py

@@ -105,6 +105,7 @@ class CustomMaskDefinition(Schema):
     nodata = fields.Integer(required=True, allow_none=False)
     saturated_data = fields.List(fields.Integer, required=False, allow_none=False)
     saturated_band = fields.String(required=False, allow_none=False)
+    bits = fields.Boolean(required=False, allow_none=False, default=False)
 
 
 class CubeParametersSchema(Schema):
@@ -114,6 +115,7 @@ class CubeParametersSchema(Schema):
     reference_day = fields.Integer(required=False, allow_none=False)
     histogram_matching = fields.Bool(required=False, allow_none=False)
     no_post_process = fields.Bool(required=False, allow_none=False)
+    band_map = fields.Dict(required=False, allow_none=False)
 
 
 class DataCubeForm(Schema):

cube_builder/maestro.py

@@ -22,6 +22,7 @@
 import json
 import logging
 import warnings
+from collections.abc import Iterable
 from contextlib import contextmanager
 from copy import deepcopy
 from pathlib import Path
@@ -64,6 +65,12 @@ def days_in_month(date):
     return td
 
 
+def _valid_channel_limit(value, expects):
+    if not isinstance(value, Iterable) or len(value) != expects:
+        raise ValueError('Invalid type for "channel_limits". '
+                         f'Expects Iterable of {expects} elements, but got "{value}"')
+
+
 @contextmanager
 def timing(description: str) -> None:
     """Measure execution time of context.
@@ -131,10 +138,10 @@ def get_stac(self, collection: str) -> STAC:
         When collection not found, search on INPE STAC.
 
         Args:
-            collection - Collection name to search
+            collection (str): Collection name to search
 
         Returns:
-            STAC client
+            STAC: The STAC which offers the given collection.
         """
         if self.properties.get('stac_url'):
             return self._stac(collection, self.properties['stac_url'], **self.properties)
@@ -158,7 +165,7 @@ def _stac(self, collection: str, url: str, **kwargs) -> STAC:
             url - STAC URL
 
         Returns:
-            STAC client
+            STAC: STAC instance
         """
         try:
             options = dict()
@@ -208,6 +215,13 @@ def orchestrate(self):
         # Validate parameters
         cube_parameters.validate()
 
+        if self.properties.get('channel_limits'):
+            channel_limits = self.properties['channel_limits']
+            # Validate the entire signature (Tuple of 3 limit elements)
+            _valid_channel_limit(channel_limits, expects=3)
+            # Validate each RGB channel (Tuple of 2 int elements)
+            _ = [_valid_channel_limit(channel, expects=2) for channel in channel_limits]
+
         # Pass the cube parameters to the data cube functions arguments
         props = deepcopy(cube_parameters.metadata_)
         props.update(self.properties)
@@ -235,8 +249,13 @@ def orchestrate(self):
         self.bands = Band.query().filter(Band.collection_id == self.warped_datacube.id).all()
 
         bands = self.datacube_bands
-        self.band_map = {b.name: dict(name=b.name, data_type=b.data_type, nodata=b.nodata,
-                                      min_value=b.min_value, max_value=b.max_value) for b in bands}
+        self.band_map = {
+            b.name: dict(name=b.name, data_type=b.data_type, nodata=b.nodata,
+                         min_value=b.min_value, max_value=b.max_value,
+                         scale=b.scale,
+                         scale_add=b._metadata.get('scale_add') if b._metadata else None)
+            for b in bands
+        }
 
         if self.properties.get('reuse_from'):
             warnings.warn(
@@ -404,6 +423,13 @@ def dispatch_celery(self):
 
             warped_datacube = self.warped_datacube.name
 
+            if self.properties.get('with_rgb'):
+                input_range = self.properties.get('input_range')
+                if input_range is None:
+                    raise ValueError('Missing valid range for RGB.')
+                if type(input_range) not in (list, tuple,):
+                    raise TypeError(f'Invalid input range for RGB. Expects Tuple[int, int], got {type(input_range)}')
+
             quality_band = None
             stac_kwargs = self.properties.get('stac_kwargs', dict())
             if self.datacube.composite_function.alias != 'IDT':
@@ -509,6 +535,7 @@ def dispatch_celery(self):
                                 srs=grid_crs,
                                 tile_id=tileid,
                                 assets=assets,
+                                platforms=self.platforms,
                                 nodata=float(band.nodata),
                                 bands=band_str_list,
                                 version=self.datacube.version,
@@ -563,6 +590,9 @@ def search_images(self, feature: dict, start: str, end: str, tile_id: str, **kwa
 
         bands = self.datacube_bands
 
+        band_mapping = self.properties.get('band_map', dict())
+        platforms = set()
+
         # Retrieve band definition in dict format.
         # TODO: Should we use from STAC?
         collection_bands = dict()
@@ -581,6 +611,11 @@ def search_images(self, feature: dict, start: str, end: str, tile_id: str, **kwa
         for dataset in self.params['collections']:
             options['collections'] = [dataset]
             stac = self.get_stac(dataset)
+            stac_collection = stac.collection(dataset)
+            if stac_collection.get('summaries') and stac_collection['summaries'].get('platform'):
+                platforms = platforms.union(set(stac_collection['summaries'].get('platform')))
+            elif stac_collection.properties.get('platform'):
+                platforms = platforms.union(set(stac_collection.properties.get('platform')))
 
             token = ''
 
@@ -593,21 +628,40 @@ def search_images(self, feature: dict, start: str, end: str, tile_id: str, **kwa
                 for feature in items['features']:
                     if feature['type'] == 'Feature':
                         date = feature['properties']['datetime'][0:10]
-                        identifier = feature['id']
                         stac_bands = feature['properties'].get('eo:bands', [])
+                        identifier = feature['id']
+                        # TODO: Add handler to deal with parse result serializer.
+                        platform = feature['properties'].get('platform')
+                        if stac.url.startswith('https://landsatlook.usgs.gov'):
+                            # Remove last SR sentence.
+                            identifier = f'{identifier[:-3]}{identifier[-3:].replace("_SR", "")}'
+                        # Special treatment for missing/invalid platform values
+                        if (platform is None or isinstance(platform, list)) and _is_landsat(identifier):
+                            platform = _detect_landsat_platform(identifier)
 
                         for band in bands:
                             band_name_href = band.name
+
+                            if platform and band_mapping:
+                                platforms.add(platform)
+                                if platform not in band_mapping or not band_mapping[platform].get(band_name_href):
+                                    continue
+
+                                band_name_href = band_mapping[platform].get(band_name_href)
+
                             if 'CBERS' in dataset and band.common_name not in ('evi', 'ndvi'):
                                 band_name_href = band.common_name
 
                             elif band.name not in feature['assets']:
-                                if f'sr_{band.name}' not in feature['assets']:
+                                # TODO: Implement asset resolver
+                                if f'{band_name_href}.TIF' in feature['assets']:
+                                    band_name_href = f'{band_name_href}.TIF'
+                                elif f'sr_{band_name_href}' not in feature['assets']:
                                     continue
                                 else:
                                     band_name_href = f'sr_{band.name}'
 
-                            feature_band = list(filter(lambda b: b['name'] == band_name_href,stac_bands))
+                            feature_band = list(filter(lambda b: b['name'] == band_name_href, stac_bands))
                             feature_band = feature_band[0] if len(feature_band) > 0 else dict()
 
                             scenes[band.name].setdefault(date, dict())
@@ -620,6 +674,9 @@ def search_images(self, feature: dict, start: str, end: str, tile_id: str, **kwa
                             scene['sceneid'] = identifier
                             scene['band'] = band.name
                             scene['dataset'] = dataset
+                            scene['platform'] = platform
+                            if band_mapping:
+                                scene['original_band_name'] = band_name_href
 
                             link = link.replace(Config.CBERS_SOURCE_URL_PREFIX, Config.CBERS_TARGET_URL_PREFIX)
 
@@ -634,4 +691,29 @@ def search_images(self, feature: dict, start: str, end: str, tile_id: str, **kwa
                             scenes[band.name][date].setdefault(dataset, [])
                             scenes[band.name][date][dataset].append(scene)
 
+        self.platforms = sorted(list(platforms))
+
         return scenes
+
+
+def _is_landsat(identifier: str) -> bool:
+    try:
+        _parse_landsat(identifier)
+        return True
+    except ValueError:
+        return False
+
+
+def _parse_landsat(identifier: str) -> List[str]:
+    entities = identifier.split('_')
+    if len(entities) != 7 or not entities[0].startswith('L') or not entities[0][-2:].isnumeric():
+        raise ValueError(f'Invalid landsat scene {identifier}')
+    return entities
+
+
+def _detect_landsat_platform(identifier: str):
+    entities = _parse_landsat(identifier)
+    satellite = int(entities[0][-2:])
+    for value in [4, 5, 7, 8, 9]:
+        if satellite == value:
+            return f'Landsat-{value}'