get_formatter_function() - Code Metrics - kinverarity1/lasio - Measure and Improve Code Quality continuously with Scrutinizer

get_formatter_function() C
last analyzed 2018-07-04 10:31 UTC

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	9
dl	0
loc	40
rs	6.5866
c	0
b	0
f	0

import logging
import textwrap

import numpy as np

from .las_items import (
    HeaderItem, CurveItem, SectionItems, OrderedDict)
from . import defaults
from . import exceptions

logger = logging.getLogger(__name__)


def write(las, file_object, version=None, wrap=None, STRT=None,
          STOP=None, STEP=None, fmt='%.5f'):
    '''Write a LAS files.

    Arguments:
        las (:class:`lasio.las.LASFile`)
        file_object (file-like object open for writing): output

    Keyword Arguments:
        version (float or None): version of written file, either 1.2 or 2. 
            If this is None, ``las.version.VERS.value`` will be used.
        wrap (bool or None): whether to wrap the output data section.
            If this is None, ``las.version.WRAP.value`` will be used.
        STRT (float or None): value to use as STRT (note the data will not
            be clipped). If this is None, the data value in the first column,
            first row will be used.
        STOP (float or None): value to use as STOP (note the data will not
            be clipped). If this is None, the data value in the first column,
            last row will be used.
        STEP (float or None): value to use as STEP (note the data will not
            be resampled and/or interpolated). If this is None, the STEP will
            be estimated from the first two rows of the first column.
        fmt (str): Python string formatting operator for numeric data to be
            used.

    You should avoid calling this function directly - instead use the 
    :meth:`lasio.las.LASFile.write` method.

    '''
    if wrap is None:
        wrap = las.version['WRAP'] == 'YES'
    elif wrap is True:
        las.version['WRAP'] = HeaderItem(
            'WRAP', '', 'YES', 'Multiple lines per depth step')
    elif wrap is False:
        las.version['WRAP'] = HeaderItem(
            'WRAP', '', 'NO', 'One line per depth step')
    lines = []

    assert version in (1.2, 2, None)
    if version is None:
        version = las.version['VERS'].value
    if version == 1.2:
        las.version['VERS'] = HeaderItem(
            'VERS', '', 1.2, 'CWLS LOG ASCII STANDARD - VERSION 1.2')
    elif version == 2:
        las.version['VERS'] = HeaderItem(
            'VERS', '', 2.0, 'CWLS log ASCII Standard -VERSION 2.0')

    if STRT is None:
        STRT = las.index[0]
    if STOP is None:
        STOP = las.index[-1]
    if STEP is None:
        STEP = las.index[1] - las.index[0]  # Faster than np.gradient
    las.well['STRT'].value = STRT
    las.well['STOP'].value = STOP
    las.well['STEP'].value = STEP

    # Check units
    if las.curves[0].unit:
        unit = las.curves[0].unit
    else:
        unit = las.well['STRT'].unit
    las.well['STRT'].unit = unit
    las.well['STOP'].unit = unit
    las.well['STEP'].unit = unit
    las.curves[0].unit = unit

    # Check for any changes in the pandas dataframe and if there are,
    # create new curves so they are reflected in the output LAS file.

    # if las.use_pandas:
    #     curve_names = lambda: [ci.mnemonic for ci in las.curves]
    #     for df_curve_name in list(las.df.columns.values):
    #         if not df_curve_name in curve_names():
    #             las.add_curve(df_curve_name, las.df[df_curve_name])

    # Write each section.

    # ~Version
    logger.debug('LASFile.write Version section')
    lines.append('~Version '.ljust(60, '-'))
    order_func = get_section_order_function('Version', version)
    section_widths = get_section_widths(
        'Version', las.version, version, order_func)
    for header_item in las.version.values():
        mnemonic = header_item.original_mnemonic
        # logger.debug('LASFile.write ' + str(header_item))
        order = order_func(mnemonic)
        # logger.debug('LASFile.write order = %s' % (order, ))
        logger.debug('LASFile.write %s order=%s section_widths=%s' % (
                        header_item, order, section_widths))
        formatter_func = get_formatter_function(order, **section_widths)
        line = formatter_func(header_item)
        lines.append(line)

    # ~Well
    logger.debug('LASFile.write Well section')
    lines.append('~Well '.ljust(60, '-'))
    order_func = get_section_order_function('Well', version)
    section_widths = get_section_widths(
        'Well', las.well, version, order_func)
    # logger.debug('LASFile.write well section_widths=%s' % section_widths)
    for header_item in las.well.values():
        mnemonic = header_item.original_mnemonic
        order = order_func(mnemonic)
        logger.debug('LASFile.write %s order=%s section_widths=%s' % (
            header_item, order, section_widths))
        formatter_func = get_formatter_function(order, **section_widths)
        line = formatter_func(header_item)
        lines.append(line)

    # ~Curves
    logger.debug('LASFile.write Curves section')
    lines.append('~Curve Information '.ljust(60, '-'))
    order_func = get_section_order_function('Curves', version)
    section_widths = get_section_widths(
        'Curves', las.curves, version, order_func)
    for header_item in las.curves:
        mnemonic = header_item.original_mnemonic
        order = order_func(mnemonic)
        formatter_func = get_formatter_function(order, **section_widths)
        line = formatter_func(header_item)
        lines.append(line)

    # ~Params
    lines.append('~Params '.ljust(60, '-'))
    order_func = get_section_order_function('Parameter', version)
    section_widths = get_section_widths(
        'Parameter', las.params, version, order_func)
    for header_item in las.params.values():
        mnemonic = header_item.original_mnemonic
        order = order_func(mnemonic)
        formatter_func = get_formatter_function(order, **section_widths)
        line = formatter_func(header_item)
        lines.append(line)

    # ~Other
    lines.append('~Other '.ljust(60, '-'))
    lines += las.other.splitlines()

    lines.append('~ASCII '.ljust(60, '-'))

    file_object.write('\n'.join(lines))
    file_object.write('\n')
    line_counter = len(lines)

    # data_arr = np.column_stack([c.data for c in las.curves])
    data_arr = las.data
    nrows, ncols = data_arr.shape

    def format_data_section_line(n, fmt, l=10, spacer=' '):
        try:
            if np.isnan(n):
                return spacer + str(las.well['NULL'].value).rjust(l)
            else:
                return spacer + (fmt % n).rjust(l)
        except TypeError:
            return spacer + str(n).rjust(l)

    twrapper = textwrap.TextWrapper(width=79)

    for i in range(nrows):
        depth_slice = ''
        for j in range(ncols):
            depth_slice += format_data_section_line(data_arr[i, j], fmt)

        if wrap:
            lines = twrapper.wrap(depth_slice)
            logger.debug('LASFile.write Wrapped %d lines out of %s' %
                         (len(lines), depth_slice))
        else:
            lines = [depth_slice]
                    
        for line in lines:
            if las.version['VERS'].value == 1.2 and len(line) > 255:
                logger.warning('[v1.2] line #{} has {} chars (>256)'.format(
                    line_counter + 1, len(line)))
            file_object.write(line + '\n')
            line_counter += 1


