Various fixes to backend. Fetch multiple stat surfaces at the same time

Author: HansKallekleiv

backend/src/backend/primary/routers/surface/router.py

@@ -195,7 +195,7 @@ async def get_property_surface_resampled_to_statistical_static_surface(
             name=name_mesh,
             attribute=attribute_mesh,
         )
-        xtgeo_surf_property = await access.get_statistical_surface_data_async(
+        xtgeo_surf_property = await access.get_statistical_surfaces_data_async(
             statistic_function=service_stat_func_to_compute,
             name=name_property,
             attribute=attribute_property,
@@ -237,41 +237,35 @@ async def well_intersection_statistics(
     access = await SurfaceAccess.from_case_uuid(
         authenticated_user.get_sumo_access_token(), ensemble_ident.case_uuid, ensemble_ident.ensemble_name
     )
-    surfaces = []
 
-    async def fetch_surface(statistic, surface_name):
-        surface = await access.get_statistical_surface_data_async(
-            statistic_function=StatisticFunction.from_string_value(statistic),
+    async def fetch_surface(statistics, surface_name):
+        surfaces = await access.get_statistical_surfaces_data_async(
+            statistic_functions=[StatisticFunction.from_string_value(statistic) for statistic in statistics],
             name=surface_name,
             attribute=statistical_surface_set_spec.surface_attribute,
             realizations=statistical_surface_set_spec.realization_nums,
         )
-        surface.name = surface_name + "_" + statistic
-        return surface
+        print(surfaces, "fetch")
+        return surfaces
 
     async def fetch_all_surfaces():
         tasks = []
-        for statistic in statistical_surface_set_spec.statistic_function:
-            for surface_name in statistical_surface_set_spec.surface_names:
-                # Create a task for each combination of statistic and surface_name
-                task = fetch_surface(statistic, surface_name)
-                tasks.append(task)
+        for surface_name in statistical_surface_set_spec.surface_names:
+            task = fetch_surface(statistical_surface_set_spec.statistic_function, surface_name)
+            tasks.append(task)
 
         # Run all the tasks concurrently
         tmp_surfaces = await asyncio.gather(*tasks)
+        print(tmp_surfaces)
         return tmp_surfaces
 
-    # for statistic in statistical_surface_set_spec.statistic_function:
-    #     for surface_name in statistical_surface_set_spec.surface_names:
-    #         surface = await access.get_statistical_surface_data_async(
-    #             statistic_function=StatisticFunction.from_string_value(statistic),
-    #             name=surface_name,
-    #             attribute=statistical_surface_set_spec.surface_attribute,
-    #             realizations=statistical_surface_set_spec.realization_nums,
-    #         )
-    #         surface.name = surface_name + "_" + statistic
-    #         surfaces.append(surface)
-    surfaces = await fetch_all_surfaces()
+    nested_surfaces = await fetch_all_surfaces()
+    surfaces = []
+    if nested_surfaces:
+        for surface_list in nested_surfaces:
+            if surface_list:
+                for surface in surface_list:
+                    surfaces.append(surface)
 
     fence_arr = np.array(
         [
@@ -282,7 +276,6 @@ async def fetch_all_surfaces():
         ]
     ).T
     intersections = await make_intersections(surfaces, fence_arr)
-    print(intersections)
     return intersections
 
 

backend/src/backend/primary/routers/well/router.py

@@ -1,7 +1,7 @@
 import logging
 from typing import List, Union
 
-from fastapi import APIRouter, Depends, Query
+from fastapi import APIRouter, Depends, Query, HTTPException
 
 from src.services.smda_access import mocked_drogon_smda_access
 from src.services.smda_access.well_access import WellAccess
@@ -24,11 +24,9 @@ async def get_well_headers(
     # fmt:on
 ) -> List[WellBoreHeader]:
     """Get well headers for all wells in the field"""
-    print(case_uuid, " case_uuid FIELD")
     case_inspector = await SumoCase.from_case_uuid(authenticated_user.get_sumo_access_token(), case_uuid)
     field_identifier = (await case_inspector.get_field_identifiers())[0]
     well_access: Union[WellAccess, mocked_drogon_smda_access.WellAccess]
-    print(field_identifier, "FIELD IDENTIFIER FIELD")
     if field_identifier == "DROGON":
         # Handle DROGON
         well_access = mocked_drogon_smda_access.WellAccess(authenticated_user.get_smda_access_token())
@@ -50,7 +48,6 @@ async def get_field_well_trajectories(
     case_inspector = await SumoCase.from_case_uuid(authenticated_user.get_sumo_access_token(), case_uuid)
     field_identifier = (await case_inspector.get_field_identifiers())[0]
     well_access: Union[WellAccess, mocked_drogon_smda_access.WellAccess]
-    print(field_identifier, "FIELD IDENTIFIER FIELD")
     if field_identifier == "DROGON":
         # Handle DROGON
         well_access = mocked_drogon_smda_access.WellAccess(authenticated_user.get_smda_access_token())
@@ -78,4 +75,8 @@ async def get_well_trajectories(
     else:
         well_access = WellAccess(authenticated_user.get_smda_access_token())
 
-    return await well_access.get_wellbore_trajectories(wellbore_uuids=wellbore_uuids)
+    try:
+        trajectories = await well_access.get_wellbore_trajectories(wellbore_uuids=wellbore_uuids)
+        return trajectories
+    except KeyError:
+        raise HTTPException(status_code=404, detail="No wellbore trajectory data found")

backend/src/services/sumo_access/surface_access.py

@@ -138,14 +138,14 @@ async def get_realization_surface_data_async(
 
         return xtgeo_surf
 
-    async def get_statistical_surface_data_async(
+    async def get_statistical_surfaces_data_async(
         self,
-        statistic_function: StatisticFunction,
+        statistic_functions: list[StatisticFunction],
         name: str,
         attribute: str,
         time_or_interval_str: Optional[str] = None,
         realizations: Optional[List[int]] = None,
-    ) -> Optional[xtgeo.RegularSurface]:
+    ) -> list[xtgeo.RegularSurface]:
         """
         Compute statistic and return surface data
         """
@@ -192,12 +192,12 @@ async def get_statistical_surface_data_async(
 
         et_collect_reals_ms = timer.lap_ms()
 
-        xtgeo_surf = await _compute_statistical_surface_async(statistic_function, surface_collection)
+        xtgeo_surfs = await _compute_statistical_surfaces_async(statistic_functions, surface_collection)
+        print(xtgeo_surfs)
         et_calc_stat_ms = timer.lap_ms()
 
-        if not xtgeo_surf:
+        if not xtgeo_surfs:
             LOGGER.warning(f"Could not calculate statistical surface using Sumo for {addr_str}")
-            return None
 
         LOGGER.debug(
             f"Calculated statistical surface using Sumo in: {timer.elapsed_ms()}ms ("
@@ -207,32 +207,50 @@ async def get_statistical_surface_data_async(
             f"({addr_str} {len(realizations)=} )"
         )
 
-        return xtgeo_surf
+        return xtgeo_surfs
 
     def _make_addr_str(self, real_num: int, name: str, attribute: str, date_str: Optional[str]) -> str:
         addr_str = f"R:{real_num}__N:{name}__A:{attribute}__D:{date_str}__I:{self._iteration_name}__C:{self._case_uuid}"
         return addr_str
 
 
-async def _compute_statistical_surface_async(
-    statistic: StatisticFunction, surface_coll: SurfaceCollection
-) -> xtgeo.RegularSurface:
-    xtgeo_surf: xtgeo.RegularSurface = None
-    if statistic == StatisticFunction.MIN:
-        xtgeo_surf = await surface_coll.min_async()
-    elif statistic == StatisticFunction.MAX:
-        xtgeo_surf = await surface_coll.max_async()
-    elif statistic == StatisticFunction.MEAN:
-        xtgeo_surf = await surface_coll.mean_async()
-    elif statistic == StatisticFunction.P10:
-        xtgeo_surf = await surface_coll.p10_async()
-    elif statistic == StatisticFunction.P90:
-        xtgeo_surf = await surface_coll.p90_async()
-    elif statistic == StatisticFunction.P50:
-        xtgeo_surf = await surface_coll.p50_async()
-    elif statistic == StatisticFunction.STD:
-        xtgeo_surf = await surface_coll.std_async()
-    else:
-        raise ValueError("Unhandled statistic function")
-
-    return xtgeo_surf
+async def _compute_statistical_surfaces_async(
+    statistics: StatisticFunction, surface_coll: SurfaceCollection
+) -> list[xtgeo.RegularSurface]:
+    # xtgeo_surf: xtgeo.RegularSurface = None
+    # if statistic == StatisticFunction.MIN:
+    #     xtgeo_surf = await surface_coll.min_async()
+    # elif statistic == StatisticFunction.MAX:
+    #     xtgeo_surf = await surface_coll.max_async()
+    # elif statistic == StatisticFunction.MEAN:
+    #     xtgeo_surf = await surface_coll.mean_async()
+    # elif statistic == StatisticFunction.P10:
+    #     xtgeo_surf = await surface_coll.p10_async()
+    # elif statistic == StatisticFunction.P90:
+    #     xtgeo_surf = await surface_coll.p90_async()
+    # elif statistic == StatisticFunction.P50:
+    #     xtgeo_surf = await surface_coll.p50_async()
+    # elif statistic == StatisticFunction.STD:
+    #     xtgeo_surf = await surface_coll.std_async()
+    # else:
+    #     raise ValueError("Unhandled statistic function")
+
+    # return xtgeo_surf
+    objects = await surface_coll._utils.get_objects_async(500, surface_coll._query, ["_id"])
+    object_ids = list(map(lambda obj: obj["_id"], objects))
+    res = await surface_coll._sumo.post_async(
+        "/surface/aggregate",
+        json={"operation": [stat.value for stat in statistics], "object_ids": object_ids},
+    )
+    res_bytes = res.content
+    # print(res.content)
+    import zipfile
+
+    surfs = []
+    with zipfile.ZipFile(BytesIO(res_bytes)) as z:
+        # return z.read(z.namelist()[0])
+        for stat_op in z.namelist():
+            xtgeo_surf = xtgeo.surface_from_file(BytesIO(z.read(stat_op)))
+            xtgeo_surf.name = StatisticFunction.from_string_value(stat_op.upper())
+            surfs.append(xtgeo_surf)
+    return surfs