Function run_device_decode_autoresearch

ci/autoresearch/decode.py:142–265

Send media file to device via JSON-RPC and decode on-device. Args: decode_path: Local file path or URL to a media file. port: Serial port for device communication. timeout: RPC timeout in seconds. serial_interface: Optional serial interface override. Returns

(
    decode_path: str,
    port: str,
    timeout: float = 120.0,
    serial_interface: "SerialInterface | None" = None,
)

async def run_device_decode_autoresearch(
    decode_path: str,
    port: str,
    timeout: float = 120.0,
    serial_interface: "SerialInterface | None" = None,
) -> int:
    """Send media file to device via JSON-RPC and decode on-device.

    Args:
        decode_path: Local file path or URL to a media file.
        port: Serial port for device communication.
        timeout: RPC timeout in seconds.
        serial_interface: Optional serial interface override.

    Returns:
        Exit code (0 = success, 1 = failure).
    """
    from ci.rpc_client import RpcClient, RpcTimeoutError
    from ci.util.global_interrupt_handler import handle_keyboard_interrupt

    print()
    print("=" * 60)
    print("DEVICE DECODE AUTORESEARCH MODE")
    print("=" * 60)
    print()

    temp_download: str | None = None
    client: RpcClient | None = None

    try:
        resolved = _resolve_file(decode_path)
        file_path = resolved.file_path
        ext = resolved.extension
        temp_download = resolved.temp_download_path

        if ext not in SUPPORTED_EXTENSIONS:
            supported = ", ".join(sorted(SUPPORTED_EXTENSIONS.keys()))
            print(
                f"{Fore.RED}Error: Unsupported file extension '{ext}'{Style.RESET_ALL}"
            )
            print(f"  Supported: {supported}")
            return 1

        if not os.path.isfile(file_path):
            print(f"{Fore.RED}Error: File not found: {file_path}{Style.RESET_ALL}")
            return 1

        codec_name = SUPPORTED_EXTENSIONS[ext]
        file_size = os.path.getsize(file_path)
        print(f"  File:   {file_path}")
        print(f"  Size:   {file_size:,} bytes")
        print(f"  Codec:  {codec_name}")
        print(f"  Port:   {port}")
        print()

        with open(file_path, "rb") as f:
            file_bytes = f.read()