def get_formatter_function(order, left_width=None, middle_width=None):
    '''Create function to format a LAS header item for output.

    Arguments:
        order: format of item, either 'descr:value' or 'value:descr'

    Keyword Arguments:
        left_width (int): number of characters to the left hand side of the
            first period
        middle_width (int): total number of characters minus 1 between the
            first period from the left and the first colon from the left.

    Returns:
        A function which takes a header item 
        (e.g. :class:`lasio.las_items.HeaderItem`) as its single argument and 
        which in turn returns a string which is the correctly formatted LAS
        header line.

    '''
    if left_width is None:
        left_width = 10
    if middle_width is None:
        middle_width = 40
    mnemonic_func = lambda mnemonic: mnemonic.ljust(left_width)
    middle_func = lambda unit, right_hand_item: (
        unit
        + ' ' * (middle_width - len(str(unit)) - len(right_hand_item))
        + right_hand_item
    )
    if order == 'descr:value':
        return lambda item: '%s.%s : %s' % (
            mnemonic_func(item.original_mnemonic),
            middle_func(str(item.unit), str(item.descr)),
            item.value
        )
    elif order == 'value:descr':
        return lambda item: '%s.%s : %s' % (
            mnemonic_func(item.original_mnemonic),
            middle_func(str(item.unit), str(item.value)),
            item.descr
        )


