Skip to content

Instantly share code, notes, and snippets.

@frafra
Last active October 8, 2018 07:49
Show Gist options
  • Save frafra/56650708033b6dd5bb3906827d0e58cc to your computer and use it in GitHub Desktop.
Save frafra/56650708033b6dd5bb3906827d0e58cc to your computer and use it in GitHub Desktop.
OSM node history to SQLite
#!/usr/bin/env python3
#
# OSM node history to SQLite
import osmium
import sqlite3
import sys
SCHEMA_OBJECT = """
CREATE TABLE %s
(id INTEGER NOT NULL,
version INTEGER NOT NULL,
uid INTEGER,
user TEXT,
date DATETIME,
PRIMARY KEY (id, version))
"""
SCHEMA_NODES = SCHEMA_OBJECT % 'nodes'
SCHEMA_WAYS = SCHEMA_OBJECT % 'ways'
LOCATION = """
SELECT AddGeometryColumn('nodes', 'Geometry', 3857, 'POINT', 'XY')
"""
SCHEMA_NODES_TAGS = """
CREATE TABLE nodes_tags
(id INTEGER NOT NULL REFERENCES nodes(id),
version INTEGER NOT NULL REFERENCES nodes(version),
key TEXT,
value TEXT,
PRIMARY KEY (id, version, key))
"""
class Converter(osmium.SimpleHandler):
def __init__(self):
osmium.SimpleHandler.__init__(self)
self.prev = {}
def insert_tag(self, n, key, value):
conn.execute('INSERT INTO tags VALUES (?, ?, ?, ?)', (
n.id, n.version, key, value))
def node(self, n):
if len(n.tags) == 0 or n.deleted:
return
coordinates = f'Transform(MakePoint({n.location.lon}, {n.location.lat}, 4326), 3857)'
conn.execute(f'INSERT INTO nodes VALUES (?, ?, ?, ?, ?, {coordinates})', (
n.id, n.version, n.uid, n.user, n.timestamp))
if n.deleted:
return
tags = dict([[tag.k, tag.v] for tag in n.tags])
if n.version == 1:
self.prev = {}
for key in self.prev.keys():
if key not in tags:
tags[key] = None
for key, value in tags.items():
if key in self.prev:
if self.prev[key] != value:
conn.execute('INSERT INTO nodes_tags VALUES (?, ?, ?, ?)', (
n.id, n.version, key, value))
else:
conn.execute('INSERT INTO nodes_tags VALUES (?, ?, ?, ?)', (
n.id, n.version, key, value))
self.nodes = []
self.attrs = []
self.prev = tags
def way(self, w):
if len(w.tags) == 0 or w.deleted:
return
conn.execute('INSERT INTO ways VALUES (?, ?, ?, ?, ?)', (
w.id, w.version, w.uid, w.user, w.timestamp))
if __name__ == '__main__':
with sqlite3.connect(sys.argv[2]) as conn:
conn.enable_load_extension(True)
conn.load_extension('/usr/lib64/mod_spatialite.so')
conn.execute(SCHEMA_NODES)
conn.execute(SCHEMA_NODES_TAGS)
conn.execute(SCHEMA_WAYS)
conn.execute('SELECT InitSpatialMetaData()')
conn.execute(LOCATION)
converter = Converter()
converter.apply_file(sys.argv[1])
conn.execute("SELECT CreateSpatialIndex('nodes', 'Geometry')")
conn.commit()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment