Files
fhd_scan_web_app/backend/config.py
T
gardentech f34817d5b6 Initial commit: scan_web integrated recording UI
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.
2026-08-07 14:26:00 +09:00

236 lines
9.5 KiB
Python

"""Workspace paths, per-camera-count topic/launch config, and tunable thresholds.
Values here are ported 1:1 from the source of truth in ~/fast_ws/scan_gui.py,
scan_gui_dual.py and scan_gui_triple.py — see docstrings in each CameraSpec
list for which file a given camera-count profile was copied from.
"""
from __future__ import annotations
import os
from dataclasses import dataclass
from pathlib import Path
HOME = Path.home()
ROS_SETUP = "/opt/ros/humble/setup.bash"
# fast_ws is deprecated (2026-08-05, user decision) — it never had its own
# livox_ros_driver2 build, it only chained to lidar2_ws/install as an
# underlay (confirmed: COLCON_CURRENT_PREFIX in fast_ws/install/setup.bash
# points at lidar2_ws/install). Source lidar2_ws directly instead.
LIDAR2_WS_SETUP = str(HOME / "lidar2_ws/install/setup.bash")
CAMERA2_WS_SETUP = str(HOME / "camera2_ws/install/setup.bash")
FAST_DUAL_WS_SETUP = str(HOME / "fast_dual_ws/install/setup.bash")
RTK_WS_SETUP = str(HOME / "rtk_ws/install/setup.bash")
# hku-mars/FAST_LIO ROS2 port, LiDAR+IMU-only mapping (no camera) — confirmed
# working 2026-08-06. Its install/setup.bash was built with only
# /opt/ros/humble as an underlay (not chained to lidar2_ws), but it subscribes
# livox_ros_driver2::msg::CustomMsg on /livox/lidar, so lidar2_ws must still
# be sourced first at launch time for that message package to resolve.
FAST_LIO_WS_SETUP = str(HOME / "fast_lio/install/setup.bash")
LIDAR_LAUNCH_CMD = (
f"source {ROS_SETUP} && source {LIDAR2_WS_SETUP} && "
"stdbuf -oL -eL ros2 launch livox_ros_driver2 mid360s_fastlivo_launch.py"
)
CAMERA_LAUNCH_SOURCE = f"source {ROS_SETUP} && source {CAMERA2_WS_SETUP}"
# rviz:=false — this is launched headlessly as a recording-lifecycle
# subprocess, not an interactive session; mapping.launch.py defaults rviz to
# true which would otherwise pop an unwanted GUI window.
FASTLIO_LAUNCH_CMD = (
f"source {ROS_SETUP} && source {LIDAR2_WS_SETUP} && source {FAST_LIO_WS_SETUP} && "
"stdbuf -oL -eL ros2 launch fast_lio mapping.launch.py rviz:=false"
)
def fastlio_env() -> dict:
"""LD_LIBRARY_PATH override for the fastlio subprocess.
run_scan_web.sh sources camera2_ws (for the Hikvision camera driver)
before exec'ing uvicorn, which puts the Hikvision SDK's lib dirs on
LD_LIBRARY_PATH for the whole backend process — and therefore, by
default, every ManagedProcess it spawns too (env = os.environ.copy()).
fastlio_mapping doesn't use the camera SDK at all, but its bundled
libusb-1.0.so.0 is old enough to be missing libusb_set_option, a symbol
PCL's libpcl_io needs — with that copy resolved first, fastlio_mapping
fails to even start (symbol lookup error, confirmed 2026-08-06).
Two separate paths carry this old libusb, not just one — /opt/MVS/lib/*
(the vendor SDK's own install) AND
camera2_ws/install/hik_camera_ros2_driver/lib (which symlinks straight
into the same vendored hikSDK/lib/amd64 copy) — both get filtered here.
"""
bad = ("mvs", "hik_camera", "hiksdk")
parts = os.environ.get("LD_LIBRARY_PATH", "").split(":")
cleaned = ":".join(p for p in parts if p and not any(b in p.lower() for b in bad))
return {"LD_LIBRARY_PATH": cleaned}
CAMERA_DELAY_SEC = 5 # let LiDAR init before starting cameras — ported from scan_gui*.py
LOG_DIR = HOME / "scan_web_logs"
LIDAR_LOG_PATH = LOG_DIR / "lidar.log"
CAMERA_LOG_PATH = LOG_DIR / "camera.log"
RECORDING_LOG_PATH = LOG_DIR / "recording.log"
FASTLIO_LOG_PATH = LOG_DIR / "fastlio.log"
GNSS_LOG_PATH = LOG_DIR / "gnss.log"
DEFAULT_SAVE_DIR = str(HOME / "bags") # confirmed default from scan_gui.py:460
@dataclass(frozen=True)
class CameraSpec:
id: str
image_topic: str
info_topic: str
node_name: str # ROS node name, no leading slash
base_params_path: str
launch_arg: str # launch-file argument name this camera's params file is passed as
HIK_CFG_DIR = HOME / "camera2_ws/src/hik_camera_ros2_driver/config"
# camera_count -> (launch_file, [CameraSpec, ...])
CAMERA_PROFILES: dict[int, tuple[str, list[CameraSpec]]] = {
1: (
"hik_camera_launch.py",
[
CameraSpec(
id="cam1",
image_topic="/camera/image",
info_topic="/camera/camera_info",
node_name="hik_camera_ros2_driver",
base_params_path=str(HIK_CFG_DIR / "camera_params.yaml"),
launch_arg="params_file",
),
],
),
2: (
"hik_camera_dual_launch.py",
[
CameraSpec(
id="cam1",
image_topic="/cam1/image",
info_topic="/cam1/camera_info",
node_name="hik_camera_cam1",
base_params_path=str(HIK_CFG_DIR / "camera_params_cam1.yaml"),
launch_arg="cam1_params_file",
),
CameraSpec(
id="cam2",
image_topic="/cam2/image",
info_topic="/cam2/camera_info",
node_name="hik_camera_cam2",
base_params_path=str(HIK_CFG_DIR / "camera_params_cam2.yaml"),
launch_arg="cam2_params_file",
),
],
),
3: (
"hik_camera_triple_launch.py",
[
CameraSpec(
id="cam1",
image_topic="/cam1/image",
info_topic="/cam1/camera_info",
node_name="hik_camera_cam1",
base_params_path=str(HIK_CFG_DIR / "camera_params_cam1.yaml"),
launch_arg="cam1_params_file",
),
CameraSpec(
id="cam2",
image_topic="/cam2/image",
info_topic="/cam2/camera_info",
node_name="hik_camera_cam2",
base_params_path=str(HIK_CFG_DIR / "camera_params_cam2.yaml"),
launch_arg="cam2_params_file",
),
CameraSpec(
id="cam3",
image_topic="/cam3/image",
info_topic="/cam3/camera_info",
node_name="hik_camera_cam3",
base_params_path=str(HIK_CFG_DIR / "camera_params_cam3.yaml"),
launch_arg="cam3_params_file",
),
],
),
}
DEFAULT_CAMERA_COUNT = 3 # matches the physical rig described by the user (3 cams + MID360s)
def camera_specs(camera_count: int) -> list[CameraSpec]:
if camera_count not in CAMERA_PROFILES:
raise ValueError(f"unsupported camera_count={camera_count}, expected one of {sorted(CAMERA_PROFILES)}")
return CAMERA_PROFILES[camera_count][1]
def camera_launch_file(camera_count: int) -> str:
return CAMERA_PROFILES[camera_count][0]
def base_record_topics(camera_count: int) -> str:
parts = ["/livox/lidar", "/livox/imu"]
for cam in camera_specs(camera_count):
parts += [cam.image_topic, cam.info_topic]
return " ".join(parts)
ROSBAG_SOURCE = f"source {ROS_SETUP} && source {LIDAR2_WS_SETUP} && source {CAMERA2_WS_SETUP}"
# GPS is out of scope for M1 (see plan milestone M2) — topic names kept here only
# as forward-reference documentation, not wired to any process/health logic yet.
GPS_TOPICS_LEGACY = "/gps/fix /gps/nmea"
GPS_TOPICS_RTK = "/ublox_driver/receiver_pvt"
DISK_LOW_WARNING_GB = 10.0 # badge turns warning below this
DISK_LOW_DANGER_GB = 2.0 # badge turns danger below this — a long scan can lose a bag here
def disk_free_gb(path: str) -> float:
"""Free space (GiB) on the filesystem backing `path` (walks up to an
existing ancestor dir first — the save dir itself may not exist yet)."""
import shutil as _shutil
p = Path(path)
while not p.exists() and p != p.parent:
p = p.parent
usage = _shutil.disk_usage(p)
return usage.free / (1024 ** 3)
# ── Health staleness thresholds (seconds) — proposed defaults, tune against
# real observed rates once M1-M3 are running (see plan §Health synthesis). ──
HEALTH_THRESHOLDS = {
"/livox/lidar": (1.0, 3.0),
# No "/livox/imu" entry — not subscribed at all anymore, see
# ros_bridge.py's module docstring. imu_health in /api/status mirrors
# lidar's own classification instead.
"camera_image": (1.5, 4.0), # applied per-camera image topic
# /Odometry, not /aft_mapped_to_init — that was FAST-LIVO2/LOAM-family
# naming, anticipated before the actual M3 backend (hku-mars FAST_LIO
# ROS2 port, confirmed 2026-08-06) was built. Confirmed via
# laserMapping.cpp: pubOdomAftMapped_ publishes nav_msgs/Odometry on
# "/Odometry".
"/Odometry": (1.0, 4.0),
}
# ── M3 live point-cloud view ────────────────────────────────────────────
# FAST-LIO's /cloud_registered is the per-scan (not cumulative) downsampled
# world-frame cloud — scan_web accumulates it server-side (see
# pointcloud_codec.py) so the operator sees full scan coverage, not a
# blinking per-scan cloud.
POINTCLOUD_TOPIC = "/cloud_registered"
ODOMETRY_TOPIC = "/Odometry"
POINTCLOUD_VOXEL_SIZE_M = 0.05 # accumulation dedup grid
POINTCLOUD_MAX_ACCUM_POINTS = 200_000 # hard ceiling; oldest points evicted past this
# Two output profiles selected at WS-connect time (?profile=desktop|phone) —
# phone gets fewer points at a lower rate, reasoned from three.js draw cost
# and WS bandwidth on field WiFi. Both sample from the same accumulated
# buffer, just decimated differently per frame.
POINTCLOUD_PROFILES = {
"desktop": {"max_points": 20_000, "hz": 5.0},
"phone": {"max_points": 7_000, "hz": 3.5},
}