def get_section_order_function(section, version,
                               order_definitions=defaults.ORDER_DEFINITIONS):
    '''Get a function that returns the order per the mnemonic and section.

    Arguments:
        section (str): either 'well', 'params', 'curves', 'version'
        version (float): either 1.2 and 2.0

    Keyword Arguments:
        order_definitions (dict): see source of defaults.py for more information

    Returns:
        A function which takes a mnemonic (str) as its only argument, and 
        in turn returns the order 'value:descr' or 'descr:value'.

    '''
    section_orders = order_definitions[version][section]
    default_order = section_orders[0]
    orders = {}
    for order, mnemonics in section_orders[1:]:
        for mnemonic in mnemonics:
            orders[mnemonic] = order
    return lambda mnemonic: orders.get(mnemonic, default_order)


def get_section_widths(section_name, items, version, order_func):
    '''Find minimum section widths fitting the content in *items*.

    Arguments:
        section_name (str): either 'version', 'well', 'curves', or 'params'
        items (SectionItems): section items
        version (float): either 1.2 or 2.0
        order_func (func): see :func:`lasio.writer.get_section_order_function`

    '''
    section_widths = {
        'left_width': None,
        'middle_width': None
    }
    if len(items) > 0:
        section_widths['left_width'] = max(
            [len(i.original_mnemonic) for i in items])
        middle_widths = []
        for i in items:
            order = order_func(i.mnemonic)
            rhs_element = order.split(':')[0]
            logger.debug(
                'get_section_widths %s\n\torder=%s rhs_element=%s' % (
                    i, order, rhs_element))
            middle_widths.append(
                len(str(i.unit)) + 1 + len(str(i[rhs_element])))
        section_widths['middle_width'] = max(middle_widths)
    return section_widths


