Send text using commands via simple cmds

requirements.txt

@@ -6,4 +6,5 @@ pillow>=11.2.1
 requests>=2.32
 pychromecast~=14.0.7
 httpx~=0.28.1
-sanitize-filename~=1.2.0
+sanitize-filename~=1.2.0
+adb-shell==0.4.4

src/adb_tv.py

@@ -0,0 +1,109 @@
+"""
+This module provides utilities for interacting with Android TVs via ADB (Android Debug Bridge).
+It includes functions for managing ADB keys, connecting to devices, retrieving installed apps,
+and verifying device authorization.
+"""
+
+import asyncio
+import os
+from pathlib import Path
+from typing import Dict, Optional
+
+from adb_shell.adb_device_async import AdbDeviceTcpAsync
+from adb_shell.auth.keygen import keygen
+from adb_shell.auth.sign_pythonrsa import PythonRSASigner
+
+ADB_CERTS_DIR = Path(os.environ.get("UC_CONFIG_HOME", "./config")) / "certs"
+ADB_CERTS_DIR.mkdir(parents=True, exist_ok=True)
+
+
+def get_adb_key_paths(device_id: str) -> tuple[Path, Path]:
+    """
+    Return the paths to the private and public ADB keys for a given device.
+
+    Args:
+        device_id (str): The unique identifier for the device.
+
+    Returns:
+        tuple[Path, Path]: Paths to the private and public key files.
+    """
+    priv = ADB_CERTS_DIR / f"adb_{device_id}"
+    pub = ADB_CERTS_DIR / f"adb_{device_id}.pub"
+    return priv, pub
+
+
+def load_or_generate_adb_keys(device_id: str) -> PythonRSASigner:
+    """
+    Ensure ADB RSA keys exist for the device and return the signer.
+
+    Args:
+        device_id (str): The unique identifier for the device.
+
+    Returns:
+        PythonRSASigner: The signer object for ADB authentication.
+    """
+    priv_path, pub_path = get_adb_key_paths(device_id)
+
+    if not priv_path.exists() or not pub_path.exists():
+        keygen(str(priv_path))
+
+    with open(priv_path) as f:
+        priv = f.read()
+    with open(pub_path) as f:
+        pub = f.read()
+    return PythonRSASigner(pub, priv)
+
+
+async def adb_connect(device_id: str, host: str, port: int = 5555) -> Optional[AdbDeviceTcpAsync]:
+    """
+    Connect to an Android device via ADB.
+
+    Args:
+        device_id (str): The unique identifier for the device.
+        host (str): The IP address or hostname of the device.
+        port (int, optional): The port number for the ADB connection. Defaults to 5555.
+
+    Returns:
+        Optional[AdbDeviceTcpAsync]: The connected ADB device object, or None if the connection fails.
+    """
+    signer = load_or_generate_adb_keys(device_id)
+    device = AdbDeviceTcpAsync(host, port, default_transport_timeout_s=9.0)
+
+    try:
+        await device.connect(rsa_keys=[signer], auth_timeout_s=20)
+        return device
+    except Exception as e:
+        print(f"ADB connection failed to {host}:{port} — {e}")
+        return None
+
+
+async def get_installed_apps(device: AdbDeviceTcpAsync) -> Dict[str, Dict[str, str]]:
+    """
+    Retrieve a list of installed non-system apps in a structured format.
+
+    Args:
+        device (AdbDeviceTcpAsync): The connected ADB device.
+
+    Returns:
+        Dict[str, Dict[str, str]]: A dictionary of app package names and their metadata.
+    """
+    output = await device.shell("pm list packages -3 -e")
+    packages = sorted(line.replace("package:", "").strip() for line in output.splitlines())
+    return {package: {"url": f"market://launch?id={package}"} for package in packages}
+
+
+async def is_authorised(device: AdbDeviceTcpAsync) -> bool:
+    """
+    Check if the connected device is authorized for ADB communication.
+
+    Args:
+        device (AdbDeviceTcpAsync): The connected ADB device.
+
+    Returns:
+        bool: True if the device is authorized, False otherwise.
+    """
+    try:
+        result = await device.shell("echo ADB_OK")
+        return "ADB_OK" in result
+    except Exception:
+        return False

src/config.py

@@ -11,13 +11,27 @@
 import logging
 import os
 from dataclasses import dataclass
+from pathlib import Path
 from typing import Iterator
 
 _LOG = logging.getLogger(__name__)
 
 _CFG_FILENAME = "config.json"
 
 
+# Paths
+def _get_config_root() -> Path:
+    config_home = Path(os.environ.get("UC_CONFIG_HOME", "./config"))
+    config_home.mkdir(parents=True, exist_ok=True)
+    return config_home
+
+
+def _get_data_root() -> Path:
+    data_home = Path(os.environ.get("UC_DATA_HOME", "./data"))
+    data_home.mkdir(parents=True, exist_ok=True)
+    return data_home
+
+
 @dataclass
 class AtvDevice:
     """Android TV device configuration."""
