a bunch of thing

Author: jackharrhy

docs/plans/2026-02-21-plowmap-plowapp-classes.md

@@ -0,0 +1,92 @@
+# src/where_the_plow/cache.py
+"""Simple file-based cache for coverage trail responses.
+
+Stores JSON in /tmp/where-the-plow-cache/ keyed by a hash of the
+(since, until) time range.  Only caches queries whose `until` is
+before today (i.e. fully historical, immutable data).  Uses LRU
+eviction by file access time when total cache size exceeds a budget.
+"""
+
+import hashlib
+import json
+import logging
+import os
+import tempfile
+from datetime import datetime, timezone
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+CACHE_DIR = Path(tempfile.gettempdir()) / "where-the-plow-cache"
+MAX_CACHE_BYTES = 200 * 1024 * 1024  # 200 MB
+
+
+def _cache_key(since: datetime, until: datetime) -> str:
+    raw = f"{since.isoformat()}|{until.isoformat()}"
+    return hashlib.sha256(raw.encode()).hexdigest()[:16]
+
+
+def _is_cacheable(until: datetime) -> bool:
+    """Only cache if the entire window is in the past (before today UTC)."""
+    today_start = datetime.now(timezone.utc).replace(
+        hour=0, minute=0, second=0, microsecond=0
+    )
+    until_utc = until if until.tzinfo else until.replace(tzinfo=timezone.utc)
+    return until_utc < today_start
+
+
+def _ensure_dir():
+    CACHE_DIR.mkdir(parents=True, exist_ok=True)
+
+
+def _evict_if_needed():
+    """Delete oldest-accessed files until total size is under budget."""
+    try:
+        files = list(CACHE_DIR.glob("*.json"))
+        if not files:
+            return
+        total = sum(f.stat().st_size for f in files)
+        if total <= MAX_CACHE_BYTES:
+            return
+        # Sort by access time, oldest first
+        files.sort(key=lambda f: f.stat().st_atime)
+        for f in files:
+            if total <= MAX_CACHE_BYTES:
+                break
+            size = f.stat().st_size
+            f.unlink(missing_ok=True)
+            total -= size
+            logger.debug("cache evict: %s (%d bytes)", f.name, size)
+    except OSError:
+        pass
+
+
+def get(since: datetime, until: datetime) -> list[dict] | None:
+    """Return cached trails or None if not cached."""
+    if not _is_cacheable(until):
+        return None
+    path = CACHE_DIR / f"{_cache_key(since, until)}.json"
+    if not path.exists():
+        return None
+    try:
+        # Touch access time for LRU
+        os.utime(path)
+        data = json.loads(path.read_text())
+        logger.debug("cache hit: %s", path.name)
+        return data
+    except (OSError, json.JSONDecodeError):
+        return None
+
+
+def put(since: datetime, until: datetime, trails: list[dict]):
+    """Store trails in cache if the query is cacheable."""
+    if not _is_cacheable(until):
+        return
+    _ensure_dir()
+    _evict_if_needed()
+    path = CACHE_DIR / f"{_cache_key(since, until)}.json"
+    try:
+        path.write_text(json.dumps(trails))
+        logger.debug("cache put: %s (%d trails)", path.name, len(trails))
+    except OSError:
+        pass

src/where_the_plow/cache.py

@@ -8,6 +8,7 @@
 from where_the_plow.client import fetch_vehicles, parse_avl_response
 from where_the_plow.db import Database
 from where_the_plow.config import settings
+from where_the_plow.snapshot import build_realtime_snapshot
 
 logger = logging.getLogger(__name__)
 
@@ -20,7 +21,7 @@ def process_poll(db: Database, response: dict) -> int:
     return inserted
 
 
-async def run(db: Database):
+async def run(db: Database, store: dict):
     logger.info("Collector starting — polling every %ds", settings.poll_interval)
 
     stats = db.get_stats()
@@ -41,6 +42,7 @@ async def run(db: Database):
                     len(features),
                     inserted,
                 )
+                store["realtime"] = build_realtime_snapshot(db)
             except asyncio.CancelledError:
                 logger.info("Collector shutting down")
                 raise

src/where_the_plow/collector.py

@@ -4,7 +4,6 @@
 import duckdb
 from datetime import datetime
 from itertools import groupby
-from operator import itemgetter
 
 
 class Database:
