A small and broken OSM Jython importer

osm_importer.py

# encoding: utf-8

"""
Created on 25 feb. 2015.

@author: pupssman
"""

import math
import logging

logging.basicConfig(level=logging.INFO)

import se.kodapan.osm as osm
import org.apache.lucene as lucene

from constructor import Node, NetBuilder


class OSMData(object):
    def __init__(self, filename):
        self.pojo_root = osm.domain.root.PojoRoot()
        self.root = osm.domain.root.indexed.IndexedRoot.newInstance(self.pojo_root)
        self.root.open()

        p = osm.parser.xml.instantiated.InstantiatedOsmXmlParser.newInstance()
        p.setRoot(self.root)
        p.parse(open(filename, 'r'))
        self.root.commit()

        self._reference = list(self.pojo_root.nodes.values())[0]  # just some node

    def ways(self, tag, value):
        query = lucene.search.BooleanQuery()
        query.add(self.root.getQueryFactories().containsTagKeyAndValueQueryFactory().setKey(tag).setValue(value).build(),
                  lucene.search.BooleanClause.Occur.MUST)  # @UndefinedVariable

        return self.root.search(query)

    def relative(self, node):
        """
        Accepts an ``osm`` node and returns relative position as ``our`` node.
        Reference point is chosen arbitrarily.
        Actual algorithm is very, ehm, "approximate"
        """
        magic = 111111  # meters per latitude

        return Node((node.getLongitude() - self._reference.getLongitude()) * magic * math.cos(node.getLatitude() / 180 * math.pi),
                    - (node.getLatitude() - self._reference.getLatitude()) * magic)


def transform(source, dest, categories=['motorway', 'trunk', 'primary', 'secondary']):
    """
    Transforms OSM xml export segment to the CRS Network XML-save.
    :arg categories: categories of OSM's highways to put into export

    So far it has no notion of lanes, traffic lights or speed limit.
    TODO: add those.
    """
    data = OSMData(source)

    net = NetBuilder()

    def trace_way(way):
        logging.info('Making way %s' % way.getId())

        nodes = map(data.relative, way.getNodes())
        trace = net.trace_from(nodes.pop(0))
        for node in nodes:
            trace.to(node.x, node.y)

    for cat in categories:
        map(trace_way, data.ways('highway', cat))
        map(trace_way, data.ways('highway', cat + '_link'))

    net.save(dest)

    logging.info('Done, see %s' % dest)


def demo():
    # TODO: need better demo
    return transform('G:/Downloads/map.osm')

if __name__ == '__main__':
    demo()