216 delta ensembles simulation time series

backend_py/primary/primary/routers/timeseries/converters.py

@@ -31,14 +31,7 @@ def to_api_vector_statistic_data(
     """
     Create API VectorStatisticData from service layer VectorStatistics
     """
-    value_objects: List[schemas.StatisticValueObject] = []
-    for api_func_enum in schemas.StatisticFunction:
-        service_func_enum = StatisticFunction.from_string_value(api_func_enum.value)
-        if service_func_enum is not None:
-            value_arr = vector_statistics.values_dict.get(service_func_enum)
-            if value_arr is not None:
-                value_objects.append(schemas.StatisticValueObject(statistic_function=api_func_enum, values=value_arr))
-
+    value_objects = _create_statistic_value_object_list(vector_statistics)
     ret_data = schemas.VectorStatisticData(
         realizations=vector_statistics.realizations,
         timestamps_utc_ms=vector_statistics.timestamps_utc_ms,
@@ -48,3 +41,36 @@ def to_api_vector_statistic_data(
     )
 
     return ret_data
+
+
+def to_api_delta_ensemble_vector_statistic_data(
+    vector_statistics: VectorStatistics, is_rate: bool, unit: str
+) -> schemas.VectorStatisticData:
+    """
+    Create API VectorStatisticData from service layer VectorStatistics
+    """
+    value_objects = _create_statistic_value_object_list(vector_statistics)
+    ret_data = schemas.VectorStatisticData(
+        realizations=vector_statistics.realizations,
+        timestamps_utc_ms=vector_statistics.timestamps_utc_ms,
+        value_objects=value_objects,
+        unit=unit,
+        is_rate=is_rate,
+    )
+
+    return ret_data
+
+
+def _create_statistic_value_object_list(vector_statistics: VectorStatistics) -> list[schemas.StatisticValueObject]:
+    """
+    Create list of statistic value objects from vector statistics object
+    """
+    value_objects: list[schemas.StatisticValueObject] = []
+    for api_func_enum in schemas.StatisticFunction:
+        service_func_enum = StatisticFunction.from_string_value(api_func_enum.value)
+        if service_func_enum is not None:
+            value_arr = vector_statistics.values_dict.get(service_func_enum)
+            if value_arr is not None:
+                value_objects.append(schemas.StatisticValueObject(statistic_function=api_func_enum, values=value_arr))
+
+    return value_objects

backend_py/primary/primary/routers/timeseries/router.py

@@ -12,8 +12,10 @@
 from primary.services.sumo_access.parameter_access import ParameterAccess
 from primary.services.sumo_access.summary_access import Frequency, SummaryAccess
 from primary.services.utils.authenticated_user import AuthenticatedUser
+from primary.services.summary_delta_vectors import create_delta_vector_table, create_realization_delta_vector_list
 
 from . import converters, schemas
+import asyncio
 
 LOGGER = logging.getLogger(__name__)
 
@@ -48,6 +50,56 @@ async def get_vector_list(
     return ret_arr
 
 
+@router.get("/delta_ensemble_vector_list/")
+async def get_delta_ensemble_vector_list(
+    response: Response,
+    authenticated_user: Annotated[AuthenticatedUser, Depends(AuthHelper.get_authenticated_user)],
+    compare_case_uuid: Annotated[str, Query(description="Sumo case uuid for compare ensemble")],
+    compare_ensemble_name: Annotated[str, Query(description="Compare ensemble name")],
+    reference_case_uuid: Annotated[str, Query(description="Sumo case uuid for reference ensemble")],
+    reference_ensemble_name: Annotated[str, Query(description="Reference ensemble name")],
+) -> list[schemas.VectorDescription]:
+    """Get list of all vectors for a delta ensemble based on all vectors in a given Sumo ensemble, excluding any historical vectors
+
+    Definition:
+
+        delta_ensemble = compare_ensemble - reference_ensemble
+    """
+
+    perf_metrics = ResponsePerfMetrics(response)
+
+    compare_ensemble_access = SummaryAccess.from_case_uuid(
+        authenticated_user.get_sumo_access_token(), compare_case_uuid, compare_ensemble_name
+    )
+    reference_ensemble_access = SummaryAccess.from_case_uuid(
+        authenticated_user.get_sumo_access_token(), reference_case_uuid, reference_ensemble_name
+    )
+    perf_metrics.record_lap("get-access")
+
+    # Get vectors parallel
+    compare_vector_info_arr, reference_vector_info_arr = await asyncio.gather(
+        compare_ensemble_access.get_available_vectors_async(),
+        reference_ensemble_access.get_available_vectors_async(),
+    )
+    perf_metrics.record_lap("get-available-vectors")
+
+    # Create intersection of vector names
+    vector_names = {vi.name for vi in compare_vector_info_arr}
+    vector_names.intersection_update({vi.name for vi in reference_vector_info_arr})
+    perf_metrics.record_lap("create-vectors-names-intersection")
+
+    # Create vector descriptions, no historical vectors!
+    ret_arr: list[schemas.VectorDescription] = [
+        schemas.VectorDescription(name=vi, descriptive_name=vi, has_historical=False) for vi in vector_names
+    ]
+
+    perf_metrics.record_lap("convert-data-to-schema")
+
+    LOGGER.info(f"Got delta ensemble vector list in: {perf_metrics.to_string()}")
+
+    return ret_arr
+
+
 @router.get("/realizations_vector_data/")
 async def get_realizations_vector_data(
     # fmt:off
@@ -91,6 +143,98 @@ async def get_realizations_vector_data(
     return ret_arr
 
 
+@router.get("/delta_ensemble_realizations_vector_data/")
+async def get_delta_ensemble_realizations_vector_data(
+    # fmt:off
+    response: Response,
+    authenticated_user: Annotated[AuthenticatedUser, Depends(AuthHelper.get_authenticated_user)],
+    compare_case_uuid: Annotated[str, Query(description="Sumo case uuid for compare ensemble")],
+    compare_ensemble_name: Annotated[str, Query(description="Compare ensemble name")],
+    reference_case_uuid: Annotated[str, Query(description="Sumo case uuid for reference ensemble")],
+    reference_ensemble_name: Annotated[str, Query(description="Reference ensemble name")],
+    vector_name:  Annotated[str, Query(description="Name of the vector")],
+    resampling_frequency: Annotated[schemas.Frequency, Query(description="Resampling frequency")],
+    realizations: Annotated[list[int] | None, Query(description="Optional list of realizations to include. If not specified, all realizations will be returned.")] = None,
+    # relative_to_timestamp: Annotated[datetime.datetime | None, Query(description="Calculate relative to timestamp")] = None,
+    # fmt:on
+) -> list[schemas.VectorRealizationData]:
+    """Get vector data per realization
+
+    Definition:
+
+        delta_ensemble = compare_ensemble - reference_ensemble
+
+    """
+
+    perf_metrics = ResponsePerfMetrics(response)
+
+    service_freq = Frequency.from_string_value(resampling_frequency.value)
+    if service_freq is None:
+        raise HTTPException(
+            status_code=400, detail="Resampling frequency must be specified to create delta ensemble vector"
+        )
+
+    compare_ensemble_access = SummaryAccess.from_case_uuid(
+        authenticated_user.get_sumo_access_token(), compare_case_uuid, compare_ensemble_name
+    )
+    reference_ensemble_access = SummaryAccess.from_case_uuid(
+        authenticated_user.get_sumo_access_token(), reference_case_uuid, reference_ensemble_name
+    )
+
+    # Get tables parallel
+    # - Resampled data is assumed to be s.t. dates/timestamps are comparable between ensembles and cases, i.e. timestamps
+    #   for a resampling of a daily vector in both ensembles should be the same
+    (compare_vector_table_pa, compare_metadata), (reference_vector_table_pa, reference_metadata) = await asyncio.gather(
+        compare_ensemble_access.get_vector_table_async(
+            vector_name=vector_name,
+            resampling_frequency=service_freq,
+            realizations=realizations,
+        ),
+        reference_ensemble_access.get_vector_table_async(
+            vector_name=vector_name,
+            resampling_frequency=service_freq,
+            realizations=realizations,
+        ),
+    )
+
+    # Check for mismatching metadata
+    if compare_metadata.is_rate != reference_metadata.is_rate:
+        raise HTTPException(
+            status_code=400, detail="Rate mismatch between ensembles for delta ensemble statistical vector data"
+        )
+    if compare_metadata.unit != reference_metadata.unit:
+        raise HTTPException(
+            status_code=400, detail="Unit mismatch between ensembles for delta ensemble statistical vector data"
+        )
+
+    # Get metadata from reference ensemble
+    is_rate = reference_metadata.is_rate
+    unit = reference_metadata.unit
+
+    # Create delta ensemble data
+    delta_vector_table = create_delta_vector_table(compare_vector_table_pa, reference_vector_table_pa, vector_name)
+    perf_metrics.record_lap("create-delta-vector-table")
+
+    realization_delta_vector_list = create_realization_delta_vector_list(delta_vector_table, vector_name, is_rate, unit)
+    perf_metrics.record_lap("create-realization-delta-vector-list")
+
+    ret_arr: list[schemas.VectorRealizationData] = []
+    for vec in realization_delta_vector_list:
+        ret_arr.append(
+            schemas.VectorRealizationData(
+                realization=vec.realization,
+                timestamps_utc_ms=vec.timestamps_utc_ms,
+                values=vec.values,
+                unit=vec.unit,
+                is_rate=vec.is_rate,
+            )
+        )
+
+    LOGGER.info(f"Loaded realization delta ensemble summary data in: {perf_metrics.to_string()}")
+
+    return ret_arr
+
+
 @router.get("/timestamps_list/")
 async def get_timestamps_list(
     authenticated_user: Annotated[AuthenticatedUser, Depends(AuthHelper.get_authenticated_user)],
@@ -181,6 +325,93 @@ async def get_statistical_vector_data(
     return ret_data
 
 
+@router.get("/delta_ensemble_statistical_vector_data/")
+async def get_delta_ensemble_statistical_vector_data(
+    # fmt:off
+    response: Response,
+    authenticated_user: Annotated[AuthenticatedUser, Depends(AuthHelper.get_authenticated_user)],
+    compare_case_uuid: Annotated[str, Query(description="Sumo case uuid for compare ensemble")],
+    compare_ensemble_name: Annotated[str, Query(description="Compare ensemble name")],
+    reference_case_uuid: Annotated[str, Query(description="Sumo case uuid for reference ensemble")],
+    reference_ensemble_name: Annotated[str, Query(description="Reference ensemble name")],
+    vector_name: Annotated[str, Query(description="Name of the vector")],
+    resampling_frequency: Annotated[schemas.Frequency, Query(description="Resampling frequency")],
+    statistic_functions: Annotated[list[schemas.StatisticFunction] | None, Query(description="Optional list of statistics to calculate. If not specified, all statistics will be calculated.")] = None,
+    realizations: Annotated[list[int] | None, Query(description="Optional list of realizations to include. If not specified, all realizations will be included.")] = None,
+    # relative_to_timestamp: Annotated[datetime.datetime | None, Query(description="Calculate relative to timestamp")] = None,
+    # fmt:on
+) -> schemas.VectorStatisticData:
+    """Get statistical vector data for an ensemble
+
+    Definition:
+
+        delta_ensemble = compare_ensemble - reference_ensemble
+
+    """
+
+    perf_metrics = ResponsePerfMetrics(response)
+
+    service_freq = Frequency.from_string_value(resampling_frequency.value)
+    service_stat_funcs_to_compute = converters.to_service_statistic_functions(statistic_functions)
+
+    if service_freq is None:
+        raise HTTPException(
+            status_code=400, detail="Resampling frequency must be specified to create delta ensemble vector"
+        )
+
+    compare_ensemble_access = SummaryAccess.from_case_uuid(
+        authenticated_user.get_sumo_access_token(), compare_case_uuid, compare_ensemble_name
+    )
+    reference_ensemble_access = SummaryAccess.from_case_uuid(
+        authenticated_user.get_sumo_access_token(), reference_case_uuid, reference_ensemble_name
+    )
+
+    # Get tables parallel
+    # - Resampled data is assumed to be s.t. dates/timestamps are comparable between ensembles and cases, i.e. timestamps
+    #   for a resampling of a daily vector in both ensembles should be the same
+    (compare_vector_table_pa, compare_metadata), (reference_vector_table_pa, reference_metadata) = await asyncio.gather(
+        compare_ensemble_access.get_vector_table_async(
+            vector_name=vector_name,
+            resampling_frequency=service_freq,
+            realizations=realizations,
+        ),
+        reference_ensemble_access.get_vector_table_async(
+            vector_name=vector_name,
+            resampling_frequency=service_freq,
+            realizations=realizations,
+        ),
+    )
+
+    # Check for mismatching metadata
+    if compare_metadata.is_rate != reference_metadata.is_rate:
+        raise HTTPException(
+            status_code=400, detail="Rate mismatch between ensembles for delta ensemble statistical vector data"
+        )
+    if compare_metadata.unit != reference_metadata.unit:
+        raise HTTPException(
+            status_code=400, detail="Unit mismatch between ensembles for delta ensemble statistical vector data"
+        )
+
+    # Get metadata from reference ensemble
+    is_rate = reference_metadata.is_rate
+    unit = reference_metadata.unit
+
+    # Create delta ensemble data and compute statistics
+    delta_vector_table = create_delta_vector_table(compare_vector_table_pa, reference_vector_table_pa, vector_name)
+    statistics = compute_vector_statistics(delta_vector_table, vector_name, service_stat_funcs_to_compute)
+    if not statistics:
+        raise HTTPException(status_code=404, detail="Could not compute statistics")
+    perf_metrics.record_lap("calc-delta-vector-stat")
+
+    ret_data: schemas.VectorStatisticData = converters.to_api_delta_ensemble_vector_statistic_data(
+        statistics, is_rate, unit
+    )
+
+    LOGGER.info(f"Loaded and computed statistical delta ensemble summary data in: {perf_metrics.to_string()}")
+
+    return ret_data
+
+
 @router.get("/statistical_vector_data_per_sensitivity/")
 async def get_statistical_vector_data_per_sensitivity(
     # fmt:off
@@ -250,18 +481,3 @@ async def get_realization_vector_at_timestamp(
         vector_name=vector_name, timestamp_utc_ms=timestamp_utc_ms, realizations=None
     )
     return ensemble_response
-
-
-# @router.get("/realizations_calculated_vector_data/")
-# def get_realizations_calculated_vector_data(
-#     authenticated_user: Annotated[AuthenticatedUser, Depends(AuthHelper.get_authenticated_user)],
-#     case_uuid: Annotated[str, Query(description="Sumo case uuid")],
-#     ensemble_name: Annotated[str, Query(description="Ensemble name")],
-#     expression: Annotated[schemas.VectorExpressionInfo, Depends()],
-#     resampling_frequency: Annotated[schemas.Frequency, Query(description="Resampling frequency")],
-#     relative_to_timestamp: Annotated[datetime.datetime | None, Query(description="Calculate relative to timestamp")] = None,
-# ) -> str:
-#     """Get calculated vector data per realization"""
-#     print(expression)
-#     print(type(expression))
-#     return "hei"