counts switches and crossovers in railway data from overpass turbo

features.py

from dataclasses import dataclass, field
from typing import Any


@dataclass
class Feature:
    type: str
    id: str


@dataclass
class Node(Feature):
    lat: float
    lon: float
    tags: dict[str, Any] = field(default_factory=dict)
    endpoint_segments: set[str] = field(default_factory=set)
    midpoint_segments: set[str] = field(default_factory=set)


@dataclass
class Way(Feature):
    nodes: list[str] = field(default_factory=list)
    tags: dict[str, Any] = field(default_factory=dict)

    @property
    def start(self):
        return self.nodes[0]

    @property
    def end(self):
        return self.nodes[-1]

process_overpass_data.py

import json
from collections import defaultdict
from dataclasses import asdict

from railways.features import Node, Way

ways = {}
nodes = {}

with open("data.json") as data_file:
    data = json.loads(data_file.read())

# first load in features

for feature in data["elements"]:
    cls = None
    container = None
    if feature["type"] == "node":
        cls = Node
        container = nodes
    elif feature["type"] == "way":
        cls = Way
        container = ways
        feature["nodes"] = list(map(lambda nid: str(nid), feature["nodes"]))

    if not cls:
        continue

    feature["id"] = str(feature["id"])

    f = cls(**feature)
    container[f.id] = f

lines_to_split = defaultdict(list)

# next seed nodes with midpoint_segments and endpoint_segments values,
# and detect ways whose endpoint(s) split another way - to set up switches properly,
# switches cannot occur in the middle of a segment

for way in ways.values():
    i = 0
    for node_id in way.nodes:
        node: Node = nodes[node_id]

        if i == 0 or i == len(way.nodes):
            node.endpoint_segments.add(way.id)
        else:
            node.midpoint_segments.add(way.id)

        i += 1

        if min(len(node.endpoint_segments), len(node.midpoint_segments)) > 0:
            for way_id in node.midpoint_segments:
                lines_to_split[way_id].append(node.id)


# utility function to create a new way split
def get_new_way(way: Way, split_id):
    w = Way(**asdict(way))
    w.nodes = []
    w.id = f"{way.id}:{split_id}"
    ways[w.id] = w

    return w


# split ways that have midpoints which end other ways
for way_id, splitter_nodes in lines_to_split.items():
    way = ways[way_id]

    split_id = 0
    new_way = get_new_way(way, split_id)
    for i in range(len(way.nodes)):
        node_id = way.nodes[i]
        node: Node = nodes[node_id]
        new_way.nodes.append(node_id)

        if node_id in splitter_nodes:
            split_id += 1

            # start next split
            node.endpoint_segments.add(new_way.id)

            new_way = get_new_way(way, split_id)
            new_way.nodes.append(node_id)
            node.endpoint_segments.add(new_way.id)

            if way_id in node.midpoint_segments:
                node.midpoint_segments.remove(way_id)

    del ways[way_id]

num_switches = 0
num_crossovers = 0
for node in nodes.values():
    if len(node.endpoint_segments) >= 3:
        num_switches += 1

    if len(node.midpoint_segments) >= 2:
        num_crossovers += 1

print("Switches: ", num_switches)
print("Crossovers: ", num_crossovers)