home / skills / plurigrid / asi / osm-topology
This skill analyzes OpenStreetMap road networks and routing topology, applying GF(3) coloring to validate and optimize geographic networks.
npx playbooks add skill plurigrid/asi --skill osm-topologyReview the files below or copy the command above to add this skill to your agents.
---
name: osm-topology
description: 'OSM Topology Skill'
version: 1.0.0
---
# OSM Topology Skill
OpenStreetMap graph analysis: road networks, routing, and topological structure with GF(3) coloring.
## Trigger
- OpenStreetMap data processing
- Road network analysis, routing
- Graph-based geographic queries
- Street network topology
## GF(3) Trit: -1 (Validator)
Validates topological consistency of geographic networks.
## OSM Data Model
OSM uses three primitives:
- **Nodes**: Points with lat/lon
- **Ways**: Ordered lists of nodes (roads, boundaries)
- **Relations**: Groups of nodes/ways (routes, multipolygons)
## DuckDB OSM Integration
```sql
-- Read OSM PBF files (requires osm extension)
-- Install from: https://github.com/duckdb/duckdb_osm
-- Alternative: Use pre-processed Parquet
CREATE TABLE osm_nodes AS
SELECT * FROM read_parquet('osm_nodes.parquet');
CREATE TABLE osm_ways AS
SELECT * FROM read_parquet('osm_ways.parquet');
-- Schema for colored OSM data
CREATE TABLE osm_network (
way_id BIGINT,
name VARCHAR,
highway_type VARCHAR,
geometry GEOMETRY,
node_ids BIGINT[],
-- Topology
start_node BIGINT,
end_node BIGINT,
length_m DOUBLE,
-- GF(3) coloring
seed BIGINT,
gay_color VARCHAR,
gf3_trit INTEGER
);
```
## Graph Extraction
```python
import duckdb
import networkx as nx
def extract_road_graph(osm_parquet_path):
"""Extract road network as colored graph."""
conn = duckdb.connect()
conn.execute("INSTALL spatial; LOAD spatial;")
# Load ways with road tags
conn.execute(f"""
CREATE TABLE roads AS
SELECT
way_id,
tags->>'name' as name,
tags->>'highway' as highway,
nodes,
ST_Length_Spheroid(ST_MakeLine(
LIST_TRANSFORM(nodes, n -> ST_Point(n.lon, n.lat))
)) as length_m
FROM read_parquet('{osm_parquet_path}')
WHERE tags->>'highway' IS NOT NULL
""")
# Build graph
G = nx.DiGraph()
roads = conn.execute("""
SELECT way_id, nodes, length_m, highway FROM roads
""").fetchall()
for way_id, nodes, length, highway in roads:
for i in range(len(nodes) - 1):
n1, n2 = nodes[i], nodes[i+1]
# Color edge from way_id
seed = way_id & 0x7FFFFFFFFFFFFFFF
hue = seed % 360
trit = 1 if (hue < 60 or hue >= 300) else (0 if hue < 180 else -1)
G.add_edge(n1['id'], n2['id'],
way_id=way_id,
length=length / (len(nodes) - 1),
highway=highway,
trit=trit)
# Add reverse for bidirectional roads
if highway not in ('motorway', 'motorway_link'):
G.add_edge(n2['id'], n1['id'],
way_id=way_id,
length=length / (len(nodes) - 1),
highway=highway,
trit=trit)
return G
```
## Topological Validation
```python
def validate_network_topology(G):
"""
Validate OSM network topology.
Returns list of issues with GF(3) classification.
"""
issues = []
# Check connectivity
if not nx.is_weakly_connected(G):
components = list(nx.weakly_connected_components(G))
issues.append({
'type': 'disconnected',
'count': len(components),
'trit': -1, # Validation failure
'severity': 'high'
})
# Check for dead ends
dead_ends = [n for n in G.nodes() if G.degree(n) == 1]
if dead_ends:
issues.append({
'type': 'dead_ends',
'count': len(dead_ends),
'nodes': dead_ends[:10],
'trit': 0, # Ergodic (may be intentional)
'severity': 'low'
})
# Check for self-loops
self_loops = list(nx.selfloop_edges(G))
if self_loops:
issues.append({
'type': 'self_loops',
'count': len(self_loops),
'trit': -1, # Validation failure
'severity': 'medium'
})
# Check for duplicate edges
multi_edges = [(u, v) for u, v in G.edges() if G.number_of_edges(u, v) > 1]
if multi_edges:
issues.append({
'type': 'multi_edges',
'count': len(multi_edges),
'trit': -1,
'severity': 'medium'
})
return issues
def gf3_balance_check(G):
"""Check if edge trits are GF(3) balanced per node."""
imbalanced = []
for node in G.nodes():
edges = list(G.edges(node, data=True))
trit_sum = sum(e[2].get('trit', 0) for e in edges)
if trit_sum % 3 != 0:
imbalanced.append({
'node': node,
'trit_sum': trit_sum,
'edge_count': len(edges)
})
return {
'total_nodes': G.number_of_nodes(),
'imbalanced_count': len(imbalanced),
'balance_ratio': 1 - len(imbalanced) / G.number_of_nodes(),
'sample_imbalanced': imbalanced[:5]
}
```
## Routing with Color
```python
def colored_route(G, start, end, weight='length'):
"""Find shortest path with GF(3) coloring."""
try:
path = nx.shortest_path(G, start, end, weight=weight)
edges = []
total_length = 0
trit_sum = 0
for i in range(len(path) - 1):
edge_data = G.edges[path[i], path[i+1]]
edges.append({
'from': path[i],
'to': path[i+1],
'length': edge_data['length'],
'highway': edge_data['highway'],
'trit': edge_data['trit']
})
total_length += edge_data['length']
trit_sum += edge_data['trit']
return {
'path': path,
'edges': edges,
'total_length_m': total_length,
'hop_count': len(path) - 1,
'gf3_sum': trit_sum,
'gf3_mod3': trit_sum % 3,
'balanced': trit_sum % 3 == 0
}
except nx.NetworkXNoPath:
return {'error': 'No path found', 'trit': -1}
```
## Overpass API Integration
```python
import requests
def query_osm_overpass(bbox, highway_types=['primary', 'secondary', 'tertiary']):
"""Query OSM via Overpass API."""
highway_filter = '|'.join(highway_types)
query = f"""
[out:json][timeout:60];
(
way["highway"~"{highway_filter}"]({bbox});
);
out body;
>;
out skel qt;
"""
response = requests.post(
'https://overpass-api.de/api/interpreter',
data={'data': query}
)
return response.json()
def osm_to_colored_graph(osm_json, seed=42):
"""Convert Overpass response to colored graph."""
import hashlib
G = nx.DiGraph()
nodes = {e['id']: e for e in osm_json['elements'] if e['type'] == 'node'}
for element in osm_json['elements']:
if element['type'] == 'way':
way_id = element['id']
node_refs = element.get('nodes', [])
# Color from way ID
seed_val = int(hashlib.sha256(str(way_id).encode()).hexdigest()[:16], 16)
hue = seed_val % 360
trit = 1 if (hue < 60 or hue >= 300) else (0 if hue < 180 else -1)
for i in range(len(node_refs) - 1):
n1, n2 = node_refs[i], node_refs[i+1]
if n1 in nodes and n2 in nodes:
G.add_edge(n1, n2,
way_id=way_id,
tags=element.get('tags', {}),
trit=trit)
# Add node coordinates
for node_id, node_data in nodes.items():
if node_id in G:
G.nodes[node_id]['lat'] = node_data['lat']
G.nodes[node_id]['lon'] = node_data['lon']
return G
```
## Triads
```
osm-topology (-1) ⊗ duckdb-spatial (0) ⊗ map-projection (+1) = 0 ✓
osm-topology (-1) ⊗ geodesic-manifold (0) ⊗ geohash-coloring (+1) = 0 ✓
osm-topology (-1) ⊗ acsets (0) ⊗ gay-mcp (+1) = 0 ✓
```
## References
- OpenStreetMap Wiki
- OSMnx library (Geoff Boeing)
- NetworkX documentation
This skill analyzes OpenStreetMap road networks to extract topology, validate structural consistency, and enable GF(3)-colored routing. It builds directed graphs from OSM data (Parquet, PBF via DuckDB, or Overpass JSON), annotates edges with a deterministic GF(3) trit, and reports topological issues. The skill is aimed at network validation, routing experiments, and topology-aware analytics.
The skill reads OSM primitives (nodes, ways, relations) and converts road ways into a directed NetworkX graph with per-edge attributes: way_id, length, highway type, and a GF(3) trit derived from a hash/hue of the way id. It provides functions to validate topology (connectivity, dead ends, self-loops, duplicate edges), check per-node GF(3) balance, and compute shortest paths that include GF(3) summaries. Data can be loaded via DuckDB (Parquet or spatial extension) or fetched live through the Overpass API and converted to the colored graph.
What does the GF(3) trit mean?
Each edge receives a value in {-1,0,1} derived deterministically from a hash/hue of the way id. It is used as lightweight categorical metadata and for per-node balance checks modulo 3.
How do I handle bidirectional vs one-way ways?
The extractor adds a reverse edge for non-motorway way types by default. Honor explicit one-way tags when present to avoid incorrect reverse edges.