gardentech
f34817d5b6
FastAPI backend + vanilla JS frontend for LiDAR/camera startup, camera settings, and rosbag recording, unifying the previous scan_gui/scan_gui_dual/ scan_gui_triple desktop tools into one web app.
142 lines
4.7 KiB
Python
142 lines
4.7 KiB
Python
"""Subprocess orchestration for lidar/camera/recording processes.
|
|
|
|
ManagedProcess.stop() is a near-literal port of the `_kill_proc` pattern used
|
|
by scan_gui.py / scan_gui_dual.py / scan_gui_triple.py: SIGINT the whole
|
|
process group (so `ros2 launch` subtrees and `ros2 bag record` get a chance
|
|
to shut down / flush cleanly), wait, then SIGKILL stragglers. This must not
|
|
regress — it's the one piece of logic explicitly called out as load-bearing
|
|
in the design plan.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import signal
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
import psutil
|
|
|
|
logger = logging.getLogger("scan_web.process_manager")
|
|
|
|
|
|
class ManagedProcess:
|
|
def __init__(self, name: str, cmd: str, log_path: Optional[Path] = None, env: Optional[dict] = None):
|
|
self.name = name
|
|
self.cmd = cmd
|
|
self.log_path = log_path
|
|
self.env = env
|
|
self._proc: Optional[subprocess.Popen] = None
|
|
self._log_file = None
|
|
self._lock = threading.Lock()
|
|
self.started_at: Optional[float] = None
|
|
|
|
def start(self) -> None:
|
|
with self._lock:
|
|
if self.is_running():
|
|
return
|
|
env = os.environ.copy()
|
|
env["PYTHONUNBUFFERED"] = "1"
|
|
if self.env:
|
|
env.update(self.env)
|
|
|
|
stdout = subprocess.DEVNULL
|
|
stderr = subprocess.DEVNULL
|
|
if self.log_path is not None:
|
|
self.log_path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._log_file = open(self.log_path, "w")
|
|
stdout = self._log_file
|
|
stderr = subprocess.STDOUT
|
|
|
|
self._proc = subprocess.Popen(
|
|
["bash", "-c", self.cmd],
|
|
stdout=stdout, stderr=stderr,
|
|
preexec_fn=os.setsid, env=env,
|
|
)
|
|
self.started_at = time.monotonic()
|
|
logger.info("started %s pid=%s", self.name, self._proc.pid)
|
|
|
|
def is_running(self) -> bool:
|
|
return self._proc is not None and self._proc.poll() is None
|
|
|
|
def pid(self) -> Optional[int]:
|
|
return self._proc.pid if self._proc else None
|
|
|
|
def stop(self, timeout: float = 3.0) -> None:
|
|
with self._lock:
|
|
proc = self._proc
|
|
if proc is None or proc.poll() is not None:
|
|
self._cleanup()
|
|
return
|
|
try:
|
|
parent = psutil.Process(proc.pid)
|
|
children = parent.children(recursive=True)
|
|
except psutil.NoSuchProcess:
|
|
self._cleanup()
|
|
return
|
|
for p in children + [parent]:
|
|
try:
|
|
p.send_signal(signal.SIGINT)
|
|
except (psutil.NoSuchProcess, psutil.AccessDenied):
|
|
pass
|
|
_, alive = psutil.wait_procs(children + [parent], timeout=timeout)
|
|
for p in alive:
|
|
try:
|
|
p.kill()
|
|
except (psutil.NoSuchProcess, psutil.AccessDenied):
|
|
pass
|
|
logger.info("stopped %s", self.name)
|
|
self._cleanup()
|
|
|
|
def _cleanup(self) -> None:
|
|
self._proc = None
|
|
self.started_at = None
|
|
if self._log_file:
|
|
self._log_file.close()
|
|
self._log_file = None
|
|
|
|
|
|
class ProcessManager:
|
|
"""App-lifetime singleton (stored on app.state) tracking all managed processes."""
|
|
|
|
def __init__(self):
|
|
self._procs: dict[str, ManagedProcess] = {}
|
|
self._lock = threading.Lock()
|
|
|
|
def start(self, name: str, cmd: str, log_path: Optional[Path] = None, env: Optional[dict] = None) -> ManagedProcess:
|
|
with self._lock:
|
|
mp = self._procs.get(name)
|
|
if mp is None or not mp.is_running():
|
|
mp = ManagedProcess(name, cmd, log_path=log_path, env=env)
|
|
self._procs[name] = mp
|
|
mp.start()
|
|
return mp
|
|
|
|
def stop(self, name: str, timeout: float = 3.0) -> None:
|
|
with self._lock:
|
|
mp = self._procs.get(name)
|
|
if mp is not None:
|
|
mp.stop(timeout=timeout)
|
|
|
|
def is_running(self, name: str) -> bool:
|
|
with self._lock:
|
|
mp = self._procs.get(name)
|
|
return mp is not None and mp.is_running()
|
|
|
|
def status(self) -> dict[str, dict]:
|
|
with self._lock:
|
|
items = list(self._procs.items())
|
|
return {
|
|
name: {"running": mp.is_running(), "pid": mp.pid()}
|
|
for name, mp in items
|
|
}
|
|
|
|
def stop_all(self, timeout: float = 3.0) -> None:
|
|
with self._lock:
|
|
names = list(self._procs.keys())
|
|
for name in names:
|
|
self.stop(name, timeout=timeout)
|