@@ -53,6 +52,22 @@ def init(self):
             CREATE INDEX IF NOT EXISTS idx_positions_time_geo
                 ON positions (timestamp, latitude, longitude)
         """)
+        cur.execute("""
+            CREATE SEQUENCE IF NOT EXISTS viewports_seq
+        """)
+        cur.execute("""
+            CREATE TABLE IF NOT EXISTS viewports (
+                id          BIGINT DEFAULT nextval('viewports_seq') PRIMARY KEY,
+                timestamp   TIMESTAMPTZ NOT NULL DEFAULT now(),
+                zoom        DOUBLE NOT NULL,
+                center_lng  DOUBLE NOT NULL,
+                center_lat  DOUBLE NOT NULL,
+                sw_lng      DOUBLE NOT NULL,
+                sw_lat      DOUBLE NOT NULL,
+                ne_lng      DOUBLE NOT NULL,
+                ne_lat      DOUBLE NOT NULL
+            )
+        """)
 
         # Migration: add geom column to existing tables that lack it
         cols = cur.execute(
@@ -134,6 +149,53 @@ def get_latest_positions(
         rows = self._cursor().execute(query, [after, limit]).fetchall()
         return [self._row_to_dict(r) for r in rows]
 
+    def get_latest_positions_with_trails(
+        self, trail_points: int = 6, max_gap_s: int = 120
+    ) -> list[dict]:
+        """Get the latest position for each vehicle plus a mini-trail of recent coords.
+
+        Positions separated by more than max_gap_s seconds are treated as a
+        discontinuity — the trail is truncated to only the contiguous segment
+        ending at the most recent position.
+        """
+        query = """
+            WITH ranked AS (
+                SELECT p.vehicle_id, p.timestamp, p.longitude, p.latitude,
+                       p.bearing, p.speed, p.is_driving,
+                       v.description, v.vehicle_type,
+                       ROW_NUMBER() OVER (PARTITION BY p.vehicle_id ORDER BY p.timestamp DESC) as rn
+                FROM positions p
+                JOIN vehicles v ON p.vehicle_id = v.vehicle_id
+            )
+            SELECT vehicle_id, timestamp, longitude, latitude, bearing, speed,
+                   is_driving, description, vehicle_type
+            FROM ranked
+            WHERE rn <= $1
+            ORDER BY vehicle_id, timestamp ASC
+        """
+        rows = self._cursor().execute(query, [trail_points]).fetchall()
+        all_dicts = [self._row_to_dict(r) for r in rows]
+
+        # Group by vehicle_id: last row is current position, all rows form trail.
+        # Walk backwards to find the contiguous segment (no gap > max_gap_s).
+        results = []
+        for _, group in groupby(all_dicts, key=lambda r: r["vehicle_id"]):
+            points = list(group)
+            # Find the start of the contiguous segment ending at the latest point
+            start = len(points) - 1
+            for i in range(len(points) - 1, 0, -1):
+                gap = (
+                    points[i]["timestamp"] - points[i - 1]["timestamp"]
+                ).total_seconds()
+                if gap > max_gap_s:
+                    break
+                start = i - 1
+            contiguous = points[start:]
+            current = contiguous[-1]  # most recent
+            current["trail"] = [[p["longitude"], p["latitude"]] for p in contiguous]
+            results.append(current)
+        return results
+
     def get_nearby_vehicles(
         self,
         lat: float,
@@ -225,72 +287,70 @@ def get_coverage_trails(
         self,
         since: datetime,
         until: datetime,
-        min_interval_s: float = 30.0,
-        max_gap_s: float = 120.0,
     ) -> list[dict]:
-        """Get per-vehicle LineString trails in a time range, downsampled and gap-split."""
+        """Get per-vehicle LineString trails in a time range.
+
+        Uses SQL-side gap detection (>120s breaks a segment) and
+        time_bucket downsampling (~1 point per 30s) to minimise
+        the number of rows transferred to Python.
+        """
         query = """
