Coverage for orchestrant/pipeline/capture/gstreamer.py: 39%

1"""GStreamer subprocess capture backend.""" 

2 

3from __future__ import annotations 

4 

5import os 

6import platform 

7import shlex 

8import shutil 

9import subprocess # nosec B404 

10import threading 

11import time 

12from contextlib import suppress 

13from pathlib import Path 

14from queue import Empty, Full, Queue 

15from typing import TYPE_CHECKING 

16 

17import numpy as np 

18from loguru import logger 

19 

20from orchestrant.pipeline.constants import ( 

21 GST_DEFAULT_TIMEOUT_SECONDS, 

22 GST_FALLBACK_HEIGHT, 

23 GST_FALLBACK_WIDTH, 

24 GST_FRAME_QUEUE_TIMEOUT, 

25 GST_PROCESS_STARTUP_DELAY, 

26 GST_PROCESS_WAIT_TIMEOUT, 

27 WINDOWS_GSTREAMER_PATHS, 

28) 

29 

30 

31if TYPE_CHECKING: 

32 from orchestrant.pipeline.types import CameraConfig 

33 

34 

35def find_gstreamer_launch() -> tuple[str | None, str]: 

36 """Find the gst-launch-1.0 executable.""" 

37 exe_name = ( 

38 "gst-launch-1.0.exe" if platform.system() == "Windows" else "gst-launch-1.0" 

39 ) 

40 

41 resolved = shutil.which("gst-launch-1.0") 

42 if resolved: 

43 return resolved, "Found in PATH" 

44 

45 if platform.system() == "Windows": 

46 config_paths = _get_config_gstreamer_paths() 

47 search_paths = config_paths or WINDOWS_GSTREAMER_PATHS 

48 for base_path in search_paths: 

49 exe_path = Path(base_path) / "bin" / exe_name 

50 if exe_path.exists(): 

51 return str(exe_path), f"Found at {exe_path}" 

52 

53 return None, "gst-launch-1.0 not found" 

54 

55 

56def _get_config_gstreamer_paths() -> list[str] | None: 

57 """Get GStreamer paths from environment configuration.""" 

58 env_path = os.environ.get("KATAGLYPHIS_GSTREAMER_PATHS", "") 

59 if not env_path: 

60 return None 

61 return [p.strip() for p in env_path.split(os.pathsep) if p.strip()] 

62 

63 

64def get_gstreamer_env() -> dict[str, str]: 

65 """Get the allowlisted environment variables GStreamer needs.""" 

66 env = { 

67 "PATH": os.environ.get("PATH", ""), 

68 "HOME": os.environ.get("HOME", ""), 

69 "USER": os.environ.get("USER", ""), 

70 "TMPDIR": os.environ.get("TMPDIR", ""), 

71 "TEMP": os.environ.get("TEMP", ""), 

72 } 

73 # A package's launcher points GStreamer at its own plugins; GST_DEBUG and the like are the user's. 

74 env.update({k: v for k, v in os.environ.items() if k.startswith("GST_")}) 

75 

76 if platform.system() != "Windows": 

77 return env 

78 

79 search_paths = _get_config_gstreamer_paths() or [ 

80 p for p in WINDOWS_GSTREAMER_PATHS if Path(p).exists() 

81 ] 

82 

83 for gst_root in search_paths: 

84 root_path = Path(gst_root) 

85 if not root_path.exists(): 

86 continue 

87 bin_path = root_path / "bin" 

88 lib_path = root_path / "lib" 

89 plugin_path = lib_path / "gstreamer-1.0" 

90 

91 current_path = env.get("PATH", "") 

92 if str(bin_path) not in current_path: 

93 env["PATH"] = str(bin_path) + os.pathsep + current_path 

94 

95 env["GST_PLUGIN_PATH"] = str(plugin_path) 

96 env["GST_PLUGIN_SYSTEM_PATH"] = str(plugin_path) 

97 

98 logger.debug("GStreamer environment configured from: {}", root_path) 

99 break 

100 

101 return env 

102 

103 

104class GStreamerSubprocessCapture: 

105 """GStreamer video capture using subprocess and raw frames on stdout.""" 

106 

107 def __init__(self, config: CameraConfig) -> None: 

108 """Initialize the capture backend.""" 

109 self.config = config 

110 self.process: subprocess.Popen[bytes] | None = None 

