OSM node history to SQLite

osh2sqlite.py

#!/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()