Hexes in Motion: H3 + Tile38 + NATS + DuckDB¶

import os, time, json, random, uuid, asyncio
from nats.aio.client import Client as NATS
import h3

async def main():
    nc = NATS()
    await nc.connect(os.getenv("NATS_URL", "nats://localhost:4222"))

    # Create JetStream context
    js = nc.jetstream()

    # Create stream for position updates
    await js.add_stream(name="positions", subjects=["pos.*"])

    # Create subject for geofence events
    await js.add_stream(name="events", subjects=["events.*"])

    # Random walker parameters
    center_lat, center_lon = 40.7589, -73.9851
    walk_radius = 0.01  # ~1km radius

    while True:
        # Random walk from center
        lat = center_lat + random.uniform(-walk_radius, walk_radius)
        lon = center_lon + random.uniform(-walk_radius, walk_radius)

        # Generate H3 index at resolution 9
        h3_index = h3.latlng_to_cell(lat, lon, 9)

        msg = {
            "id": f"walker-{random.randint(1, 10)}",
            "lat": lat,
            "lon": lon,
            "ts": int(time.time() * 1000),
            "h3_index": h3_index
        }

        # Publish to NATS JetStream
        await js.publish("pos.walkers", json.dumps(msg).encode())

        await asyncio.sleep(0.1)

if __name__ == "__main__":
    asyncio.run(main())

import os, json, asyncio, duckdb
from nats.aio.client import Client as NATS
import h3

class AnalyticsProcessor:
    def __init__(self):
        self.nats_url = os.getenv("NATS_URL", "nats://localhost:4222")
        self.db_path = os.getenv("DUCKDB_PATH", "/data/positions.db")
        self.conn = duckdb.connect(self.db_path)
        self.setup_database()

    def setup_database(self):
        """Create tables and views for analytics."""
        # Create positions table
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS positions (
                id TEXT,
                lat DOUBLE,
                lon DOUBLE,
                ts BIGINT,
                h3_index TEXT,
                hour_bucket BIGINT
            )
        """)

        # Create geofence events table
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS geofence_events (
                event_type TEXT,
                fence_id TEXT,
                device_id TEXT,
                lat DOUBLE,
                lon DOUBLE,
                timestamp BIGINT,
                hour_bucket BIGINT
            )
        """)

        # Create H3 aggregation view
        self.conn.execute("""
            CREATE OR REPLACE VIEW h3_heatmap AS
            SELECT 
                h3_index,
                COUNT(*) as position_count,
                AVG(lat) as avg_lat,
                AVG(lon) as avg_lon,
                MIN(ts) as first_seen,
                MAX(ts) as last_seen
            FROM positions
            GROUP BY h3_index
        """)

    async def process_positions(self, msg):
        """Process position updates and store in DuckDB."""
        try:
            data = json.loads(msg.data.decode())

            # Calculate hour bucket for partitioning
            hour_bucket = (data['ts'] // 3600000) * 3600000

            # Insert into DuckDB
            self.conn.execute("""
                INSERT INTO positions (id, lat, lon, ts, h3_index, hour_bucket)
                VALUES (?, ?, ?, ?, ?, ?)
            """, [
                data['id'], data['lat'], data['lon'], 
                data['ts'], data['h3_index'], hour_bucket
            ])

            print(f"Processed position: {data['id']} at {data['lat']}, {data['lon']}")

        except Exception as e:
            print(f"Error processing position: {e}")

    async def process_geofence_events(self, msg):
        """Process geofence events and store in DuckDB."""
        try:
            data = json.loads(msg.data.decode())

            # Calculate hour bucket
            hour_bucket = (data['timestamp'] // 3600000) * 3600000

            # Insert into DuckDB
            self.conn.execute("""
                INSERT INTO geofence_events (event_type, fence_id, device_id, lat, lon, timestamp, hour_bucket)
                VALUES (?, ?, ?, ?, ?, ?, ?)
            """, [
                data['event_type'], data['fence_id'], data['device_id'],
                data['position']['lat'], data['position']['lon'],
                data['timestamp'], hour_bucket
            ])

            print(f"Geofence event: {data['event_type']} {data['fence_id']} for {data['device_id']}")

        except Exception as e:
            print(f"Error processing geofence event: {e}")

    async def run(self):
        """Main processing loop."""
        nc = NATS()
        await nc.connect(self.nats_url)

        # Subscribe to position updates
        await nc.subscribe("pos.*", cb=self.process_positions)

        # Subscribe to geofence events
        await nc.subscribe("events.*", cb=self.process_geofence_events)

        print("Analytics processor started")

        # Keep running
        while True:
            await asyncio.sleep(1)

if __name__ == "__main__":
    processor = AnalyticsProcessor()
    asyncio.run(processor.run())