新北市垃圾車提醒 CLI
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
143
garbage.py
Executable file
143
garbage.py
Executable file
@@ -0,0 +1,143 @@
|
||||
#!/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收工")
|
||||
Reference in New Issue
Block a user