Feature: TCPUART api (#31)

src/linkplay/consts.py

@@ -4,6 +4,8 @@
 API_TIMEOUT: int = 2
 UNKNOWN_TRACK_PLAYING: str = "Unknown"
 UPNP_DEVICE_TYPE = "urn:schemas-upnp-org:device:MediaRenderer:1"
+TCPPORT = 8899
+TCP_MESSAGE_LENGTH = 1024
 
 MTLS_CERTIFICATE_CONTENTS = """
 -----BEGIN PRIVATE KEY-----
@@ -92,6 +94,16 @@ class LinkPlayCommand(StrEnum):
     PLAY_PRESET = "MCUKeyShortClick:{}"
 
 
+class LinkPlayTcpUartCommand(StrEnum):
+    """Defined LinkPlay TCPUART commands."""
+
+    GET_METADATA = "MCU+MEA+GET"
+    PRESET_PLAY = "MCU+KEY+{:03}"
+    PRESET_NEXT = "MCU+KEY+NXT"
+    INPUT_WIFI = "MCU+PLM+000"
+    INPUT_BLUETOOTH = "MCU+PLM+006"
+
+
 class SpeakerType(StrEnum):
     """Defines the speaker type."""
 

src/linkplay/endpoint.py

@@ -1,8 +1,15 @@
+import asyncio
 from abc import ABC, abstractmethod
 
 from aiohttp import ClientSession
 
-from linkplay.utils import session_call_api_json, session_call_api_ok
+from linkplay.consts import TCPPORT
+from linkplay.utils import (
+    call_tcpuart,
+    call_tcpuart_json,
+    session_call_api_json,
+    session_call_api_ok,
+)
 
 
 class LinkPlayEndpoint(ABC):
@@ -38,3 +45,20 @@ async def json_request(self, command: str) -> dict[str, str]:
 
     def __str__(self) -> str:
         return self._endpoint
+
+
+class LinkPlayTcpUartEndpoint(LinkPlayEndpoint):
+    """Represents a LinkPlay TCPUART API endpoint."""
+
+    def __init__(
+        self, *, connection: tuple[asyncio.StreamReader, asyncio.StreamWriter]
+    ):
+        self._connection = connection
+
+    async def request(self, command: str) -> None:
+        reader, writer = self._connection
+        await call_tcpuart(reader, writer, command)
+
+    async def json_request(self, command: str) -> dict[str, str]:
+        reader, writer = self._connection
+        return await call_tcpuart_json(reader, writer, command)

src/linkplay/utils.py

@@ -1,6 +1,7 @@
 import asyncio
 import contextlib
 import json
+import logging
 import os
 import socket
 import ssl
@@ -17,11 +18,14 @@
     API_ENDPOINT,
     API_TIMEOUT,
     MTLS_CERTIFICATE_CONTENTS,
+    TCP_MESSAGE_LENGTH,
     PlayerAttribute,
     PlayingStatus,
 )
 from linkplay.exceptions import LinkPlayRequestException
 
+_LOGGER = logging.getLogger(__name__)
+
 
 async def session_call_api(endpoint: str, session: ClientSession, command: str) -> str:
     """Calls the LinkPlay API and returns the result as a string.
@@ -74,6 +78,44 @@ async def session_call_api_ok(
         raise LinkPlayRequestException(f"Didn't receive expected OK from {endpoint}")
 
 
+async def call_tcpuart(
+    reader: asyncio.StreamReader, writer: asyncio.StreamWriter, cmd: str
+) -> str:
+    """Get the latest data from TCP UART service."""
+    payload_header: str = "18 96 18 20 "
+    payload_length: str = format(len(cmd), "02x")
+    payload_command_header: str = " 00 00 00 c1 02 00 00 00 00 00 00 00 00 00 00 "
+    payload_command_content: str = " ".join(hex(ord(c))[2:] for c in cmd)
+
+    async with async_timeout.timeout(API_TIMEOUT):
+        writer.write(
+            bytes.fromhex(
+                payload_header
+                + payload_length
+                + payload_command_header
+                + payload_command_content
+            )
+        )
+
+        data: bytes = await reader.read(TCP_MESSAGE_LENGTH)
+
+        if data == b"":
+            raise LinkPlayRequestException("No data received from socket")
+
+        return str(repr(data))
+
+
+async def call_tcpuart_json(
+    reader: asyncio.StreamReader, writer: asyncio.StreamWriter, cmd: str
+) -> dict[str, str]:
+    """Get JSON data from TCPUART service."""
+    raw_response: str = await call_tcpuart(reader, writer, cmd)
+    strip_start = raw_response.find("{")
+    strip_end = raw_response.find("}", strip_start) + 1
+    data = raw_response[strip_start:strip_end]
+    return json.loads(data)  # type: ignore
+
+
 def decode_hexstr(hexstr: str) -> str:
     """Decode a hex string."""
     try: