From 96f918445c639d1987101f9c245a934b7be4b0b6 Mon Sep 17 00:00:00 2001
From: Piotr Oleszczyk <piotr@oleszczyk.eu>
Date: Mon, 12 May 2025 14:46:33 +0200
Subject: [PATCH] correct daily snapshot: flatten stops, add fetch date

---
 src/ptscrapper/scheduler.py | 18 +++++++++++++++++-
 src/ptscrapper/storage.py   | 34 ++++++++++++++++++++++++++++++----
 2 files changed, 47 insertions(+), 5 deletions(-)

diff --git a/src/ptscrapper/scheduler.py b/src/ptscrapper/scheduler.py
index 9ff59bf..893694e 100644
--- a/src/ptscrapper/scheduler.py
+++ b/src/ptscrapper/scheduler.py
@@ -196,7 +196,23 @@ async def daily_snapshot_job():
     try:
         # stops
         posts = await fetch_posts(client)
-        table = pa.Table.from_pylist(posts)
+        rows = []
+        for grp in posts:
+            # grp: { "n": name, "s": group_id, "t": group_type, "p": [ {s,x,y,t}, … ] }
+            for stop in grp.get("p", []):
+                rows.append(
+                    {
+                        "fetch_date": ts,  # utc fetch timestamp
+                        "group_name": grp["n"],
+                        "group_id": grp["s"],
+                        "group_type": grp["t"],
+                        "stop_id": stop["s"],
+                        "stop_type": stop["t"],
+                        "x": stop["x"],
+                        "y": stop["y"],
+                    }
+                )
+        table = pa.Table.from_pylist(rows)
         fp = write_stops(table, ts)
         push_to_nas(fp)
 
diff --git a/src/ptscrapper/storage.py b/src/ptscrapper/storage.py
index f5ae915..3691fc0 100644
--- a/src/ptscrapper/storage.py
+++ b/src/ptscrapper/storage.py
@@ -78,21 +78,47 @@ def write_course_posts(table: pa.Table, course_id: int, ts: datetime) -> str:
 
 def write_stops(table: pa.Table, snapshot_dt: datetime) -> str:
     """
-    Write the daily snapshot of stops (getPosts).
+    Take a flat table with columns
+      fetch_date, group_name, group_id, group_type,
+      stop_id, stop_type, x, y
+    and convert x/y → a WKB Point geometry,
+    drop the raw coords, and write GeoParquet.
     """
+    # 1) pull x,y into numpy
+    xs = table.column("x").to_numpy()
+    ys = table.column("y").to_numpy()
+
+    # 2) build PointType with CRS
+    pt_builder = ga.point().with_crs("EPSG:4326")
+
+    # 3) build WKB geometry
+    geom = ga.as_wkb(pt_builder.from_geobuffers(None, xs, ys))
+
+    # 4) drop coords & append geometry column
+    table = table.drop(["x", "y"]).append_column("geometry", geom)
+
+    # 5) write out
     subdir = snapshot_dt.strftime("stops/%Y-%m-%d")
     local_dir = os.path.join(LOCAL_DATA_DIR, subdir)
     _ensure_dir(local_dir)
 
-    local_path = os.path.join(local_dir, f"stops_{HOST_ID}.parquet")
-    pq.write_table(table, local_path, compression="snappy")
-    return local_path
+    filename = f"stops_{HOST_ID}.parquet"
+    out_path = os.path.join(local_dir, filename)
+    pq.write_table(table, out_path, compression="snappy")
+    return out_path
 
 
 def write_posts_lines(table: pa.Table, snapshot_dt: datetime) -> str:
     """
     Write the daily stop→lines mapping (getPostsLines).
     """
+    # ensure every row gets a fetch_date
+    if "fetch_date" not in table.column_names:
+        table = table.append_column(
+            "fetch_date",
+            pa.array([snapshot_dt] * table.num_rows, type=pa.timestamp("ns", tz="UTC")),
+        )
+
     subdir = snapshot_dt.strftime("stops_lines/%Y-%m-%d")
     local_dir = os.path.join(LOCAL_DATA_DIR, subdir)
     _ensure_dir(local_dir)