""" Android hardware Serial 组件 — SerialManager, SerialPort, SerialUsb 方法签名完全对齐 Android SDK """ from __future__ import annotations import io import os import time from typing import Any, Callable, Dict, List, Optional, Tuple from .._core import Stub, AndroidOnlyError __all__ = [ "SerialManager", "SerialPort", "SerialUsb", ] # ═══════════════════════════════════════════════════════════════ # 串口常量 # ═══════════════════════════════════════════════════════════════ SERIAL_PORT_MODE_READ = 0 SERIAL_PORT_MODE_WRITE = 1 SERIAL_PORT_MODE_READ_WRITE = 2 SERIAL_PARITY_NONE = 0 SERIAL_PARITY_ODD = 1 SERIAL_PARITY_EVEN = 2 SERIAL_PARITY_MARK = 3 SERIAL_PARITY_SPACE = 4 SERIAL_STOP_BITS_1 = 1 SERIAL_STOP_BITS_1_5 = 3 SERIAL_STOP_BITS_2 = 2 SERIAL_DATA_BITS_5 = 5 SERIAL_DATA_BITS_6 = 6 SERIAL_DATA_BITS_7 = 7 SERIAL_DATA_BITS_8 = 8 SERIAL_FLOW_CONTROL_NONE = 0 SERIAL_FLOW_CONTROL_RTS_CTS = 1 SERIAL_FLOW_CONTROL_XON_XOFF = 2 SERIAL_FLOW_CONTROL_DTR_DSR = 4 SERIAL_BAUD_300 = 300 SERIAL_BAUD_1200 = 1200 SERIAL_BAUD_2400 = 2400 SERIAL_BAUD_4800 = 4800 SERIAL_BAUD_9600 = 9600 SERIAL_BAUD_19200 = 19200 SERIAL_BAUD_38400 = 38400 SERIAL_BAUD_57600 = 57600 SERIAL_BAUD_115200 = 115200 SERIAL_BAUD_230400 = 230400 SERIAL_BAUD_460800 = 460800 SERIAL_BAUD_921600 = 921600 SERIAL_ERROR_NONE = 0 SERIAL_ERROR_PORT_NOT_FOUND = -1 SERIAL_ERROR_PORT_IN_USE = -2 SERIAL_ERROR_PERMISSION_DENIED = -3 SERIAL_ERROR_IO_ERROR = -4 SERIAL_ERROR_UNSUPPORTED = -5 SERIAL_ERROR_TIMEOUT = -6 SERIAL_ERROR_DISCONNECTED = -7 # ═══════════════════════════════════════════════════════════════ # SerialPort # ═══════════════════════════════════════════════════════════════ class SerialPort(Stub): """SerialPort — 串口""" def __init__(self, name: str = "", baudRate: int = SERIAL_BAUD_9600, dataBits: int = SERIAL_DATA_BITS_8, parity: int = SERIAL_PARITY_NONE, stopBits: int = SERIAL_STOP_BITS_1, flowControl: int = SERIAL_FLOW_CONTROL_NONE, mode: int = SERIAL_PORT_MODE_READ_WRITE): super().__init__() self._name = name self._baud_rate = baudRate self._data_bits = dataBits self._parity = parity self._stop_bits = stopBits self._flow_control = flowControl self._mode = mode self._is_open = False self._fd: Optional[Any] = None self._timeout_ms = 1000 self._buffer_size = 4096 self._read_buffer = b"" self._write_buffer = b"" def getName(self) -> str: return self._name def getBaudRate(self) -> int: return self._baud_rate def setBaudRate(self, baudRate: int): self._baud_rate = baudRate def getDataBits(self) -> int: return self._data_bits def setDataBits(self, dataBits: int): self._data_bits = dataBits def getParity(self) -> int: return self._parity def setParity(self, parity: int): self._parity = parity def getStopBits(self) -> int: return self._stop_bits def setStopBits(self, stopBits: int): self._stop_bits = stopBits def getFlowControl(self) -> int: return self._flow_control def setFlowControl(self, flowControl: int): self._flow_control = flowControl def getMode(self) -> int: return self._mode def setTimeout(self, timeoutMs: int): self._timeout_ms = timeoutMs def getTimeout(self) -> int: return self._timeout_ms def isOpen(self) -> bool: return self._is_open def open(self) -> int: if self._is_open: return SERIAL_ERROR_NONE try: self._fd = io.FileIO(self._name, "rb+" if self._mode == SERIAL_PORT_MODE_READ_WRITE else "rb" if self._mode == SERIAL_PORT_MODE_READ else "wb") self._is_open = True return SERIAL_ERROR_NONE except (IOError, OSError) as e: if "Permission denied" in str(e): return SERIAL_ERROR_PERMISSION_DENIED if "No such file" in str(e): return SERIAL_ERROR_PORT_NOT_FOUND return SERIAL_ERROR_IO_ERROR def close(self): if self._fd: try: self._fd.close() except Exception: pass self._fd = None self._is_open = False def read(self, size: int = -1) -> Optional[bytes]: if not self._is_open or not self._fd: return None try: if self._mode == SERIAL_PORT_MODE_WRITE: return b"" if size < 0: data = self._fd.read() else: data = self._fd.read(size) return data if data else b"" except Exception: return None def write(self, data: bytes) -> int: if not self._is_open or not self._fd: return SERIAL_ERROR_PORT_NOT_FOUND try: if self._mode == SERIAL_PORT_MODE_READ: return SERIAL_ERROR_PERMISSION_DENIED written = self._fd.write(data) return written if written else len(data) except Exception: return SERIAL_ERROR_IO_ERROR def flush(self) -> int: if not self._is_open: return SERIAL_ERROR_PORT_NOT_FOUND try: if hasattr(self._fd, 'flush'): self._fd.flush() return SERIAL_ERROR_NONE except Exception: return SERIAL_ERROR_IO_ERROR def getAvailableBytes(self) -> int: if not self._is_open or not self._fd: return 0 try: import os if hasattr(self._fd, 'fileno'): fd = self._fd.fileno() import select r, _, _ = select.select([fd], [], [], 0) if r: import struct import fcntl try: buf = fcntl.ioctl(fd, 0x541B, struct.pack('I', 0)) # FIONREAD return struct.unpack('I', buf)[0] except Exception: pass except Exception: pass return 0 def setDTR(self, value: bool): pass def setRTS(self, value: bool): pass def getDCD(self) -> bool: return False def getDSR(self) -> bool: return False def getCTS(self) -> bool: return False def getRI(self) -> bool: return False def sendBreak(self, durationMs: int = 100): pass def toString(self) -> str: return (f"SerialPort(name={self._name}, baud={self._baud_rate}, " f"open={self._is_open})") def equals(self, other: Any) -> bool: if not isinstance(other, SerialPort): return False return self._name == other._name def hashCode(self) -> int: return hash(self._name) & 0x7FFFFFFF # ═══════════════════════════════════════════════════════════════ # SerialUsb # ═══════════════════════════════════════════════════════════════ class SerialUsb(Stub): """SerialUsb — USB 串口""" USB_VENDOR_FTDI = 0x0403 USB_VENDOR_CP210X = 0x10C4 USB_VENDOR_CH340 = 0x1A86 USB_VENDOR_PL2303 = 0x067B USB_VENDOR_ARDUINO = 0x2341 USB_VENDOR_SILABS = 0x10C4 USB_PRODUCT_FTDI_FT232 = 0x6001 USB_PRODUCT_CP2102 = 0xEA60 USB_PRODUCT_CH340 = 0x7523 USB_PRODUCT_PL2303 = 0x2303 def __init__(self, vendorId: int = 0, productId: int = 0, serialNumber: str = "", manufacturer: str = "", product: str = "", devicePort: str = "", deviceNode: str = ""): super().__init__() self._vendor_id = vendorId self._product_id = productId self._serial_number = serialNumber self._manufacturer = manufacturer self._product = product self._device_port = devicePort self._device_node = deviceNode self._port: Optional[SerialPort] = None def getVendorId(self) -> int: return self._vendor_id def getProductId(self) -> int: return self._product_id def getSerialNumber(self) -> str: return self._serial_number def getManufacturer(self) -> str: return self._manufacturer def getProduct(self) -> str: return self._product def getDevicePort(self) -> str: return self._device_port def getDeviceNode(self) -> str: return self._device_node def createSerialPort(self, baudRate: int = SERIAL_BAUD_9600, dataBits: int = SERIAL_DATA_BITS_8, parity: int = SERIAL_PARITY_NONE, stopBits: int = SERIAL_STOP_BITS_1, flowControl: int = SERIAL_FLOW_CONTROL_NONE) -> Optional[SerialPort]: if not self._device_node and not self._device_port: return None port_name = self._device_node or f"/dev/{self._device_port}" port = SerialPort(name=port_name, baudRate=baud_rate, dataBits=dataBits, parity=parity, stopBits=stopBits, flowControl=flowControl) self._port = port return port def getSerialPort(self) -> Optional[SerialPort]: return self._port def toString(self) -> str: return (f"SerialUsb(vendor=0x{self._vendor_id:04X}, " f"product=0x{self._product_id:04X})") def equals(self, other: Any) -> bool: if not isinstance(other, SerialUsb): return False return (self._vendor_id == other._vendor_id and self._product_id == other._product_id and self._serial_number == other._serial_number) def hashCode(self) -> int: return hash((self._vendor_id, self._product_id, self._serial_number)) & 0x7FFFFFFF # ═══════════════════════════════════════════════════════════════ # SerialManager # ═══════════════════════════════════════════════════════════════ class SerialManager(Stub): """SerialManager — 串口管理器""" ACTION_SERIAL_PORT_ADDED = "android.hardware.usb.action.USB_DEVICE_ATTACHED" ACTION_SERIAL_PORT_REMOVED = "android.hardware.usb.action.USB_DEVICE_DETACHED" EXTRA_SERIAL_PORT = "serial_port" EXTRA_USB_DEVICE = "device" def __init__(self): super().__init__() self._ports: Dict[str, SerialPort] = {} self._usb_devices: Dict[str, SerialUsb] = {} self._listeners: Dict[int, Callable] = {} def getSerialPorts(self) -> List[SerialPort]: return list(self._ports.values()) def getSerialPort(self, name: str) -> Optional[SerialPort]: return self._ports.get(name) def openSerialPort(self, name: str, baudRate: int = SERIAL_BAUD_9600, dataBits: int = SERIAL_DATA_BITS_8, parity: int = SERIAL_PARITY_NONE, stopBits: int = SERIAL_STOP_BITS_1, flowControl: int = SERIAL_FLOW_CONTROL_NONE, mode: int = SERIAL_PORT_MODE_READ_WRITE) -> Optional[SerialPort]: if name in self._ports and self._ports[name].isOpen(): return None port = SerialPort(name=name, baudRate=baudRate, dataBits=dataBits, parity=parity, stopBits=stopBits, flowControl=flowControl, mode=mode) err = port.open() if err != SERIAL_ERROR_NONE: return None self._ports[name] = port return port def closeSerialPort(self, name: str): port = self._ports.get(name) if port: port.close() self._ports.pop(name, None) def getUsbDevices(self) -> List[SerialUsb]: return list(self._usb_devices.values()) def addUsbDevice(self, device: SerialUsb): self._usb_devices[device.getDeviceNode() or device.getDevicePort()] = device def removeUsbDevice(self, device: SerialUsb): key = device.getDeviceNode() or device.getDevicePort() self._usb_devices.pop(key, None) def getAvailableDevices(self) -> List[str]: """列出系统可用的串口设备路径""" ports = [] for base in ("/dev/ttyS", "/dev/ttyUSB", "/dev/ttyACM", "/dev/ttyAMA", "/dev/ttyO", "/dev/cu.usbserial", "/dev/tty.wchusbserial"): for i in range(256): path = f"{base}{i}" if os.path.exists(path): ports.append(path) return ports + list(self._ports.keys()) def getSerialPortsForDevice(self, usbDevice: SerialUsb) -> List[SerialPort]: return [p for n, p in self._ports.items() if usbDevice.getDeviceNode() in n or usbDevice.getDevicePort() in n] def registerListener(self, listener: Callable): lid = id(listener) self._listeners[lid] = listener def unregisterListener(self, listener: Callable): lid = id(listener) self._listeners.pop(lid, None) def isSerialPortName(self, name: str) -> bool: return name.startswith(("ttyS", "ttyUSB", "ttyACM", "ttyAMA")) def toString(self) -> str: return f"SerialManager(ports={len(self._ports)}, usb={len(self._usb_devices)})" def equals(self, other: Any) -> bool: return self is other def hashCode(self) -> int: return id(self) & 0x7FFFFFFF def getSerialPortCount(self) -> int: return len(self._ports) def getUsbDeviceCount(self) -> int: return len(self._usb_devices) def hasSerialPort(self, name: str) -> bool: return name in self._ports def hasUsbDevice(self, vendorId: int, productId: int) -> bool: for dev in self._usb_devices.values(): if dev.getVendorId() == vendorId and dev.getProductId() == productId: return True return False def findUsbDeviceBySerial(self, serialNumber: str) -> Optional[SerialUsb]: for dev in self._usb_devices.values(): if dev.getSerialNumber() == serialNumber: return dev return None def findUsbDeviceByVendorProduct(self, vendorId: int, productId: int) -> List[SerialUsb]: return [dev for dev in self._usb_devices.values() if dev.getVendorId() == vendorId and dev.getProductId() == productId] def closeAllPorts(self) -> None: for name, port in list(self._ports.items()): port.close() self._ports.pop(name, None) def getPortStatus(self, name: str) -> int: port = self._ports.get(name) if port is None: return SERIAL_ERROR_PORT_NOT_FOUND if not port.isOpen(): return SERIAL_ERROR_PORT_NOT_FOUND return SERIAL_ERROR_NONE def reconfigurePort(self, name: str, baudRate: int = 0, dataBits: int = 0, parity: int = -1, stopBits: int = 0, flowControl: int = -1) -> int: port = self._ports.get(name) if port is None: return SERIAL_ERROR_PORT_NOT_FOUND if baudRate > 0: port.setBaudRate(baudRate) if dataBits > 0: port.setDataBits(dataBits) if parity >= 0: port.setParity(parity) if stopBits > 0: port.setStopBits(stopBits) if flowControl >= 0: port.setFlowControl(flowControl) return SERIAL_ERROR_NONE def addSerialPort(self, port: SerialPort) -> None: self._ports[port.getName()] = port def removeSerialPort(self, name: str) -> None: self._ports.pop(name, None) def getSerialPortNames(self) -> List[str]: return list(self._ports.keys()) def getUsbDeviceSerialNumbers(self) -> List[str]: return [dev.getSerialNumber() for dev in self._usb_devices.values()] def clearAllDevices(self) -> None: self._usb_devices.clear() for port in self._ports.values(): if port.isOpen(): port.close() self._ports.clear()