Add hold_key to remote commands

requirements.txt

@@ -11,3 +11,4 @@ black>=22.1
 mypy>=0.931
 pre-commit>=2.17
 pytest>=6.2.5
+pytest-asyncio>=0.18.2

samsungtvws/async_connection.py

@@ -19,19 +19,19 @@
     Boston, MA  02110-1335  USA
 
 """
-from asyncio import Task, ensure_future, sleep
+import asyncio
 import contextlib
 import json
 import logging
 import ssl
 from types import TracebackType
-from typing import Any, Awaitable, Callable, Dict, Optional, Union
+from typing import Any, Awaitable, Callable, Dict, List, Optional, Union
 
 from websockets.client import WebSocketClientProtocol, connect
 from websockets.exceptions import ConnectionClosed
 
 from . import connection, exceptions, helper
-from .command import SamsungTVCommand
+from .command import SamsungTVCommand, SamsungTVSleepCommand
 from .event import MS_CHANNEL_CONNECT
 
 _LOGGING = logging.getLogger(__name__)
@@ -40,7 +40,7 @@
 class SamsungTVWSAsyncConnection(connection.SamsungTVWSBaseConnection):
 
     connection: Optional[WebSocketClientProtocol]
-    _recv_loop: Optional[Task[None]]
+    _recv_loop: Optional["asyncio.Task[None]"]
 
     async def __aenter__(self) -> "SamsungTVWSAsyncConnection":
         return self
@@ -87,7 +87,7 @@ async def start_listening(
 
         self.connection = await self.open()
 
-        self._recv_loop = ensure_future(
+        self._recv_loop = asyncio.ensure_future(
             self._do_start_listening(callback, self.connection)
         )
 
@@ -117,22 +117,38 @@ async def close(self) -> None:
         self.connection = None
         _LOGGING.debug("Connection closed.")
 
+    async def _send_command_sequence(
+        self, commands: List[SamsungTVCommand], key_press_delay: float
+    ) -> None:
+        assert self.connection
+        for command in commands:
+            if isinstance(command, SamsungTVSleepCommand):
+                await asyncio.sleep(command.delay)
+            else:
+                payload = command.get_payload()
+                await self.connection.send(payload)
+                await asyncio.sleep(key_press_delay)
+
     async def send_command(
         self,
-        command: Union[SamsungTVCommand, Dict[str, Any]],
+        command: Union[List[SamsungTVCommand], SamsungTVCommand, Dict[str, Any]],
         key_press_delay: Optional[float] = None,
     ) -> None:
         if self.connection is None:
             self.connection = await self.open()
 
+        delay = self.key_press_delay if key_press_delay is None else key_press_delay
+
+        if isinstance(command, list):
+            await self._send_command_sequence(command, delay)
+            return
+
         if isinstance(command, SamsungTVCommand):
-            payload = command.get_payload()
+            await self.connection.send(command.get_payload())
         else:
-            payload = json.dumps(command)
-        await self.connection.send(payload)
+            await self.connection.send(json.dumps(command))
 
-        delay = self.key_press_delay if key_press_delay is None else key_press_delay
-        await sleep(delay)
+        await asyncio.sleep(delay)
 
     def is_alive(self) -> bool:
         return self.connection is not None and not self.connection.closed

samsungtvws/command.py

@@ -15,3 +15,15 @@ def as_dict(self) -> Dict[str, Any]:
 
     def get_payload(self) -> str:
         return json.dumps(self.as_dict())
+
+
+class SamsungTVSleepCommand(SamsungTVCommand):
+    def __init__(self, delay: float) -> None:
+        super().__init__("sleep", {})
+        self.delay = delay
+
+    def as_dict(self) -> Dict[str, Any]:
+        raise NotImplementedError("Cannot use as_dict on SamsungTVSleepCommand")
+
+    def get_payload(self) -> str:
+        raise NotImplementedError("Cannot use get_payload on SamsungTVSleepCommand")

samsungtvws/connection.py

@@ -26,12 +26,12 @@
 import threading
 import time
 from types import TracebackType
-from typing import Any, Callable, Dict, Optional, Union
+from typing import Any, Callable, Dict, List, Optional, Union
 
 import websocket
 
 from . import exceptions, helper
-from .command import SamsungTVCommand
+from .command import SamsungTVCommand, SamsungTVSleepCommand
 from .event import MS_CHANNEL_CONNECT, MS_ERROR_EVENT
 
 _LOGGING = logging.getLogger(__name__)
@@ -213,21 +213,37 @@ def close(self) -> None:
         self.connection = None
         _LOGGING.debug("Connection closed.")
 
+    def _send_command_sequence(
+        self, commands: List[SamsungTVCommand], key_press_delay: float
+    ) -> None:
+        assert self.connection
+        for command in commands:
+            if isinstance(command, SamsungTVSleepCommand):
+                time.sleep(command.delay)
+            else:
+                payload = command.get_payload()
+                self.connection.send(payload)
+                time.sleep(key_press_delay)
+
     def send_command(
         self,
-        command: Union[SamsungTVCommand, Dict[str, Any]],
+        command: Union[List[SamsungTVCommand], SamsungTVCommand, Dict[str, Any]],
         key_press_delay: Optional[float] = None,
     ) -> None:
         if self.connection is None:
             self.connection = self.open()
 
+        delay = self.key_press_delay if key_press_delay is None else key_press_delay
+
+        if isinstance(command, list):
+            self._send_command_sequence(command, delay)
+            return
+
         if isinstance(command, SamsungTVCommand):
-            payload = command.get_payload()
+            self.connection.send(command.get_payload())
         else:
-            payload = json.dumps(command)
-        self.connection.send(payload)
+            self.connection.send(json.dumps(command))
 
-        delay = self.key_press_delay if key_press_delay is None else key_press_delay
         time.sleep(delay)
 
     def is_alive(self) -> bool:

samsungtvws/remote.py

@@ -26,7 +26,7 @@
 from samsungtvws.event import ED_INSTALLED_APP_EVENT, parse_installed_app
 
 from . import art, connection, helper, rest, shortcuts
-from .command import SamsungTVCommand
+from .command import SamsungTVCommand, SamsungTVSleepCommand
 
 _LOGGING = logging.getLogger(__name__)
 
@@ -82,6 +82,36 @@ def click(key: str) -> "SendRemoteKey":
             }
         )
 
+    @staticmethod
+    def press(key: str) -> "SendRemoteKey":
+        return SendRemoteKey(
+            {
+                "Cmd": "Press",
+                "DataOfCmd": key,
+                "Option": "false",
+                "TypeOfRemote": "SendRemoteKey",
+            }
+        )
+
+    @staticmethod
+    def release(key: str) -> "SendRemoteKey":
+        return SendRemoteKey(
+            {
+                "Cmd": "Release",
+                "DataOfCmd": key,
+                "Option": "false",
+                "TypeOfRemote": "SendRemoteKey",
+            }
+        )
+
+    @staticmethod
+    def hold_key(key: str, seconds: float) -> List["SamsungTVCommand"]:
+        return [
+            SendRemoteKey.press(key),
+            SamsungTVSleepCommand(seconds),
+            SendRemoteKey.release(key),
+        ]
+
     # power
     @staticmethod
     def power() -> "SendRemoteKey":
@@ -244,9 +274,7 @@ def send_key(
             )
 
     def hold_key(self, key: str, seconds: float) -> None:
-        self.send_key(key, cmd="Press")
-        time.sleep(seconds)
-        self.send_key(key, cmd="Release")
+        self.send_command(SendRemoteKey.hold_key(key, seconds))
 
     def move_cursor(self, x: int, y: int, duration: int = 0) -> None:
         self._ws_send(

tests/conftest.py

@@ -1,7 +1,9 @@
 """Tests for remote module."""
+import asyncio
 from unittest.mock import Mock, patch
 
 import pytest
+from websockets.client import WebSocketClientProtocol
 
 
 @pytest.fixture(autouse=True)
@@ -22,3 +24,23 @@ def get_connection():
     ) as connection_class:
         connection_class.return_value = connection
         yield connection
+
+
+@pytest.fixture(autouse=True)
+def override_asyncio_sleep():
+    """Ignore asyncio sleep in tests."""
+    sleep_future = asyncio.Future()
+    sleep_future.set_result(None)
+    with patch("samsungtvws.async_connection.asyncio.sleep", return_value=sleep_future):
+        yield
+
+
+@pytest.fixture(name="async_connection")
+def get_async_connection():
+    """Open a websockets connection."""
+    connection = Mock(WebSocketClientProtocol)
+    with patch(
+        "samsungtvws.async_connection.connect", return_value=asyncio.Future()
+    ) as connection_class:
+        connection_class.return_value.set_result(connection)
+        yield connection

tests/test_async_remote.py

@@ -0,0 +1,62 @@
+"""Tests for remote module."""
+import asyncio
+from unittest.mock import Mock, call, patch
+
+import pytest
+
+from samsungtvws.async_remote import SamsungTVWSAsyncRemote
+from samsungtvws.remote import SendRemoteKey
+
+
+@pytest.mark.asyncio
+async def test_send_key(async_connection: Mock) -> None:
+    """Ensure simple data can be parsed."""
+    open_response = (
+        '{"data": {"token": 123456789}, "event": "ms.channel.connect", "from": "host"}'
+    )
+    open_response_future = asyncio.Future()
+    open_response_future.set_result(open_response)
+
+    send_command_future = asyncio.Future()
+    send_command_future.set_result(None)
+
+    async_connection.recv = Mock(side_effect=[open_response_future])
+    async_connection.send = Mock(return_value=send_command_future)
+    tv = SamsungTVWSAsyncRemote("127.0.0.1")
+    await tv.send_command(SendRemoteKey.click("KEY_POWER"))
+    async_connection.send.assert_called_once_with(
+        '{"method": "ms.remote.control", "params": {'
+        '"Cmd": "Click", '
+        '"DataOfCmd": "KEY_POWER", '
+        '"Option": "false", '
+        '"TypeOfRemote": "SendRemoteKey"'
+        "}}"
+    )
+
+
+@pytest.mark.asyncio
+async def test_send_hold_key(async_connection: Mock) -> None:
+    """Ensure simple data can be parsed."""
+    open_response = (
+        '{"data": {"token": 123456789}, "event": "ms.channel.connect", "from": "host"}'
+    )
+    open_response_future = asyncio.Future()
+    open_response_future.set_result(open_response)
+
+    send_command_future = asyncio.Future()
+    send_command_future.set_result(None)
+
+    async_connection.recv = Mock(side_effect=[open_response_future])
+    async_connection.send = Mock(return_value=send_command_future)
+
+    sleep_future = asyncio.Future()
+    sleep_future.set_result(None)
+
+    tv = SamsungTVWSAsyncRemote("127.0.0.1")
+    with patch(
+        "samsungtvws.async_connection.asyncio.sleep", return_value=sleep_future
+    ) as patch_sleep:
+        await tv.send_command(SendRemoteKey.hold_key("KEY_POWER", 3))
+
+    assert patch_sleep.call_count == 3
+    assert patch_sleep.call_args_list == [call(1), call(3), call(1)]

tests/test_remote.py

@@ -1,5 +1,5 @@
 """Tests for remote module."""
-from unittest.mock import Mock
+from unittest.mock import Mock, call, patch
 
 from samsungtvws.remote import SamsungTVWS
 
@@ -63,3 +63,18 @@ def test_app_list_invalid(connection: Mock) -> None:
     connection.send.assert_called_once_with(
         '{"method": "ms.channel.emit", "params": {"event": "ed.installedApp.get", "to": "host"}}'
     )
+
+
+def test_send_hold_key(connection: Mock) -> None:
+    """Ensure simple data can be parsed."""
+    open_response = (
+        '{"data": {"token": 123456789}, "event": "ms.channel.connect", "from": "host"}'
+    )
+    connection.recv.side_effect = [open_response]
+
+    tv = SamsungTVWS("127.0.0.1")
+    with patch("samsungtvws.connection.time.sleep") as patch_sleep:
+        tv.hold_key("KEY_POWER", 3)
+
+    assert patch_sleep.call_count == 3
+    assert patch_sleep.call_args_list == [call(1), call(3), call(1)]