diff --git a/data_only_viz/iphone_usb_source.py b/data_only_viz/iphone_usb_source.py index 9bdf9ec..32e985b 100644 --- a/data_only_viz/iphone_usb_source.py +++ b/data_only_viz/iphone_usb_source.py @@ -101,46 +101,75 @@ class IphoneUSBSource: return self._opened def _run(self, sock) -> None: - try: - for tag, pid, payload in iter_frames(sock): - if self._stop.is_set(): - break - if tag == TAG_VIDEO and len(payload) > 1: - annexb = _to_annexb(payload[1:]) - if self._codec is not None and av is not None: - try: - for fr in self._codec.decode(av.Packet(annexb)): - img = fr.to_ndarray(format="bgr24") - img = cv2.resize(img, (self.target_w, self.target_h)) - with self._lock: - self._frame = img - except Exception as e: # av.AVError is a removal-prone alias - LOG.debug("hevc decode: %s", e) - elif tag == TAG_SKELETON and self.state is not None: - joints = decode_skeleton(payload) - if joints is not None: - with self.state.lock(): - arr = self.state.persons_arkit_joints.get(pid) - if arr is None or arr.shape != (_ARKIT_JOINTS, 3): - arr = np.zeros((_ARKIT_JOINTS, 3), dtype=np.float32) - for i, (x, y, z, valid) in enumerate(joints): - if valid: - arr[i] = (x, y, z) - self.state.persons_arkit_joints[pid] = arr - self.state.persons_arkit_last_t[pid] = time.perf_counter() - elif tag == TAG_HANDS: - hands = _decode_hands(payload) - if hands is not None and self._write_hands and self.state is not None: - with self.state.lock(): - self.state.persons_hands = hands - except OSError as e: - LOG.warning("iphone usb stream ended: %s", e) - finally: + while not self._stop.is_set(): try: - sock.close() - except OSError: - pass + for tag, pid, payload in iter_frames(sock): + if self._stop.is_set(): + break + if tag == TAG_VIDEO and len(payload) > 1: + annexb = _to_annexb(payload[1:]) + if self._codec is not None and av is not None: + try: + for fr in self._codec.decode(av.Packet(annexb)): + img = fr.to_ndarray(format="bgr24") + img = cv2.resize(img, (self.target_w, self.target_h)) + with self._lock: + self._frame = img + except Exception as e: # av.AVError is a removal-prone alias + LOG.debug("hevc decode: %s", e) + elif tag == TAG_SKELETON and self.state is not None: + joints = decode_skeleton(payload) + if joints is not None: + with self.state.lock(): + arr = self.state.persons_arkit_joints.get(pid) + if arr is None or arr.shape != (_ARKIT_JOINTS, 3): + arr = np.zeros((_ARKIT_JOINTS, 3), dtype=np.float32) + for i, (x, y, z, valid) in enumerate(joints): + if valid: + arr[i] = (x, y, z) + self.state.persons_arkit_joints[pid] = arr + self.state.persons_arkit_last_t[pid] = time.perf_counter() + elif tag == TAG_HANDS: + hands = _decode_hands(payload) + if hands is not None and self._write_hands and self.state is not None: + with self.state.lock(): + self.state.persons_hands = hands + except OSError as e: + LOG.warning("iphone usb stream error: %s", e) + finally: + try: + sock.close() + except OSError: + pass + # Stream ended. Drop the stale frame so read() reports no-frame, + # then reconnect (unless stopping) so a transient USB/app drop + # doesn't freeze the pose pipeline (the iPhone backgrounding or a + # USB hiccup ends iter_frames; we re-establish :7000). + with self._lock: + self._frame = None self._opened = False + if self._stop.is_set(): + break + LOG.info("iphone usb stream ended — reconnecting...") + sock = self._reconnect() + if sock is None: + break + self._sock = sock + self._opened = True + LOG.info("iphone usb reconnected") + self._opened = False + + def _reconnect(self): + """Retry connect_device() with capped backoff until it succeeds or + release() is called. Returns the new socket, or None if stopping.""" + delay = 0.5 + while not self._stop.is_set(): + sock = connect_device() + if sock is not None: + return sock + self._stop.wait(delay) + delay = min(delay * 2, 3.0) + return None def read(self): with self._lock: