Hugging Face's logo

Spaces:
Stouching-Robotics
/
DataFlowing
Running

App Files Community
DataFlowing
File size: 4,740 Bytes
d646dd7
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
"""热加载 Worker — watchfiles 监听代码变化,自动重启全部 worker 进程。

worker 本身无热重载(`python -m worker` 启动时加载模块代码,改代码必须重启)。
本脚本与 hot_reload_backend.py 同模式:监听 app/worker/.env 文件变化 →
干净杀掉全部 worker → 重新启动,避免"改了代码 worker 还在跑旧逻辑"。

中断安全:worker 执行中被 kill 的任务,后端靠租约(默认 90s)到期后重新
放回队列,由重启后的 worker 重新领取 —— 任务不丢,最多从头重跑一遍。

用法:
  python scripts/hot_reload_worker.py
  (API Key 未设置时自动从项目 .env 读 WORKER_API_KEY,不会打印)

可选环境变量:
  EGODATA_WORKER_COUNT=3        worker 数量(默认 3,与后端进程布局一致)
  EGODATA_WORKER_API_KEY=...    显式指定(优先于 .env)
  EGODATA_SERVER_URL=...        默认 http://127.0.0.1:8000
  EGODATA_DEVICE=auto           上报的设备类型
"""

import os
import socket
import subprocess
import sys
import time
import uuid
from pathlib import Path

import watchfiles

ROOT = Path(__file__).resolve().parent.parent
# .env 也在监视内:改配置(如 WORKER_API_KEY / 临时目录)后自动重启生效
WATCH_DIRS = [ROOT / "app", ROOT / "worker", ROOT / ".env"]
LOG_PATH = ROOT / "data" / "tmp" / "worker" / "worker-hot.log"
WORKER_COUNT = int(os.getenv("EGODATA_WORKER_COUNT", "3"))


def _read_env_file(path: Path) -> dict[str, str]:
    out: dict[str, str] = {}
    if not path.is_file():
        return out
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        k, _, v = line.partition("=")
        out[k.strip()] = v.strip().strip('"').strip("'")
    return out


def _log(msg: str) -> None:
    line = f"[{time.strftime('%H:%M:%S')}] {msg}"
    print(line, flush=True)
    try:
        LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
        with open(LOG_PATH, "a", encoding="utf-8") as f:
            f.write(line + "\n")
    except Exception:
        pass


def _resolve_api_key() -> str:
    env = _read_env_file(ROOT / ".env")
    return (os.getenv("EGODATA_WORKER_API_KEY")
            or env.get("EGODATA_WORKER_API_KEY")
            or env.get("WORKER_API_KEY")
            or "")


API_KEY = _resolve_api_key()


def start_worker(index: int, generation: str) -> subprocess.Popen:
    env = dict(os.environ)
    env.setdefault("PYTHONPATH", str(ROOT))
    env["EGODATA_WORKER_API_KEY"] = API_KEY
    env.setdefault("EGODATA_SERVER_URL", "http://127.0.0.1:8000")
    env.setdefault("EGODATA_DEVICE", "auto")
    host = os.getenv("EGODATA_WORKER_ID_PREFIX", f"linux-{socket.gethostname()}")
    env["EGODATA_WORKER_ID"] = f"{host}-g{generation}-{index + 1}"
    # 每个 worker 独立临时子目录,避免共目录残留冲突
    env.setdefault("EGODATA_WORK_DIR",
                   str(ROOT / "data" / "tmp" / "worker" / f"w{index + 1}"))
    log_path = LOG_PATH.parent / f"worker-hot-{index + 1}.log"
    log_file = open(log_path, "a", encoding="utf-8")
    proc = subprocess.Popen([sys.executable, "-m", "worker"],
                            cwd=str(ROOT), env=env,
                            stdout=log_file, stderr=log_file)
    _log(f"started worker #{index + 1} pid={proc.pid} id={env['EGODATA_WORKER_ID']}")
    return proc


def stop_worker(proc: subprocess.Popen) -> None:
    if proc is None or proc.poll() is not None:
        return
    proc.terminate()
    try:
        proc.wait(timeout=8)
    except subprocess.TimeoutExpired:
        proc.kill()
        try:
            proc.wait(timeout=5)
        except Exception:
            pass


def main() -> None:
    if not API_KEY:
        print("[error] EGODATA_WORKER_API_KEY 未设置,且 .env 里没有 WORKER_API_KEY",
              file=sys.stderr)
        sys.exit(1)
    generation = uuid.uuid4().hex[:10]
    procs = [start_worker(i, generation) for i in range(WORKER_COUNT)]
    last_restart = time.monotonic()
    try:
        for changes in watchfiles.watch(*WATCH_DIRS, step=500):
            if time.monotonic() - last_restart < 1.5:
                continue  # 防抖:合并连续变更
            files = sorted({str(p) for _, p in changes})
            _log("detected changes: " + ", ".join(files[:5]))
            for p in procs:
                stop_worker(p)
            generation = uuid.uuid4().hex[:10]
            procs = [start_worker(i, generation) for i in range(WORKER_COUNT)]
            last_restart = time.monotonic()
    except KeyboardInterrupt:
        for p in procs:
            stop_worker(p)
        _log("hot reload stopped")


if __name__ == "__main__":
    main()