How to use osmium - 10 common examples

#!/usr/bin/env python3

import sys
from osmium.replication import server

if __name__ == '__main__':
    if len(sys.argv) != 3:
        print("Usage: python check_server_for_updates.py  ")
        sys.exit(254)

    seqid = int(sys.argv[2])

    state = server.ReplicationServer(sys.argv[1]).get_state_info()

    if state is None:
        print("ERROR: Cannot get state from URL %s." % (sys.argv[1], ))
        sys.exit(253)

    if state.sequence <= seqid:
        print("Database up to date.")
        sys.exit(1)

    print("New data available (%i => %i)." % (seqid, state.sequence))
    sys.exit(0)

def way(self, way):
        # callback method for ways
        if 'highway' in way.tags and (not self.roads or way.tags['highway'] in self.roads):
            # ignore single-point ways
            if len(way.nodes) < 2:
                self.log('\nSkipping single-point way: id: {}, tags: {}, nodes: {}\n'.format(way.id, way.tags, way.nodes))
                return

            new_way = {'id': way.id, 'tags': {}, 'refs': [], 'coords': []}
            for tag in way.tags:
                new_way['tags'][tag.k] = tag.v
            for node in way.nodes:
                try:
                    new_way['refs'].append(node.ref)
                    new_way['coords'].append((node.lat, node.lon))
                except InvalidLocationError as e:
                    self.log('\nSkipping node: {} (x={}, y={}) because of error: {}\n'.format(node.ref, node.x, node.y, e))

            # Add our ways to a route collection if we can match them either
            # by route-number or alternatively, by name. These route-collections
            # will later be joined into longer segments so that curvature
            # calculations can extend over multiple way-segments that might be
            # split due to bridges, local names, speed limits, or other metadata
            # changes.
            if 'ref' in new_way['tags']:
                routes = new_way['tags']['ref'].split(';')
                for route in routes:
                    if route not in self.routes:
                        self.routes[route] = {  'join_type': 'ref',
                                                'join_data': route,
                                                'ways': []}
                    self.routes[route]['ways'].append(new_way)

"""


import sys
import os.path as osp
from collections import defaultdict
from datetime import datetime, timezone
#if sys.version_info[0] == 3:
#    from datetime import timezone
import pandas as pd
import osmium as osm

DEFAULT_START = pd.Timestamp("2000-01-01T00:00:00Z")

########################################
class NodeTimelineHandler(osm.SimpleHandler):
    def __init__(self):
        osm.SimpleHandler.__init__(self)
        self.ids = set()
        # list of lists (node_id, lon, lat, version, visible, ts, uid, changesetid, ntags, tagkeys)
        self.nodetimeline = []
        
    def node(self,n):
        self.ids.add(n.id)
        nodeloc = n.location
        if nodeloc.valid():
            self.nodetimeline.append([n.id,
                                      nodeloc.lon,
                                      nodeloc.lat,
                                      n.version,
                                      not(n.deleted),
                                      pd.Timestamp(n.timestamp),

"""Extract some stats for OSM ways, from a history OSM data file
The file has to be runned through the following format:                            
python   
"""

import sys
import os.path as osp
from collections import defaultdict
from datetime import datetime, timezone
import pandas as pd
import osmium as osm

DEFAULT_START = pd.Timestamp("2000-01-01T00:00:00Z")

class RelationTimelineHandler(osm.SimpleHandler):
    def __init__(self):
        osm.SimpleHandler.__init__(self)
        self.ids = set()
        # list of lists (way_id, version, visible, ts, uid, changesetid, nb_members, mids, nb_tags, tagkeys)
        self.relationtimeline = []
        
    def relation(self,r):
        self.ids.add(r.id)
        self.relationtimeline.append([r.id,
                                  r.version,
                                  r.visible,
                                  pd.Timestamp(r.timestamp),
                                  r.uid,
                                  r.changeset,
                                  len(r.members),
                                  [(m.ref,m.role,m.type) for m in r.members],

def __init__(self):
        osmium.SimpleHandler.__init__(self)

def __init__(self, node_count: int) -> None:
		osmium.SimpleHandler.__init__(self)
		self.node_count = node_count
		self.location_list: List[common.Location] = []
		self.node_index = 0  # indexes nodes while iterating over them

import sys

import osmium
import geojson
import shapely.geometry

from robosat.osm.core import FeatureStorage, is_polygon


class BuildingHandler(osmium.SimpleHandler):
    """Extracts building polygon features (visible in satellite imagery) from the map.
    """

    # building=* to discard because these features are not vislible in satellite imagery
    building_filter = set(
        ["construction", "houseboat", "static_caravan", "stadium", "conservatory", "digester", "greenhouse", "ruins"]
    )

    # location=* to discard because these features are not vislible in satellite imagery
    location_filter = set(["underground", "underwater"])

    def __init__(self, out, batch):
        super().__init__()
        self.storage = FeatureStorage(out, batch)

    def way(self, w):

layer = self.layers[(layer_name,geom_type)]
        feature = ogr.Feature(layer.defn)
        feature.SetGeometry(geom)
        if layer.osm_id:
            feature.SetField('osm_id',osm_id)
        for column_name in layer.columns:
            if column_name in tags:
                feature.SetField(column_name,tags[column_name])
        layer.ogr_layer.CreateFeature(feature)

    def finalize(self):
        for k, layer in self.layers.items():
            layer.ds.CommitTransaction()
        self.layers = None

class Handler(o.SimpleHandler):
    def __init__(self,outputs,mapping,clipping_geom=None):
        super(Handler, self).__init__()
        self.outputs = outputs
        self.mapping = mapping
        self.clipping_geom = clipping_geom
        if clipping_geom:
            self.prepared_clipping_geom = prep(clipping_geom)

    def node(self,n):
        if len(n.tags) == 0:
            return
        geom = None
        for theme in self.mapping.themes:
            if theme.matches(GeomType.POINT,n.tags):
                if not geom:
                    wkb = fab.create_point(n)

def __init__(self):
        """ Class default constructor"""
        osm.SimpleHandler.__init__(self)
        self.elemtimeline = [] # Dictionnary of OSM elements

#! /usr/bin/env python

from typing import List
import sys
import random
import osmium

import common

class RandomNodeSelector(osmium.SimpleHandler):
	"""Picks 'node_count' uniformly randomly selected OSM nodes."""
	def __init__(self, node_count: int) -> None:
		osmium.SimpleHandler.__init__(self)
		self.node_count = node_count
		self.location_list: List[common.Location] = []
		self.node_index = 0  # indexes nodes while iterating over them

	def _location(self, node: osmium.osm.Node) -> common.Location:
		return common.Location(latitude = node.location.lat,
		                       longitude = node.location.lon)

	def node(self, n):
		self.node_index += 1
		if self.node_index <= self.node_count:
			# We're still looking at the initial few nodes
			self.location_list.append(self._location(n))