csv-to-geo.py

#!/usr/bin/env python3

import csv
import json

from lxml import etree
from pykml.factory import KML_ElementMaker as K


OPERATIONS = {
        'is': (lambda x, y: x == y),
        'is not': (lambda x, y: x != y),
        'in': (lambda x, y: x in y),
        'not in': (lambda x, y: x not in y),
        '>': (lambda x, y: x > y),
        '>=': (lambda x, y: x >= y),
        '<': (lambda x, y: x <= y),
        '<=': (lambda x, y: x <= y)
        }



def extended_data(data):
    """ Return an ExtendedData tag. """
    data_elements = []
    for key, value in data.items():
        data_elements.append(K.Data(K.value(value), name=key))
    return K.ExtendedData(*data_elements)


class KMLDoc:

    def __init__(self, config=None):
        self.config = config
        self.folder = K.Folder()
        if self.config.title:
            self.folder.append(K.name(self.config.title))
        if self.config.description:
            self.folder.append(K.description(self.config.description))

    def _properties(self, row):
        props = {self.config.url_label: self.config.url_format % row}
        props.update({label: row[column] for column, label in self.config.properties_to_labels.items()})
        return props

    def append(self, row):
        """ Map the fields in a row to a KML Point Placemark. """
        lon, lat = row['Location'].split(',')
        if not self.config.display_filter(row, lat, lon):
            return
        placemark = K.Placemark(
                K.name(row['Name']),
                K.Point(
                    K.coordinates(','.join((lat,lon)))
                    ),
                extended_data(self._properties(row))
                )
        if self.config.conditional_heading:
            icon_style = self.config.highlight_icon(row)
            placemark.append(K.styleUrl(icon_style))
        self.folder.append(placemark)

    def __str__(self):
        return etree.tostring(K.kml(self.folder), pretty_print=True, encoding='UTF-8', xml_declaration=True).decode()


def transform_csv(source_file, config):
    with source_file as csv_file:
        reader = csv.DictReader(csv_file)
        doc = KMLDoc(config)
        for row in reader:
            doc.append(row)
    return doc


class Config:

    def __init__(self, config_file):
        self.values = json.load(config_file)

    def display_filter(self, row, lat, lon):
        row['lat'] = lat
        row['lon'] = lon
        return self._filter(row, self.values['display_conditions'])

    def highlight_filter(self, row):
        return self._filter(row, self.values['highlight_conditions'])

    def highlight_icon(self, row):
        return self.values['icon_map'][str(self.highlight_filter(row)).lower()]

    def _filter(self, row, conditions):
        for column, operator, value in conditions:
            operation = OPERATIONS[operator]
            if not operation(row[column], value):
                return False
        return True

    def __getattr__(self, attr):
        return self.values.get(attr)


if __name__ == '__main__':
    import argparse

    parser = argparse.ArgumentParser()
    parser.add_argument('-i', '--input', type=open, help='A CSV file with your geographic information.', required=True)
    parser.add_argument('-c', '--config', type=open, help='A json file containing the required configuration info.', required=True)
    args = parser.parse_args()
    print(transform_csv(args.input, Config(args.config)))