def read_data_from_log_file()

in src/loading_manifest/osdu_laslog_manifest.py [0:0]

• 93 lines of code
• 25 McCabe index (conditional complexity)

def read_data_from_log_file(log_file):

    def item_strvalue(section, field_name, check_float=False):
        try:
            val = str(section[field_name].value).strip()
            if check_float:
                try:
                    float(val)
                except ValueError:
                    logging.warning("Non-number depth: " + val + "  in log file: " + log_file)
                    val = None
            return val
        except Exception:
            return None

    def item_unit(section, field_name):
        try:
            return str(section[field_name].unit).strip()
        except Exception:
            return None

    with open(log_file, "r") as fp:
        las = lasio.read(fp)

        # version
        las_version = item_strvalue(las.version, 'VERS')

        # well section
        # osdu log_uniqid = item_strvalue(las.well, 'UWBI')
        log_uniqid = item_strvalue(las.params, 'UBID')
        if log_uniqid is None or len(log_uniqid) == 0:
            log_uniqid = item_strvalue(las.well, 'UWI')
        if log_uniqid is None or len(log_uniqid) == 0:
            log_uniqid = item_strvalue(las.well, 'WELL')
        well_name = item_strvalue(las.well, 'WELL')
        logging_contractor = item_strvalue(las.well, 'SRVC')
        date_logged = item_strvalue(las.well, 'DATE')
        log_depth_units = item_unit(las.well, 'STRT')
        log_start_depth = item_strvalue(las.well, 'STRT', True)
        log_stop_depth = item_strvalue(las.well, 'STOP', True)
        log_depth_step = item_strvalue(las.well, 'STEP', True)
        log_null_value = item_strvalue(las.well, 'NULL', True)

        # parameters section
        log_name = item_strvalue(las.params, 'LNAM')
        log_type = item_strvalue(las.params, 'LTYP')
        log_source = item_strvalue(las.params, 'LSOU')
        log_activity = item_strvalue(las.params, 'LACT')
        log_version = item_strvalue(las.params, 'LVSN')

        # top/bottom depths
        log_top_depth = log_start_depth
        log_bottom_depth = log_stop_depth
        if (log_start_depth is not None and log_stop_depth is not None
                and float(log_start_depth) > float(log_stop_depth)):
            log_top_depth = log_stop_depth
            log_bottom_depth = log_start_depth

        # curve section
        log_curves = []
        # 'mnemonic' = curves[].mnemonic
        # 'curve_type' =  seems not in the las file ????
        # 'curve_unit' =  curves[].unit
        # 'curve_top_depth' =  remove top/bottom null values
        # 'curve_bottom_depth' =  remove top/bottom null values
        if len(las.curves) > 0:  # > 1: for osdu, not to skip the first channel
            # skip first channel
            # for osdu, not to skip the first channel
            for curveitem in las.curves:
                log_curve = dict()
                log_curves.append(log_curve)
                log_curve['mnemonic'] = curveitem.mnemonic
                log_curve['unit'] = curveitem.unit
                log_curve['depth_units'] = log_depth_units
                # derive top/bottom depths for each curve -- stripping off the start/end null values
                curve_start_depth = None
                curve_stop_depth = None
                for i in range(len(curveitem.data)):
                    if not math.isnan(curveitem.data[i]):
                        curve_start_depth = las.curves[0].data[i]
                        break
                for i in reversed(range(len(curveitem.data))):
                    if not math.isnan(curveitem.data[i]):
                        curve_stop_depth = las.curves[0].data[i]
                        break

                if curve_start_depth is None:
                    log_curve['top_depth'] = None
                    log_curve['bottom_depth'] = None
                else:
                    log_curve['top_depth'] = '{:f}'.format(curve_start_depth).rstrip('0').rstrip('.')
                    log_curve['bottom_depth'] = '{:f}'.format(curve_stop_depth).rstrip('0').rstrip('.')
                if (curve_start_depth is not None and curve_stop_depth is not None
                        and float(curve_start_depth) > float(curve_stop_depth)):
                    log_curve['top_depth'] = '{:f}'.format(curve_stop_depth).rstrip('0').rstrip('.')
                    log_curve['bottom_depth'] = '{:f}'.format(curve_start_depth).rstrip('0').rstrip('.')

    # return as a dict
    return {
        'las_version': las_version,
        'log_uniqid': log_uniqid,
        'well_name': well_name,
        'log_name': log_name,
        'log_type': log_type,
        'log_source': log_source,
        'log_activity': log_activity,
        'log_version': log_version,
        'logging_contractor': logging_contractor,
        'date_logged': date_logged,
        'log_depth_units': log_depth_units,
        'log_start_depth': log_start_depth,
        'log_stop_depth': log_stop_depth,
        'log_depth_step': log_depth_step,
        'log_top_depth': log_top_depth,
        'log_bottom_depth': log_bottom_depth,
        'log_null_value': log_null_value,
        'log_curves': log_curves
    }