-            SELECT p.vehicle_id, p.timestamp, p.longitude, p.latitude,
-                   v.description, v.vehicle_type
-            FROM positions p
-            JOIN vehicles v ON p.vehicle_id = v.vehicle_id
-            WHERE p.timestamp >= $1
-            AND p.timestamp <= $2
-            ORDER BY p.vehicle_id, p.timestamp ASC
+            WITH with_gap AS (
+                SELECT
+                    p.vehicle_id,
+                    p.timestamp,
+                    p.longitude,
+                    p.latitude,
+                    v.description,
+                    v.vehicle_type,
+                    EPOCH(p.timestamp - LAG(p.timestamp) OVER (
+                        PARTITION BY p.vehicle_id ORDER BY p.timestamp
+                    )) AS gap_s
+                FROM positions p
+                JOIN vehicles v ON p.vehicle_id = v.vehicle_id
+                WHERE p.timestamp >= $1
+                AND p.timestamp <= $2
+            ),
+            with_segment AS (
+                SELECT *,
+                    SUM(CASE WHEN gap_s IS NULL OR gap_s > 120 THEN 1 ELSE 0 END)
+                        OVER (PARTITION BY vehicle_id ORDER BY timestamp) AS segment_id
+                FROM with_gap
+            ),
+            bucketed AS (
+                SELECT *,
+                    ROW_NUMBER() OVER (
+                        PARTITION BY vehicle_id, segment_id,
+                            time_bucket(INTERVAL '30 seconds', timestamp)
+                        ORDER BY timestamp
+                    ) AS bucket_rn
+                FROM with_segment
+            )
+            SELECT vehicle_id, segment_id, timestamp, longitude, latitude,
+                   description, vehicle_type
+            FROM bucketed
+            WHERE bucket_rn = 1
+            ORDER BY vehicle_id, segment_id, timestamp
         """
         rows = self._cursor().execute(query, [since, until]).fetchall()
 
         trails = []
-        for vid, group in groupby(rows, key=itemgetter(0)):
+        for (vid, seg_id), group in groupby(rows, key=lambda r: (r[0], r[1])):
             points = list(group)
             if len(points) < 2:
                 continue
-
-            # Downsample: keep first point, then skip until >= min_interval_s
-            sampled = [points[0]]
-            for pt in points[1:]:
-                elapsed = (pt[1] - sampled[-1][1]).total_seconds()
-                if elapsed >= min_interval_s:
-                    sampled.append(pt)
-            # Always include the last point
-            if sampled[-1] != points[-1]:
-                sampled.append(points[-1])
-
-            if len(sampled) < 2:
-                continue
-
-            # Split at gaps > max_gap_s
-            segments = []
-            current_segment = [sampled[0]]
-            for pt in sampled[1:]:
-                gap = (pt[1] - current_segment[-1][1]).total_seconds()
-                if gap > max_gap_s:
-                    if len(current_segment) >= 2:
-                        segments.append(current_segment)
-                    current_segment = [pt]
-                else:
-                    current_segment.append(pt)
-            if len(current_segment) >= 2:
-                segments.append(current_segment)
-
-            description = sampled[0][4]
-            vehicle_type = sampled[0][5]
-
-            for seg in segments:
-                trails.append(
-                    {
-                        "vehicle_id": vid,
-                        "description": description,
-                        "vehicle_type": vehicle_type,
-                        "coordinates": [[p[2], p[3]] for p in seg],
-                        "timestamps": [
-                            p[1].isoformat()
-                            if isinstance(p[1], datetime)
-                            else str(p[1])
-                            for p in seg
-                        ],
-                    }
-                )
+            trails.append(
+                {
+                    "vehicle_id": vid,
+                    "description": points[0][5],
+                    "vehicle_type": points[0][6],
+                    "coordinates": [[p[3], p[4]] for p in points],
+                    "timestamps": [
+                        p[2].isoformat() if isinstance(p[2], datetime) else str(p[2])
+                        for p in points
+                    ],
+                }
+            )
 
         return trails
 
@@ -339,5 +399,24 @@ def get_stats(self) -> dict:
                 result["latest"] = row[1]
         return result
 
+    def insert_viewport(
+        self,
+        zoom: float,
+        center_lng: float,
+        center_lat: float,
+        sw_lng: float,
+        sw_lat: float,
+        ne_lng: float,
+        ne_lat: float,
+    ):
+        """Record a user viewport focus event."""
+        self._cursor().execute(
+            """
+            INSERT INTO viewports (zoom, center_lng, center_lat, sw_lng, sw_lat, ne_lng, ne_lat)
+            VALUES (?, ?, ?, ?, ?, ?, ?)
+            """,
+            [zoom, center_lng, center_lat, sw_lng, sw_lat, ne_lng, ne_lat],
+        )
+
     def close(self):
         self.conn.close()

src/where_the_plow/db.py

@@ -25,9 +25,10 @@ async def lifespan(app: FastAPI):
     db = Database(settings.db_path)
     db.init()
     app.state.db = db
+    app.state.store = {}
     logger.info("Database initialized at %s", settings.db_path)
 
-    task = asyncio.create_task(collector.run(db))
+    task = asyncio.create_task(collector.run(db, app.state.store))
     yield
     task.cancel()
     try:

src/where_the_plow/main.py

@@ -27,6 +27,9 @@ class FeatureProperties(BaseModel):
     bearing: int | None = Field(None, description="Heading in degrees (0-360)")
     is_driving: str | None = Field(None, description="Driving status: 'maybe' or 'no'")
     timestamp: str = Field(..., description="Position timestamp (ISO 8601)")
+    trail: list[list[float]] | None = Field(
+        None, description="Recent trail coordinates [[lng, lat], ...]"
+    )
 
 
 class Feature(BaseModel):
@@ -80,6 +83,20 @@ class CoverageFeatureCollection(BaseModel):
     features: list[CoverageFeature]
 
 
+class ViewportTrack(BaseModel):
+    zoom: float = Field(..., description="Current map zoom level")
+    center: list[float] = Field(
+        ...,
+        description="[longitude, latitude] of map center",
+        min_length=2,
+        max_length=2,
+    )
+    bounds: dict = Field(
+        ...,
+        description="Map bounds: {sw: [lng, lat], ne: [lng, lat]}",
+    )
+
+
 class StatsResponse(BaseModel):
     total_positions: int = Field(..., description="Total position records stored")
     total_vehicles: int = Field(..., description="Total unique vehicles seen")