1			import logging
2			import textwrap
3
4			import numpy as np
5
6			from .las_items import (
7			HeaderItem, CurveItem, SectionItems, OrderedDict)
8			from . import defaults
9			from . import exceptions
10
11			logger = logging.getLogger(__name__)
12
13
14			def write(las, file_object, version=None, wrap=None, STRT=None,
15			STOP=None, STEP=None, fmt='%.5f'):
16			'''Write a LAS files.
17
18			Arguments:
19			las (:class:`lasio.las.LASFile`)
20			file_object (file-like object open for writing): output
21
22			Keyword Arguments:
23			version (float or None): version of written file, either 1.2 or 2.
24			If this is None, ``las.version.VERS.value`` will be used.
25			wrap (bool or None): whether to wrap the output data section.
26			If this is None, ``las.version.WRAP.value`` will be used.
27			STRT (float or None): value to use as STRT (note the data will not
28			be clipped). If this is None, the data value in the first column,
29			first row will be used.
30			STOP (float or None): value to use as STOP (note the data will not
31			be clipped). If this is None, the data value in the first column,
32			last row will be used.
33			STEP (float or None): value to use as STEP (note the data will not
34			be resampled and/or interpolated). If this is None, the STEP will
35			be estimated from the first two rows of the first column.
36			fmt (str): Python string formatting operator for numeric data to be
37			used.
38
39			You should avoid calling this function directly - instead use the
40			:meth:`lasio.las.LASFile.write` method.
41
42			'''
43			if wrap is None:
44			wrap = las.version['WRAP'] == 'YES'
45			elif wrap is True:
46			las.version['WRAP'] = HeaderItem(
47			'WRAP', '', 'YES', 'Multiple lines per depth step')
48			elif wrap is False:
49			las.version['WRAP'] = HeaderItem(
50			'WRAP', '', 'NO', 'One line per depth step')
51			lines = []
52
53			assert version in (1.2, 2, None)
54			if version is None:
55			version = las.version['VERS'].value
56			if version == 1.2:
57			las.version['VERS'] = HeaderItem(
58			'VERS', '', 1.2, 'CWLS LOG ASCII STANDARD - VERSION 1.2')
59			elif version == 2:
60			las.version['VERS'] = HeaderItem(
61			'VERS', '', 2.0, 'CWLS log ASCII Standard -VERSION 2.0')
62
63			if STRT is None:
64			STRT = las.index[0]
65			if STOP is None:
66			STOP = las.index[-1]
67			if STEP is None:
68			STEP = las.index[1] - las.index[0] # Faster than np.gradient
69			las.well['STRT'].value = STRT
70			las.well['STOP'].value = STOP
71			las.well['STEP'].value = STEP
72
73			# Check units
74			if las.curves[0].unit:
75			unit = las.curves[0].unit
76			else:
77			unit = las.well['STRT'].unit
78			las.well['STRT'].unit = unit
79			las.well['STOP'].unit = unit
80			las.well['STEP'].unit = unit
81			las.curves[0].unit = unit
82
83			# Check for any changes in the pandas dataframe and if there are,
84			# create new curves so they are reflected in the output LAS file.
85
86			# if las.use_pandas:
87			# curve_names = lambda: [ci.mnemonic for ci in las.curves]
88			# for df_curve_name in list(las.df.columns.values):
89			# if not df_curve_name in curve_names():
90			# las.add_curve(df_curve_name, las.df[df_curve_name])
91
92			# Write each section.
93
94			# ~Version
95			logger.debug('LASFile.write Version section')
96			lines.append('~Version '.ljust(60, '-'))
97			order_func = get_section_order_function('Version', version)
98			section_widths = get_section_widths(
99			'Version', las.version, version, order_func)
100			for header_item in las.version.values():
101			mnemonic = header_item.original_mnemonic
102			# logger.debug('LASFile.write ' + str(header_item))
103			order = order_func(mnemonic)
104			# logger.debug('LASFile.write order = %s' % (order, ))
105			logger.debug('LASFile.write %s order=%s section_widths=%s' % (
106			header_item, order, section_widths))
107			formatter_func = get_formatter_function(order, **section_widths)
108			line = formatter_func(header_item)
109			lines.append(line)
110
111			# ~Well
112			logger.debug('LASFile.write Well section')
113			lines.append('~Well '.ljust(60, '-'))
114			order_func = get_section_order_function('Well', version)
115			section_widths = get_section_widths(
116			'Well', las.well, version, order_func)
117			# logger.debug('LASFile.write well section_widths=%s' % section_widths)
118			for header_item in las.well.values():
119			mnemonic = header_item.original_mnemonic
120			order = order_func(mnemonic)
121			logger.debug('LASFile.write %s order=%s section_widths=%s' % (
122			header_item, order, section_widths))
123			formatter_func = get_formatter_function(order, **section_widths)
124			line = formatter_func(header_item)
125			lines.append(line)
126
127			# ~Curves
128			logger.debug('LASFile.write Curves section')
129			lines.append('~Curve Information '.ljust(60, '-'))
130			order_func = get_section_order_function('Curves', version)
131			section_widths = get_section_widths(
132			'Curves', las.curves, version, order_func)
133			for header_item in las.curves:
134			mnemonic = header_item.original_mnemonic
135			order = order_func(mnemonic)
136			formatter_func = get_formatter_function(order, **section_widths)
137			line = formatter_func(header_item)
138			lines.append(line)
139
140			# ~Params
141			lines.append('~Params '.ljust(60, '-'))
142			order_func = get_section_order_function('Parameter', version)
143			section_widths = get_section_widths(
144			'Parameter', las.params, version, order_func)
145			for header_item in las.params.values():
146			mnemonic = header_item.original_mnemonic
147			order = order_func(mnemonic)
148			formatter_func = get_formatter_function(order, **section_widths)
149			line = formatter_func(header_item)
150			lines.append(line)
151
152			# ~Other
153			lines.append('~Other '.ljust(60, '-'))
154			lines += las.other.splitlines()
155
156			lines.append('~ASCII '.ljust(60, '-'))
157
158			file_object.write('\n'.join(lines))
159			file_object.write('\n')
160			line_counter = len(lines)
161
162			# data_arr = np.column_stack([c.data for c in las.curves])
163			data_arr = las.data
164			nrows, ncols = data_arr.shape
165
166			def format_data_section_line(n, fmt, l=10, spacer=' '):
167			try:
168			if np.isnan(n):
169			return spacer + str(las.well['NULL'].value).rjust(l)
170			else:
171			return spacer + (fmt % n).rjust(l)
172			except TypeError:
173			return spacer + str(n).rjust(l)
174
175			twrapper = textwrap.TextWrapper(width=79)
176
177			for i in range(nrows):
178			depth_slice = ''
179			for j in range(ncols):
180			depth_slice += format_data_section_line(data_arr[i, j], fmt)
181
182			if wrap:
183			lines = twrapper.wrap(depth_slice)
184			logger.debug('LASFile.write Wrapped %d lines out of %s' %
185			(len(lines), depth_slice))
186			else:
187			lines = [depth_slice]
188
189			for line in lines:
190			if las.version['VERS'].value == 1.2 and len(line) > 255:
191			logger.warning('[v1.2] line #{} has {} chars (>256)'.format(
192			line_counter + 1, len(line)))
193			file_object.write(line + '\n')
194			line_counter += 1
195
196
197			def get_formatter_function(order, left_width=None, middle_width=None):
198			'''Create function to format a LAS header item for output.
199
200			Arguments:
201			order: format of item, either 'descr:value' or 'value:descr'
202
203			Keyword Arguments:
204			left_width (int): number of characters to the left hand side of the
205			first period
206			middle_width (int): total number of characters minus 1 between the
207			first period from the left and the first colon from the left.
208
209			Returns:
210			A function which takes a header item
211			(e.g. :class:`lasio.las_items.HeaderItem`) as its single argument and
212			which in turn returns a string which is the correctly formatted LAS
213			header line.
214
215			'''
216			if left_width is None:
217			left_width = 10
218			if middle_width is None:
219			middle_width = 40
220			mnemonic_func = lambda mnemonic: mnemonic.ljust(left_width)
221			middle_func = lambda unit, right_hand_item: (
222			unit
223			+ ' ' * (middle_width - len(str(unit)) - len(right_hand_item))
224			+ right_hand_item
225			)
226			if order == 'descr:value':
227			return lambda item: '%s.%s : %s' % (
228			mnemonic_func(item.original_mnemonic),
229			middle_func(str(item.unit), str(item.descr)),
230			item.value
231			)
232			elif order == 'value:descr':
233			return lambda item: '%s.%s : %s' % (
234			mnemonic_func(item.original_mnemonic),
235			middle_func(str(item.unit), str(item.value)),
236			item.descr
237			)
238
239
240			def get_section_order_function(section, version,
241			order_definitions=defaults.ORDER_DEFINITIONS):
242			'''Get a function that returns the order per the mnemonic and section.
243
244			Arguments:
245			section (str): either 'well', 'params', 'curves', 'version'
246			version (float): either 1.2 and 2.0
247
248			Keyword Arguments:
249			order_definitions (dict): see source of defaults.py for more information
250
251			Returns:
252			A function which takes a mnemonic (str) as its only argument, and
253			in turn returns the order 'value:descr' or 'descr:value'.
254
255			'''
256			section_orders = order_definitions[version][section]
257			default_order = section_orders[0]
258			orders = {}
259			for order, mnemonics in section_orders[1:]:
260			for mnemonic in mnemonics:
261			orders[mnemonic] = order
262			return lambda mnemonic: orders.get(mnemonic, default_order)
263
264
265			def get_section_widths(section_name, items, version, order_func):
266			'''Find minimum section widths fitting the content in items.
267
268			Arguments:
269			section_name (str): either 'version', 'well', 'curves', or 'params'
270			items (SectionItems): section items
271			version (float): either 1.2 or 2.0
272			order_func (func): see :func:`lasio.writer.get_section_order_function`
273
274			'''
275			section_widths = {
276			'left_width': None,
277			'middle_width': None
278			}
279			if len(items) > 0:
280			section_widths['left_width'] = max(
281			[len(i.original_mnemonic) for i in items])
282			middle_widths = []
283			for i in items:
284			order = order_func(i.mnemonic)
285			rhs_element = order.split(':')[0]
286			logger.debug(
287			'get_section_widths %s\n\torder=%s rhs_element=%s' % (
288			i, order, rhs_element))
289			middle_widths.append(
290			len(str(i.unit)) + 1 + len(str(i[rhs_element])))
291			section_widths['middle_width'] = max(middle_widths)
292			return section_widths
293

kinverarity1 / lasio

get_formatter_function() C last analyzed 2018-07-04 10:31 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

get_formatter_function() C
last analyzed 2018-07-04 10:31 UTC