feat: encapsulate all bbox computation parameters and separate bbox c…

…omputation method

backend/geonature/core/gn_synthese/imports/__init__.py

@@ -11,6 +11,7 @@
 from geonature.core.gn_synthese.models import Synthese, TSources
 
 from geonature.core.imports.models import Entity, EntityField, BibFields, TImports
+import geonature.core.imports.bbox as bbox
 from geonature.core.imports.utils import (
     load_transient_data_in_dataframe,
     update_transient_data_from_dataframe,
@@ -354,16 +355,19 @@ def remove_data_from_synthese(imprt):
         db.session.delete(source)
 
 
-def get_name_geom_4326_field():
-    """Return the name of the field that contains the 4326 geometry.
-    For synthese, the name is actually the same for import transient table
-    `gn_imports.t_imports_synthese` and for the destination table `gn_synthese.synthese`.
+def get_bbox_computation_infos():
+    """Return the infos that will be used for the bbox computation.
+    For synthese, the bbox computation is relevant. The entity is "observation", and the name_field is actually the same for import transient table `gn_imports.t_imports_synthese` and for the destination table `gn_synthese.synthese`.
     Returns
     -------
-    str
-        The name of the field
+    dict
+        A dictionary indexed by the keys from bbox.Key enum
     """
-    return "the_geom_4326"
+    return {
+        bbox.Key.IS_RELEVANT: True,
+        bbox.Key.NAME_FIELD: "the_geom_4326",
+        bbox.Key.CODE_ENTITY: "observation",
+    }
 
 
 def get_where_clause_id_import(imprt):

backend/geonature/core/gn_synthese/module.py

@@ -3,7 +3,7 @@
     check_transient_data,
     import_data_to_synthese,
     remove_data_from_synthese,
-    get_name_geom_4326_field,
+    get_bbox_computation_infos,
     get_where_clause_id_import,
     report_plot,
 )
@@ -21,7 +21,7 @@ def generate_input_url_for_dataset(self, dataset):
         "check_transient_data": check_transient_data,
         "import_data_to_destination": import_data_to_synthese,
         "remove_data_from_destination": remove_data_from_synthese,
-        "get_name_geom_4326_field": get_name_geom_4326_field,
+        "get_bbox_computation_infos": get_bbox_computation_infos,
         "get_where_clause_id_import": get_where_clause_id_import,
         "statistics_labels": [
             {"key": "taxa_count", "value": "Nombre de taxons importés"},

backend/geonature/core/imports/bbox.py

@@ -0,0 +1,88 @@
+from enum import Enum
+from geonature.core.imports.models import Entity
+from sqlalchemy import func, select
+from geonature.utils.env import db
+import json
+
+
+class Key(str, Enum):
+    CODE_ENTITY = "code_entity"
+    NAME_FIELD = "field_name"
+    IS_RELEVANT = "is_relevant"
+
+
+def get_valid_bbox(imprt):
+    """Get the valid bounding box for a given import.
+    Parameters
+    ----------
+    imprt : geonature.core.imports.models.TImports
+        The import object.
+    entity : geonature.core.imports.models.Entity
+        The entity object (e.g.: observation, station...).
+    Returns
+    -------
+    dict or None
+        The valid bounding box as a JSON object, or None if no valid bounding box.
+    Raises
+    ------
+    NotImplementedError
+        If the function is not implemented for the destination of the import.
+    """
+    # Retrieve the name of the field and the name of the entity to retrieve geometries from
+    if "get_bbox_computation_infos" not in imprt.destination.module._imports_:
+        raise NotImplementedError(
+            f"function get_valid_bbox not implemented for an import with destination '{imprt.destination.code}, needs `get_bbox_computation_infos` function"
+        )
+    infos = imprt.destination.module._imports_["get_bbox_computation_infos"]()
+
+    if not Key.IS_RELEVANT in infos:
+        raise NotImplementedError(
+            f"The function get_valid_bbox implementation for an import with destination '{imprt.destination.code}' is incomplete. It requires '{Key.IS_RELEVANT}' field"
+        )
+    if not infos[Key.IS_RELEVANT]:
+        return None
+
+    if not Key.CODE_ENTITY in infos:
+        raise NotImplementedError(
+            f"The function get_valid_bbox implementation for an import with destination '{imprt.destination.code}' is incomplete. It requires '{Key.CODE_ENTITY}' field"
+        )
+    code_entity = infos[Key.CODE_ENTITY]
+
+    if not Key.NAME_FIELD in infos:
+        raise NotImplementedError(
+            f"The function get_valid_bbox implementation for an import with destination '{imprt.destination.code}' is incomplete. It requires '${Key.NAME_FIELD}' field"
+        )
+    name_geom_4326_field = infos[Key.NAME_FIELD]
+
+    # Retrieve the where clause to filter data for the given import
+    if "get_where_clause_id_import" not in imprt.destination.module._imports_:
+        raise NotImplementedError(
+            f"function get_valid_bbox not implemented for an import with destination '{imprt.destination.code}, needs `get_where_clause_id_import` function"
+        )
+
+    entity = Entity.query.filter_by(destination=imprt.destination, code=code_entity).one()
+
+    where_clause_id_import = imprt.destination.module._imports_["get_where_clause_id_import"](imprt)
+
+    # Build the statement to retrieve the valid bounding box
+    statement = None
+    if imprt.loaded == True:
+        # Compute from entries in the transient table and related to the import
+        transient_table = imprt.destination.get_transient_table()
+        statement = (
+            select(func.ST_AsGeojson(func.ST_Extent(transient_table.c[name_geom_4326_field])))
+            .where(where_clause_id_import)
+            .where(transient_table.c[entity.validity_column] == True)
+        )
+    else:
+        # Compute from entries in the destination table and related to the import
+        statement = select(
+            func.ST_AsGeojson(func.ST_Extent(destination_table.c[name_geom_4326_field]))
+        ).where(where_clause_id_import)
+
+    # Execute the statement to eventually retrieve the valid bounding box
+    (valid_bbox,) = db.session.execute(statement).fetchone()
+
+    # Return the valid bounding box or None
+    if valid_bbox:
+        return json.loads(valid_bbox)

backend/geonature/core/imports/routes/imports.py

@@ -39,14 +39,14 @@
 from geonature.core.imports.blueprint import blueprint
 from geonature.core.imports.utils import (
     ImportStep,
-    get_valid_bbox,
     detect_encoding,
     detect_separator,
     insert_import_data_in_transient_table,
     get_file_size,
     clean_import,
     generate_pdf_from_template,
 )
+from geonature.core.imports.bbox import get_valid_bbox
 from geonature.core.imports.tasks import do_import_checks, do_import_in_destination
 
 IMPORTS_PER_PAGE = 15
@@ -465,21 +465,8 @@ def preview_valid_data(scope, imprt):
     if not imprt.processed:
         raise Conflict("Import must have been prepared before executing this action.")
 
-    # FIXME FIXME FIXME
-    if imprt.destination.code == "synthese":
-        code_entity = "observation"
-        name_field_geom_4326 = "the_geom_4326"
-    elif imprt.destination.code == "occhab":
-        code_entity = "station"
-        name_field_geom_4326 = "geom_4326"
-
-    entity = Entity.query.filter_by(destination=imprt.destination, code=code_entity).one()
-    geom_4326_field = BibFields.query.filter_by(
-        destination=imprt.destination, name_field=name_field_geom_4326
-    ).one()
-
     data = {
-        "valid_bbox": get_valid_bbox(imprt, entity, geom_4326_field),
+        "valid_bbox": get_valid_bbox(imprt),
         "entities": [],
     }
 

backend/geonature/core/imports/utils.py

@@ -6,7 +6,7 @@
 from datetime import datetime, timedelta
 
 from flask import current_app, render_template
-from sqlalchemy import func, delete
+from sqlalchemy import delete
 from chardet.universaldetector import UniversalDetector
 from sqlalchemy.sql.expression import select, insert
 import pandas as pd
@@ -18,8 +18,6 @@
 
 from geonature.utils.sentry import start_sentry_child
 from geonature.core.imports.models import ImportUserError, BibFields
-from geonature.core.gn_commons.models.base import TModules
-from geonature.core.gn_synthese.models import TSources
 
 
 class ImportStep(IntEnum):
@@ -103,61 +101,6 @@ def detect_separator(f, encoding):
     return dialect.delimiter
 
 
-def get_valid_bbox(imprt, entity):
-    """Get the valid bounding box for a given import.
-    Parameters
-    ----------
-    imprt : geonature.core.imports.models.TImports
-        The import object.
-    entity : geonature.core.imports.models.Entity
-        The entity object (e.g.: observation, station...).
-    Returns
-    -------
-    dict or None
-        The valid bounding box as a JSON object, or None if no valid bounding box.
-    Raises
-    ------
-    NotImplementedError
-        If the function is not implemented for the destination of the import.
-    """
-    # Retrieve the name of the geom field to retrieve geometries of data from
-    if "get_name_geom_4326_field" not in imprt.destination.module._imports_:
-        raise NotImplementedError(
-            f"function get_valid_bbox not implemented for an import with destination '{imprt.destination.code}, needs `get_name_geom_4326_field` function"
-        )
-    name_geom_4326_field = imprt.destination.module._imports_["get_name_geom_4326_field"]()
-
-    # Retrieve the where clause to filter data for the given import
-    if "get_where_clause_id_import" not in imprt.destination.module._imports_:
-        raise NotImplementedError(
-            f"function get_valid_bbox not implemented for an import with destination '{imprt.destination.code}, needs `get_where_clause_id_import` function"
-        )
-    where_clause_id_import = imprt.destination.module._imports_["get_where_clause_id_import"](imprt)
-
-    # Build the statement to retrieve the valid bounding box
-    statement = None
-    if imprt.loaded == True:
-        # Compute from entries in the transient table and related to the import
-        transient_table = imprt.destination.get_transient_table()
-        statement = (
-            select(func.ST_AsGeojson(func.ST_Extent(transient_table.c[name_geom_4326_field])))
-            .where(where_clause_id_import)
-            .where(transient_table.c[entity.validity_column] == True)
-        )
-    else:
-        # Compute from entries in the destination table and related to the import
-        statement = select(
-            func.ST_AsGeojson(func.ST_Extent(destination_table.c[name_geom_4326_field]))
-        ).where(where_clause_id_import)
-
-    # Execute the statement to eventually retrieve the valid bounding box
-    (valid_bbox,) = db.session.execute(statement).fetchone()
-
-    # Return the valid bounding box or None
-    if valid_bbox:
-        return json.loads(valid_bbox)
-
-
 def preprocess_value(df, field, source_col):
     if field.multi:
         assert type(source_col) is list

contrib/gn_module_occhab/backend/gn_module_occhab/imports/__init__.py

@@ -6,6 +6,7 @@
 from geonature.utils.sentry import start_sentry_child
 
 from geonature.core.imports.models import Entity, EntityField, BibFields
+import geonature.core.imports.bbox as bbox
 from geonature.core.imports.utils import (
     load_transient_data_in_dataframe,
     update_transient_data_from_dataframe,
@@ -309,16 +310,19 @@ def remove_data_from_occhab(imprt):
         )
 
 
-def get_name_geom_4326_field():
-    """Return the name of the field that contains the 4326 geometry.
-    For occhab, the name is actually the same for import transient table
-    `gn_imports.t_imports_occhab` and for the destination table `pr_occhab.t_stations`.
+def get_bbox_computation_infos():
+    """Return the infos that will be used for the bbox computation.
+    For occhab, the bbox computation is relevant. The entity is "station", and the name_field is actually the same for import transient table `gn_imports.t_imports_synthese` and for the destination table `gn_synthese.synthese`.
     Returns
     -------
-    str
-        The name of the field
+    dict
+        A dictionary indexed by the keys from bbox.Key enum
     """
-    return "geom_4326"
+    return {
+        bbox.Key.IS_RELEVANT: True,
+        bbox.Key.NAME_FIELD: "geom_4326",
+        bbox.Key.CODE_ENTITY: "station",
+    }
 
 
 def get_where_clause_id_import(imprt):

contrib/gn_module_occhab/backend/gn_module_occhab/module.py

@@ -4,7 +4,7 @@
     check_transient_data,
     import_data_to_occhab,
     remove_data_from_occhab,
-    get_name_geom_4326_field,
+    get_bbox_computation_infos,
     get_where_clause_id_import,
 )
 
@@ -22,7 +22,7 @@ def generate_input_url_for_dataset(self, dataset):
         "check_transient_data": check_transient_data,
         "import_data_to_destination": import_data_to_occhab,
         "remove_data_from_destination": remove_data_from_occhab,
-        "get_name_geom_4326_field": get_name_geom_4326_field,
+        "get_bbox_computation_infos": get_bbox_computation_infos,
         "get_where_clause_id_import": get_where_clause_id_import,
         "statistics_labels": [
             {"key": "station_count", "value": "Nombre de stations importées"},