111 self.frame_queue: Queue[np.ndarray] = Queue(maxsize=2) 

112 self.running = False 

113 self.reader_thread: threading.Thread | None = None 

114 self._frame_buffer: np.ndarray | None = None 

115 

116 self.actual_width = config.width 

117 self.actual_height = config.height 

118 self.actual_fps = float(config.fps) 

119 self.pipeline_string = "" 

120 self.frame_size = 0 

121 

122 self.gst_launch_path, self.gst_status = find_gstreamer_launch() 

123 self.gst_env = get_gstreamer_env() 

124 

125 def _allocate_frame_buffer(self) -> None: 

126 """Pre-allocate reusable frame buffer for zero-copy reading.""" 

127 frame_size = self.actual_width * self.actual_height * 3 

128 self._frame_buffer = np.empty(frame_size, dtype=np.uint8) 

129 

130 def _build_pipeline_string( 

131 self, 

132 source: str, 

133 width: int, 

134 height: int, 

135 fps: int, 

136 pipeline_type: str = "strict", 

137 *, 

138 with_device: bool = True, 

139 include_size: bool = True, 

140 include_fps: bool = True, 

141 include_format: bool = True, 

142 ) -> str: 

143 """Build a GStreamer pipeline string for the requested settings.""" 

144 sink = "fdsink fd=1 sync=false" 

145 device = f" device-index={self.config.device_index}" if with_device else "" 

146 source_prefix = f"{source}{device}" 

147 

148 caps_parts = [] 

149 if include_size: 

150 caps_parts.append(f"width={width}") 

151 caps_parts.append(f"height={height}") 

152 if include_fps: 

153 caps_parts.append(f"framerate={fps}/1") 

154 if include_format: 

155 caps_parts.append("format=BGR") 

156 

157 caps = f"video/x-raw,{','.join(caps_parts)}" if caps_parts else "video/x-raw" 

158 

159 if pipeline_type == "strict": 

160 return f"{source_prefix} ! {caps} ! videoconvert ! {caps} ! {sink}" 

161 if pipeline_type == "flexible": 

162 return ( 

163 f"{source_prefix} ! " 

164 f"videoconvert ! videoscale ! " 

165 f"{caps} ! " 

166 f"videorate ! {caps} ! " 

167 f"videoconvert ! {caps} ! {sink}" 

168 ) 

169 return f"{source_prefix} ! videoconvert ! videoscale ! {caps} ! {sink}" 

170 

171 def _log_short_read(self, bytes_read: int, frame_size: int) -> bool: 

172 """Log an under-sized read; True means the stream ended (0 bytes).""" 

173 if bytes_read == 0: 

174 logger.warning("GStreamer process ended (no data)") 

175 return True 

176 logger.warning("Incomplete frame: {}/{}", bytes_read, frame_size) 

177 return False 

178 

179 def _read_one_frame(self, frame_size: int) -> tuple[np.ndarray | None, bool]: 

180 """Read one raw BGR frame; return (frame or None on a short read, stream_ended).""" 

181 # A pipeline that died after the caller's check ends the stream, like a closed pipe. 

182 process = self.process 

183 if process is None or process.stdout is None: 

184 logger.warning("GStreamer process is gone before read") 

185 return None, True 

186 stdout = process.stdout 

187 

188 # getattr, not hasattr: hasattr gives the type checker nothing to narrow. 

189 readinto = getattr(stdout, "readinto", None) 

190 # One read: open() allocates the buffer while this thread already runs. 

191 buffer = self._frame_buffer 

192 if ( 

193 buffer is not None 

194 and buffer.size == frame_size 

195 and readinto is not None 

196 and callable(readinto) 

197 ): 

198 bytes_read = readinto(buffer) 

199 if bytes_read != frame_size: 

200 return None, self._log_short_read(bytes_read, frame_size) 

201 # A copy: the queued frame must not be the buffer the next readinto overwrites. 

202 frame = buffer.reshape((self.actual_height, self.actual_width, 3)).copy() 

203 else: 

204 raw_data = stdout.read(frame_size) 

205 if len(raw_data) != frame_size: 

206 return None, self._log_short_read(len(raw_data), frame_size) 

207 frame = np.frombuffer(raw_data, dtype=np.uint8).reshape( 

208 (self.actual_height, self.actual_width, 3) 

209 ) 

