feat: archive CW-033 deployed sign map prototype v0.1.0

source/scripts/verify_snapshot.py

@@ -0,0 +1,21 @@
+#!/usr/bin/env python3
+"""Validate map coordinate, ID, service and shape integrity against the captured source inputs."""
+import argparse,collections,csv,io,json,pathlib,zipfile
+p=argparse.ArgumentParser();p.add_argument('--data',type=pathlib.Path,required=True);p.add_argument('--native',type=pathlib.Path,required=True);p.add_argument('--gtfs',type=pathlib.Path,required=True);a=p.parse_args();d=json.loads(a.data.read_text());n=json.loads(a.native.read_text());z=zipfile.ZipFile(a.gtfs)
+def rows(name):return list(csv.DictReader(io.TextIOWrapper(z.open(name+'.txt'),encoding='utf-8-sig')))
+stops={r['stop_id']:r for r in rows('stops')};trips={r['trip_id']:r for r in rows('trips')};routes={r['route_id']:r for r in rows('routes')};active=set(d['meta']['active_services']);sr=collections.defaultdict(set);sh=collections.defaultdict(list);rs=collections.defaultdict(set)
+for r in rows('stop_times'):
+ if trips[r['trip_id']]['service_id'] in active:sr[r['stop_id']].add(trips[r['trip_id']]['route_id'])
+for t in trips.values():
+ if t['service_id'] in active and t['shape_id']:rs[t['route_id']].add(t['shape_id'])
+for r in rows('shapes'):sh[r['shape_id']].append((int(r['shape_pt_sequence']),float(r['shape_pt_lat']),float(r['shape_pt_lon'])))
+for player in d['players']:
+ assert player['resolved_stop_ids']==[str(n['stops'][sid]['SrcStopID']) for sid in player['source_ids']]
+ assert set(player['routes'])==(set().union(*(sr[s] for s in player['resolved_stop_ids'])) if player['source_ids'] else set())
+ if player['kind']=='kiosk':assert all(stops[s]['parent_station']==player['candidate_ids'][0] for s in player['resolved_stop_ids'])
+for sid,s in d['stops'].items():assert [s['lat'],s['lon']]==[float(stops[sid]['stop_lat']),float(stops[sid]['stop_lon'])]
+for r in d['routes']:
+ assert r['color']=='#'+routes[r['id']]['route_color'] and set(r['shape_ids'])==rs[r['id']]
+ for sid,line in zip(r['shape_ids'],r['lines']):assert line==[[lat,lon] for _,lat,lon in sorted(sh[sid])]
+assert len(d['audit'])==sum(len(p['source_ids']) for p in d['players'])
+print('PASS: native ID joins, GTFS coordinates, mapped-bay routes, active-service geometry and route colors')