friction_loss.inputs.read_excel_data

  1import copy
  2from io import BytesIO
  3from itertools import islice
  4from typing import Optional, Dict, Union, List
  5
  6import numpy as np
  7import pandas as pd
  8from custom_plugin_base.services.clients.unisum_client import UnisumClient
  9from custom_plugin_base.utils.service_enum import LangEnum
 10from polypres.depths_keeper import DepthSortedWellPointsKeeper
 11from polypres.form_vlp_inputs import form_polypres_core_inputs
 12from polypres.pump import Pump as PolypresPump
 13from polypres.rate_internal import NumbaRate, rates_units
 14
 15from friction_loss.utils.const import PARAMETERS_NAMES, MM_TO_INCHES
 16from friction_loss.utils.errors import (get_message_empty_column, get_message_incorrect_table_cell,
 17                                        get_message_negative_value, get_message_incorrect_md_tvdss,
 18                                        get_message_incorrect_template, get_message_incorrect_coefs,
 19                                        get_message_incorrect_pump_parameter, get_message_incorrect_pump_name,
 20                                        get_message_incorrect_ref_gauge, get_message_incorrect_depth_gauge)
 21from friction_loss.utils.plugin_dataclasses import PolypresInputParameter, Pump
 22from friction_loss.utils.pumps import check_pump
 23from friction_loss.utils.set_language import (PARAMETERS_NAMES_TRANSLATOR, CONVERT_UNITS, NAMES_COLUMNS_TRANSLATOR,
 24                                              COLUMNS_PUMP_DATABASE_TRANSLATOR, UNIT_SYSTEMS)
 25
 26
 27def read_excel_data(data: bytes) -> (np.ndarray, np.ndarray, np.ndarray, bool, Dict):
 28    """
 29    Function reads the data and brings it to the required format.
 30    Detailed description parameters of input data: http://mtm.polykod.ru/pages/viewpage.action?pageId=89623327
 31
 32    Parameters
 33    ----------
 34    data : bytes
 35        File with input parameters in the form of bytes.
 36
 37    Returns
 38    -------
 39    input_data : ndarray(DepthSortedWellPointsKeeper)
 40             Structure of the elements of the Polypres library well construction.
 41    table_initial_unit : ndarray(dict)
 42        keys : parameter names
 43        values : dataclass with rows of table
 44    md_tvdss_coords: np.ndarray(dict)
 45        Array with MD and TVDss coordinates
 46    report_flag : bool
 47        Flag of generation web-report.
 48    depths : dict[str, float]
 49        Depths at which the pressure value is set and at which the pressure must be calculated
 50    """
 51
 52    # Converting bytes to text.
 53    sheets = pd.read_excel(BytesIO(data), engine='openpyxl', sheet_name=None)
 54
 55    # Determining the language of an Excel file.
 56    if NAMES_COLUMNS_TRANSLATOR[LangEnum.RU]['pump'] in sheets and \
 57            NAMES_COLUMNS_TRANSLATOR[LangEnum.RU]['params sheet'] in sheets:
 58        language = LangEnum.RU
 59    elif NAMES_COLUMNS_TRANSLATOR[LangEnum.US]['pump'] in sheets and\
 60            NAMES_COLUMNS_TRANSLATOR[LangEnum.US]['params sheet'] in sheets:
 61        language = LangEnum.US
 62    else:
 63        raise ValueError(get_message_incorrect_template())
 64
 65    # TODO перейти на openpyxl, тк pandas импортится только для одной функции
 66
 67    # Reading pump
 68    pumps_database = pd.read_excel(BytesIO(data), engine='openpyxl',
 69                                   sheet_name=NAMES_COLUMNS_TRANSLATOR[language]['pump'])
 70
 71    col_names = NAMES_COLUMNS_TRANSLATOR[language]
 72    params_pd = sheets[col_names['params sheet']]
 73    params_dict = dict(params_pd.to_dict(orient='split')['data'])
 74
 75    if params_dict[col_names['report']] in ['No', 'Нет']:
 76        report_flag = False
 77    elif params_dict[col_names['report']] in ['Yes', 'Да']:
 78        report_flag = True
 79    else:
 80        raise ValueError(get_message_incorrect_template())
 81
 82    initial_unit = {'len': params_dict[col_names['units len']],
 83                    'diam': params_dict[col_names['units diam']]}
 84
 85    units = _convert_units(initial_unit)
 86
 87    initial_depths = {'gauge': params_dict[col_names['start point']],
 88                      'ref': params_dict[col_names['ref point']]}
 89
 90    depths = _convert_depths(depths=initial_depths, units=units)
 91
 92    del sheets[col_names['params sheet']]
 93    del sheets[col_names['pump']]
 94    input_data = np.empty(len(sheets), object)
 95    table_initial_unit = np.empty(len(sheets) + 1, object)
 96    md_tvdss_coords = np.empty(len(sheets), object)
 97
 98    # Converting a dataframe to a dictionary with parameters and translating the table elements.
 99    for idx, well in enumerate(sheets):
