109 lines
3.1 KiB
Python
109 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
|
|
@dataclass
|
|
class PlaybackInfo:
|
|
state: str = "idle"
|
|
position_ms: int = 0
|
|
server_updated_at: int = 0
|
|
target_latency_ms: int | None = None
|
|
|
|
@staticmethod
|
|
def from_json(data: dict[str, Any] | None) -> "PlaybackInfo":
|
|
data = data or {}
|
|
return PlaybackInfo(
|
|
state=data.get("state", "idle"),
|
|
position_ms=int(data.get("positionMs", 0)),
|
|
server_updated_at=int(data.get("serverUpdatedAt", 0)),
|
|
target_latency_ms=data.get("targetLatencyMs"),
|
|
)
|
|
|
|
def to_json(self) -> dict[str, Any]:
|
|
payload: dict[str, Any] = {
|
|
"state": self.state,
|
|
"positionMs": self.position_ms,
|
|
"serverUpdatedAt": self.server_updated_at,
|
|
}
|
|
if self.target_latency_ms is not None:
|
|
payload["targetLatencyMs"] = self.target_latency_ms
|
|
return payload
|
|
|
|
|
|
@dataclass
|
|
class SourceInfo:
|
|
type: str = "url"
|
|
mode: str = "direct"
|
|
url: str = ""
|
|
is_live: bool = False
|
|
headers: dict[str, str] = field(default_factory=dict)
|
|
username: str = ""
|
|
password: str = ""
|
|
|
|
@staticmethod
|
|
def from_json(data: dict[str, Any] | None) -> "SourceInfo | None":
|
|
if not data:
|
|
return None
|
|
return SourceInfo(
|
|
type=data.get("type", "url"),
|
|
mode=data.get("mode", "direct"),
|
|
url=data.get("url", ""),
|
|
is_live=bool(data.get("isLive", False)),
|
|
headers={str(k): str(v) for k, v in (data.get("headers") or {}).items()},
|
|
username=data.get("username") or "",
|
|
password=data.get("password") or "",
|
|
)
|
|
|
|
def to_json(self) -> dict[str, Any]:
|
|
payload: dict[str, Any] = {
|
|
"type": self.type,
|
|
"mode": self.mode,
|
|
"url": self.url,
|
|
"isLive": self.is_live,
|
|
}
|
|
if self.headers:
|
|
payload["headers"] = self.headers
|
|
if self.username:
|
|
payload["username"] = self.username
|
|
if self.password:
|
|
payload["password"] = self.password
|
|
return payload
|
|
|
|
|
|
@dataclass
|
|
class RoomInfo:
|
|
code: str
|
|
owner_device_id: str = ""
|
|
online_count: int = 0
|
|
source: SourceInfo | None = None
|
|
playback: PlaybackInfo = field(default_factory=PlaybackInfo)
|
|
|
|
@staticmethod
|
|
def from_json(data: dict[str, Any]) -> "RoomInfo":
|
|
return RoomInfo(
|
|
code=data["code"],
|
|
owner_device_id=data.get("ownerDeviceId") or "",
|
|
online_count=int(data.get("onlineCount", 0)),
|
|
source=SourceInfo.from_json(data.get("source")),
|
|
playback=PlaybackInfo.from_json(data.get("playback")),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class RoomSnapshot:
|
|
room: RoomInfo
|
|
is_owner: bool
|
|
device_id: str
|
|
server_now: int
|
|
|
|
@staticmethod
|
|
def from_json(data: dict[str, Any]) -> "RoomSnapshot":
|
|
return RoomSnapshot(
|
|
room=RoomInfo.from_json(data["room"]),
|
|
is_owner=bool(data.get("isOwner", False)),
|
|
device_id=data.get("deviceId", ""),
|
|
server_now=int(data.get("serverNow", 0)),
|
|
)
|