@@ -38,6 +52,8 @@ class AtvDevice:
     """Enable External Metadata."""
     use_chromecast: bool = False
     """Enable Chromecast features."""
+    use_adb: bool = False
+    """Enable ADB features."""
 
 
 class _EnhancedJSONEncoder(json.JSONEncoder):
@@ -133,24 +149,25 @@ def update(self, atv: AtvDevice) -> bool:
                 item.auth_error = atv.auth_error
                 item.use_external_metadata = atv.use_external_metadata
                 item.use_chromecast = atv.use_chromecast
+                item.use_adb = atv.use_adb
                 return self.store()
         return False
 
     def default_certfile(self) -> str:
         """Return the default certificate file for initializing a device."""
-        return os.path.join(self._data_path, "androidtv_remote_cert.pem")
+        return os.path.join(self._data_path + "/certs", "androidtv_remote_cert.pem")
 
     def default_keyfile(self) -> str:
         """Return the default key file for initializing a device."""
-        return os.path.join(self._data_path, "androidtv_remote_key.pem")
+        return os.path.join(self._data_path + "/certs", "androidtv_remote_key.pem")
 
     def certfile(self, atv_id: str) -> str:
         """Return the certificate file of the device."""
-        return os.path.join(self._data_path, f"androidtv_{atv_id}_remote_cert.pem")
+        return os.path.join(self._data_path + "/certs", f"androidtv_{atv_id}_remote_cert.pem")
 
     def keyfile(self, atv_id: str) -> str:
         """Return the key file of the device."""
-        return os.path.join(self._data_path, f"androidtv_{atv_id}_remote_key.pem")
+        return os.path.join(self._data_path + "/certs", f"androidtv_{atv_id}_remote_key.pem")
 
     def remove(self, atv_id: str) -> bool:
         """Remove the given device configuration."""
@@ -236,6 +253,7 @@ def load(self) -> bool:
                     item.get("auth_error", False),
                     item.get("use_external_metadata", False),
                     item.get("use_chromecast", False),
+                    item.get("use_adb", False),
                 )
                 self._config.append(atv)
             return True

src/external_metadata.py

@@ -21,6 +21,7 @@
 from PIL.Image import Resampling
 from pychromecast.controllers.media import MediaImage
 from sanitize_filename import sanitize
+from config import _get_config_root, _get_data_root
 
 _LOG = logging.getLogger(__name__)
 
@@ -30,27 +31,14 @@
 
 
 # Paths
-def _get_config_root() -> Path:
-    config_home = Path(os.environ.get("UC_CONFIG_HOME", "./config"))
-    config_home.mkdir(parents=True, exist_ok=True)
-    return config_home
-
-
-def _get_cache_root() -> Path:
-    data_home = Path(os.environ.get("UC_DATA_HOME", "./data"))
-    cache_root = data_home / CACHE_ROOT
-    cache_root.mkdir(parents=True, exist_ok=True)
-    return cache_root
-
-
 def _get_metadata_dir() -> Path:
-    metadata_dir = _get_cache_root()
+    metadata_dir = _get_data_root() / CACHE_ROOT
     metadata_dir.mkdir(parents=True, exist_ok=True)
     return metadata_dir
 
 
 def _get_icon_dir() -> Path:
-    icon_dir = _get_cache_root() / ICON_SUBDIR
+    icon_dir = _get_data_root() / CACHE_ROOT / ICON_SUBDIR
     icon_dir.mkdir(parents=True, exist_ok=True)
     return icon_dir
 

src/profiles.py

@@ -263,6 +263,25 @@ def match(self, manufacturer: str, model: str, use_chromecast: bool) -> Profile:
             select_profile = copy.copy(select_profile)
             select_profile.features.extend(CHROMECAST_FEATURES)
 
+        import string
+        for char in string.ascii_uppercase + " ":
+            # Use 'SPACE' as the key name for the space character
+            key = "SPACE" if char == " " else char
+            command_name = f"TEXT_{key}"
+
+            # Append to simple_commands list
+            select_profile.simple_commands.append(command_name)
+
+            # Map the command: space gets ' ', others get lowercase letter
+            command_value = " " if char == " " else char.lower()
+            select_profile.command_map[command_name] = Command(command_value, 'TEXT')
+
+        select_profile.simple_commands.append('TEXT_BACKSPACE')
+        select_profile.command_map['TEXT_BACKSPACE'] = Command('DEL', KeyPress.SHORT)
+
+        select_profile.simple_commands.append('TEXT_ENTER')
+        select_profile.command_map['TEXT_ENTER'] = Command('ENTER', KeyPress.SHORT)
+
         return select_profile