210 return frame, False 

211 

212 def _offer_frame(self, frame: np.ndarray) -> None: 

213 """Queue a frame, dropping the oldest while full; never blocks, never raises Full.""" 

214 while True: 

215 try: 

216 self.frame_queue.put_nowait(frame) 

217 except Full: 

218 with suppress(Empty): 

219 self.frame_queue.get_nowait() 

220 else: 

221 return 

222 

223 def _frame_reader(self) -> None: 

224 frame_size = self.actual_width * self.actual_height * 3 

225 

226 while self.running: 

227 # One read of self.process: release() on another thread sets it to None. 

228 process = self.process 

229 if process is None or process.poll() is not None: 

230 break 

231 try: 

232 if process.stdout is None: 

233 break 

234 

235 frame, stream_ended = self._read_one_frame(frame_size) 

236 if frame is None: 

237 if stream_ended: 

238 break 

239 continue 

240 

241 self._offer_frame(np.ascontiguousarray(frame)) 

242 

243 except (OSError, ValueError) as exc: 

244 if self.running: 

245 logger.error("Frame reader error: {}", exc) 

246 break 

247 

248 self.running = False 

249 

250 def _start_pipeline(self, pipeline_str: str) -> bool: 

251 if self.gst_launch_path is None: 

252 return False 

253 

254 cmd = [self.gst_launch_path, "-q", *shlex.split(pipeline_str)] 

255 try: 

256 # nosec before noqa: ruff reads codes to end-of-line, bandit stops at "#". 

257 self.process = subprocess.Popen( # nosec B603 # noqa: S603 

258 cmd, 

259 stdout=subprocess.PIPE, 

260 stderr=subprocess.PIPE, 

261 bufsize=0, 

262 env=self.gst_env, 

263 ) 

264 except OSError as exc: 

265 logger.warning("Failed to start gst-launch: {}", exc) 

266 self.process = None 

267 return False 

268 

269 time.sleep(GST_PROCESS_STARTUP_DELAY) 

270 if self.process.poll() is not None: 

271 stderr_text = "" 

272 if self.process.stderr is not None: 

273 stderr_text = self.process.stderr.read().decode( 

274 "utf-8", errors="ignore" 

275 ) 

276 logger.warning("Pipeline failed: {}", stderr_text[:300]) 

277 self.release() 

278 return False 

279 

280 self.running = True 

281 self.reader_thread = threading.Thread(target=self._frame_reader, daemon=True) 

282 self.reader_thread.start() 

283 

284 try: 

285 frame = self.frame_queue.get(timeout=GST_FRAME_QUEUE_TIMEOUT) 

286 except Empty: 

287 logger.warning("No frames from pipeline") 

288 self.release() 

289 return False 

290 

291 if frame.shape[0] != self.actual_height or frame.shape[1] != self.actual_width: 

292 logger.info("Adjusting dimensions: {}", frame.shape[:2]) 

293 self.actual_height, self.actual_width = frame.shape[:2] 

294 self.frame_size = self.actual_width * self.actual_height * 3 

295 

296 self._offer_frame(frame) 

297 self._allocate_frame_buffer() 

298 return True 

299 

300 def _try_pipelines( 

301 self, 

302 sources: list[tuple[str, bool]], 

303 specs: list[tuple[str, bool, bool, bool]], 

304 width: int, 

305 height: int, 

306 fps: int, 

307 deadline: float, 

308 ) -> bool: 

309 """Try multiple pipeline configurations until one succeeds or deadline passes.""" 

310 for source, with_device in sources: 

311 for pipeline_type, include_size, include_fps, include_format in specs: 

312 if time.monotonic() > deadline: 

313 return False 

314 

315 pipeline_str = self._build_pipeline_string( 

316 source, 

317 width, 

318 height, 

319 fps, 

320 pipeline_type, 

321 with_device=with_device, 

322 include_size=include_size, 

323 include_fps=include_fps, 

324 include_format=include_format, 

325 ) 

326 logger.debug("Pipeline: {}", pipeline_str) 

327 

328 if self._start_pipeline(pipeline_str): 

329 self.pipeline_string = pipeline_str 

330 return True 

331 

332 logger.warning("Pipeline failed: {} ({})", pipeline_type, source) 

333 return False 

334 

