probe

#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 验证工具 —— SenseCraft HMI 的 Image/HTML 控件对**同一 URL** 是否每个刷新周期都重新抓取。原理：本地起一个极简 HTTP 服务，固定路径 /probe.png 每 --every 秒重绘一张 800x480 探针图（大号生成时刻 HH:MM:SS + 递增序号 + 六色校准块），响应头 Cache-Control: no-store；每次有人请求就在终端打一行日志（时间、路径、User-Agent、来源 IP，含经隧道透传的 CF-Connecting-IP）。判定： - 终端日志每个设备刷新周期都出现一次 /probe.png 请求，且设备画面上的时刻/序号随之推进 → **每周期重抓，方案二成立**；此后只要不断覆盖同一 URL 指向的文件即可，无需再经 HTML 页。 - 日志只在「部署」那一刻出现一次、之后再无请求 → 平台只在部署时抓一次并缓存，方案二不成立。 - 也可同时把 /index.html（HTML 控件方案）放到另一台画布对照。用法： python3 probe_server.py [--port 8787] [--every 60] # 另开终端暴露成公网 HTTPS（云端抓不到局域网/明文 HTTP）。cloudflared 快速隧道无需账号： brew install cloudflared cloudflared tunnel --url http://localhost:8787 # 把打印出的 https://xxxx.trycloudflare.com/probe.png 填入 SenseCraft 画布的 Image 控件（铺满 800x480）→ 部署。 """ import argparse import io import threading import time from datetime import datetime from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from PIL import Image, ImageDraw, ImageFont import config BLACK, WHITE, RED = (0, 0, 0), (255, 255, 255), (255, 0, 0) SIX = [BLACK, WHITE, (255, 255, 0), RED, (0, 255, 0), (0, 0, 255)] _state = {"png": b"", "seq": 0, "stamp": ""} _lock = threading.Lock() def render_probe(seq, now): """800x480 探针图：序号（左上）/ 大号时刻（居中，F96）/ 日期（下）/ 底部六色块。全部纯色。""" img = Image.new("RGB", (800, 480), WHITE) d = ImageDraw.Draw(img) d.fontmode = "1" f24 = ImageFont.truetype(config.FONT_PATH, 24) f96 = ImageFont.truetype(config.FONT_PATH, 96) d.text((20, 16), f"probe #{seq:04d}", font=f24, fill=BLACK) d.text((20, 48), "SenseCraft re-fetch test / 同一 URL 每周期是否重抓", font=f24, fill=BLACK) t = now.strftime("%H:%M:%S") w = d.textlength(t, font=f96) for dx in range(3): # 伪粗体 d.text(((800 - w) / 2 + dx, 150), t, font=f96, fill=RED) ds = now.strftime("%Y/%m/%d %A") d.text(((800 - d.textlength(ds, font=f24)) / 2, 290), ds, font=f24, fill=BLACK) d.text((20, 400), "画面上的时刻若随设备刷新周期推进 → 平台每周期重新抓取了同一 URL", font=f24, fill=BLACK) n = len(SIX) for i, c in enumerate(SIX): d.rectangle([round(800 / n * i), 456, round(800 / n * (i + 1)) - 1, 479], fill=c) buf = io.BytesIO() img.save(buf, "PNG") return buf.getvalue() def regenerate(every): while True: now = datetime.now() with _lock: _state["seq"] += 1 _state["png"] = render_probe(_state["seq"], now) _state["stamp"] = now.strftime("%H:%M:%S") print(f"[{now:%H:%M:%S}] 重绘 probe.png seq={_state['seq']}", flush=True) time.sleep(every) _INDEX = """ probe

""" class Handler(BaseHTTPRequestHandler): server_version = "probe/1.0" def log_message(self, fmt, *args): # 关闭默认日志，用自定义格式 pass def _log(self, code): ua = self.headers.get("User-Agent", "-") ip = self.headers.get("CF-Connecting-IP") or self.headers.get("X-Forwarded-For") or self.client_address[0] print(f"[{datetime.now():%H:%M:%S}] {code} GET {self.path} ip={ip} ua={ua}", flush=True) def _send(self, code, ctype, body): self.send_response(code) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store, no-cache, must-revalidate, max-age=0") self.send_header("Pragma", "no-cache") self.send_header("Expires", "0") self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(body) def do_GET(self): path = self.path.split("?", 1)[0] if path == "/probe.png": with _lock: body = _state["png"] self._log(200) return self._send(200, "image/png", body) if path in ("/", "/index.html"): self._log(200) return self._send(200, "text/html; charset=utf-8", _INDEX.encode("utf-8")) if path == "/favicon.ico": return self._send(204, "image/x-icon", b"") self._log(404) self._send(404, "text/plain; charset=utf-8", b"not found") do_HEAD = do_GET def main(): ap = argparse.ArgumentParser(description="SenseCraft 同 URL 重抓验证服务") ap.add_argument("--port", type=int, default=8787) ap.add_argument("--every", type=int, default=60, help="探针图重绘间隔（秒）") args = ap.parse_args() threading.Thread(target=regenerate, args=(args.every,), daemon=True).start() while not _state["png"]: time.sleep(0.05) srv = ThreadingHTTPServer(("0.0.0.0", args.port), Handler) print(f"探针服务已启动：http://localhost:{args.port}/probe.png （HTML 对照页：/index.html）") print("下一步：另开终端执行 cloudflared tunnel --url http://localhost:%d ，把得到的 https 地址 + /probe.png 填入 Image 控件。" % args.port) print("观察本终端：每个设备刷新周期是否出现一条 GET /probe.png。Ctrl+C 退出。", flush=True) try: srv.serve_forever() except KeyboardInterrupt: pass if __name__ == "__main__": main()