diff --git a/automate.py b/automate.py index f3c309e..12a4540 100644 --- a/automate.py +++ b/automate.py @@ -3,10 +3,12 @@ 萤石录像自动下载脚本 通过 ADB 自动化模拟器中的萤石 App，下载前一天的所有报警录像。 """ -import subprocess, time, os, shutil, struct, zlib +import subprocess, time, os, shutil, struct, zlib, io, hashlib, math from datetime import datetime, timedelta from pathlib import Path +from PIL import Image + ADB = os.path.expanduser("~/Library/Android/sdk/platform-tools/adb") DEVICE = "emulator-5554" OUTPUT_DIR = Path("./recordings") @@ -53,7 +55,7 @@ BLUE_THRESHOLD = 500 # 窗口内强蓝像素数 >= 此值判为已打勾（打 # 灰占位是纯色 std≈0，据此切出"纹理带"，每个带的竖直中心就是一行卡片的点击中心。 ROW_Y0, ROW_Y1 = 1000, 2300 # 列表纵向扫描范围（避开顶部状态栏/预览/标题栏、底部手势条） ROW_STD_X0, ROW_STD_W = 350, 700 # 横向 std 采样区（避开左右圆角/阴影） -ROW_STD_COLS = 175 # 降采样列数（一次 magick 折成每行 175 个样本算 std，避免逐行上百次 magick） +ROW_STD_COLS = 175 # 降采样列数（Pillow 一次折成每行 175 个样本算 std，避免逐行上百次采样） ROW_STD_LOW = 28 # std 阈值：>= 判为卡片纹理，< 判为间隙/标签/占位（实测卡片≥46、非卡片=0，余量足） ROW_MIN_H = 90 # 纹理带最小高度（滤掉分组标签、半截卡片、噪声） ROW_MAX_H = 260 # 纹理带最大高度（超过视为两行粘连，按行距二次切分） @@ -204,18 +206,10 @@ def tap_safe(x, y, expect_activity_keyword, hint="", delay=1.5): def sample_pixel_rgb(x, y, png_path="/tmp/_pixchk.png"): - """截图并采样指定坐标 RGB。需要系统装了 ImageMagick (`magick`)。""" - import re + """截图并采样指定坐标 RGB。用 Pillow 实现，不依赖系统 ImageMagick。""" adb("shell", "screencap", "-p", "/sdcard/_pixchk.png") adb("pull", "/sdcard/_pixchk.png", png_path) - out = subprocess.run( - ["magick", png_path, "-format", "%[pixel:p{" + str(x) + "," + str(y) + "}]", "info:"], - capture_output=True, text=True, - ) - m = re.match(r"srgba?\((\d+),(\d+),(\d+)", out.stdout.strip()) - if not m: - return (0, 0, 0) - return tuple(map(int, m.groups())) + return sample_rgb_from_file(png_path, x, y) def screenshot(path="/tmp/ezviz_screen.png"): @@ -228,35 +222,35 @@ def screenshot(path="/tmp/ezviz_screen.png"): def sample_rgb_from_file(png_path, x, y): - """从已经拉取到本地的截图文件采样指定坐标 RGB（不重新截图，供批量取点复用）。""" - import re - out = subprocess.run( - ["magick", png_path, "-format", "%[pixel:p{" + str(x) + "," + str(y) + "}]", "info:"], - capture_output=True, text=True, - ) - m = re.match(r"srgba?\((\d+),(\d+),(\d+)", out.stdout.strip()) - if not m: + """从已经拉取到本地的截图文件采样指定坐标 RGB（不重新截图，供批量取点复用）。 + 用 Pillow 实现，不依赖系统 ImageMagick。""" + try: + with Image.open(png_path) as img: + img = img.convert("RGB") + r, g, b = img.getpixel((x, y)) + return (r, g, b) + except Exception: return (0, 0, 0) - return tuple(map(int, m.groups())) def count_blue_pixels(png_path, cx, cy): """统计以 (cx,cy) 为中心、大小 BLUE_WIN 的窗口内"强蓝"像素数量。 强蓝判据：b>150 且 b-r>80（蓝勾角标的高饱和蓝，灰/黑/彩色画面几乎不出现）。 - 用窗口统计而非单点采样，是为了吸收卡片垂直位置的漂移（见 BADGE_OFFSET 说明）。""" - import re + 用窗口统计而非单点采样，是为了吸收卡片垂直位置的漂移（见 BADGE_OFFSET 说明）。 + 用 Pillow 实现，不依赖系统 ImageMagick。""" w, h = BLUE_WIN x0, y0 = cx - w // 2, cy - h // 2 - out = subprocess.run( - ["magick", png_path, "-crop", f"{w}x{h}+{x0}+{y0}", "txt:-"], - capture_output=True, text=True, - ) - cnt = 0 - for m in re.finditer(r"srgba?\((\d+),(\d+),(\d+)", out.stdout): - r, g, b = map(int, m.groups()) - if b > 150 and b - r > 80: - cnt += 1 - return cnt + try: + with Image.open(png_path) as img: + img = img.convert("RGB") + crop = img.crop((x0, y0, x0 + w, y0 + h)) + cnt = 0 + for r, g, b in crop.getdata(): + if b > 150 and b - r > 80: + cnt += 1 + return cnt + except Exception: + return 0 def get_checked_cells(png_path): @@ -277,29 +271,27 @@ def std_profile(png_path): 卡片间水平间隙、整点分组标签、灰色占位图都是纯色 → std≈0。据此可把"卡片行" 从"间隙/标签/占位"里分出来。 - 性能：不逐行调 magick（一屏上百次太慢），而是用一次 magick 把 ROW_STD_W 宽的 - 采样区灰度图横向降采样到 ROW_STD_COLS 列（每列是原若干像素的平均），导出像素文本， - 再在 Python 里按行算这 ROW_STD_COLS 个样本的标准差。实测降采样法的低谷/高峰位置 - 与逐行 %[fx:standard_deviation] 完全一致，且对比度更大（间隙 0 vs 卡片 60+）。""" - import re, math + 性能：不逐行采样（一屏上百次太慢），而是用 Pillow 把 ROW_STD_W 宽的采样区灰度图 + 横向降采样到 ROW_STD_COLS 列（每列是原若干像素的平均），再在 Python 里按行算这 + ROW_STD_COLS 个样本的标准差。用 Pillow 实现，不依赖系统 ImageMagick。""" h = ROW_Y1 - ROW_Y0 - out = subprocess.run( - ["magick", png_path, "-crop", f"{ROW_STD_W}x{h}+{ROW_STD_X0}+{ROW_Y0}", "+repage", - "-colorspace", "Gray", "-resize", f"{ROW_STD_COLS}x{h}!", "txt:-"], - capture_output=True, text=True, - ) - rows = {} - # txt 行形如 "x,y: (gray,alpha) ..."，灰度值取第一个数 - for m in re.finditer(r"^\d+,(\d+): \((\d+)", out.stdout, re.M): - rr, val = int(m.group(1)), int(m.group(2)) - rows.setdefault(rr, []).append(val) - prof = [] - for rr in sorted(rows): - vals = rows[rr] - mean = sum(vals) / len(vals) - var = sum((v - mean) ** 2 for v in vals) / len(vals) - prof.append((ROW_Y0 + rr, math.sqrt(var))) - return prof + try: + with Image.open(png_path) as img: + img = img.convert("L") + crop = img.crop((ROW_STD_X0, ROW_Y0, ROW_STD_X0 + ROW_STD_W, ROW_Y0 + h)) + small = crop.resize((ROW_STD_COLS, h), Image.Resampling.NEAREST) + pixels = list(small.getdata()) + prof = [] + for rr in range(h): + row_vals = pixels[rr * ROW_STD_COLS:(rr + 1) * ROW_STD_COLS] + if not row_vals: + continue + mean = sum(row_vals) / len(row_vals) + var = sum((v - mean) ** 2 for v in row_vals) / len(row_vals) + prof.append((ROW_Y0 + rr, math.sqrt(var))) + return prof + except Exception: + return [] def detect_card_rows(png_path): @@ -412,11 +404,11 @@ def date_label_hash(png_path="/tmp/_datechk.png"): import hashlib adb("shell", "screencap", "-p", "/sdcard/_datechk.png") adb("pull", "/sdcard/_datechk.png", png_path) - crop_path = png_path.replace(".png", "_crop.png") - subprocess.run(["magick", png_path, "-crop", "400x90+40+1080", crop_path], - capture_output=True) - with open(crop_path, "rb") as f: - return hashlib.md5(f.read()).hexdigest() + with Image.open(png_path) as img: + crop = img.crop((40, 1080, 440, 1170)) # 400x90+40+1080 + buf = io.BytesIO() + crop.save(buf, format="PNG") + return hashlib.md5(buf.getvalue()).hexdigest() def tap_and_check_date_changed(arrow_coord, timeout=12, interval=1.0): @@ -523,12 +515,11 @@ def screenshot_hash(png_path="/tmp/_scrollchk.png"): import hashlib adb("shell", "screencap", "-p", "/sdcard/_scrollchk.png") adb("pull", "/sdcard/_scrollchk.png", png_path) - crop_path = png_path.replace(".png", "_crop.png") - # 缩略图网格区域：y=1000 到屏幕底部，避开上方标题/预览/状态栏（Pixel 6 模拟器 1080x2400） - subprocess.run(["magick", png_path, "-crop", "1080x1400+0+1000", crop_path], - capture_output=True) - with open(crop_path, "rb") as f: - return hashlib.md5(f.read()).hexdigest() + with Image.open(png_path) as img: + crop = img.crop((0, 1000, 1080, 2400)) # 1080x1400+0+1000 + buf = io.BytesIO() + crop.save(buf, format="PNG") + return hashlib.md5(buf.getvalue()).hexdigest() def dismiss_dialog_if_any(): @@ -699,11 +690,18 @@ def tap_all_thumbnails(baseline, max_rounds=300, poll_interval=4, max_wait_per_r # 的总根源），再对每行 × 固定 3 列逐格判打勾，只点还没下载完的。 grid_shot = screenshot("/tmp/_gridchk.png") full_rows = detect_card_rows(grid_shot) - if not full_rows: - # 偶发一张都没检出（仍在加载/异常画面），重拍一次 - time.sleep(1.5) + # 检测不到行时重试最多3次：缩略图加载慢、ANR刚恢复、画面切换中都会导致 + # 临时检测不到行。之前只重试1次就放弃，会误判"全部已打勾"直接滚动， + # 漏掉当前屏所有卡片（这是漏下载的重要原因之一）。 + retry = 0 + while not full_rows and retry < 3: + retry += 1 + time.sleep(2.0 + retry * 1.0) # 逐次加长等待，给缩略图加载留时间 grid_shot = screenshot("/tmp/_gridchk.png") full_rows = detect_card_rows(grid_shot) + if not full_rows: + print(f" 轮 {r+1}: ⚠ 连续 {retry+1} 次未检出卡片行，跳过本轮滚动（不盲点）") + continue # 每轮最多处理最上面 3 行（9 格）。萤石同时下载上限约 10 个，一屏可见 5~6 行， # 若把检出的所有行都点了（12~14 个）会超并发：多点的卡片不进下载队列、滚上来 # 仍未打勾，被反复重点，实测会让"跳过已打勾"退化成 0、重复率回升。多出的行 @@ -751,10 +749,29 @@ def tap_all_thumbnails(baseline, max_rounds=300, poll_interval=4, max_wait_per_r done = len(tapped_cells) - len(pending) total = get_remote_new_file_count(baseline) if pending: - # 个别卡片下载失败/超时一直不打勾，不能无限等，照常滚动。这几张 - # 下一轮可能被重复点一次，靠下游 analyze.py 的 MD5 查重兜底。 + # 超时未打勾的卡片，先重试点击一次：可能是点击落在卡片间隙/分组标签上 + # 没生效（下载根本没触发），而不是下载慢。之前直接滚动会导致这些卡片 + # 永远漏下载。重试后仍超时才滚动，靠下游 MD5 查重兜底重复。 print(f" 轮 {r+1}: 等待 {waited}s 超时，{done}/{len(tapped_cells)} 个已打勾，" - f"仍有 {len(pending)} 个未完成，照常滚动（这几张可能重复，靠下游去重兜底）") + f"仍有 {len(pending)} 个未完成，重试点击未完成的卡片...") + for (gx, cy) in pending: + adb("shell", "input", "tap", str(gx), str(cy)) + time.sleep(0.5) + # 重试后再等一轮 + retry_waited = 0 + while retry_waited < max_wait_per_round: + time.sleep(poll_interval) + retry_waited += poll_interval + ensure_not_anr() + retry_shot = screenshot("/tmp/_retrywait.png") + retry_chk = get_checked_cells_dynamic(retry_shot, row_centers) + still_pending = [c for c in pending if not retry_chk.get(c)] + if not still_pending: + break + if still_pending: + print(f" 轮 {r+1}: 重试后仍有 {len(still_pending)} 个未完成，照常滚动（靠下游去重兜底）") + else: + print(f" 轮 {r+1}: 重试后全部打勾完成") else: print(f" 轮 {r+1}: {len(tapped_cells)} 个全部打勾完成（等待 {waited}s），累计新下载 {total} 个，开始滚动") else: @@ -800,6 +817,17 @@ def tap_all_thumbnails(baseline, max_rounds=300, poll_interval=4, max_wait_per_r time.sleep(THUMB_LOAD_WAIT) sshot = screenshot("/tmp/_sweep.png") srows = detect_card_rows(sshot) + # 清扫时检测不到行要重试：底部缩略图可能还在加载，直接判"无未下载" + # 会提前结束、漏掉底部最后几个卡片。 + sweep_retry = 0 + while not srows and sweep_retry < 3: + sweep_retry += 1 + time.sleep(2.0) + sshot = screenshot("/tmp/_sweep.png") + srows = detect_card_rows(sshot) + if not srows: + print(f" → 清扫第 {sweep+1} 批：连续 {sweep_retry+1} 次未检出卡片行，视为底部已无内容") + break schk = get_checked_cells_dynamic(sshot, srows) todo = [(gx, cy) for cy in srows for gx in THUMB_X if not schk.get((gx, cy))] if not todo: @@ -957,27 +985,33 @@ def sync_recent_days(n_days=7): for d in range(n_days): rec_date = (datetime.now() - timedelta(days=d)).strftime("%Y%m%d") print(f"\n===== [{d+1}/{n_days}] 录像日期 {rec_date}（今天往前第 {d} 天）=====") - try: - # 清空本地该天旧目录：重下的文件名是新的下载时刻，和旧文件名不同， - # 不清会产生"内容相同、文件名不同"的重复副本。SD 卡是真源，重下可恢复。 - day_dir = OUTPUT_DIR / "BH5455783" / rec_date - if day_dir.exists(): - shutil.rmtree(day_dir) - print(f" 已清空本地旧目录 {day_dir}（覆盖式重下）") - launch_app() - clear_device_recordings() # 下载前清空，保证有空间、且设备只剩这天的 - navigate_to_yesterday_batch(days_back=d) - tap_all_thumbnails(set()) - wait_for_downloads(set()) - day_files = get_all_remote_files() # 这一天下载的全部录像 - if day_files: - pull_files(list(day_files), date_str=rec_date) - else: - print(f"→ {rec_date} 没有下载到文件") - clear_device_recordings() # 拉完即时清空，释放空间给下一天 - except Exception as e: - print(f"⚠ {rec_date} 处理失败，跳过：{e}") - continue + for attempt in range(2): # 每天最多重试1次（共2次尝试），避免偶发ANR/网络问题导致整天漏下载 + try: + # 清空本地该天旧目录：重下的文件名是新的下载时刻，和旧文件名不同， + # 不清会产生"内容相同、文件名不同"的重复副本。SD 卡是真源，重下可恢复。 + day_dir = OUTPUT_DIR / "BH5455783" / rec_date + if day_dir.exists(): + shutil.rmtree(day_dir) + print(f" 已清空本地旧目录 {day_dir}（覆盖式重下）") + launch_app() + clear_device_recordings() # 下载前清空，保证有空间、且设备只剩这天的 + navigate_to_yesterday_batch(days_back=d) + tap_all_thumbnails(set()) + wait_for_downloads(set()) + day_files = get_all_remote_files() # 这一天下载的全部录像 + if day_files: + pull_files(list(day_files), date_str=rec_date) + else: + print(f"→ {rec_date} 没有下载到文件") + clear_device_recordings() # 拉完即时清空，释放空间给下一天 + break # 成功则跳出重试循环 + except Exception as e: + print(f"⚠ {rec_date} 第 {attempt+1} 次尝试失败：{e}") + if attempt == 0: + print(f" 重试一次...") + time.sleep(5) + else: + print(f" {rec_date} 两次尝试均失败，跳过该天") print(f"\n===== 最近 {n_days} 天下载完成 =====")