335 def open(self, timeout: float = GST_DEFAULT_TIMEOUT_SECONDS) -> bool: 

336 """Start the GStreamer subprocess, trying pipelines for up to ``timeout`` seconds.""" 

337 if self.gst_launch_path is None: 

338 logger.error("GStreamer not available: {}", self.gst_status) 

339 return False 

340 

341 logger.info("GStreamer: {}", self.gst_status) 

342 logger.debug("Executable: {}", self.gst_launch_path) 

343 

344 self.frame_size = self.actual_width * self.actual_height * 3 

345 deadline = time.monotonic() + timeout 

346 

347 sources: list[tuple[str, bool]] = [ 

348 ("mfvideosrc", True), 

349 ("ksvideosrc", True), 

350 ("dshowvideosrc", False), 

351 ("autovideosrc", False), 

352 ] 

353 pipeline_specs: list[tuple[str, bool, bool, bool]] = [ 

354 ("strict", True, True, True), 

355 ("strict", True, False, True), 

356 ("strict", False, False, True), 

357 ("strict", False, False, False), 

358 ("flexible", True, True, True), 

359 ("flexible", True, False, True), 

360 ("flexible", False, False, True), 

361 ("auto", True, True, True), 

362 ("auto", False, False, True), 

363 ("auto", False, False, False), 

364 ] 

365 

366 if self._try_pipelines( 

367 sources, 

368 pipeline_specs, 

369 self.actual_width, 

370 self.actual_height, 

371 int(self.actual_fps), 

372 deadline, 

373 ): 

374 logger.success( 

375 "GStreamer started: {}x{}", self.actual_width, self.actual_height 

376 ) 

377 return True 

378 

379 if time.monotonic() > deadline: 

380 logger.warning("GStreamer pipeline search timed out after {}s", timeout) 

381 return False 

382 

383 fallback_width = GST_FALLBACK_WIDTH 

384 fallback_height = GST_FALLBACK_HEIGHT 

385 fallback_fps = int(self.actual_fps) if self.actual_fps > 0 else 30 

386 logger.info("Trying fallback pipeline ({}x{})", fallback_width, fallback_height) 

387 

388 self.actual_width = fallback_width 

389 self.actual_height = fallback_height 

390 self.frame_size = self.actual_width * self.actual_height * 3 

391 

392 if self._try_pipelines( 

393 sources, 

394 pipeline_specs, 

395 fallback_width, 

396 fallback_height, 

397 fallback_fps, 

398 deadline, 

399 ): 

400 logger.success( 

401 "GStreamer started (fallback): {}x{}", 

402 self.actual_width, 

403 self.actual_height, 

404 ) 

405 return True 

406 

407 logger.error("All GStreamer pipelines failed!") 

408 return False 

409 

410 def read(self) -> tuple[bool, np.ndarray | None]: 

411 """Read a frame from the capture queue.""" 

412 if not self.running: 

413 return False, None 

414 

415 try: 

416 frame = self.frame_queue.get(timeout=1.0) 

417 except Empty: 

418 return False, None 

419 else: 

420 return True, frame 

421 

422 def release(self) -> None: 

423 """Stop capture and release subprocess resources.""" 

424 self.running = False 

425 if self.process is not None: 

426 try: 

427 self.process.terminate() 

428 self.process.wait(timeout=GST_PROCESS_WAIT_TIMEOUT) 

429 except subprocess.TimeoutExpired: 

430 self.process.kill() 

431 except Exception as exc: 

432 logger.debug("Failed to terminate GStreamer process: {}", exc) 

433 self.process = None 

434 

435 if self.reader_thread and self.reader_thread.is_alive(): 

436 self.reader_thread.join(timeout=1.0) 

437 

438 while not self.frame_queue.empty(): 

439 with suppress(Empty): 

440 self.frame_queue.get_nowait() 

441 

442 def is_opened(self) -> bool: 

443 """Return True if the subprocess is running.""" 

444 return self.running and self.process is not None and self.process.poll() is None 

445 

446 def get_info(self) -> dict: 

447 """Return backend metadata for diagnostics.""" 

448 return { 

449 "backend": "GStreamer (subprocess)", 

450 "pipeline": self.pipeline_string, 

451 "width": self.actual_width, 

452 "height": self.actual_height, 

453 "fps": self.actual_fps, 

454 }