Translating features with Fiona

fiona-translate.py

# Translation of Shapefile record geometries in a functional style.

from fiona import collection
from functools import partial
from itertools import imap
import logging

log = logging.getLogger()

# To begin: a few functions to translate geometry coordinates. They call each
# other semi-recursively. GeoJSON was designed to make this possible, BTW.

def translate_Point(coords, delta):
    """Returns translated Point coordinates, preserving order, number,
    and dimensionality.
    
    delta is a (delta_x, delta_y [, delta_y]) tuple."""
    return tuple(c + d for c, d in zip(coords, delta))

def translate_LineString(coords, delta):
    """Returns translated LineString or Ring coordinates, preserving 
    order, number, and dimensionality.
    
    delta is a (delta_x, delta_y [, delta_y]) tuple."""
    # Calls translate_Point.
    return list(translate_Point(pt_coords, delta) for pt_coords in coords)

def translate_Polygon(coords, delta):
    """Returns translated Polygon coordinates, preserving order, number,
    and dimensionality.
    
    delta is a (delta_x, delta_y [, delta_y]) tuple."""
    # Calls tanslate_LineString.
    return list(
        translate_LineString(ring_coords, delta) for ring_coords in coords)

# The next function is applied to each record in the following
# processing code.

def translate(delta, rec):
    """Returns a record after translating its geometry's coordinates.
    
    delta is a (delta_x, delta_y [, delta_y]) tuple."""
    
    # Use lexical dispatch on geometry type to get the proper translation
    # function.
    handlers = {
        'Point': translate_Point,
        'LineString': translate_LineString,
        'Polygon': translate_Polygon }
    try:
        g = rec['geometry']
        g['coordinates'] = handlers[g['type']](g['coordinates'], delta )
        rec['geometry'] = g
        return rec
    except Exception, e:
        log.exception("Error processing record %s:", rec)

# And now the processing code. First, open a source file.
with collection("docs/data/test_uk.shp", "r") as source:
    
    # Create a sink file for processed features with the same format and
    # coordinate reference system as the source.
    with collection(
            "translated.shp", 
            "w",
            driver=source.driver,
            schema=source.schema,
            crs=source.crs
            ) as sink:

        # Example 2D translation, 1 unit eastward and northward.
        results = imap(partial(translate, (1.0, 1.0)), source)
        sink.writerecords(results)