100        input_data[idx], table_initial_unit[idx], md_tvdss_coords[idx] = _read_well_worksheet(sheets[well], language,
101                                                                                              pumps_database, depths,
102                                                                                              units, well)
103
104    table_initial_unit[-1] = initial_depths
105
106    return input_data, table_initial_unit, md_tvdss_coords, report_flag, depths
107
108
109def _convert_units(data: Dict[str, str]):
110    """
111    The function converts the units of measurement read from Excel to the required language
112
113    Parameters
114    ----------
115    data: dict[str, str]
116        Units of measurement read from Excel file
117
118    Returns
119    -------
120    ret: dict[str, str]
121        Units of measurement reduced to the required form
122    """
123    ret = {}
124    # If the UI in the Excel file is specified in Russian, then we translate them into English
125    for key, val in data.items():
126        if val in CONVERT_UNITS.values():
127            ret[key] = [k for k, value in CONVERT_UNITS.items() if value == val][0]
128        else:
129            ret[key] = val
130
131    return ret
132
133
134def _convert_depths(depths: Dict[str, float], units: Dict[str, str]) -> Dict[str, float]:
135    """
136
137    Parameters
138    ----------
139    depths: Dict[str, float]
140        Dictionary containing reference depth and gauge depth
141    units: Dict[str, str]
142        Dictionary containing units of length and diameter
143
144    Returns
145    -------
146    ret: Dict[str, float]
147        Dictionary containing depths converted to meters
148    """
149    ret = {}
150    unisum_convert = UnisumClient()
151
152    # If the sensor depth is not specified, then we assume that it is located at the mouth (MD coordinate = 0)
153    if np.isnan(depths['gauge']):
154        depths['gauge'] = 0
155
156    # Converting specified depth points to meters
157    for name, val in depths.items():
158        ret[name] = unisum_convert.convert_one(units['len'], 'm', depths[name])
159
160    return ret
161
162
163def _read_well_worksheet(input_df: pd.DataFrame, language: LangEnum, pumps_database: pd.DataFrame,
164                         depths: Dict[str, float], units: Dict[str, str], well_name: str):
165    """
166    Reading well design data from Excel pages
167
168    Parameters
169    ----------
170    input_df: pd.DataFrame
171        Excel file page with well construction
172    language: LangEnum
173        Language of Excel file
174    pumps_database: pd.DataFrame
175        Excel file page with pump database
176    depths: Dict[str, float]
177        Dictionary containing reference depth and gauge depth
178    units: dict:
179        Units of length and diameter
180    well_name: str
181        Name of well
182
183    Returns
184    -------
185    input_data: DepthSortedWellPointsKeeper
186         Structure of the elements of the Polypres library well construction.
187    table_initial_unit: dict[str, PolypresInputParameter]
188        keys : parameter names.
189        values : dataclass with rows of table.
190    trajectory: dict[str, np.ndarray]
191        Dictionary containing MD and TVDss coordinates
192    """
193    parameters_dict, md, tvdss = _get_dict_from_df_well(input_df=input_df, language=language, units=units)
194
195    assert not (isinstance(parameters_dict['pump'].value, float) and np.isnan(parameters_dict['pump'].value)), (
196        get_message_incorrect_pump_name())
197
198    selected_pump = None
199    if parameters_dict['pump'].value == NAMES_COLUMNS_TRANSLATOR[language]['no_pump']:
200        parameters_dict['pump'].value = None
201    else:
202        pump_dict, unit = _read_pump_from_excel(input_df=pumps_database, name=parameters_dict['pump'].value,
203                                                language=language)
204        selected_pump = _pump_params_unit_si(pump=pump_dict, unit=unit, language=language)
205
206    # Checking the correctness of data entry.
207    validated_parameters_dict, md, tvdss = _validation_well_input_data(parameters_dict=parameters_dict,
208                                                                       md=md, tvdss=tvdss, language=language)
209    table_initial_unit = copy.deepcopy(validated_parameters_dict)
210
211    # Convertation to si units.
212    data_unit_si, md, tvdss,  = _well_input_data_to_unit_si(parameters_dict=validated_parameters_dict,
213                                                            md=md, tvdss=tvdss, units=units)
214
215    # Checking the physicality of data.
216    md, tvdss = _physical_check_well_input_data(parameters_dict=data_unit_si, md=md, tvdss=tvdss,
217                                                name=well_name, depths=depths)
218
219    # Converting data to the input format of the Polypres library.
220    input_data = _polypres_input_format_data(param=data_unit_si, selected_pump=selected_pump,
221                                             depth=depths['gauge'], md=md, tvdss=tvdss)
222
223    trajectory = {'md': md, 'tvdss': tvdss}
224
225    return input_data, table_initial_unit, trajectory
226
227
228def _get_dict_from_df_well(input_df: pd.DataFrame, language: LangEnum, units: Dict[str, str]) -> \
229        (Dict[str, PolypresInputParameter], np.ndarray, np.ndarray):
230    """
231    Function creates dict from DataFrame
232
233    Parameters
234    ----------
235    input_df : pd.DataFrame
236        Page of Excel table with input data.
237    language: LangEnum
238        Language of Excel file
239    units: dict:
240        Units of length and diameter
241    Returns
242    -------
243    parameters_dict: dict[str, PolypresInputParameter]
244                        keys : parameter names
245                        values : dataclass with rows of table
246    md: np.ndarray
247        MD координаты скважины
248    tvdss: np.ndarray
249        TVDss координаты скважины
250    """
251    names_columns = NAMES_COLUMNS_TRANSLATOR[language]
252    input_dict = input_df.to_dict(orient='records')
253
254    md = np.array([i['MD'] for i in input_dict])
255    tvdss = np.array([i['TVDss'] for i in input_dict])
256
257    parameters_dict = {}
258    # For each parameter we will create a dataclass.
259    for obj in input_dict:
260        if isinstance(obj[names_columns['param']], str):
261            # Translating table columns
262            parameter = [key for key, value in PARAMETERS_NAMES_TRANSLATOR[language].items() if
263                         value == obj[names_columns['param']]][0]
264
265            if parameter in ['d', 'd_choke', 'd_tbg', 'd_csg', 'e']:
266                unit = units['diam']
267            else:
268                unit = units['len']
269
270            parameters_dict[parameter] = PolypresInputParameter(name=parameter,
271                                                                value=obj[names_columns['val']],
272                                                                unit=unit)
273    return parameters_dict, md, tvdss
274
275
276def _validation_well_input_data(parameters_dict: Dict[str, PolypresInputParameter], md: np.ndarray, tvdss: np.ndarray,
277                                language: LangEnum) -> (Dict[str, PolypresInputParameter], np.ndarray, np.ndarray):
278    """
279    Function of checking the correctness of reading / filling in the table elements.
280
281    Parameters
282    ----------
283    parameters_dict : dict[str, PolypresInputParameter]
284        keys: parameter names.
285        values: dataclass with rows of table.
286    md: np.ndarray
287        MD array of well coordinates
288    tvdss: np.ndarray
289        TVDss array of well coordinates
290    language: LangEnum
291        Language of Excel file
292
293    Returns
294    -------
295    dict[str, PolypresInputParameter]
296        keys : parameter names.
297        values : dataclass with rows of table.
298    md: np.ndarray
299        MD array of well coordinates
300    tvdss: np.ndarray
301        TVDss array of well coordinates
302    """
303
304    # Run through the entire table.
305    for param in parameters_dict.values():
306        if param.name in PARAMETERS_NAMES and param.name not in ['pump', 'report']:
307            # Processing of the remaining values. if there is a comma separator in the table column, then the
308            # read_excel function immediately converts the value to the float data type when reading, and if
309            if isinstance(param.value, str):
310                number = param.value.replace('.', '').replace(',', '')
311                assert number.isnumeric(), \
312                    get_message_incorrect_table_cell(PARAMETERS_NAMES_TRANSLATOR[language][param.name])
313                param.value = float(param.value.replace(',', '.'))
314            else:
315                assert ~np.isnan(param.value), \
316                    get_message_empty_column(PARAMETERS_NAMES_TRANSLATOR[language][param.name])
317            # Checking values for negativity.
318            if param.name not in ['alt_manifold', 'alt_wellhead']:
319                assert float(param.value) >= 0.0, \
320                    get_message_negative_value(PARAMETERS_NAMES_TRANSLATOR[language][param.name])
321
322    md = np.array([i.replace('.', '').replace(',', '') if isinstance(i, str) else i for i in md])
323    tvdss = np.array([i.replace('.', '').replace(',', '') if isinstance(i, str) else i for i in tvdss])
324
325    return parameters_dict, md, tvdss
326
327
328def _well_input_data_to_unit_si(parameters_dict: Dict[str, PolypresInputParameter], md: np.ndarray,
329                                tvdss: np.ndarray, units: Dict[str, str]) -> (Dict[str, PolypresInputParameter],
330                                                                              np.ndarray, np.ndarray):
331    """
332    Function to bring units of measurement to the si system
333
334    Parameters
335    ----------
336    parameters_dict : dict[str, PolypresInputParameter]
337        keys: parameter names.
338        values: dataclass with rows of table.
339    md: np.ndarray
340        MD array of well coordinates
341    tvdss: np.ndarray
342        TVDss array of well coordinates
343    units: dict:
344        Units of length and diameter
345
346    Returns
347    -------
348    dict[str, PolypresInputParameter]
349        keys : parameter names.
350        values : dataclass with rows of table.
351    md: np.ndarray
352        MD array of well coordinates
353    tvdss: np.ndarray
354        TVDss array of well coordinates
355    """
356
357    unisum_convert = UnisumClient()
358
359    # Run through the entire table.
360    for param in parameters_dict.values():
361
362        # Convertation lenghts.
363        if param.name not in ['angle', 'pump', 'report']:
364            if param.unit == 'mm':
365                param.value *= 0.001  # converting millimeters -> meters
366            elif param.unit == 'in':
367                param.value *= 0.0254  # converting inches -> meters
368            else:
369                param.value = unisum_convert.convert_one(param.unit, 'm', param.value)
370            continue
371
372    md = unisum_convert.convert_many_to_numpy_array(units['len'], 'm', md)
373    tvdss = unisum_convert.convert_many_to_numpy_array(units['len'], 'm', tvdss)
374
375    return parameters_dict, md, tvdss
376
377
378def _physical_check_well_input_data(parameters_dict: Dict[str, PolypresInputParameter], md: np.ndarray,
379                                    tvdss: np.ndarray, name: str, depths: Dict[str, float]) -> (np.ndarray, np.ndarray):
380    """
381    Function checks the physicality of the input data.
382
383    Parameters
384    ----------
385    parameters_dict : dict[str, PolypresInputParameter]
386        keys: parameter names.
387        values: dataclass with rows of table.
388    md: np.ndarray
389        MD array of well coordinates
390    tvdss: np.ndarray
391        TVDss array of well coordinates
392    name: str
393        Name of well
394    depths: dict
395        Depths at which the pressure value is set and at which the pressure must be calculated
396
397    Returns
398    -------
399    md: np.ndarray
400        MD array of well coordinates
401    tvdss: np.ndarray
402        TVDss array of well coordinates
403
404    """
405    if any(np.isnan(md)) or any(np.isnan(tvdss)):
406        md = None
407        tvdss = None
408
409        if ~np.isnan(depths['gauge']):
410            assert (depths['gauge'] > 0 or depths['gauge'] < parameters_dict['L_csg'].value),\
411                get_message_incorrect_depth_gauge(name)
412
413        if ~np.isnan(depths['ref']):
414            assert (depths['ref'] > 0 or depths['ref'] < parameters_dict['L_csg'].value),\
415                get_message_incorrect_ref_gauge(name)
416    else:
417        if ~np.isnan(depths['gauge']):
418            assert (depths['gauge'] > md[0] or depths['gauge'] < md[-1]), \
419                get_message_incorrect_depth_gauge(name)
420
421        if ~np.isnan(depths['ref']):
422            assert (depths['ref'] > md[0] or depths['ref'] < md[-1]), \
423                get_message_incorrect_ref_gauge(name)
424
425        assert len(md) == len(tvdss), get_message_incorrect_md_tvdss(name)
426
427    return md, tvdss
428
429
430def _polypres_input_format_data(param: Dict[str, PolypresInputParameter], selected_pump: Optional[Pump], depth: float,
431                                md: Optional[np.ndarray], tvdss: Optional[np.ndarray]) -> DepthSortedWellPointsKeeper:
432    """
433    The function from the input data elements creates a dictionary necessary for Polypres.
434    Using the form_data_frame function, a structure of well elements is created in a special format.
435    Detailed description parameters: http://mtm.polykod.ru/pages/viewpage.action?pageId=89623327
436
437    Parameters
438    ----------
439    param : dict[str, PolypresInputParameter]
440        keys: parameter names.
441        values: dataclass with rows of table.
442    selected_pump : Pump
443        Selected pump.
444    depth : float
445        Depth at which pressure needs to be calculated
446    md: np.ndarray
447        MD array of well coordinates
448    tvdss: np.ndarray
449        TVDss array of well coordinates
450
451    Returns
452    -------
453    polypres_inputs : DepthSortedWellPointsKeeper
454         Structure of the elements of the Polypres library well construction.
455    """
456
457    gzu_coords = {'K1': [0, 0, -param['alt_manifold'].value]}
458
459    if selected_pump is not None:
460        selected_pump = PolypresPump(check_pump(name=param['pump'].value, selected_pump=selected_pump))
461
462    well = {
463        'name': 'W1',
464        'md': md,
465        'tvdss': tvdss,
466        'head': [0, 0, param['alt_wellhead'].value],
467        'hill': [0, 0, param['L_tbg'].value],
468        'toe': [0, 0, param['L_csg'].value],
469        'trajectory': [],
470        'cluster_id': 'K1',
471        'tubing_depth': param['L_tbg'].value,
472        'casing': [param['alt_wellhead'].value, param['L_csg'].value],
473        'tubing_diameter': param['d_tbg'].value,
474        'casing_diameter': param['d_csg'].value,
475        'choke': param['d_choke'].value,
476        'pump': selected_pump
477    }
478
479    d_flowline = param['d'].value
480    tvdss_ipr = param['d_csg'].value
481    rates_all = np.array([
482        NumbaRate(q_O, q_G, q_W, rates_units['metric'])
483        for q_O, q_G, q_W in zip([0], [0], [0])
484    ])
485    polypres_inputs = form_polypres_core_inputs(well, d_flowline, rates_all, 0.0,
486                                                gzu_coords, param['e'].value, tvdss_ipr, depth)
487
488    return polypres_inputs
489
490
491def _read_pump_from_excel(input_df: pd.DataFrame, name: str, language: LangEnum) -> (Dict[str, Union[float, List]], str):
492    """
493    Function for reading and processing Excel table page with pump database.
494
495    Parameters
496    ----------
497    input_df: pd.DataFrame
498        Page of Excel table with pump database.
499    name: str
500        Name of pump.
501    language: LangEnum
502        Language of Excel file.
503
504    Returns
505    -------
506    custom_pump: dict[str, float | list]
507        keys: Pump parameter names.
508        values: Pump parameter values.
509    unit: str
510        Units of measurement in which the pump database is presented
511    """
512    input_dict = input_df.to_dict(orient='records')
513    col_names = COLUMNS_PUMP_DATABASE_TRANSLATOR[language]
514
515    # Поиск насоса в таблице по названию
516    selected_pump = next((p for p in input_dict if p[col_names['name']] == name), None)
517
518    # Определение единиц измерения таблицы ( ['bpd', 'psi', 'in'] на обоих языках одинаково записываются)
519    if {'bpd', 'psi', 'in'} <= set([item.split(', ')[-1] for item in selected_pump.keys()]):
520        unit = 'Field'
521    else:
522        unit = 'Metric'
523
524    # Валидация входных данных
525    for param, value in islice(selected_pump.items(), 1, None):
526        name = param.split(',')[0]
527        if isinstance(value, str):
528            if param in [col_names['pump_koeffs'], col_names['hp_koeffs']]:
529                val = [float(i.split('^')[0]) ** float(i.split('^')[1]) if '^' in i else float(i) for i in value.strip('[]').split(',')]
530            else:
531                assert value.replace('.', '').replace(',', '').isnumeric(),\
532                    get_message_incorrect_pump_parameter(name)
533                val = float(value.replace(',', '.'))
534
535            selected_pump[param] = val
536        else:
537            if name not in [col_names['k_f'], col_names['pump_koeffs'], col_names['hp_koeffs']]:
538                assert ~np.isnan(value), get_message_incorrect_pump_parameter(name)
539
540    assert ~np.all([np.isnan(selected_pump[col_names['k_f']]), np.all(np.isnan(selected_pump[col_names['pump_koeffs']])),
541                   np.all(np.isnan(selected_pump[col_names['hp_koeffs']]))]), get_message_incorrect_coefs()
542
543    return selected_pump, unit
544
545
546def _pump_params_unit_si(pump: Dict[str, Union[float, List]], unit: str, language: LangEnum) -> Pump:
547    """
548    Function for converting pump parameters into system SI.
549
550    Parameters
551    ----------
552    pump: dict[str, float | list]
553        keys: Pump parameter names.
554        values: Pump parameter values.
555    unit: str
556        Units of measurement in which the pump database is presented
557
558    Returns
559    -------
560    custom_pump: Pump
561        Dataclass of pump
562    """
563    col_names = COLUMNS_PUMP_DATABASE_TRANSLATOR[language]
564    unisum_convert = UnisumClient()
565
566    q_nominal = pump[col_names['q_nominal']+UNIT_SYSTEMS[language][unit]['Flow rate']]
567    q_nominal = unisum_convert.convert_one(UNIT_SYSTEMS[LangEnum.US][unit]['Flow rate'], 'm^3/day', q_nominal)
568
569    dp_nom = pump[col_names['head_nominal']+UNIT_SYSTEMS[language][unit]['Pressure']]
570    dp_nom = unisum_convert.convert_one(UNIT_SYSTEMS[LangEnum.US][unit]['Pressure'], 'Pa', dp_nom)
571
572    if unit == 'Field':
573        pump_size = pump[col_names['pumpsize']+UNIT_SYSTEMS[language][unit]['Diameter']] * MM_TO_INCHES  # in -> mm
574    else:
575        pump_size = pump[col_names['pumpsize']+UNIT_SYSTEMS[language][unit]['Diameter']]
576
577    custom_pump = Pump(q_nominal=float(q_nominal),
578                       dp_nominal=float(dp_nom),
579                       f_nominal=float(pump[col_names['f_nominal']]),
580                       f_actual=float(pump[col_names['f_nominal']]),
581                       pumpsize=float(pump_size),
582                       k_f=float(pump[col_names['k_f']]),
583                       stages_count=float(pump[col_names['stages_count']]),
584                       custom_stages_count=int(pump[col_names['stages_count']]),
585                       pump_koeffs=pump[col_names['pump_koeffs']],
586                       hp_koeffs=pump[col_names['hp_koeffs']])
587    return custom_pump