picoKVM/samsung_monitor.py at main · FuNK3Y/picoKVM · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
import ubinascii
import network
import aiohttp
import asyncio
import ssl
from device import Device


class SamsungMonitor(Device):
    def __init__(
        self,
        hostname,
        command_sequences,
        command_delay=0.1,
        token=None,
    ):
        super().__init__()
        self.hostname = hostname
        self.command_delay = command_delay
        self.command_sequences = command_sequences
        self.token = token
        self._ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        self._clientSession = aiohttp.ClientSession()
        self._client_encoded_name = ubinascii.b2a_base64(network.hostname()).decode().strip()

    async def send_commands(self, commands):
        channel_uri = f"wss://{self.hostname}:8002/api/v2/channels/samsung.remote.control?name={self._client_encoded_name}"
        if not self.token:
            async with self._clientSession.ws_connect(channel_uri, self._ssl_context) as ws:
                self.token = (await ws.receive_json())["data"]["token"]
        channel_uri += f"&token={self.token}"
        async with self._clientSession.ws_connect(channel_uri) as ws:
            await ws.receive_json()
            for command in commands:
                if isinstance(command, dict):
                    command_delay = command.get("delay", self.command_delay)
                    repeat = command.get("repeat", 1)
                    command = command["command"]
                else:
                    command_delay = self.command_delay
                    repeat = 1
                message = {
                    "method": "ms.remote.control",
                    "params": {
                        "Cmd": "Click",
                        "DataOfCmd": command,
                        "Option": "false",
                        "TypeOfRemote": "SendRemoteKey",
                    },
                }
                for _ in range(repeat):
                    await ws.send_json(message)
                    await asyncio.sleep(command_delay)

    async def set_active_input(self, input):
        await self.send_commands(self.command_sequences[input])