"""Query monthly polygon event totals using the World V2.0-preview API. Usage: NOSIBLE_API_KEY=... python query-monthly-chokepoints.py \ --start 2025-10 --end 2026-09 --output oil-chokepoints-monthly.json Place oil-chokepoints-named.geojson alongside this script. No event text or credentials are written to the output. Each request uses limit=1 and reads `total`, which counts all matching events rather than just that first page. """ import argparse import calendar from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timezone import hashlib import json import os from pathlib import Path import time import urllib.error import urllib.request PUBLIC_ENDPOINT = 'https://nosible.world/api/v2.0-preview/polygon/events' def month_range(start, end): year, month = map(int, start.split('-')) while f'{year:04}-{month:02}' <= end: yield f'{year:04}-{month:02}', f'{year:04}-{month:02}-01', f'{year:04}-{month:02}-{calendar.monthrange(year, month)[1]}' year, month = (year + 1, 1) if month == 12 else (year, month + 1) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--start', required=True, help='First calendar month, YYYY-MM') parser.add_argument('--end', required=True, help='Last calendar month, YYYY-MM, inclusive') parser.add_argument('--output', type=Path, default=Path('oil-chokepoints-monthly.json')) args = parser.parse_args() for value in [args.start, args.end]: assert datetime.strptime(value, '%Y-%m').strftime('%Y-%m') == value assert args.start <= args.end key = os.environ['NOSIBLE_API_KEY'] endpoint = os.environ.get('NOSIBLE_POLYGON_EVENTS_URL', PUBLIC_ENDPOINT) named = json.loads((Path(__file__).parent / 'oil-chokepoints-named.geojson').read_text()) ids = ['hormuz', 'malacca', 'suez', 'little-belt', 'great-belt', 'oresund', 'bab'] polygons = [] for polygon_id, feature in zip(ids, named['features'], strict=True): polygons.append({'id': polygon_id, 'areaId': feature['properties']['chokepoint_id'], 'name': feature['properties']['name'], 'geometry': feature['geometry']}) areas = [] for polygon in polygons: if any(a['id'] == polygon['areaId'] for a in areas): continue parts = [p for p in polygons if p['areaId'] == polygon['areaId']] areas.append({'id': polygon['areaId'], 'polygonIds': [p['id'] for p in parts]}) geometries = {p['id']: p['geometry'] for p in polygons} danish = [p for p in polygons if p['areaId'] == 'danish'] geometries['danish'] = {'type': 'MultiPolygon', 'coordinates': [p['geometry']['coordinates'] for p in danish]} geometries['combined'] = {'type': 'MultiPolygon', 'coordinates': [p['geometry']['coordinates'] for p in polygons]} months = list(month_range(args.start, args.end)) checkpoint = args.output.with_suffix('.receipts.json') cache = json.loads(checkpoint.read_text()) if checkpoint.exists() else [] receipts = {(r['month'], r['queryId']): r for r in cache} def query(month, start, end, query_id, geometry): geometry_hash = hashlib.sha256(json.dumps(geometry, sort_keys=True).encode()).hexdigest() existing = receipts.get((month, query_id)) if existing and existing['geometrySha256'] == geometry_hash and existing['dateWindow']['from'] == start and existing['dateWindow']['to'] == end: return existing body = {'polygon': geometry, 'from': start, 'to': end, 'limit': 1, 'view': 'lite', 'order': 'asc'} for attempt in range(4): req = urllib.request.Request(endpoint, data=json.dumps(body).encode(), headers={ 'Authorization': 'Bearer ' + key, 'Content-Type': 'application/json'}) try: with urllib.request.urlopen(req, timeout=90) as response: result = json.load(response) break except urllib.error.HTTPError as error: if error.code not in [429, 502, 503, 504] or attempt == 3: raise RuntimeError(f'{month} {query_id}: HTTP {error.code}') from None time.sleep(min(15, 2 ** attempt)) assert result['schema'] == 'nosible_world_polygon_events_v1' assert result['date_window']['from'] == start and result['date_window']['to'] == end, 'API changed the requested date range' assert result.get('hydration_misses', 0) == 0 assert result['count'] == min(result['total'], 1) if result['events']: assert start <= result['events'][0]['event']['date'][:10] <= end return {'month': month, 'queryId': query_id, 'total': result['total'], 'geometrySha256': geometry_hash, 'dateWindow': result['date_window'], 'indexBuild': result['as_of']['index_build'], 'hydrationMisses': result.get('hydration_misses', 0), 'queriedAt': datetime.now(timezone.utc).isoformat(), 'tookMs': result['took_ms']} with ThreadPoolExecutor(max_workers=2) as pool: tasks = [pool.submit(query, month, start, end, query_id, geometry) for month, start, end in months for query_id, geometry in geometries.items()] for task in as_completed(tasks): receipt = task.result() receipts[(receipt['month'], receipt['queryId'])] = receipt checkpoint.write_text(json.dumps(list(receipts.values()), indent=2) + '\n') print(f"{len(receipts)}/{len(tasks)} {receipt['month']} {receipt['queryId']}: {receipt['total']}", flush=True) used = [receipts[(m, q)] for m, _, _ in months for q in geometries] builds = {r['indexBuild'] for r in used} assert len(builds) == 1, 'Archive generation changed; remove the checkpoint and repeat the collection' rows = [] for month, start, end in months: counts = {p['id']: receipts[(month, p['id'])]['total'] for p in polygons} grouped = {a['id']: receipts[(month, a['id'])]['total'] for a in areas} combined = receipts[(month, 'combined')]['total'] assert max(counts[p['id']] for p in danish) <= grouped['danish'] <= sum(counts[p['id']] for p in danish) assert max(grouped.values()) <= combined <= sum(grouped.values()) rows.append({'month': month, 'from': start, 'to': end, 'polygons': counts, 'chokepoints': grouped, 'combinedUnique': combined}) output = {'schema': 'nosible_chokepoint_monthly_events_v1', 'from': months[0][1], 'to': months[-1][2], 'endpoint': PUBLIC_ENDPOINT, 'collectedAt': datetime.now(timezone.utc).isoformat(), 'archiveBuild': next(iter(builds)), 'archiveCutoff': used[0]['dateWindow']['archive_cutoff'], 'countBasis': 'Distinct matching events per geometry per calendar month; any stored location; all topics.', 'queryCount': len(used), 'polygons': polygons, 'areas': areas, 'months': rows, 'receipts': used} args.output.write_text(json.dumps(output, indent=2) + '\n') print('Saved', args.output, flush=True) if __name__ == '__main__': main()