Spaces:
Running
Running
File size: 4,543 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 | """
会话后台加载器 —— 会话时间线与手部关键点的 parquet 读取全部在 worker 线程。
与 UI 无关（QtCore QObject + 信号，不依赖 QtWidgets）：
- SessionLoader 后台加载（gen 防过期协议由调用方维护）
- 可选手部关键点经 core.hand_tracking 惰性导入
"""
from __future__ import annotations
import threading
from PyQt5.QtCore import QObject, pyqtSignal
from core.session_timeline import load_timeline, SensorTimeline
# 手部关键点（可选模块；core.hand_tracking 引入 ultralytics/torch 等重依赖）
try:
from core.hand_tracking import load_hand_kpts as _load_hand_kpts
from core.hand_tracking import load_hand_kpts_pooled as _load_hand_kpts_pooled
_HAND_KPTS_AVAILABLE = True
except ImportError:
_load_hand_kpts = None
_load_hand_kpts_pooled = None
_HAND_KPTS_AVAILABLE = False
class SessionLoader(QObject):
"""后台加载会话时间线与手部关键点（parquet 读取全部在 worker 线程）。
信号
----
finished(int, object) — (gen, {"timeline": SensorTimeline, "hand_kpts": dict})
payload 按引用传递，worker emit 后不再触碰
failed(int, str) — (gen, error)
VideoCapture 不在 worker 中创建/读取（OpenCV FFmpeg 后端不承诺跨线程
安全），视频打开与播放全部留在调用方主线程。
"""
finished = pyqtSignal(int, object)
failed = pyqtSignal(int, str)
def __init__(self, parent=None):
super().__init__(parent)
self._running = False
self._pending = None # 排队的下一次加载请求（最新覆盖旧的）
self._lock = threading.Lock()
self._thread: threading.Thread = None
def start(self, gen: int, session_dir: str, sensor_names: list[str],
load_kpts: bool = False, load_timeline: bool = True,
episode_index: int = 0):
"""启动后台加载。
已有加载在跑时把请求排队（新请求覆盖旧请求——过期结果本就会被
gen 丢弃，只跑最新一次即可）。
load_timeline=False → kpts-only 模式（手部处理完成后重载关键点）。
episode_index > 0 时按池化布局读本 episode 的文件组。
"""
with self._lock:
if self._running:
self._pending = (gen, session_dir, sensor_names,
load_kpts, load_timeline, episode_index)
return
self._running = True
self._spawn(gen, session_dir, sensor_names, load_kpts, load_timeline,
episode_index)
def is_running(self) -> bool:
return self._running
def _spawn(self, gen, session_dir, sensor_names, load_kpts, load_timeline,
episode_index):
self._thread = threading.Thread(
target=self._run,
args=(gen, session_dir, sensor_names, load_kpts, load_timeline,
episode_index),
daemon=True, name="session-loader",
)
self._thread.start()
# ── 后台 ──────────────────────────────────────────
def _run(self, gen: int, session_dir: str, sensor_names: list[str],
load_kpts: bool, want_timeline: bool, episode_index: int):
try:
timeline = None
if want_timeline:
timeline = load_timeline(session_dir, sensor_names,
episode_index)
hand_kpts = {}
if load_kpts and _HAND_KPTS_AVAILABLE:
if episode_index > 0 and _load_hand_kpts_pooled:
# 池化布局：镜像目录 keypoints_output/<task>/episode_NNNNNN/
hand_kpts = (_load_hand_kpts_pooled(
session_dir, episode_index) or {})
elif _load_hand_kpts:
hand_kpts = _load_hand_kpts(session_dir) or {}
self.finished.emit(gen, {"timeline": timeline, "hand_kpts": hand_kpts})
except Exception as e:
self.failed.emit(gen, str(e)[:300])
finally:
with self._lock:
self._running = False
pending = self._pending
self._pending = None
if pending is not None:
self._running = True
if pending is not None:
self._spawn(*pending)
|