Merge branch '168-invalid-parameter-used-in-eodag-search-request-for-polarisation-mode' into 'develop'

Resolve "Invalid parameter used in eodag search request for polarisation mode"

Closes #168

See merge request s1-tiling/s1tiling!138

Author: Luc Hermitte

docs/FAQ.rst

@@ -62,7 +62,7 @@ This Q/A is tracked in `S1Tiling issue #70
 Q: How can I overcome timeouts when searching for online products?
 ------------------------------------------------------------------
 
-Some data providers like PEPS may fail to obtain in time the list of products
+Some data providers like Geodes may fail to obtain in time the list of products
 matching our criteria.
 
 Since `EODAG <https://github.com/CS-SI/eodag>`_ v2.11.0, we can override the

docs/configuration.rst

@@ -160,12 +160,12 @@ You can use this :download:`this template
       See :external+eodag:std:doc:`EODAG § on Configure EODAG
       <getting_started_guide/configure>`
 
-      For instance, given a PEPS account, :file:`$HOME/.config/eodag/eodag.yml`
+      For instance, given a geodes account, :file:`$HOME/.config/eodag/eodag.yml`
       could contain
 
       .. code-block:: yaml
 
-          peps:
+          geodes:
               auth:
                   credentials:
                       username: THEUSERNAME

docs/intro.rst

@@ -40,14 +40,14 @@ existing CNES open source project `Orfeo Tool Box <https://www.orfeo-toolbox.org
 
 The resulting images are gridded to Sentinel-2 MGRS geographic reference grid (`S2 tiling system - kml file <https://sentinel.esa.int/documents/247904/1955685/S2A_OPER_GIP_TILPAR_MPC__20151209T095117_V20150622T000000_21000101T000000_B00.kml>`_).
 Thanks to `EODAG <https://eodag.readthedocs.io/>`_ , different Sentinel-1 data providers can be used
-like `PEPS <https://peps.cnes.fr/>`_ or `Copernicus Data Space <https://scihub.copernicus.eu>`_.
+like `Geodes <https://geodes-portal.cnes.fr/>`_ or `Copernicus Data Space <https://scihub.copernicus.eu>`_.
 It can be used on any type of platform, from a large computing cluster to a
 laptop (the fan will make some noise during processing). It is considerably
 faster than the ortho-rectification tool in ESA SNAP software with similar results and can be easily used in
 script form.
 
-S1Tiling is currently used for many applications, such deforestation detection
-in the Amazon, monitoring of rice crops in Southeast Asia or monitoring of
-water stocks in India. In addition, this software is accessible as an on-demand
-processing service on the French PEPS collaborative ground segment, in order to
-make it easier for users to use.
+S1Tiling is currently used as Sentinel-1 data pre-processing for many
+applications, such deforestation detection in the Amazon, monitoring of rice
+crops in Southeast Asia or monitoring of water stocks in India. In addition,
+this software will be implemented in GEODES, the French portal for Earth
+Observation data, in order to provide SAR Ready Analysis Data to users.

docs/release_notes.rst

@@ -105,6 +105,9 @@ v1.2.0 Improvements
   uppercase or lowercase. A new time stamp is also defined to hold the first
   time stamp among the ones from the input S1 images
   (`#188 <https://gitlab.orfeo-toolbox.org/s1-tiling/s1tiling/-/issues/188>`_).
+- Product downloading has been fixed to work with all data providers supported
+  by EODAG v3.9.0+
+  (`#168 <https://gitlab.orfeo-toolbox.org/s1-tiling/s1tiling/-/issues/168>`_).
 
 v1.2.0 Bugs fixed
 +++++++++++++++++

s1tiling/libs/S1FileManager.py

