541 lines
18 KiB
Python
541 lines
18 KiB
Python
"""Typed application configuration, loaded once from the environment / .env file.
|
|
|
|
Everything that could differ between "my laptop with no robot" and "the show floor
|
|
with a live AGIBOT A3" lives here. No module outside this package reads os.environ
|
|
directly, and no robot address is ever hard-coded in source.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
import json
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Literal, Optional
|
|
|
|
from dotenv import load_dotenv
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
|
ENV_FILE = PROJECT_ROOT / ".env"
|
|
|
|
RobotMode = Literal["mock", "real"]
|
|
Transport = Literal["aimdk", "http", "ws", "ros2", "ssh"]
|
|
|
|
TRANSPORTS = ("aimdk", "http", "ws", "ros2", "ssh")
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# env helpers
|
|
# --------------------------------------------------------------------------- #
|
|
def _raw(key: str) -> Optional[str]:
|
|
value = os.environ.get(key)
|
|
if value is None:
|
|
return None
|
|
value = value.strip()
|
|
# Treat `KEY=` and quoted-empty as "unset" rather than "empty string".
|
|
if value in ('', '""', "''"):
|
|
return None
|
|
if len(value) >= 2 and value[0] == value[-1] and value[0] in ('"', "'"):
|
|
value = value[1:-1]
|
|
return value or None
|
|
|
|
|
|
def env_str(key: str, default: str = "") -> str:
|
|
value = _raw(key)
|
|
return default if value is None else value
|
|
|
|
|
|
def env_opt(key: str) -> Optional[str]:
|
|
return _raw(key)
|
|
|
|
|
|
def env_int(key: str, default: int) -> int:
|
|
value = _raw(key)
|
|
if value is None:
|
|
return default
|
|
try:
|
|
return int(value)
|
|
except ValueError:
|
|
return default
|
|
|
|
|
|
def env_float(key: str, default: float) -> float:
|
|
value = _raw(key)
|
|
if value is None:
|
|
return default
|
|
try:
|
|
return float(value)
|
|
except ValueError:
|
|
return default
|
|
|
|
|
|
def env_bool(key: str, default: bool) -> bool:
|
|
value = _raw(key)
|
|
if value is None:
|
|
return default
|
|
return value.lower() in ("1", "true", "yes", "on", "y")
|
|
|
|
|
|
def env_json(key: str, default: Any) -> Any:
|
|
"""Parse a JSON-valued env var; fall back to `default` on malformed input."""
|
|
value = _raw(key)
|
|
if value is None:
|
|
return default
|
|
try:
|
|
return json.loads(value)
|
|
except json.JSONDecodeError:
|
|
return default
|
|
|
|
|
|
def env_list(key: str, default: Optional[List[str]] = None) -> List[str]:
|
|
value = _raw(key)
|
|
if value is None:
|
|
return list(default or [])
|
|
return [item.strip() for item in value.split(",") if item.strip()]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# configuration issues (surfaced in the UI instead of crashing the app)
|
|
# --------------------------------------------------------------------------- #
|
|
@dataclass(frozen=True)
|
|
class ConfigIssue:
|
|
level: Literal["error", "warning"]
|
|
key: str
|
|
message: str
|
|
|
|
def to_dict(self) -> Dict[str, str]:
|
|
return {"level": self.level, "key": self.key, "message": self.message}
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# settings groups
|
|
# --------------------------------------------------------------------------- #
|
|
@dataclass(frozen=True)
|
|
class ServerSettings:
|
|
host: str
|
|
port: int
|
|
log_level: str
|
|
cors_origins: List[str]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RobotSettings:
|
|
"""Vendor-neutral robot connection settings."""
|
|
|
|
mode: RobotMode
|
|
name: str
|
|
model: str
|
|
ip: Optional[str]
|
|
port: int
|
|
use_tls: bool
|
|
connect_timeout: float
|
|
request_timeout: float
|
|
health_interval: float
|
|
reconnect_min_delay: float
|
|
reconnect_max_delay: float
|
|
|
|
@property
|
|
def scheme(self) -> str:
|
|
return "https" if self.use_tls else "http"
|
|
|
|
@property
|
|
def ws_scheme(self) -> str:
|
|
return "wss" if self.use_tls else "ws"
|
|
|
|
@property
|
|
def base_url(self) -> str:
|
|
return "{0}://{1}:{2}".format(self.scheme, self.ip, self.port)
|
|
|
|
@property
|
|
def ws_base_url(self) -> str:
|
|
return "{0}://{1}:{2}".format(self.ws_scheme, self.ip, self.port)
|
|
|
|
@property
|
|
def address(self) -> str:
|
|
return "{0}:{1}".format(self.ip, self.port) if self.ip else "(not configured)"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class A3Settings:
|
|
"""AGIBOT A3 adapter settings.
|
|
|
|
IMPORTANT: every endpoint below is *configuration*, never a guess baked into
|
|
code. See docs/AGIBOT_A3_INTEGRATION.md - once the interface for your unit is
|
|
confirmed you fill these in and the adapter works unchanged.
|
|
"""
|
|
|
|
transport: Transport
|
|
|
|
# -- AimDK transport (AgiBot's documented A3 speech RPC) ------------------
|
|
# POST http://<robot>:<port>/rpc/<service>/<method>
|
|
# https://open.agibot.com/docs/en/aimdk/a3/v3_2/dev_guide/07-02-audio_play
|
|
aimdk_service: str
|
|
aimdk_play_method: str
|
|
aimdk_stop_method: str
|
|
aimdk_status_method: str
|
|
aimdk_priority: str
|
|
aimdk_domain: str
|
|
aimdk_interrupt: bool
|
|
aimdk_max_bytes: int
|
|
|
|
# -- HTTP / REST transport ------------------------------------------------
|
|
http_speak_path: str
|
|
http_speak_method: str
|
|
http_speak_payload: Any
|
|
http_stop_path: str
|
|
http_stop_method: str
|
|
http_stop_payload: Any
|
|
http_status_path: str
|
|
http_status_method: str
|
|
http_headers: Dict[str, str]
|
|
http_auth_token: Optional[str]
|
|
http_success_field: Optional[str]
|
|
|
|
# -- WebSocket transport --------------------------------------------------
|
|
ws_path: str
|
|
ws_speak_payload: Any
|
|
ws_stop_payload: Any
|
|
ws_ping_interval: float
|
|
ws_done_field: Optional[str]
|
|
ws_done_value: Optional[str]
|
|
|
|
# -- ROS 2 transport ------------------------------------------------------
|
|
ros_domain_id: int
|
|
ros_speak_topic: str
|
|
ros_speak_msg_type: str
|
|
ros_speak_msg_field: str
|
|
ros_stop_topic: str
|
|
ros_use_service: bool
|
|
ros_service_name: str
|
|
ros_service_type: str
|
|
|
|
# -- SSH / on-robot command transport -------------------------------------
|
|
ssh_user: str
|
|
ssh_password: Optional[str]
|
|
ssh_key_path: Optional[str]
|
|
ssh_port: int
|
|
ssh_speak_command: str
|
|
ssh_stop_command: str
|
|
ssh_probe_command: str
|
|
|
|
# -- shared TTS parameters (only sent when set) ---------------------------
|
|
voice: Optional[str]
|
|
language: Optional[str]
|
|
volume: Optional[float]
|
|
speed: Optional[float]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MockSettings:
|
|
connect_delay_ms: int
|
|
network_latency_ms: int
|
|
processing_ms: int
|
|
words_per_minute: int
|
|
failure_rate: float
|
|
local_audio: bool
|
|
flaky_connection: bool
|
|
voice: Optional[str]
|
|
speech_rate: int
|
|
speech_volume: int
|
|
speech_pitch: int
|
|
voice_engine: str
|
|
gemini_api_key: Optional[str]
|
|
gemini_model: str
|
|
gemini_voice: str
|
|
gemini_style: str
|
|
gemini_chunk_chars: int
|
|
pronunciation: bool
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SpeechSettings:
|
|
max_length: int
|
|
min_length: int
|
|
allow_interrupt: bool
|
|
history_limit: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Settings:
|
|
server: ServerSettings
|
|
robot: RobotSettings
|
|
a3: A3Settings
|
|
mock: MockSettings
|
|
speech: SpeechSettings
|
|
env_file: str
|
|
issues: List[ConfigIssue] = field(default_factory=list)
|
|
|
|
@property
|
|
def is_mock(self) -> bool:
|
|
return self.robot.mode == "mock"
|
|
|
|
@property
|
|
def has_blocking_issue(self) -> bool:
|
|
return any(issue.level == "error" for issue in self.issues)
|
|
|
|
def public_dict(self) -> Dict[str, Any]:
|
|
"""Safe-to-expose subset for the browser. Never leaks credentials."""
|
|
return {
|
|
"mode": self.robot.mode,
|
|
"robotName": self.robot.name,
|
|
"robotModel": self.robot.model,
|
|
"transport": self.a3.transport if self.robot.mode == "real" else "mock",
|
|
"address": self.robot.address,
|
|
"ipConfigured": bool(self.robot.ip),
|
|
"maxLength": self.speech.max_length,
|
|
"allowInterrupt": self.speech.allow_interrupt,
|
|
"historyLimit": self.speech.history_limit,
|
|
"healthInterval": self.robot.health_interval,
|
|
"envFile": self.env_file,
|
|
"issues": [issue.to_dict() for issue in self.issues],
|
|
}
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# validation
|
|
# --------------------------------------------------------------------------- #
|
|
def _looks_like_host(value: str) -> bool:
|
|
try:
|
|
ipaddress.ip_address(value)
|
|
return True
|
|
except ValueError:
|
|
pass
|
|
# Permissive hostname check - mDNS names such as "agibot-a3.local" are valid.
|
|
return bool(re.fullmatch(r"[A-Za-z0-9]([A-Za-z0-9\-._]{0,251}[A-Za-z0-9])?", value))
|
|
|
|
|
|
def _validate(robot: RobotSettings, a3: A3Settings, speech: SpeechSettings) -> List[ConfigIssue]:
|
|
issues: List[ConfigIssue] = []
|
|
|
|
if robot.mode not in ("mock", "real"):
|
|
issues.append(
|
|
ConfigIssue("error", "ROBOT_MODE", "Unknown ROBOT_MODE '%s'. Use 'mock' or 'real'." % robot.mode)
|
|
)
|
|
return issues
|
|
|
|
if robot.mode == "real":
|
|
if not robot.ip:
|
|
issues.append(
|
|
ConfigIssue(
|
|
"error",
|
|
"ROBOT_IP",
|
|
"ROBOT_MODE=real but ROBOT_IP is empty. Set the robot's LAN IP address in .env.",
|
|
)
|
|
)
|
|
elif not _looks_like_host(robot.ip):
|
|
issues.append(
|
|
ConfigIssue("error", "ROBOT_IP", "'%s' is not a valid IP address or hostname." % robot.ip)
|
|
)
|
|
|
|
if not 0 < robot.port < 65536:
|
|
issues.append(
|
|
ConfigIssue("error", "ROBOT_PORT", "ROBOT_PORT must be 1-65535, got %s." % robot.port)
|
|
)
|
|
|
|
if a3.transport not in TRANSPORTS:
|
|
issues.append(
|
|
ConfigIssue(
|
|
"error",
|
|
"A3_TRANSPORT",
|
|
"Unknown A3_TRANSPORT '%s'. Use one of: %s."
|
|
% (a3.transport, ", ".join(TRANSPORTS)),
|
|
)
|
|
)
|
|
elif a3.transport == "aimdk":
|
|
if not a3.aimdk_service or not a3.aimdk_play_method:
|
|
issues.append(
|
|
ConfigIssue(
|
|
"error",
|
|
"A3_AIMDK_SERVICE",
|
|
"AimDK transport needs a service name and a play method.",
|
|
)
|
|
)
|
|
if robot.port != 59301:
|
|
issues.append(
|
|
ConfigIssue(
|
|
"warning",
|
|
"ROBOT_PORT",
|
|
"AgiBot documents the A3 TTS RPC on port 59301; ROBOT_PORT is %s. "
|
|
"Confirm the port for your firmware." % robot.port,
|
|
)
|
|
)
|
|
elif a3.transport == "http" and not a3.http_speak_path:
|
|
issues.append(
|
|
ConfigIssue(
|
|
"error",
|
|
"A3_HTTP_SPEAK_PATH",
|
|
"HTTP transport selected but no speak endpoint is configured. "
|
|
"Fill it in from the robot's API reference - see docs/AGIBOT_A3_INTEGRATION.md.",
|
|
)
|
|
)
|
|
elif a3.transport == "ws" and not a3.ws_path:
|
|
issues.append(
|
|
ConfigIssue("error", "A3_WS_PATH", "WebSocket transport selected but A3_WS_PATH is empty.")
|
|
)
|
|
elif a3.transport == "ros2" and not (a3.ros_speak_topic or a3.ros_service_name):
|
|
issues.append(
|
|
ConfigIssue(
|
|
"error",
|
|
"A3_ROS_SPEAK_TOPIC",
|
|
"ROS 2 transport selected but neither a topic nor a service name is configured.",
|
|
)
|
|
)
|
|
elif a3.transport == "ssh":
|
|
if not a3.ssh_speak_command:
|
|
issues.append(
|
|
ConfigIssue(
|
|
"error",
|
|
"A3_SSH_SPEAK_COMMAND",
|
|
"SSH transport selected but A3_SSH_SPEAK_COMMAND is empty.",
|
|
)
|
|
)
|
|
if not a3.ssh_password and not a3.ssh_key_path:
|
|
issues.append(
|
|
ConfigIssue(
|
|
"warning",
|
|
"A3_SSH_PASSWORD",
|
|
"No SSH password or key configured; authentication will rely on an ssh agent.",
|
|
)
|
|
)
|
|
|
|
if speech.max_length < 1:
|
|
issues.append(ConfigIssue("error", "SPEECH_MAX_LENGTH", "SPEECH_MAX_LENGTH must be >= 1."))
|
|
|
|
return issues
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# loading
|
|
# --------------------------------------------------------------------------- #
|
|
def load_settings(env_file: Optional[Path] = None, override: bool = True) -> Settings:
|
|
"""Read .env + process environment into an immutable Settings object."""
|
|
path = Path(env_file) if env_file else ENV_FILE
|
|
if path.exists():
|
|
load_dotenv(path, override=override)
|
|
|
|
server = ServerSettings(
|
|
host=env_str("HOST", "127.0.0.1"),
|
|
port=env_int("PORT", 8000),
|
|
log_level=env_str("LOG_LEVEL", "info").lower(),
|
|
cors_origins=env_list("CORS_ORIGINS", []),
|
|
)
|
|
|
|
robot = RobotSettings(
|
|
mode=env_str("ROBOT_MODE", "mock").lower(), # type: ignore[arg-type]
|
|
name=env_str("ROBOT_NAME", "AGIBOT A3"),
|
|
model=env_str("ROBOT_MODEL", "AgiBot A3"),
|
|
ip=env_opt("ROBOT_IP"),
|
|
port=env_int("ROBOT_PORT", 59301),
|
|
use_tls=env_bool("ROBOT_USE_TLS", False),
|
|
connect_timeout=env_float("ROBOT_CONNECT_TIMEOUT", 3.0),
|
|
request_timeout=env_float("ROBOT_REQUEST_TIMEOUT", 8.0),
|
|
health_interval=env_float("ROBOT_HEALTH_INTERVAL", 5.0),
|
|
reconnect_min_delay=env_float("ROBOT_RECONNECT_MIN_DELAY", 1.0),
|
|
reconnect_max_delay=env_float("ROBOT_RECONNECT_MAX_DELAY", 15.0),
|
|
)
|
|
|
|
a3 = A3Settings(
|
|
transport=env_str("A3_TRANSPORT", "aimdk").lower(), # type: ignore[arg-type]
|
|
aimdk_service=env_str("A3_AIMDK_SERVICE", "aimdk.protocol.TTSService"),
|
|
aimdk_play_method=env_str("A3_AIMDK_PLAY_METHOD", "PlayTTS"),
|
|
aimdk_stop_method=env_str("A3_AIMDK_STOP_METHOD", "StopTTSTraceId"),
|
|
aimdk_status_method=env_str("A3_AIMDK_STATUS_METHOD", "GetAudioStatus"),
|
|
aimdk_priority=env_str("A3_AIMDK_PRIORITY", "INTERACTION_L6"),
|
|
aimdk_domain=env_str("A3_AIMDK_DOMAIN", "voice_control"),
|
|
aimdk_interrupt=env_bool("A3_AIMDK_INTERRUPT", True),
|
|
aimdk_max_bytes=env_int("A3_AIMDK_MAX_BYTES", 1024),
|
|
http_speak_path=env_str("A3_HTTP_SPEAK_PATH", ""),
|
|
http_speak_method=env_str("A3_HTTP_SPEAK_METHOD", "POST").upper(),
|
|
http_speak_payload=env_json("A3_HTTP_SPEAK_PAYLOAD", {"text": "{text}"}),
|
|
http_stop_path=env_str("A3_HTTP_STOP_PATH", ""),
|
|
http_stop_method=env_str("A3_HTTP_STOP_METHOD", "POST").upper(),
|
|
http_stop_payload=env_json("A3_HTTP_STOP_PAYLOAD", {}),
|
|
http_status_path=env_str("A3_HTTP_STATUS_PATH", ""),
|
|
http_status_method=env_str("A3_HTTP_STATUS_METHOD", "GET").upper(),
|
|
http_headers=env_json("A3_HTTP_HEADERS", {}),
|
|
http_auth_token=env_opt("A3_HTTP_AUTH_TOKEN"),
|
|
http_success_field=env_opt("A3_HTTP_SUCCESS_FIELD"),
|
|
ws_path=env_str("A3_WS_PATH", ""),
|
|
ws_speak_payload=env_json("A3_WS_SPEAK_PAYLOAD", {"text": "{text}"}),
|
|
ws_stop_payload=env_json("A3_WS_STOP_PAYLOAD", {}),
|
|
ws_ping_interval=env_float("A3_WS_PING_INTERVAL", 20.0),
|
|
ws_done_field=env_opt("A3_WS_DONE_FIELD"),
|
|
ws_done_value=env_opt("A3_WS_DONE_VALUE"),
|
|
ros_domain_id=env_int("A3_ROS_DOMAIN_ID", 0),
|
|
ros_speak_topic=env_str("A3_ROS_SPEAK_TOPIC", ""),
|
|
ros_speak_msg_type=env_str("A3_ROS_SPEAK_MSG_TYPE", "std_msgs/msg/String"),
|
|
ros_speak_msg_field=env_str("A3_ROS_SPEAK_MSG_FIELD", "data"),
|
|
ros_stop_topic=env_str("A3_ROS_STOP_TOPIC", ""),
|
|
ros_use_service=env_bool("A3_ROS_USE_SERVICE", False),
|
|
ros_service_name=env_str("A3_ROS_SERVICE_NAME", ""),
|
|
ros_service_type=env_str("A3_ROS_SERVICE_TYPE", ""),
|
|
ssh_user=env_str("A3_SSH_USER", "root"),
|
|
ssh_password=env_opt("A3_SSH_PASSWORD"),
|
|
ssh_key_path=env_opt("A3_SSH_KEY_PATH"),
|
|
ssh_port=env_int("A3_SSH_PORT", 22),
|
|
ssh_speak_command=env_str("A3_SSH_SPEAK_COMMAND", ""),
|
|
ssh_stop_command=env_str("A3_SSH_STOP_COMMAND", ""),
|
|
ssh_probe_command=env_str("A3_SSH_PROBE_COMMAND", "true"),
|
|
voice=env_opt("A3_VOICE"),
|
|
language=env_opt("A3_LANGUAGE"),
|
|
volume=env_float("A3_VOLUME", 1.0) if env_opt("A3_VOLUME") else None,
|
|
speed=env_float("A3_SPEED", 1.0) if env_opt("A3_SPEED") else None,
|
|
)
|
|
|
|
mock = MockSettings(
|
|
connect_delay_ms=env_int("MOCK_CONNECT_DELAY_MS", 350),
|
|
network_latency_ms=env_int("MOCK_NETWORK_LATENCY_MS", 45),
|
|
processing_ms=env_int("MOCK_PROCESSING_MS", 180),
|
|
words_per_minute=env_int("MOCK_WORDS_PER_MINUTE", 150),
|
|
failure_rate=env_float("MOCK_FAILURE_RATE", 0.0),
|
|
local_audio=env_bool("MOCK_LOCAL_AUDIO", False),
|
|
flaky_connection=env_bool("MOCK_FLAKY_CONNECTION", False),
|
|
voice=env_opt("MOCK_VOICE"),
|
|
speech_rate=env_int("MOCK_SPEECH_RATE", 0),
|
|
speech_volume=env_int("MOCK_SPEECH_VOLUME", 100),
|
|
speech_pitch=env_int("MOCK_SPEECH_PITCH", 0),
|
|
voice_engine=env_str("MOCK_VOICE_ENGINE", "system").lower(),
|
|
gemini_api_key=env_opt("GEMINI_API_KEY"),
|
|
gemini_model=env_str("GEMINI_TTS_MODEL", "gemini-3.1-flash-tts-preview"),
|
|
gemini_voice=env_str("GEMINI_VOICE", "Puck"),
|
|
gemini_style=env_str("GEMINI_TTS_STYLE", ""),
|
|
gemini_chunk_chars=env_int("GEMINI_CHUNK_CHARS", 0),
|
|
pronunciation=env_bool("SPEECH_PRONUNCIATION", True),
|
|
)
|
|
|
|
speech = SpeechSettings(
|
|
max_length=env_int("SPEECH_MAX_LENGTH", 1000),
|
|
min_length=env_int("SPEECH_MIN_LENGTH", 1),
|
|
allow_interrupt=env_bool("SPEECH_ALLOW_INTERRUPT", True),
|
|
history_limit=env_int("SPEECH_HISTORY_LIMIT", 100),
|
|
)
|
|
|
|
return Settings(
|
|
server=server,
|
|
robot=robot,
|
|
a3=a3,
|
|
mock=mock,
|
|
speech=speech,
|
|
env_file=str(path),
|
|
issues=_validate(robot, a3, speech),
|
|
)
|
|
|
|
|
|
_settings: Optional[Settings] = None
|
|
|
|
|
|
def get_settings() -> Settings:
|
|
global _settings
|
|
if _settings is None:
|
|
_settings = load_settings()
|
|
return _settings
|
|
|
|
|
|
def reload_settings() -> Settings:
|
|
"""Re-read .env from disk (used by the /api/config/reload endpoint)."""
|
|
global _settings
|
|
_settings = load_settings(override=True)
|
|
return _settings
|