144 lines
5.1 KiB
Python
Executable File
144 lines
5.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""新北市垃圾車提醒 CLI。純 stdlib，無外部相依。
|
||
|
||
用法:
|
||
./garbage.py find <關鍵字> # 搜尋清運點，取得 lineid 與經緯度
|
||
./garbage.py watch --lat 25.01 --lng 121.52 --line 234007 [--radius 300] [--interval 120]
|
||
|
||
find 幫你從表定路線挑到你家最近的清運點，watch 定時比對出勤車輛距離、靠近就提醒。
|
||
"""
|
||
import argparse
|
||
import json
|
||
import math
|
||
import ssl
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
import urllib.request
|
||
|
||
# ponytail: 政府伺服器憑證缺 Subject Key Identifier，Python 3.14 嚴格模式會拒絕。
|
||
# 這是公開、唯讀、免驗證的開放資料，關閉憑證驗證風險僅止於被餵假座標，可接受。
|
||
_CTX = ssl._create_unverified_context()
|
||
|
||
LIVE = "https://data.ntpc.gov.tw/api/datasets/28ab4122-60e1-4065-98e5-abccb69aaca6/json"
|
||
ROUTES = "https://data.ntpc.gov.tw/api/datasets/edc3ad26-8ae7-4916-a00b-bc6048d19bf8/json"
|
||
|
||
|
||
def fetch(url, size=1000, all_pages=False):
|
||
out, page = [], 0
|
||
while True:
|
||
u = f"{url}?page={page}&size={size}"
|
||
with urllib.request.urlopen(u, timeout=30, context=_CTX) as r:
|
||
batch = json.load(r)
|
||
out += batch
|
||
if not all_pages or len(batch) < size:
|
||
return out
|
||
page += 1
|
||
|
||
|
||
def haversine(lat1, lng1, lat2, lng2):
|
||
"""兩點間公尺距離。"""
|
||
R = 6371000
|
||
p1, p2 = math.radians(lat1), math.radians(lat2)
|
||
dp = math.radians(lat2 - lat1)
|
||
dl = math.radians(lng2 - lng1)
|
||
a = math.sin(dp / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dl / 2) ** 2
|
||
return 2 * R * math.asin(math.sqrt(a))
|
||
|
||
|
||
def cmd_find(args):
|
||
kw = args.keyword
|
||
hits = [
|
||
p for p in fetch(ROUTES, all_pages=True)
|
||
if kw in p["name"] or kw in p["village"] or kw in p["linename"] or kw in p["city"]
|
||
]
|
||
if not hits:
|
||
print(f"找不到含「{kw}」的清運點")
|
||
return
|
||
for p in hits:
|
||
print(f'lineid={p["lineid"]:>7} rank={p["rank"]:>3} {p["time"]} '
|
||
f'{p["city"]}{p["village"]} {p["name"]} '
|
||
f'({p["latitude"]},{p["longitude"]})')
|
||
print(f"\n共 {len(hits)} 筆。挑一筆，把它的 lineid 和經緯度餵給 watch。")
|
||
|
||
|
||
def notify(title, msg):
|
||
print(f"\a🚛 {title}: {msg}") # \a 響鈴
|
||
# ponytail: macOS 原生通知；非 macOS 就只有上面那行
|
||
try:
|
||
subprocess.run(
|
||
["osascript", "-e", f'display notification "{msg}" with title "{title}"'],
|
||
check=False,
|
||
)
|
||
except FileNotFoundError:
|
||
pass
|
||
|
||
|
||
def cmd_watch(args):
|
||
alerted = False
|
||
while True:
|
||
try:
|
||
cars = [c for c in fetch(LIVE, size=200)
|
||
if not args.line or c["lineid"] == args.line]
|
||
except Exception as e:
|
||
print(f"抓取失敗，稍後重試：{e}", file=sys.stderr)
|
||
time.sleep(args.interval)
|
||
continue
|
||
|
||
if not cars:
|
||
print(f"[{time.strftime('%H:%M:%S')}] 目前沒有出勤車輛（離峰或還沒發車）")
|
||
else:
|
||
nearest = min(
|
||
cars,
|
||
key=lambda c: haversine(args.lat, args.lng,
|
||
float(c["latitude"]), float(c["longitude"])),
|
||
)
|
||
d = haversine(args.lat, args.lng,
|
||
float(nearest["latitude"]), float(nearest["longitude"]))
|
||
print(f"[{time.strftime('%H:%M:%S')}] 最近車輛 {nearest['car']} "
|
||
f"距離 {d:.0f}m（{nearest['location']}，定位 {nearest['time']}）")
|
||
if d <= args.radius and not alerted:
|
||
notify("垃圾車快到了", f"約 {d:.0f} 公尺，{nearest['location']}")
|
||
alerted = True
|
||
elif d > args.radius * 1.5:
|
||
alerted = False # 車開遠了，重置以便下一趟再提醒
|
||
|
||
time.sleep(args.interval)
|
||
|
||
|
||
def demo():
|
||
# 台北車站 ↔ 永和保生路，約 4.3 公里
|
||
d = haversine(25.0478, 121.5170, 25.010365, 121.504743)
|
||
assert 4000 < d < 4700, d
|
||
assert haversine(25.0, 121.0, 25.0, 121.0) == 0
|
||
print("ok")
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(description="新北市垃圾車提醒")
|
||
sub = ap.add_subparsers(dest="cmd", required=True)
|
||
|
||
f = sub.add_parser("find", help="搜尋清運點")
|
||
f.add_argument("keyword", help="路線/里/點名/行政區關鍵字")
|
||
f.set_defaults(func=cmd_find)
|
||
|
||
w = sub.add_parser("watch", help="監看並提醒")
|
||
w.add_argument("--lat", type=float, required=True, help="你的清運點緯度")
|
||
w.add_argument("--lng", type=float, required=True, help="你的清運點經度")
|
||
w.add_argument("--line", help="只看這個 lineid（強烈建議填）")
|
||
w.add_argument("--radius", type=float, default=300, help="提醒距離公尺，預設 300")
|
||
w.add_argument("--interval", type=int, default=120, help="輪詢秒數，預設 120")
|
||
w.set_defaults(func=cmd_watch)
|
||
|
||
sub.add_parser("demo", help="自我測試").set_defaults(func=lambda a: demo())
|
||
|
||
args = ap.parse_args()
|
||
args.func(args)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
main()
|
||
except KeyboardInterrupt:
|
||
print("\n收工")
|