131 lines
3.1 KiB
Python
131 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from enum import Enum
|
|
|
|
from pydantic import BaseModel, Field, ConfigDict
|
|
|
|
|
|
def to_camel(value: str) -> str:
|
|
parts = value.split("_")
|
|
return parts[0] + "".join(part.capitalize() for part in parts[1:])
|
|
|
|
|
|
class ApiModel(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)
|
|
|
|
|
|
class AgentStatus(str, Enum):
|
|
IDLE = "IDLE"
|
|
HAS_UPDATE = "HAS_UPDATE"
|
|
WAIT_USER_CONFIRM = "WAIT_USER_CONFIRM"
|
|
BACKING_UP_DATABASE = "BACKING_UP_DATABASE"
|
|
PULLING_IMAGE = "PULLING_IMAGE"
|
|
RESTARTING_SERVICE = "RESTARTING_SERVICE"
|
|
HEALTH_CHECKING = "HEALTH_CHECKING"
|
|
SUCCESS = "SUCCESS"
|
|
FAILED = "FAILED"
|
|
ROLLED_BACK = "ROLLED_BACK"
|
|
|
|
|
|
class ComponentImages(ApiModel):
|
|
images: dict[str, str] = Field(default_factory=dict)
|
|
|
|
def to_env_mapping(self) -> dict[str, str]:
|
|
return {key: value for key, value in self.images.items() if value}
|
|
|
|
|
|
class UpdateManifest(ApiModel):
|
|
release_version: str
|
|
release_notes: str = ""
|
|
components: ComponentImages = Field(default_factory=ComponentImages)
|
|
upgrade_mode: str = "manual_confirm"
|
|
|
|
|
|
class AgentState(BaseModel):
|
|
vehicle_id: str
|
|
vin: str
|
|
current_release: str
|
|
status: AgentStatus = AgentStatus.IDLE
|
|
available_update: UpdateManifest | None = None
|
|
last_check_at: str | None = None
|
|
last_heartbeat_at: str | None = None
|
|
last_result: str | None = None
|
|
updated_at: str = Field(default_factory=lambda: utc_now())
|
|
|
|
|
|
class HeartbeatPayload(ApiModel):
|
|
vehicle_id: str
|
|
vin: str
|
|
current_release: str
|
|
agent_status: str
|
|
target_release: str | None = None
|
|
last_result: str | None = None
|
|
images: dict[str, str] = Field(default_factory=dict)
|
|
backup_file: str | None = None
|
|
updated_at: str
|
|
|
|
|
|
class UpdateCheckRequest(ApiModel):
|
|
vehicle_id: str
|
|
vin: str
|
|
current_release: str
|
|
|
|
|
|
class UpdateCheckResponse(ApiModel):
|
|
has_update: bool = False
|
|
manifest: UpdateManifest | None = None
|
|
message: str = ""
|
|
|
|
|
|
class ReportPayload(ApiModel):
|
|
vehicle_id: str
|
|
vin: str
|
|
current_release: str
|
|
target_release: str | None = None
|
|
agent_status: str
|
|
success: bool
|
|
message: str = ""
|
|
images: dict[str, str] = Field(default_factory=dict)
|
|
backup_file: str | None = None
|
|
updated_at: str = Field(default_factory=lambda: utc_now())
|
|
|
|
|
|
class ConfirmUpgradeRequest(BaseModel):
|
|
confirmed_by: str = "android-app"
|
|
|
|
|
|
class PostponeUpgradeRequest(BaseModel):
|
|
reason: str = "user_postpone"
|
|
|
|
|
|
class LocalStatusResponse(BaseModel):
|
|
vehicle_id: str
|
|
vin: str
|
|
current_release: str
|
|
status: str
|
|
available_update: UpdateManifest | None = None
|
|
last_result: str | None = None
|
|
updated_at: str
|
|
|
|
|
|
class CommandResult(BaseModel):
|
|
success: bool
|
|
stdout: str = ""
|
|
stderr: str = ""
|
|
returncode: int = 0
|
|
|
|
|
|
class HealthcheckResult(BaseModel):
|
|
success: bool
|
|
detail: str
|
|
|
|
|
|
class OperationResult(BaseModel):
|
|
success: bool
|
|
detail: str
|
|
|
|
|
|
def utc_now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|