central-coast-aerial-survey/lastools.py

import os
import subprocess


def call_lastools(tool_name, las_in, las_out=None, args=None, verbose=True):
    """Send commands to the lastools library.

    Requires lastools in system path.

    Args:
        tool_name:  name of lastools binary
        las_in:     bytes from stdout, or path to input point cloud
        las_out:    '-stdout' to pipe output, or path to output point cloud
        args:       list of additional arguments, formatted for lastools
        verbose:    show all warnings and messages from lastools (boolean)

    Returns:
        bytes of output las, if las_out='-stdout'
        None, if las_out='path/to/las/file'

    Examples:

    # Convert xyz file to las and pipe stdout to a python bytes object
    las_data = call_lastools('txt2las', las_in='points.xyz', las_out='-stdout',
                             args=['-parse', 'sxyz'])

    # Clip las_data with a shapefile, and save to a new las file
    call_lastools('lasclip', las_in=las_data, las_out='points.las',
                  args=['-poly', 'boundary.shp'])
    """

    # Start building command string
    cmd = [tool_name]

    # Parse input
    if type(las_in) == bytes:
        # Pipe input las bytes to stdin
        cmd += ['-stdin']
        stdin = las_in
    else:
        # Load las from file path
        cmd += ['-i', las_in]
        stdin = None

    # Parse output
    if las_out == '-stdout':
        # Pipe output las to stdout
        cmd += ['-stdout']
    elif las_out:
        # Save output las to file
        cmd += ['-o', las_out]

    # Append additional lastools arguments, if provided
    if args:
        cmd += args

    process = subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        stdin=subprocess.PIPE)
    stdout, stderr = process.communicate(input=stdin)

    # Handle errors, if detected
    if process.returncode != 0:
        print("Error: {} failed on {}".format(tool_name,
                                              os.path.basename(las_in)))
        print(stderr.decode())

    elif verbose:
        # Print addional messages if verbose mode is being used
        print(stderr.decode())

    # Output piped stdout if required
    if las_out == '-stdout':
        return stdout
    else:
        return None