Spaces:
Running
Running
File size: 5,648 Bytes
06a1901 | 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 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 | import logging
import argparse
import warnings
from typing import Any, Tuple, Optional
from reachy_mini import ReachyMini
from reachy_mini_receptionist.camera_worker import CameraWorker
def parse_args() -> Tuple[argparse.Namespace, list]: # type: ignore
"""Parse command line arguments."""
parser = argparse.ArgumentParser("Reachy Mini Receptionist App")
parser.add_argument(
"--head-tracker",
choices=["yolo", "mediapipe", None],
default=None,
help="Choose head tracker (default: None)",
)
parser.add_argument("--no-camera", default=False, action="store_true", help="Disable camera usage")
parser.add_argument(
"--local-vision",
default=False,
action="store_true",
help="Use local vision model instead of gpt-realtime vision",
)
parser.add_argument("--gradio", default=False, action="store_true", help="Open gradio interface")
parser.add_argument("--debug", default=False, action="store_true", help="Enable debug logging")
parser.add_argument(
"--robot-name",
type=str,
default=None,
help="[Optional] Robot name/prefix for Zenoh topics (must match daemon's --robot-name). Only needed for development with multiple robots.",
)
return parser.parse_known_args()
def handle_vision_stuff(args: argparse.Namespace, current_robot: ReachyMini) -> Tuple[CameraWorker | None, Any, Any]:
"""Initialize camera, head tracker, camera worker, and vision manager.
By default, vision is handled by gpt-realtime model when camera tool is used.
If --local-vision flag is used, a local vision model will process images periodically.
"""
camera_worker = None
head_tracker = None
vision_manager = None
if not args.no_camera:
# Initialize head tracker if specified
if args.head_tracker is not None:
if args.head_tracker == "yolo":
from reachy_mini_receptionist.vision.yolo_head_tracker import HeadTracker
head_tracker = HeadTracker()
elif args.head_tracker == "mediapipe":
from reachy_mini_toolbox.vision import HeadTracker # type: ignore[no-redef]
head_tracker = HeadTracker()
# Initialize camera worker
camera_worker = CameraWorker(current_robot, head_tracker)
# Initialize vision manager only if local vision is requested
if args.local_vision:
try:
from reachy_mini_receptionist.vision.processors import initialize_vision_manager
vision_manager = initialize_vision_manager(camera_worker)
except ImportError as e:
raise ImportError(
"To use --local-vision, please install the extra dependencies: pip install '.[local_vision]'",
) from e
else:
logging.getLogger(__name__).info(
"Using gpt-realtime for vision (default). Use --local-vision for local processing.",
)
return camera_worker, head_tracker, vision_manager
# Dashboard polling routes that spam the console — suppress them from uvicorn access logs.
_SUPPRESSED_ROUTES = (
"/api/face_status",
"/api/logs",
"/api/best_face_jpeg",
"/api/guests",
"/api/calendar",
"/api/outbox",
"/video_feed",
"/guest_images/",
)
class _SuppressPollingRoutes(logging.Filter):
"""Drop uvicorn access-log records for high-frequency dashboard polling endpoints."""
def filter(self, record: logging.LogRecord) -> bool: # noqa: A003
msg = record.getMessage()
return not any(route in msg for route in _SUPPRESSED_ROUTES)
def setup_logger(debug: bool) -> logging.Logger:
"""Set up the logger."""
log_level = "DEBUG" if debug else "INFO"
logging.basicConfig(
level=getattr(logging, log_level, logging.INFO),
format="%(asctime)s %(levelname)s %(name)s:%(lineno)d | %(message)s",
)
logger = logging.getLogger(__name__)
# Suppress WebRTC warnings
warnings.filterwarnings("ignore", message=".*AVCaptureDeviceTypeExternal.*")
warnings.filterwarnings("ignore", category=UserWarning, module="aiortc")
# Tame third-party noise (looser in DEBUG)
if log_level == "DEBUG":
logging.getLogger("aiortc").setLevel(logging.INFO)
logging.getLogger("fastrtc").setLevel(logging.INFO)
logging.getLogger("aioice").setLevel(logging.INFO)
logging.getLogger("openai").setLevel(logging.INFO)
logging.getLogger("websockets").setLevel(logging.INFO)
else:
logging.getLogger("aiortc").setLevel(logging.ERROR)
logging.getLogger("fastrtc").setLevel(logging.ERROR)
logging.getLogger("aioice").setLevel(logging.WARNING)
# Silence high-frequency dashboard polling from uvicorn access logs
logging.getLogger("uvicorn.access").addFilter(_SuppressPollingRoutes())
return logger
def log_connection_troubleshooting(logger: logging.Logger, robot_name: Optional[str]) -> None:
"""Log troubleshooting steps for connection issues."""
logger.error("Troubleshooting steps:")
logger.error(" 1. Verify reachy-mini-daemon is running")
if robot_name is not None:
logger.error(
f" 2. Daemon must be started with: --robot-name '{robot_name}'"
)
else:
logger.error(
" 2. If daemon uses --robot-name, add the same flag here: "
"--robot-name <name>"
)
logger.error(" 3. For wireless: check network connectivity")
logger.error(" 4. Review daemon logs")
logger.error(" 5. Restart the daemon")
|