Files
Frobot-OTA-Agent/ota_agent/config.py
2026-04-23 14:40:29 +08:00

89 lines
2.7 KiB
Python

from __future__ import annotations
from pathlib import Path
from typing import Any
import yaml
from pydantic import BaseModel, Field
from pydantic_settings import BaseSettings, SettingsConfigDict
class ServerConfig(BaseModel):
host: str = "0.0.0.0"
port: int = 19090
class VehicleConfig(BaseModel):
vehicle_id: str = "vehicle-test-001"
vin: str = "vehicle-test-001"
current_release: str = "vehicle-release-0.0.1"
class CloudConfig(BaseModel):
base_url: str = "http://127.0.0.1:8080"
heartbeat_path: str = "/api/agent/heartbeat"
update_check_path: str = "/api/agent/update-check"
report_path: str = "/api/agent/report"
timeout_seconds: int = 10
token: str = "change-me"
token_header: str = "X-OTA-TOKEN"
class RegistryConfig(BaseModel):
enabled: bool = False
server: str = ""
username: str = ""
password: str = ""
timeout_seconds: int = 30
class ComposeConfig(BaseModel):
working_dir: str
file: str = "docker-compose.yml"
env_file: str = ".env"
backup_env_file: str = ".env.bak"
pull_command: str = "docker compose pull"
up_command: str = "docker compose up -d"
health_check_seconds: int = 15
healthcheck_url: str | None = None
healthcheck_interval_seconds: int = 3
class PollingConfig(BaseModel):
update_interval_seconds: int = 30
heartbeat_interval_seconds: int = 30
class StorageConfig(BaseModel):
state_file: str
manifest_dir: str
log_dir: str = "./runtime/logs"
class MysqlBackupConfig(BaseModel):
enabled: bool = False
backup_dir: str = "./runtime/mysql-backups"
dump_command: str = "docker exec mysql mysqldump -uroot -p123456 app_db > {backup_file}"
timeout_seconds: int = 120
class AgentConfig(BaseSettings):
model_config = SettingsConfigDict(extra="ignore")
server: ServerConfig = Field(default_factory=ServerConfig)
vehicle: VehicleConfig = Field(default_factory=VehicleConfig)
cloud: CloudConfig = Field(default_factory=CloudConfig)
registry: RegistryConfig = Field(default_factory=RegistryConfig)
compose: ComposeConfig = Field(default_factory=lambda: ComposeConfig(working_dir="./runtime"))
polling: PollingConfig = Field(default_factory=PollingConfig)
storage: StorageConfig = Field(default_factory=lambda: StorageConfig(state_file="./runtime/state.json", manifest_dir="./runtime/manifests", log_dir="./runtime/logs"))
mysql_backup: MysqlBackupConfig = Field(default_factory=MysqlBackupConfig)
def load_config(config_path: str | Path) -> AgentConfig:
path = Path(config_path)
raw: dict[str, Any] = {}
if path.exists():
raw = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
return AgentConfig(**raw)