roborock.data.camera

  1"""Data models for Roborock camera live view."""
  2
  3from dataclasses import dataclass, field
  4from enum import StrEnum
  5from typing import Any, Self
  6
  7from .containers import RoborockBase
  8
  9__all__ = ["CameraIceCandidate", "CameraQuality", "CameraSdpDescription", "CameraTurnServer"]
 10
 11_VALID_TURN_SCHEMES = ("turn:", "turns:", "stun:", "stuns:")
 12
 13
 14class CameraQuality(StrEnum):
 15    """Camera video streaming resolution setting."""
 16
 17    HD = "HD"
 18    SD = "SD"
 19    FHD = "FHD"
 20    AUTO = "auto"
 21
 22    @classmethod
 23    def from_str(cls, value: str) -> Self:
 24        """Parse CameraQuality case-insensitively."""
 25        normalized = value.strip()
 26        # "auto" is lowercase in the Roborock app enum, while HD/SD/FHD are uppercase
 27        for member in cls:
 28            if member.value.lower() == normalized.lower():
 29                return member
 30        valid = ", ".join(repr(m.value) for m in cls)
 31        raise ValueError(f"Invalid camera quality: {value!r}. Expected one of {valid}.")
 32
 33
 34@dataclass
 35class CameraTurnServer(RoborockBase):
 36    """Ephemeral TURN server credentials for WebRTC peer connection relay."""
 37
 38    urls: list[str]
 39    username: str | None = None
 40    credential: str | None = field(default=None, repr=False)
 41
 42    @classmethod
 43    def from_dict(cls, data: dict[str, Any] | list[dict[str, Any]]) -> Self:
 44        """Create CameraTurnServer from get_turn_server response or dict.
 45
 46        Handles both device RPC responses (single-item list or dict with 'url', 'user', 'pwd')
 47        and standard WebRTC configuration dicts ('urls', 'username', 'credential').
 48        """
 49        if isinstance(data, list):
 50            if not data:
 51                raise ValueError("Empty TURN server data list")
 52            item = data[0]
 53            if not isinstance(item, dict):
 54                raise TypeError(f"Expected dict inside TURN server list, got {type(item).__name__}")
 55            data = item
 56        if not isinstance(data, dict):
 57            raise TypeError(f"Expected dict or list of dicts for TURN server, got {type(data).__name__}")
 58
 59        if "urls" in data:
 60            urls_val = data["urls"]
 61        elif "url" in data:
 62            urls_val = data["url"]
 63        else:
 64            raise ValueError("Missing 'url' or 'urls' in TURN server response")
 65
 66        urls: list[str] = []
 67        if isinstance(urls_val, str):
 68            urls_clean = urls_val.strip()
 69            if not urls_clean:
 70                raise ValueError("TURN server URL cannot be empty")
 71            urls = [urls_clean]
 72        elif isinstance(urls_val, list):
 73            if not urls_val:
 74                raise ValueError("TURN server 'urls' list cannot be empty")
 75            for u in urls_val:
 76                if not isinstance(u, str) or not u.strip():
 77                    raise ValueError(f"Each URL in 'urls' must be a non-empty string, got {u!r}")
 78                urls.append(u.strip())
 79        else:
 80            raise TypeError(f"Expected str or list for urls, got {type(urls_val).__name__}")
 81
 82        # Validate URL scheme (stun:, stuns:, turn:, turns:)
 83        for u in urls:
 84            if not any(u.lower().startswith(scheme) for scheme in _VALID_TURN_SCHEMES):
 85                raise ValueError(f"Invalid ICE server URL scheme in {u!r}. Expected turn:, turns:, stun:, or stuns:")
 86
 87        user_val = data.get("username") if "username" in data else data.get("user")
 88        pwd_val = data.get("credential") if "credential" in data else data.get("pwd")
 89
 90        if user_val is not None and not isinstance(user_val, str):
 91            raise TypeError(f"Expected str for username, got {type(user_val).__name__}")
 92        if pwd_val is not None and not isinstance(pwd_val, str):
 93            raise TypeError(f"Expected str for credential, got {type(pwd_val).__name__}")
 94
 95        return cls(
 96            urls=urls,
 97            username=user_val,
 98            credential=pwd_val,
 99        )
100
101
102@dataclass
103class CameraIceCandidate(RoborockBase):
104    """Parsed WebRTC ICE candidate representation (RFC 5245 / RFC 8839)."""
105
106    foundation: str
107    component: int
108    protocol: str
109    priority: int
110    ip: str
111    port: int
112    candidate_type: str
113    related_address: str | None = None
114    related_port: int | None = None
115    tcp_type: str | None = None
116    sdp_mid: str | None = None
117    sdp_mline_index: int | None = None
118    extensions: dict[str, str] = field(default_factory=dict)
119
120    def to_sdp(self) -> str:
121        """Format candidate as an SDP attribute line value (RFC 5245 / RFC 8839)."""
122        parts = [
123            f"candidate:{self.foundation}",
124            str(self.component),
125            self.protocol.lower(),
126            str(self.priority),
127            self.ip,
128            str(self.port),
129            "typ",
130            self.candidate_type.lower(),
131        ]
132        if self.related_address is not None and self.related_port is not None:
133            parts.extend(["raddr", self.related_address, "rport", str(self.related_port)])
134        if self.tcp_type is not None:
135            parts.extend(["tcptype", self.tcp_type.lower()])
136        for ext_k, ext_v in self.extensions.items():
137            parts.extend([ext_k, ext_v])
138        return " ".join(parts)
139
140    def to_signaling_dict(self) -> dict[str, str | int]:
141        """Convert to Roborock camera signaling dictionary format."""
142        return {
143            "candidate": self.to_sdp(),
144            "sdpMid": self.sdp_mid if self.sdp_mid is not None else "0",
145            "sdpMLineIndex": self.sdp_mline_index if self.sdp_mline_index is not None else 0,
146        }
147
148    def __str__(self) -> str:
149        """Return the SDP candidate string."""
150        return self.to_sdp()
151
152
153@dataclass
154class CameraSdpDescription(RoborockBase):
155    """WebRTC session description exchanged during camera signaling."""
156
157    sdp: str
158    type: str