friction_loss.outputs.send_gauge_data

View Source

 1from typing import Dict
 2
 3import numpy as np
 4from custom_plugin_base.server.calculation import SolveRequest
 5from custom_plugin_base.services.clients.well.gauge.gauge_graph_client import (GaugeGraphClient, GaugeGraphCreateData,
 6                                                                               GaugeGraphUpdateData, Plot2DData)
 7from custom_plugin_base.services.clients.well.gauge.gauge_pocket_client import GaugePocketClient, GaugePocketCreateData
 8from custom_plugin_base.utils.service_enum import TypeGraphicEnum, PocketTypeEnum
 9
10from friction_loss.utils.custom_plugin_utils import get_entities
11
12
13def send_gauge_data(request: SolveRequest, data: Dict[str, np.ndarray]) -> None:
14    """
15    Sending the recalculated pressure to the calc pocket
16
17    Parameters
18    ----------
19    request: SolveRequest
20        Input request
21    data: dict[str, np.ndarray]
22        'pressure': np.ndarray(np.ndarray),
23        'time': np.ndarray(np.ndarray)
24        'gauge message': np.ndarray(GaugeMessage),
25        'calendar time': np.array(datetime)}
26    """
27    gauge_pocket_client = GaugePocketClient()
28    gauge_graph_client = GaugeGraphClient()
29
30    for idx in range(len(data['pressure'])):
31        list_gauges_calc = [get_entities(request)[i] for i in data['gauge message'][idx].GaugePocketIds if
32                            get_entities(request)[i].PocketType == PocketTypeEnum.CALCULATION.get_int_value()]
33
34        if len(list_gauges_calc) == 0:
35            # Создаем карман calc, если его нет
36            gauge_pocket_data = [GaugePocketCreateData(gauge_id=data['gauge message'][idx].Id,
37                                                       pocket_type=PocketTypeEnum.CALCULATION)]
38            gauge_pocket = [x for x in gauge_pocket_client.create_gauge_pockets(gauge_pocket_data)][0]
39            gauge_graph_data = [GaugeGraphCreateData(
40                gauge_pocket_id=gauge_pocket.Id,
41                graph_type=TypeGraphicEnum.PRESSURE,
42                calendar_time=data['calendar_time'][idx],
43                plot=Plot2DData(x=data['time'][idx], values=data['pressure'][idx]))]
44            gauge_graph = list(gauge_graph_client.create_graphs(gauge_graph_data))
45        else:
46            # Обновляем карман calc, если он есть
47            gauge_pocket = list_gauges_calc[-1]
48            gauge_graph_id = gauge_pocket.GaugeGraphIds[0]
49            gauge_graph_data = GaugeGraphUpdateData(
50                gauge_graph_id=gauge_graph_id,
51                calendar_time=data['calendar_time'][idx],
52                plot=Plot2DData(x=data['time'][idx], values=data['pressure'][idx]))
53            gauge_graph_client.update_graph(gauge_graph_data)

def send_gauge_data( request: custom_plugin_base.server.calculation.SolveRequest, data: Dict[str, numpy.ndarray]) -> None: View Source

14def send_gauge_data(request: SolveRequest, data: Dict[str, np.ndarray]) -> None:
15    """
16    Sending the recalculated pressure to the calc pocket
17
18    Parameters
19    ----------
20    request: SolveRequest
21        Input request
22    data: dict[str, np.ndarray]
23        'pressure': np.ndarray(np.ndarray),
24        'time': np.ndarray(np.ndarray)
25        'gauge message': np.ndarray(GaugeMessage),
26        'calendar time': np.array(datetime)}
27    """
28    gauge_pocket_client = GaugePocketClient()
29    gauge_graph_client = GaugeGraphClient()
30
31    for idx in range(len(data['pressure'])):
32        list_gauges_calc = [get_entities(request)[i] for i in data['gauge message'][idx].GaugePocketIds if
33                            get_entities(request)[i].PocketType == PocketTypeEnum.CALCULATION.get_int_value()]
34
35        if len(list_gauges_calc) == 0:
36            # Создаем карман calc, если его нет
37            gauge_pocket_data = [GaugePocketCreateData(gauge_id=data['gauge message'][idx].Id,
38                                                       pocket_type=PocketTypeEnum.CALCULATION)]
39            gauge_pocket = [x for x in gauge_pocket_client.create_gauge_pockets(gauge_pocket_data)][0]
40            gauge_graph_data = [GaugeGraphCreateData(
41                gauge_pocket_id=gauge_pocket.Id,
42                graph_type=TypeGraphicEnum.PRESSURE,
43                calendar_time=data['calendar_time'][idx],
44                plot=Plot2DData(x=data['time'][idx], values=data['pressure'][idx]))]
45            gauge_graph = list(gauge_graph_client.create_graphs(gauge_graph_data))
46        else:
47            # Обновляем карман calc, если он есть
48            gauge_pocket = list_gauges_calc[-1]
49            gauge_graph_id = gauge_pocket.GaugeGraphIds[0]
50            gauge_graph_data = GaugeGraphUpdateData(
51                gauge_graph_id=gauge_graph_id,
52                calendar_time=data['calendar_time'][idx],
53                plot=Plot2DData(x=data['time'][idx], values=data['pressure'][idx]))
54            gauge_graph_client.update_graph(gauge_graph_data)

Sending the recalculated pressure to the calc pocket

Parameters

request: SolveRequest Input request data: dict[str, np.ndarray] 'pressure': np.ndarray(np.ndarray), 'time': np.ndarray(np.ndarray) 'gauge message': np.ndarray(GaugeMessage), 'calendar time': np.array(datetime)}