@@ -499,7 +499,7 @@ def _search_products(  # pylint: disable=too-many-arguments, too-many-locals
         assert polarization in ['VV VH', 'VV', 'VH', 'HH HV', 'HH', 'HV']
         # In case only 'VV' or 'VH' is requested, we still need to
         # request 'VV VH' to the data provider through eodag.
-        dag_polarization_param  = 'VV VH' if polarization in ['VV VH', 'VV', 'VH'] else 'HH HV'
+        dag_polarization_param  = 'VV+VH' if polarization in ['VV VH', 'VV', 'VH'] else 'HH+HV'
         dag_orbit_dir_param     = k_dir_assoc.get(orbit_direction or "", None)  # None => all ; <<or "">> used to silence mypy
         dag_orbit_list_param    = relative_orbit_list[0] if len(relative_orbit_list) == 1 else None
         dag_platform_list_param = platform_list[0] if len(platform_list) == 1 else None
@@ -515,7 +515,7 @@ def _search_products(  # pylint: disable=too-many-arguments, too-many-locals
                             start=first_date, end=last_date,
                             box=extent,
                             # If we have eodag v1.6+, we try to filter product during the search request
-                            polarizationMode=dag_polarization_param,
+                            polarizationChannels=dag_polarization_param,
                             sensorMode="IW",
                             orbitDirection=dag_orbit_dir_param,        # None => all
                             relativeOrbitNumber=dag_orbit_list_param,  # List doesn't work. Single number yes!

s1tiling/libs/configuration.py

@@ -275,6 +275,14 @@ def getboolean(self, section: str, name: str, **kwargs) -> bool:
         """Helper function to report errors while extracting boolean configuration options"""
         return getboolean_opt(self.__config, self.config_file, section, name, **kwargs)
 
+    def getoptionalboolean(self, section: str, name: str, **kwargs) -> Optional[bool]:
+        """Helper function to report errors while extracting optional boolean configuration options"""
+        try:
+            return getboolean_opt(self.__config, self.config_file, section, name, **kwargs)
+        except Exception:  # pylint: disable=broad-except
+            # We cannot use "fallback=None" to handle ": None" w/ getboolean()
+            return None
+
     def get_items(self, section: str) -> Dict:
         """Helper function to return configuration items from a section"""
         res = {}
@@ -453,16 +461,13 @@ def __init_processing(self, accessor: _ConfigAccessor) -> None:
             accessor.throw(f"Unexpected value for Processing.cache_dem_by option: '{self.cache_dem_by}' is neither 'copy' nor 'symlink'")
 
         # - - - - - - - - - -[ Cut margins
-        try:
-            self.override_azimuth_cut_threshold_to: Optional[bool] = accessor.getboolean('Processing', 'override_azimuth_cut_threshold_to')
-        except Exception:  # pylint: disable=broad-except
-            # We cannot use "fallback=None" to handle ": None" w/ getboolean()
-            #: Internal to override analysing of top/bottom cutting: See :ref:`[Processing.override_azimuth_cut_threshold_to] <Processing.override_azimuth_cut_threshold_to>`
-            self.override_azimuth_cut_threshold_to = None
+        #: Internal to override analysing of top/bottom cutting: See :ref:`[Processing.override_azimuth_cut_threshold_to] <Processing.override_azimuth_cut_threshold_to>`
+        self.override_azimuth_cut_threshold_to: Optional[bool] = accessor.getoptionalboolean('Processing', 'override_azimuth_cut_threshold_to')
 
         # - - - - - - - - - -[ Calibration
         #: SAR Calibration applied: See :ref:`[Processing.calibration] <Processing.calibration>`
         self.calibration_type     = accessor.get('Processing', 'calibration')
+
         #: Shall we remove thermal noise: :ref:`[Processing.remove_thermal_noise] <Processing.remove_thermal_noise>`
         self.removethermalnoise   = accessor.getboolean('Processing', 'remove_thermal_noise')
         if self.removethermalnoise and otb_version() < '7.4.0':
@@ -477,27 +482,27 @@ def __init_processing(self, accessor: _ConfigAccessor) -> None:
 
         # - - - - - - - - - -[ Gamma area computation
         #: Resampling: See :ref:`[Processing.use_resampled_dem] <Processing.use_resampled_dem>`
-        self.use_resampled_dem                          = accessor.getboolean('Processing', 'use_resampled_dem', fallback=True)
-        no_use_resampled_dem                            = accessor.getboolean('Processing', 'no_use_resampled_dem', fallback=None)
+        self.use_resampled_dem            = accessor.getboolean('Processing', 'use_resampled_dem', fallback=True)
+        no_use_resampled_dem              = accessor.getboolean('Processing', 'no_use_resampled_dem', fallback=None)
         if no_use_resampled_dem is not None:
             accessor.throw("'no_use_resampled_dem' has be deprecated, please use the positive option: 'use_resampled_dem' instead")
 
         #: Resampling: See :ref:`[Processing.factor_x] <Processing.resample_dem_factor_x>`
-        self.resample_dem_factor_x :float               = accessor.getfloat('Processing', 'resample_dem_factor_x', fallback=2.0)
+        self.resample_dem_factor_x :float = accessor.getfloat('Processing', 'resample_dem_factor_x', fallback=2.0)
         #: Resampling: See :ref:`[Processing.factor_y] <Processing.resample_dem_factor_y>`
-        self.resample_dem_factor_y :float               = accessor.getfloat('Processing', 'resample_dem_factor_y', fallback=2.0)
+        self.resample_dem_factor_y :float = accessor.getfloat('Processing', 'resample_dem_factor_y', fallback=2.0)
 
         #: Gamma area: See :ref:`[Processing.distribute_area] <Processing.distribute_area>`
-        self.distribute_area :bool                      = accessor.getboolean('Processing', 'distribute_area', fallback=False)
+        self.distribute_area :bool        = accessor.getboolean('Processing', 'distribute_area', fallback=False)
         #: Gamma area: See :ref:`[Processing.inner_margin_ratio] <Processing.inner_margin_ratio>`
-        self.inner_margin_ratio :float                  = accessor.getfloat('Processing', 'inner_margin_ratio', fallback=0.01)
+        self.inner_margin_ratio :float    = accessor.getfloat('Processing', 'inner_margin_ratio', fallback=0.01)
         #: Gamma area: See :ref:`[Processing.outer_margin_ratio] <Processing.outer_margin_ratio>`
-        self.outer_margin_ratio :float                  = accessor.getfloat('Processing', 'outer_margin_ratio', fallback=0.04)
+        self.outer_margin_ratio :float    = accessor.getfloat('Processing', 'outer_margin_ratio', fallback=0.04)
 
         #: Gamma area to gamma naught rtc: See :ref:`[Processing.min_gamma_area] <Processing.min_gamma_area>`
-        self.min_gamma_area :float                      = accessor.getfloat('Processing', 'min_gamma_area', fallback=1.0)
+        self.min_gamma_area :float        = accessor.getfloat('Processing', 'min_gamma_area', fallback=1.0)
         #: Gamma area to gamma naught rtc: See :ref:`[Processing.calibration_factor] <Processing.calibration_factor>`
-        self.calibration_factor :float                  = accessor.getfloat('Processing', 'calibration_factor', fallback=1.0)
+        self.calibration_factor :float    = accessor.getfloat('Processing', 'calibration_factor', fallback=1.0)
 
         # - - - - - - - - - -[ Orthorectification
         #: Pixel size (in meters) of the output images: :ref:`[Processing.output_spatial_resolution] <Processing.output_spatial_resolution>`
@@ -514,21 +519,9 @@ def __init_processing(self, accessor: _ConfigAccessor) -> None:
         if not os.path.isfile(self.output_grid):
             accessor.throw(f"output_grid={self.output_grid} is not a valid path")
 
-        # IF tiles_list_in_file is set, use the option, and throw if there is an error
-        # ELSE: if unset, then use "tiles" option
-        tiles_file = accessor.get('Processing', 'tiles_list_in_file', fallback=None)
-        if tiles_file:
-            try:
-                with open(tiles_file, 'r', encoding='utf-8') as tiles_file_handle:
-                    tile_list = tiles_file_handle.readlines()
-                self.tile_list: List[str] = [s.rstrip() for s in tile_list]
-                logging.info("The following tiles will be processed: %s", self.tile_list)
-            except Exception as e:  # pylint: disable=broad-exception-caught
-                accessor.throw(f"Cannot read tile list file {tiles_file!r}", e)
-        else:
-            tiles = accessor.get('Processing', 'tiles')
-            #: List of S2 tiles to process: See :ref:`[Processing.tiles] <Processing.tiles>`
-            self.tile_list = _split_option(tiles)
+        #: List of S2 tiles to process: See :ref:`[Processing.tiles] <Processing.tiles>`
+        self.tile_list : List[str] = self.__extract_tile_list_option(accessor)
+        # logging.info("The following tiles will be processed: %s", self.tile_list)
 
         # - - - - - - - - - -[ Parallelization & RAM
         #: Number of tasks executed in parallel: See :ref:`[Processing.nb_parallel_processes] <Processing.nb_parallel_processes>`
@@ -586,6 +579,25 @@ def __init_filtering(self, accessor: _ConfigAccessor) -> None:
             else:
                 accessor.throw(f"Invalid despeckling filter value '{self.filter}'. Select one among none/lee/frost/gammamap/kuan")
 
+    def __extract_tile_list_option(self, accessor: _ConfigAccessor) -> List[str]:  # pylint: disable=inconsistent-return-statements
+        # NB: pylint is unable to see accessor.throw() is NoReturn, hence the disable=inconsistent-return-statements
+        # It seems related to https://github.com/pylint-dev/pylint/issues/9692
+
+        # IF tiles_list_in_file is set, use the option, and throw if there is an error
+        # ELSE: if unset, then use "tiles" option
+        tiles_file = accessor.get('Processing', 'tiles_list_in_file', fallback=None)
+        if tiles_file:
+            try:
+                with open(tiles_file, 'r', encoding='utf-8') as tiles_file_handle:
+                    tile_list = tiles_file_handle.readlines()
+                    return [s.rstrip() for s in tile_list]
+            except Exception as e:  # pylint: disable=broad-exception-caught
+                accessor.throw(f"Cannot read tile list file {tiles_file!r}", e)
+        else:
+            tiles = accessor.get('Processing', 'tiles')
+            return _split_option(tiles)
+
+
     # ----------------------------------------------------------------------
     def __init_fname_fmt(self, accessor: _ConfigAccessor) -> None:
         # Permit to override default file name formats

s1tiling/libs/orbit/_manager.py

@@ -302,6 +302,7 @@ def _fetch_eof_files(  # pylint: disable=too-many-arguments
     def search_for(  # pylint: disable=too-many-arguments
             self,
             relative_orbits: List[int],
+            *,
             missions       : Iterable[str] = (),
             first_date     : Optional[datetime] = None,
             last_date      : Optional[datetime] = None,

s1tiling/libs/otbpipeline.py

@@ -43,7 +43,6 @@
 from itertools import filterfalse
 import logging
 import logging.handlers
-import multiprocessing
 from typing import Dict, Generic, List, Optional, Protocol, Set, Tuple, Type, TypeVar, Union, runtime_checkable
 
 from distributed import get_worker
@@ -1061,8 +1060,6 @@ def generate_tasks(self, do_watch_ram=False) -> Tuple[TaskNodeDict, List[str], L
         return tasks, final_products, []
 
 
-# ======================================================================
-# Multi processing related (old) code
 def mp_worker_config(queue):
     """
     Worker configuration function called by Pool().
@@ -1077,110 +1074,3 @@ def mp_worker_config(queue):
     global logger
     logger = logging.getLogger()
     logger.addHandler(qh)
-
-
-# TODO: try to make it static...
-def execute4mp(pipeline):
-    """
-    Internal worker function used by multiprocess to execute a pipeline.
-    """
-    return pipeline.do_execute()
-
-
-class PoolOfOTBExecutions:
-    """
-    Internal multiprocess Pool of OTB pipelines.
-    """
-    def __init__(self,
-            title,
-            do_measure,
-            nb_procs, nb_threads,
-            log_queue, log_queue_listener,
-            debug_otb) -> None:
-        """
-        constructor
-        """
-        self.__pool = []
-        self.__title              = title
-        self.__do_measure         = do_measure
-        self.__nb_procs           = nb_procs
-        self.__nb_threads         = nb_threads
-        self.__log_queue          = log_queue
-        self.__log_queue_listener = log_queue_listener
-        self.__debug_otb          = debug_otb
-
-    def new_pipeline(self, **kwargs):
-        """
-        Register a new pipeline.
-        """
-        in_memory    = kwargs.get('in_memory', True)
-        do_watch_ram = kwargs.get('do_watch_ram', False)
-        pipeline = Pipeline(self.__do_measure, in_memory, do_watch_ram)
-        self.__pool.append(pipeline)
-        return pipeline
-
-    def process(self):
-        """
-        Executes all the pipelines in parallel.
-        """
-        nb_cmd = len(self.__pool)
-
-        os.environ["ITK_GLOBAL_DEFAULT_NUMBER_OF_THREADS"] = str(self.__nb_threads)
-        os.environ["GDAL_NUM_THREADS"] = str(self.__nb_threads)
-        os.environ['OTB_LOGGER_LEVEL'] = 'DEBUG'
-        if self.__debug_otb:  # debug OTB applications with gdb => do not spawn process!
-            execute4mp(self.__pool[0])
-        else:
-            with multiprocessing.Pool(self.__nb_procs, mp_worker_config, [self.__log_queue]) as pool:
-                self.__log_queue_listener.start()
-                for count, result in enumerate(pool.imap_unordered(execute4mp, self.__pool), 1):
-                    logger.info("%s correctly finished", result)
-                    logger.info(' --> %s... %s%%', self.__title, count * 100. / nb_cmd)
-
-                pool.close()
-                pool.join()
-                self.__log_queue_listener.stop()
-
-
-class Processing:
-    """
-    Entry point for executing multiple instance of the same pipeline of
-    different inputs.
-
-    1. The object is initialized with a log queue and its listener
-    2. The pipeline is registered with a list of :class`StepFactory` s
-    3. The processing is done on a list of :class:`FirstStep` s
-    """
-    def __init__(self, cfg, debug_otb) -> None:
-        self.__log_queue          = cfg.log_queue
-        self.__log_queue_listener = cfg.log_queue_listener
-        self.__cfg                = cfg
-        self.__factory_steps      = []
-        self.__debug_otb          = debug_otb
-
-    def register_pipeline(self, factory_steps):
-        """
-        Register a list of :class:`StepFactory` s that describes a pipeline.
-        """
-        # Automatically append the final storing step
-        self.__factory_steps = factory_steps + [Store]
-
-    def process(self, startpoints):
-        """
-        Defines pipelines from the registered steps. Each pipeline is instanciated with a
-        startpoint. Then they registered into the PoolOfOTBExecutions.
-        The pool is finally executed.
-        """
-        assert self.__factory_steps
-        pool = PoolOfOTBExecutions("testpool", True,
-                self.__cfg.nb_procs, self.__cfg.OTBThreads,
-                self.__log_queue, self.__log_queue_listener, debug_otb=self.__debug_otb)
-        for startpoint in startpoints:
-            logger.info("register processing of %s", startpoint.basename)
-            pipeline = pool.new_pipeline(in_memory=True)
-            pipeline.set_inputs(startpoint)
-            for factory in self.__factory_steps:
-                pipeline.push(factory(self.__cfg))
-
-        logger.debug('Launch pipelines')
-        pool.process()