api.py

"""WebSocket API message models."""

from __future__ import annotations

from dataclasses import dataclass, field
from enum import StrEnum
from typing import Any

from mashumaro.mixins.orjson import DataClassORJSONMixin


class ErrorCode(StrEnum):
    """WebSocket API error codes."""

    INVALID_MESSAGE = "invalid_message"
    UNKNOWN_COMMAND = "unknown_command"
    INVALID_ARGS = "invalid_args"
    NOT_FOUND = "not_found"
    ALREADY_EXISTS = "already_exists"
    INTERNAL_ERROR = "internal_error"
    NOT_AUTHENTICATED = "not_authenticated"
    RATE_LIMITED = "rate_limited"
    # Transient external dependency failed — distinct from
    # ``INTERNAL_ERROR`` (backend bug) and ``INVALID_ARGS`` (user
    # typo). The frontend renders this as a "couldn't reach the
    # receiver / try again" toast rather than a stack-trace
    # diagnostic. Used by the offloader-side peer-link commands
    # (``preview_pair`` / ``request_pair``) when the remote
    # dashboard isn't reachable, the Noise handshake fails to
    # authenticate, or the post-handshake frame doesn't decrypt.
    UNAVAILABLE = "unavailable"
    # State precondition not met — the operation is well-formed
    # and the remote is reachable, but the current state of one
    # side disqualifies the request. Used by the offloader's
    # ``request_pair`` for: (a) pin mismatch between the value
    # the user OOB-confirmed in ``preview_pair`` and the actual
    # pubkey from the live handshake (TOCTOU defense — the
    # receiver may have rotated identity, or there's an active
    # MITM), and (b) the receiver returning ``rejected`` (admin
    # explicitly declined or there's a stale "rejected" memo
    # within the soft-block window). The frontend rendering
    # distinguishes these via the ``details`` field; both share
    # the same "you can't proceed past this without out-of-band
    # action" semantic.
    PRECONDITION_FAILED = "precondition_failed"
    # Pairing window on the receiver is closed — the offloader's
    # ``intent="pair_request"`` arrived outside the receiver's
    # admin-supervised acceptance window. The frontend prompts
    # the user to ask the receiver-side admin to open the
    # Pairing requests screen, then retry. Distinct from
    # ``UNAVAILABLE`` (transport failure: receiver unreachable)
    # and ``PRECONDITION_FAILED`` (receiver reachable + made a
    # decision); this is "receiver reachable but not currently
    # listening."
    NO_PAIRING_WINDOW = "no_pairing_window"


@dataclass
class CommandMessage(DataClassORJSONMixin):
    """Client -> Server: a command request."""

    command: str
    message_id: str
    args: dict[str, Any] = field(default_factory=dict)


@dataclass
class ResultMessage(DataClassORJSONMixin):
    """Server -> Client: successful command result."""

    message_id: str
    result: Any = None


@dataclass
class ErrorMessage(DataClassORJSONMixin):
    """Server -> Client: command error."""

    message_id: str
    error_code: ErrorCode
    details: str = ""


@dataclass
class EventMessage(DataClassORJSONMixin):
    """Server -> Client: streaming output or push event."""

    message_id: str
    event: str
    data: Any = None


@dataclass
class ServerInfoMessage(DataClassORJSONMixin):
    """Server -> Client: sent on connection."""

    server_version: str
    esphome_version: str
    port: int
    ha_addon: bool = False
    requires_auth: bool = False