Android-pyApi/android_pyapi/android/os.py

"""
Android OS 组件 — Bundle, Handler, Looper, Message, Build 等
方法签名完全对齐 Android SDK
"""

import asyncio
import functools
import heapq
import os as _os
import sys
import threading
import time as _time
from datetime import datetime
from typing import Any, Callable, Dict, List, Optional, Set, Tuple, TypeVar, Union

from .._core import (
    Stub, AndroidOnlyError, android_context,
    resolve_android_path, to_android_path,
    _sdcard_root, _app_data_dir, _package_name,
    HAS_PSUTIL,
)

# Linux 真实实现导入
try:
    from .._linux_impl import IS_LINUX, LinuxPower, LinuxProcess, HAS_PSUTIL as _HAS_PSUTIL
    _USE_LINUX = IS_LINUX
except ImportError:
    _USE_LINUX = False


# ═══════════════════════════════════════════════════════════════
# Build
# ═══════════════════════════════════════════════════════════════

class Build(Stub):
    """Build — 设备信息"""

    class VERSION(Stub):
        """版本信息"""
        SDK_INT = 34  # Android 14
        RELEASE = "14"
        INCREMENTAL = "20240301"
        CODENAME = "REL"
        SECURITY_PATCH = "2024-03-05"
        BASE_OS = ""
        PREVIEW_SDK_INT = 0
        RESOURCES_SDK_INT = 34
        MEDIA_PERFORMANCE_CLASS = 0

    class VERSION_CODES(Stub):
        """版本代码常量"""
        BASE = 1
        BASE_1_1 = 2
        CUPCAKE = 3
        DONUT = 4
        ECLAIR = 5
        ECLAIR_0_1 = 6
        ECLAIR_MR1 = 7
        FROYO = 8
        GINGERBREAD = 9
        GINGERBREAD_MR1 = 10
        HONEYCOMB = 11
        HONEYCOMB_MR1 = 12
        HONEYCOMB_MR2 = 13
        ICE_CREAM_SANDWICH = 14
        ICE_CREAM_SANDWICH_MR1 = 15
        JELLY_BEAN = 16
        JELLY_BEAN_MR1 = 17
        JELLY_BEAN_MR2 = 18
        KITKAT = 19
        KITKAT_WATCH = 20
        LOLLIPOP = 21
        LOLLIPOP_MR1 = 22
        M = 23
        N = 24
        N_MR1 = 25
        O = 26
        O_MR1 = 27
        P = 28
        Q = 29
        R = 30
        S = 31
        S_V2 = 32
        TIRAMISU = 33
        UPSIDE_DOWN_CAKE = 34

    VERSION = VERSION()
    BOARD = "redfin"
    BOOTLOADER = "r7.0-12345"
    BRAND = "google"
    CPU_ABI = "arm64-v8a"
    CPU_ABI2 = ""
    DEVICE = "redfin"
    DISPLAY = "AP1A.240305.001"
    FINGERPRINT = "google/redfin/redfin:14/AP1A.240305.001/12345678:user/release-keys"
    HARDWARE = "redfin"
    HOST = "build-test-01"
    ID = "AP1A.240305.001"
    MANUFACTURER = "Google"
    MODEL = "Pixel 5"
    ODM_SKU = None
    PRODUCT = "redfin"
    RADIO = ""
    SERIAL = "unknown"
    SKU = None
    SOC_MANUFACTURER = "Qualcomm"
    SOC_MODEL = "Snapdragon 765G"
    SUPPORTED_32_BIT_ABIS = ["armeabi-v7a", "armeabi"]
    SUPPORTED_64_BIT_ABIS = ["arm64-v8a"]
    SUPPORTED_ABIS = ["arm64-v8a", "armeabi-v7a", "armeabi"]
    TAGS = "release-keys"
    TIME = int(_time.time() * 1000)
    TYPE = "user"
    USER = "build-user"

    @staticmethod
    def getRadioVersion() -> str:
        """返回基带/无线电版本。桌面端无无线电硬件，返回空字符串。"""
        return Build.RADIO

    @staticmethod
    def getSerial() -> str:
        """返回设备序列号。尝试从 /sys 或 /proc 读取真实硬件序列号。"""
        # 尝试从 /sys/class/android_usb 读取
        try:
            for name in _os.listdir("/sys/class/android_usb"):
                serial_path = f"/sys/class/android_usb/{name}/iSerial"
                if _os.path.exists(serial_path):
                    with open(serial_path, "r") as f:
                        serial = f.read().strip()
                        if serial:
                            return serial
        except (OSError, IOError):
            pass
        # 尝试从 /proc/cpuinfo 读取 Serial 字段
        try:
            with open("/proc/cpuinfo", "r") as f:
                for line in f:
                    if line.lower().startswith("serial"):
                        parts = line.split(":", 1)
                        if len(parts) == 2:
                            serial = parts[1].strip()
                            if serial and serial != "0" * len(serial):
                                return serial
        except (OSError, IOError):
            pass
        # 尝试从 /sys/class/dmi/id/product_serial 读取
        try:
            path = "/sys/class/dmi/id/product_serial"
            if _os.path.exists(path):
                with open(path, "r") as f:
                    serial = f.read().strip()
                    if serial:
                        return serial
        except (OSError, IOError):
            pass
        return Build.SERIAL

    @staticmethod
    def getFingerprint() -> str:
        return Build.FINGERPRINT

    @classmethod
    def _populate_from_linux(cls):
        """从 LinuxProcess.get_system_info() 填充真实设备信息"""
        try:
            info = LinuxProcess.get_system_info() or {}
            import platform as _platform
            import socket as _socket
            hostname = info.get("hostname") or _socket.gethostname()
            system = info.get("platform") or _platform.system()
            arch = info.get("arch") or _platform.machine()
            kernel = info.get("kernel") or _platform.release()
            cls.MODEL = hostname
            cls.MANUFACTURER = "Linux"
            cls.BRAND = system
            cls.DEVICE = arch
            cls.BOARD = arch
            cls.HARDWARE = arch
            cls.HOST = hostname
            cls.PRODUCT = hostname
            cls.DISPLAY = kernel
            cls.ID = kernel
            # FINGERPRINT: brand/device/device:version/id/build:user/release-keys
            cls.FINGERPRINT = (
                f"{system}/{hostname}/{arch}:Linux/{kernel}/{hostname}:user/release-keys"
            )
            # CPU ABI 用架构映射
            _abi_map = {
                "x86_64": "x86_64", "amd64": "x86_64",
                "aarch64": "arm64-v8a", "arm64": "arm64-v8a",
                "x86": "x86", "i386": "x86", "i686": "x86",
                "armv7l": "armeabi-v7a", "arm": "armeabi-v7a",
            }
            cls.CPU_ABI = _abi_map.get(arch, arch)
            cls.SUPPORTED_ABIS = [cls.CPU_ABI]
            cls.SUPPORTED_64_BIT_ABIS = [cls.CPU_ABI] if "64" in cls.CPU_ABI else []
            cls.SUPPORTED_32_BIT_ABIS = [cls.CPU_ABI] if "64" not in cls.CPU_ABI else []
        except Exception:
            pass


# 在 Linux 上用真实系统信息填充 Build
if _USE_LINUX:
    Build._populate_from_linux()


# ═══════════════════════════════════════════════════════════════
# SystemClock
# ═══════════════════════════════════════════════════════════════

class SystemClock(Stub):
    """SystemClock — 系统时钟"""

    @staticmethod
    def elapsedRealtime() -> int:
        """返回自启动以来的毫秒数"""
        return int(_time.monotonic() * 1000)

    @staticmethod
    def elapsedRealtimeNanos() -> int:
        return int(_time.monotonic_ns())

    @staticmethod
    def uptimeMillis() -> int:
        """返回自启动以来（不含深度睡眠）的毫秒数"""
        return int(_time.monotonic() * 1000)

    @staticmethod
    def currentThreadTimeMillis() -> int:
        return int(_time.clock_gettime(_time.CLOCK_THREAD_CPUTIME_ID) * 1000)

    @staticmethod
    def sleep(ms: int):
        _time.sleep(ms / 1000.0)

    @staticmethod
    def setCurrentTimeMillis(millis: int) -> bool:
        """尝试设置系统时间。需要 root 权限，通常在桌面端会失败。"""
        try:
            import subprocess
            seconds = millis / 1000.0
            # 尝试使用 date 命令设置时间
            result = subprocess.run(
                ["date", "-s", f"@{int(seconds)}"],
                capture_output=True, timeout=5)
            return result.returncode == 0
        except Exception:
            return False


# ═══════════════════════════════════════════════════════════════
# Environment
# ═══════════════════════════════════════════════════════════════

class Environment(Stub):
    """Environment — 环境目录"""

    MEDIA_BAD_REMOVAL = "bad_removal"
    MEDIA_CHECKING = "checking"
    MEDIA_EJECTING = "ejecting"
    MEDIA_MOUNTED = "mounted"
    MEDIA_MOUNTED_READ_ONLY = "mounted_ro"
    MEDIA_NOFS = "nofs"
    MEDIA_REMOVED = "removed"
    MEDIA_SHARED = "shared"
    MEDIA_UNKNOWN = "unknown"
    MEDIA_UNMOUNTABLE = "unmountable"
    MEDIA_UNMOUNTED = "unmounted"

    DIRECTORY_ALARMS = "Alarms"
    DIRECTORY_DCIM = "DCIM"
    DIRECTORY_DOCUMENTS = "Documents"
    DIRECTORY_DOWNLOADS = "Download"
    DIRECTORY_MOVIES = "Movies"
    DIRECTORY_MUSIC = "Music"
    DIRECTORY_NOTIFICATIONS = "Notifications"
    DIRECTORY_PICTURES = "Pictures"
    DIRECTORY_PODCASTS = "Podcasts"
    DIRECTORY_RINGTONES = "Ringtones"
    DIRECTORY_AUDIOBOOKS = "Audiobooks"
    DIRECTORY_SCREENSHOTS = "Screenshots"

    @staticmethod
    def getExternalStorageDirectory() -> str:
        return _sdcard_root

    @staticmethod
    def getExternalStoragePublicDirectory(type_: str) -> str:
        path = _os.path.join(_sdcard_root, type_)
        _os.makedirs(path, exist_ok=True)
        return path

    @staticmethod
    def getDataDirectory() -> str:
        return _os.path.join(_app_data_dir, "data")

    @staticmethod
    def getDownloadCacheDirectory() -> str:
        return _os.path.join(_app_data_dir, "cache", "download")

    @staticmethod
    def getRootDirectory() -> str:
        return _os.path.abspath(_os.sep)

    @staticmethod
    def getStorageDirectory(user_handle: Any = None) -> str:
        return _sdcard_root

    @staticmethod
    def getExternalStorageState(path: Optional[str] = None) -> str:
        return Environment.MEDIA_MOUNTED

    @staticmethod
    def isExternalStorageRemovable(path: Optional[str] = None) -> bool:
        """判断外部存储是否可移除。通过检查 /proc/mounts 判断。"""
        try:
            check_path = path or _sdcard_root
            with open("/proc/mounts", "r") as f:
                for line in f:
                    parts = line.split()
                    if len(parts) >= 3 and check_path in parts[1]:
                        # 检查挂载选项中是否包含 SD 卡相关设备
                        device = parts[0]
                        if "/sd" in device or "/mmc" in device or "/usb" in device:
                            return True
                        return False
        except (OSError, IOError):
            pass
        return False

    @staticmethod
    def isExternalStorageEmulated(path: Optional[str] = None) -> bool:
        return True

    @staticmethod
    def isExternalStorageLegacy(path: Optional[str] = None) -> bool:
        return True


# ═══════════════════════════════════════════════════════════════
# Process
# ═══════════════════════════════════════════════════════════════

class Process(Stub):
    """Process — 进程信息"""

    # 进程优先级
    THREAD_PRIORITY_DEFAULT = 0
    THREAD_PRIORITY_LOWEST = -19
    THREAD_PRIORITY_BACKGROUND = 10
    THREAD_PRIORITY_FOREGROUND = -2
    THREAD_PRIORITY_DISPLAY = -4
    THREAD_PRIORITY_URGENT_DISPLAY = -8
    THREAD_PRIORITY_AUDIO = -16
    THREAD_PRIORITY_URGENT_AUDIO = -19
    THREAD_PRIORITY_MORE_FAVORABLE = -1
    THREAD_PRIORITY_LESS_FAVORABLE = 1

    SIGNAL_KILL = 9
    SIGNAL_QUIT = 3
    SIGNAL_USR1 = 10
    SIGNAL_USR2 = 12

    @staticmethod
    def myTid() -> int:
        return threading.get_ident()

    @staticmethod
    def myUid() -> int:
        """返回当前进程的真实 UID。"""
        return _os.getuid()

    @staticmethod
    def myPid() -> int:
        return _os.getpid()

    @staticmethod
    def killProcess(pid):
        import signal
        try:
            _os.kill(pid, signal.SIGKILL)
        except (OSError, ProcessLookupError):
            pass

    @staticmethod
    def sendSignal(pid, signal_num):
        try:
            _os.kill(pid, signal_num)
        except (OSError, ProcessLookupError):
            pass

    @staticmethod
    def setThreadPriority(*args):
        """设置线程优先级。支持 setThreadPriority(priority) 和 setThreadPriority(tid, priority)。"""
        if len(args) == 1:
            tid = 0
            priority = args[0]
        elif len(args) == 2:
            tid, priority = args
        else:
            return
        try:
            _os.setpriority(_os.PRIO_PROCESS, tid, priority)
        except (OSError, PermissionError):
            pass

    @staticmethod
    def getElapsedCpuTime() -> int:
        return int(_time.clock_gettime(_time.CLOCK_PROCESS_CPUTIME_ID) * 1000)

    @staticmethod
    def supportsProcesses() -> bool:
        return True


# ═══════════════════════════════════════════════════════════════
# BaseBundle / PersistableBundle
# ═══════════════════════════════════════════════════════════════

class BaseBundle(Stub):
    """BaseBundle — Bundle 基类"""

    def __init__(self):
        super().__init__()
        self._map: Dict[str, Any] = {}
        self._class_loader = None

    def getString(self, key: str, default: Optional[str] = None) -> Optional[str]:
        v = self._map.get(key, default)
        return str(v) if v is not None else default

    def getInt(self, key: str, default: int = 0) -> int:
        v = self._map.get(key, default)
        try:
            return int(v) if v is not None else default
        except (ValueError, TypeError):
            return default

    def getLong(self, key: str, default: int = 0) -> int:
        return self.getInt(key, default)

    def getFloat(self, key: str, default: float = 0.0) -> float:
        v = self._map.get(key, default)
        try:
            return float(v) if v is not None else default
        except (ValueError, TypeError):
            return default

    def getDouble(self, key: str, default: float = 0.0) -> float:
        return self.getFloat(key, default)

    def getBoolean(self, key: str, default: bool = False) -> bool:
        v = self._map.get(key, default)
        if isinstance(v, bool):
            return v
        if isinstance(v, str):
            return v.lower() in ("true", "1", "yes")
        return bool(v) if v is not None else default

    def getByte(self, key: str, default: int = 0) -> int:
        return self.getInt(key, default)

    def getChar(self, key: str, default: str = "") -> str:
        return str(self._map.get(key, default))

    def getShort(self, key: str, default: int = 0) -> int:
        return self.getInt(key, default)

    def getCharSequence(self, key: str, default: Optional[str] = None) -> Optional[str]:
        return self.getString(key, default)

    def getSerializable(self, key: str, default: Any = None) -> Any:
        return self._map.get(key, default)

    def getStringArrayList(self, key: str) -> Optional[List[str]]:
        v = self._map.get(key)
        if isinstance(v, list):
            return [str(x) for x in v]
        return None

    def getIntegerArrayList(self, key: str) -> Optional[List[int]]:
        v = self._map.get(key)
        if isinstance(v, list):
            return [int(x) for x in v if isinstance(x, (int, float))]
        return None

    def containsKey(self, key: str) -> bool:
        return key in self._map

    def isEmpty(self) -> bool:
        return len(self._map) == 0

    def size(self) -> int:
        return len(self._map)

    def clear(self):
        self._map.clear()

    def keySet(self) -> Set[str]:
        return set(self._map.keys())

    def hasKey(self, key: str) -> bool:
        return key in self._map

    def toString(self) -> str:
        return f"Bundle({self._map})"

    def __len__(self) -> int:
        return len(self._map)

    def __contains__(self, key: str) -> bool:
        return key in self._map

    def __getitem__(self, key: str):
        return self._map[key]

    def __setitem__(self, key: str, value: Any):
        self._map[key] = value

    def __delitem__(self, key: str):
        self._map.pop(key, None)


class PersistableBundle(BaseBundle):
    """PersistableBundle — 可持久化的 Bundle"""

    def __init__(self, capacity: int = 0):
        super().__init__()

    def putPersistableBundle(self, key: str, value: "PersistableBundle"):
        self._map[key] = value

    def getPersistableBundle(self, key: str) -> Optional["PersistableBundle"]:
        v = self._map.get(key)
        return v if isinstance(v, PersistableBundle) else None


# ═══════════════════════════════════════════════════════════════
# Bundle
# ═══════════════════════════════════════════════════════════════

class Bundle(BaseBundle):
    """Bundle — 键值对数据容器"""

    EMPTY: "Bundle" = None  # 在类加载后初始化

    def __init__(self, capacity: int = 0):
        super().__init__()
        self._capacity = capacity

    def putString(self, key: str, value: Optional[str]):
        self._map[key] = value

    def putInt(self, key: str, value: int):
        self._map[key] = value

    def putLong(self, key: str, value: int):
        self._map[key] = value

    def putFloat(self, key: str, value: float):
        self._map[key] = value

    def putDouble(self, key: str, value: float):
        self._map[key] = value

    def putBoolean(self, key: str, value: bool):
        self._map[key] = value

    def putByte(self, key: str, value: int):
        self._map[key] = value

    def putChar(self, key: str, value: str):
        self._map[key] = value

    def putShort(self, key: str, value: int):
        self._map[key] = value

    def putCharSequence(self, key: str, value: Optional[str]):
        self._map[key] = value

    def putSerializable(self, key: str, value: Any):
        self._map[key] = value

    def putParcelable(self, key: str, value: Any):
        self._map[key] = value

    def putStringArrayList(self, key: str, value: Optional[List[str]]):
        self._map[key] = value

    def putIntegerArrayList(self, key: str, value: Optional[List[int]]):
        self._map[key] = value

    # ---- 数组类型 ----
    def putIntArray(self, key: str, value: Optional[List[int]]):
        self._map[key] = list(value) if value is not None else None

    def getIntArray(self, key: str) -> Optional[List[int]]:
        v = self._map.get(key)
        if isinstance(v, list):
            return [int(x) for x in v]
        return None

    def putStringArray(self, key: str, value: Optional[List[str]]):
        self._map[key] = list(value) if value is not None else None

    def getStringArray(self, key: str) -> Optional[List[str]]:
        v = self._map.get(key)
        if isinstance(v, list):
            return [str(x) for x in v]
        return None

    def putLongArray(self, key: str, value: Optional[List[int]]):
        self._map[key] = list(value) if value is not None else None

    def getLongArray(self, key: str) -> Optional[List[int]]:
        v = self._map.get(key)
        if isinstance(v, list):
            return [int(x) for x in v]
        return None

    def putBooleanArray(self, key: str, value: Optional[List[bool]]):
        self._map[key] = list(value) if value is not None else None

    def getBooleanArray(self, key: str) -> Optional[List[bool]]:
        v = self._map.get(key)
        if isinstance(v, list):
            return [bool(x) for x in v]
        return None

    def putDoubleArray(self, key: str, value: Optional[List[float]]):
        self._map[key] = list(value) if value is not None else None

    def getDoubleArray(self, key: str) -> Optional[List[float]]:
        v = self._map.get(key)
        if isinstance(v, list):
            return [float(x) for x in v]
        return None

    def putByteArray(self, key: str, value: Optional[List[int]]):
        self._map[key] = list(value) if value is not None else None

    def getByteArray(self, key: str) -> Optional[List[int]]:
        v = self._map.get(key)
        if isinstance(v, list):
            return [int(x) & 0xFF for x in v]
        return None

    def putAll(self, bundle: "Bundle"):
        if bundle:
            self._map.update(bundle._map)

    def getParcelable(self, key: str, default: Any = None) -> Any:
        return self._map.get(key, default)

    def getBundle(self, key: str) -> Optional["Bundle"]:
        v = self._map.get(key)
        return v if isinstance(v, Bundle) else None

    def putBundle(self, key: str, value: Optional["Bundle"]):
        self._map[key] = value

    def describeContents(self) -> int:
        return 0

    def clone(self) -> "Bundle":
        b = Bundle()
        b._map = dict(self._map)
        return b

    def deepCopy(self) -> "Bundle":
        import copy
        b = Bundle()
        b._map = copy.deepcopy(self._map)
        return b

    def filterValues(self):
        pass

    def getPairValue(self) -> Optional[str]:
        return None

    def setClassLoader(self, loader: Any):
        self._class_loader = loader

    def getClassLoader(self) -> Any:
        return self._class_loader

    @staticmethod
    def forPair(key: str, value: str) -> "Bundle":
        b = Bundle()
        b.putString(key, value)
        return b


Bundle.EMPTY = Bundle()


# ═══════════════════════════════════════════════════════════════
# Parcel / Parcelable
# ═══════════════════════════════════════════════════════════════

class Parcelable(Stub):
    """Parcelable 接口"""

    CONTENTS_FILE_DESCRIPTOR = "CONTENTS_FILE_DESCRIPTOR"
    PARCELABLE_WRITE_RETURN_VALUE = 1
    PARCELABLE_ELIDE_DUPLICATES = 2

    def describeContents(self) -> int:
        return 0

    def writeToParcel(self, dest: "Parcel", flags: int):
        pass

    class Creator:
        def createFromParcel(self, source: "Parcel"):
            return None
        def newArray(self, size: int):
            return [None] * size

    class ClassLoaderCreator:
        def createFromParcel(self, source: "Parcel", loader: Any):
            return None
        def newArray(self, size: int):
            return [None] * size


class Parcel(Stub):
    """Parcel — 序列化数据容器"""

    STRING_CREATOR = "STRING_CREATOR"
    STRING_LIST_CREATOR = "STRING_LIST_CREATOR"
    INT_CREATOR = "INT_CREATOR"

    def __init__(self):
        super().__init__()
        self._data: Dict[str, Any] = {}
        self._position = 0
        self._size = 0

    @staticmethod
    def obtain() -> "Parcel":
        return Parcel()

    def recycle(self):
        self._data.clear()
        self._position = 0
        self._size = 0

    def writeInt(self, value: int):
        self._data.setdefault("_ints", []).append(value)
        self._size += 4

    def writeLong(self, value: int):
        self._data.setdefault("_longs", []).append(value)
        self._size += 8

    def writeFloat(self, value: float):
        self._data.setdefault("_floats", []).append(value)
        self._size += 4

    def writeDouble(self, value: float):
        self._data.setdefault("_doubles", []).append(value)
        self._size += 8

    def writeString(self, value: Optional[str]):
        strs = self._data.setdefault("_strings", [])
        strs.append(value)
        self._size += len(value) + 4 if value else 4

    def writeBoolean(self, value: bool):
        self.writeInt(1 if value else 0)

    def writeByte(self, value: int):
        self.writeInt(value & 0xFF)

    def writeParcelable(self, p: Parcelable, flags: int):
        pass

    def writeStringArray(self, values: Optional[List[str]]):
        self._data["_string_array"] = values

    def writeIntArray(self, values: Optional[List[int]]):
        self._data["_int_array"] = values

    def writeLongArray(self, values: Optional[List[int]]):
        self._data["_long_array"] = values

    def readInt(self) -> int:
        ints = self._data.get("_ints", [])
        if ints:
            return ints.pop(0)
        return 0

    def readLong(self) -> int:
        longs = self._data.get("_longs", [])
        if longs:
            return longs.pop(0)
        return 0

    def readFloat(self) -> float:
        floats = self._data.get("_floats", [])
        if floats:
            return floats.pop(0)
        return 0.0

    def readDouble(self) -> float:
        doubles = self._data.get("_doubles", [])
        if doubles:
            return doubles.pop(0)
        return 0.0

    def readString(self) -> Optional[str]:
        strs = self._data.get("_strings", [])
        if strs:
            return strs.pop(0)
        return None

    def readBoolean(self) -> bool:
        return self.readInt() != 0

    def readByte(self) -> int:
        return self.readInt() & 0xFF

    def dataAvail(self) -> int:
        return self._size - self._position

    def dataPosition(self) -> int:
        return self._position

    def setDataPosition(self, pos: int):
        self._position = pos

    def dataSize(self) -> int:
        return self._size

    def dataCapacity(self) -> int:
        return self._size

    def setDataCapacity(self, capacity: int):
        pass

    def appendFrom(self, parcel: "Parcel", offset: int, length: int):
        pass

    def marshall(self) -> bytes:
        import json
        return json.dumps(self._data).encode("utf-8")

    @staticmethod
    def unmarshall(data: bytes) -> "Parcel":
        import json
        p = Parcel()
        try:
            p._data = json.loads(data.decode("utf-8"))
        except Exception:
            pass
        return p

    def writeValue(self, v: Any):
        self._data["_value"] = v

    def readValue(self) -> Any:
        return self._data.get("_value")

    def hasFileDescriptors(self) -> bool:
        return False

    def toString(self) -> str:
        return f"Parcel({self._size} bytes)"


# ═══════════════════════════════════════════════════════════════
# Message
# ═══════════════════════════════════════════════════════════════

class Message(Stub):
    """Message — Handler 消息"""

    def __init__(self):
        super().__init__()
        self.what = 0
        self.arg1 = 0
        self.arg2 = 0
        self.obj: Any = None
        self.data: Optional[Bundle] = None
        self.replyTo: Optional["Messenger"] = None
        self._target: Optional["Handler"] = None
        self._callback: Optional[Callable] = None
        self.when: int = 0
        self._sending = False

    @staticmethod
    def obtain(handler: Optional["Handler"] = None, what: int = 0) -> "Message":
        msg = Message()
        if handler:
            msg._target = handler
            msg.what = what
        return msg

    def sendToTarget(self):
        if self._target:
            self._target.sendMessage(self)

    def getTarget(self) -> Optional["Handler"]:
        return self._target

    def setTarget(self, handler: "Handler"):
        self._target = handler

    @property
    def target(self) -> Optional["Handler"]:
        return self._target

    @target.setter
    def target(self, handler: "Handler"):
        self._target = handler

    def getCallback(self) -> Optional[Callable]:
        return self._callback

    def copyFrom(self, other: "Message"):
        self.what = other.what
        self.arg1 = other.arg1
        self.arg2 = other.arg2
        self.obj = other.obj
        self.data = other.data
        self.replyTo = other.replyTo

    def recycle(self):
        self.what = 0
        self.arg1 = 0
        self.arg2 = 0
        self.obj = None
        self.data = None
        self._target = None
        self._callback = None

    def isAsynchronous(self) -> bool:
        return False

    def setAsynchronous(self, async_: bool):
        pass

    def toString(self) -> str:
        return f"Message(what={self.what}, arg1={self.arg1}, arg2={self.arg2})"


# ═══════════════════════════════════════════════════════════════
# Looper
# ═══════════════════════════════════════════════════════════════

class Looper(Stub):
    """Looper — 消息循环"""

    _main_looper: Optional["Looper"] = None
    _thread_local = threading.local()

    def __init__(self, quit_allowed: bool = True):
        super().__init__()
        self._thread = threading.current_thread()
        self._running = False
        self._message_queue: List[Message] = []
        self._queue_lock = threading.Lock()
        self._quit_allowed = quit_allowed

    @staticmethod
    def prepare():
        """为当前线程准备一个 Looper"""
        looper = Looper()
        Looper._thread_local.looper = looper
        return looper

    @staticmethod
    def prepareMainLooper():
        if Looper._main_looper is None:
            Looper._main_looper = Looper.prepare()

    @staticmethod
    def getMainLooper() -> "Looper":
        if Looper._main_looper is None:
            Looper.prepareMainLooper()
        return Looper._main_looper

    @staticmethod
    def myLooper() -> Optional["Looper"]:
        return getattr(Looper._thread_local, "looper", None)

    @staticmethod
    def loop():
        """运行消息循环，处理到期的消息"""
        looper = Looper.myLooper()
        if looper is None:
            raise RuntimeError("No Looper prepared")
        looper._running = True
        while looper._running:
            msg = looper._next()
            if msg is not None:
                target = msg._target
                if target is not None:
                    target.dispatchMessage(msg)
                msg.recycle()
            else:
                # 无到期消息，短暂 sleep 避免忙等
                _time.sleep(0.001)

    def _next(self) -> Optional[Message]:
        """非阻塞地取出一条到期消息（when <= now）"""
        with self._queue_lock:
            now = SystemClock.uptimeMillis()
            for i, msg in enumerate(self._message_queue):
                if msg.when <= now:
                    return self._message_queue.pop(i)
            return None

    def _enqueue(self, msg: Message):
        """将消息按 when 时间顺序入队"""
        with self._queue_lock:
            inserted = False
            for i, m in enumerate(self._message_queue):
                if msg.when < m.when:
                    self._message_queue.insert(i, msg)
                    inserted = True
                    break
            if not inserted:
                self._message_queue.append(msg)

    def _remove_callbacks(self, callback):
        """从队列中移除匹配 callback 的消息"""
        with self._queue_lock:
            self._message_queue = [
                m for m in self._message_queue
                if m._callback is not callback
            ]

    def _remove_messages(self, what: int):
        """从队列中移除匹配 what 的消息"""
        with self._queue_lock:
            self._message_queue = [
                m for m in self._message_queue
                if m.what != what
            ]

    def _has_messages(self, what: int) -> bool:
        """检查队列中是否有匹配 what 的消息"""
        with self._queue_lock:
            return any(m.what == what for m in self._message_queue)

    def quit(self):
        """退出消息循环"""
        if not self._quit_allowed:
            raise RuntimeError("Main Looper is not allowed to quit")
        with self._queue_lock:
            self._running = False
            self._message_queue.clear()

    def quitSafely(self):
        """安全退出，仅停止循环（已入队消息不再处理）"""
        with self._queue_lock:
            self._running = False

    def isCurrentThread(self) -> bool:
        return threading.current_thread() == self._thread

    def getThread(self) -> threading.Thread:
        return self._thread

    def getQueue(self):
        return self._message_queue

    def dump(self, pw: Any, prefix: str):
        """将 DropBox 条目转储到给定的 writer。"""
        try:
            if hasattr(pw, "write"):
                pw.write(f"{prefix}DropBoxManager entries:\n")
                for entry in self._entries:
                    pw.write(f"{prefix}  [{entry.getTag()}] @ {entry.getTimeMillis()}: "
                             f"{entry.getText(1024)}\n")
            elif hasattr(pw, "println"):
                pw.println(f"{prefix}DropBoxManager entries:")
                for entry in self._entries:
                    pw.println(f"{prefix}  [{entry.getTag()}] @ {entry.getTimeMillis()}: "
                               f"{entry.getText(1024)}")
        except Exception:
            pass

    def toString(self) -> str:
        return f"Looper({self._thread.name})"


# ═══════════════════════════════════════════════════════════════
# Handler
# ═══════════════════════════════════════════════════════════════

class Handler(Stub):
    """Handler — 消息处理器"""

    def __init__(self, looper: Optional[Looper] = None, callback: Optional[Callable] = None):
        super().__init__()
        self._looper = looper or Looper.getMainLooper()
        self._callback = callback
        self._async = False

    def handleMessage(self, msg: Message):
        pass

    def dispatchMessage(self, msg: Message):
        if msg._callback:
            msg._callback()
        elif self._callback:
            self._callback(msg)
        else:
            self.handleMessage(msg)

    def post(self, r: Callable) -> bool:
        msg = Message.obtain(self)
        msg._callback = r
        return self.sendMessage(msg)

    def postDelayed(self, r: Callable, delay_ms: int) -> bool:
        msg = Message.obtain(self)
        msg._callback = r
        return self.sendMessageDelayed(msg, delay_ms)

    def postAtTime(self, r: Callable, uptime_ms: int) -> bool:
        msg = Message.obtain(self)
        msg._callback = r
        msg.when = uptime_ms
        return self.sendMessageAtTime(msg, uptime_ms)

    def sendMessage(self, msg: Message) -> bool:
        msg._target = self
        looper = self._looper
        if looper is not None and looper._running:
            looper._enqueue(msg)
            return True
        # 当前线程无运行中的 Looper，回退到同步执行（保持兼容性）
        self.dispatchMessage(msg)
        return True

    def sendEmptyMessage(self, what: int) -> bool:
        msg = Message.obtain(self, what)
        return self.sendMessage(msg)

    def sendMessageDelayed(self, msg: Message, delay_ms: int) -> bool:
        msg._target = self
        msg.when = SystemClock.uptimeMillis() + delay_ms
        looper = self._looper
        if looper is not None and looper._running:
            looper._enqueue(msg)
            return True
        # 回退到同步执行
        self.dispatchMessage(msg)
        return True

    def sendMessageAtTime(self, msg: Message, uptime_ms: int) -> bool:
        msg._target = self
        msg.when = uptime_ms
        looper = self._looper
        if looper is not None and looper._running:
            looper._enqueue(msg)
            return True
        self.dispatchMessage(msg)
        return True

    def sendMessageAtFrontOfQueue(self, msg: Message) -> bool:
        msg._target = self
        msg.when = 0
        looper = self._looper
        if looper is not None and looper._running:
            with looper._queue_lock:
                looper._message_queue.insert(0, msg)
            return True
        self.dispatchMessage(msg)
        return True

    def removeCallbacks(self, r: Callable):
        looper = self._looper
        if looper is not None:
            looper._remove_callbacks(r)

    def removeMessages(self, what: int):
        looper = self._looper
        if looper is not None:
            looper._remove_messages(what)

    def obtainMessage(self, what: int = 0) -> Message:
        return Message.obtain(self, what)

    def hasMessages(self, what: int) -> bool:
        looper = self._looper
        if looper is not None:
            return looper._has_messages(what)
        return False

    def getLooper(self) -> Looper:
        return self._looper

    def toString(self) -> str:
        return f"Handler({self.__class__.__name__})"


# ═══════════════════════════════════════════════════════════════
# HandlerThread
# ═══════════════════════════════════════════════════════════════

class HandlerThread(threading.Thread, Stub):
    """HandlerThread — 带 Looper 的工作线程"""

    def __init__(self, name: str, priority: int = Process.THREAD_PRIORITY_DEFAULT):
        super().__init__()
        Stub.__init__(self)
        self._name = name
        self._priority = priority
        self._looper: Optional[Looper] = None
        self._started = threading.Event()

    def run(self):
        Looper.prepare()
        self._looper = Looper.myLooper()
        self._started.set()
        Looper.loop()

    def getLooper(self) -> Optional[Looper]:
        self._started.wait()
        return self._looper

    def getThreadId(self) -> int:
        return self.ident or 0

    def quit(self) -> bool:
        if self._looper:
            self._looper.quit()
            return True
        return False

    def quitSafely(self) -> bool:
        return self.quit()


# ═══════════════════════════════════════════════════════════════
# AsyncTask Executor (守护线程池)
# ═══════════════════════════════════════════════════════════════

class _AsyncTaskExecutor:
    """守护线程池执行器 — 用于 AsyncTask 后台执行，不阻塞程序退出。

    使用 BoundedSemaphore 限制并发数，每个任务在独立的守护线程中运行。
    """

    def __init__(self, max_workers: int = 4):
        self._max_workers = max_workers
        self._semaphore = threading.BoundedSemaphore(max_workers)
        self._shutdown_flag = False

    def submit(self, fn, *args, **kwargs):
        if self._shutdown_flag:
            raise RuntimeError("Executor has been shut down")
        self._semaphore.acquire()

        def _run():
            try:
                fn(*args, **kwargs)
            finally:
                self._semaphore.release()

        t = threading.Thread(target=_run, daemon=True)
        t.start()

    def execute(self, command):
        self.submit(command)

    def shutdown(self, wait: bool = True):
        self._shutdown_flag = True


# ═══════════════════════════════════════════════════════════════
# AsyncTask
# ═══════════════════════════════════════════════════════════════

class AsyncTask(Stub):
    """AsyncTask — 异步任务"""

    Status = type("Status", (), {"PENDING": 0, "RUNNING": 1, "FINISHED": 2})

    SERIAL_EXECUTOR = None
    THREAD_POOL_EXECUTOR = None

    def __init__(self):
        super().__init__()
        self._status = AsyncTask.Status.PENDING
        self._result: Any = None
        self._cancelled = False
        self._handler = Handler(Looper.getMainLooper())

    def doInBackground(self, *params) -> Any:
        raise NotImplementedError

    def onPreExecute(self):
        pass

    def onPostExecute(self, result: Any):
        pass

    def onProgressUpdate(self, *values):
        pass

    def onCancelled(self, result: Any = None):
        pass

    def execute(self, *params) -> "AsyncTask":
        return self.executeOnExecutor(AsyncTask.THREAD_POOL_EXECUTOR, *params)

    def executeOnExecutor(self, exec: Any, *params) -> "AsyncTask":
        self._status = AsyncTask.Status.RUNNING
        self.onPreExecute()
        exec.submit(self._run_background, params)
        return self

    def _run_background(self, params):
        """在后台线程中执行 doInBackground，完成后通过 Handler 回调主线程"""
        try:
            self._result = self.doInBackground(*params)
        except Exception as e:
            self._result = e
        self._status = AsyncTask.Status.FINISHED
        if self._cancelled:
            self._handler.post(lambda: self.onCancelled(self._result))
        else:
            self._handler.post(lambda: self.onPostExecute(self._result))

    def get(self) -> Any:
        return self._result

    def getStatus(self) -> int:
        return self._status

    def isCancelled(self) -> bool:
        return self._cancelled

    def cancel(self, may_interrupt_if_running: bool) -> bool:
        self._cancelled = True
        return True

    def publishProgress(self, *values):
        self._handler.post(lambda: self.onProgressUpdate(*values))


# 初始化默认执行器
AsyncTask.THREAD_POOL_EXECUTOR = _AsyncTaskExecutor(max_workers=4)
AsyncTask.SERIAL_EXECUTOR = _AsyncTaskExecutor(max_workers=1)


# ═══════════════════════════════════════════════════════════════
# CountDownTimer
# ═══════════════════════════════════════════════════════════════

class CountDownTimer(Stub):
    """CountDownTimer — 倒计时"""

    def __init__(self, millis_in_future: int, count_down_interval: int):
        super().__init__()
        self._millis_in_future = millis_in_future
        self._interval = count_down_interval
        self._cancelled = False

    def onTick(self, millis_until_finished: int):
        pass

    def onFinish(self):
        pass

    def start(self) -> "CountDownTimer":
        self._cancelled = False
        remaining = self._millis_in_future
        while remaining > 0 and not self._cancelled:
            self.onTick(remaining)
            _time.sleep(self._interval / 1000.0)
            remaining -= self._interval
        if not self._cancelled:
            self.onFinish()
        return self

    def cancel(self):
        self._cancelled = True


# ═══════════════════════════════════════════════════════════════
# Messenger
# ═══════════════════════════════════════════════════════════════

class Messenger(Stub):
    """Messenger — 跨进程消息传递"""

    def __init__(self, handler: Handler):
        super().__init__()
        self._handler = handler

    def send(self, message: Message):
        self._handler.sendMessage(message)

    def getBinder(self):
        return None

    @staticmethod
    def obtain(handler: Handler) -> "Messenger":
        return Messenger(handler)


# ═══════════════════════════════════════════════════════════════
# PowerManager / PowerManagerWakeLock
# ═══════════════════════════════════════════════════════════════

class PowerManager(Stub):
    """PowerManager — 电源管理"""

    ACQUIRE_CAUSES_WAKEUP = 0x10000000
    DOZE_WAKE_LOCK = 0x00000040
    DRAW_WAKE_LOCK = 0x00000080
    FULL_WAKE_LOCK = 0x0000001A
    LOCATION_HARDWARE_POWER_MODE = 0x00000080
    ON_AFTER_RELEASE = 0x20000000
    PARTIAL_WAKE_LOCK = 0x00000001
    PROXIMITY_SCREEN_OFF_WAKE_LOCK = 0x00000020
    SCREEN_BRIGHT_WAKE_LOCK = 0x0000000A
    SCREEN_DIM_WAKE_LOCK = 0x00000006
    WAKE_LOCK_LEVEL_MASK = 0x0000FFFF

    def __init__(self):
        super().__init__()
        self._is_interactive = True
        self._is_power_save = False
        self._is_device_idle = False

    def newWakeLock(self, level_and_flags: int, tag: str) -> "PowerManagerWakeLock":
        return PowerManagerWakeLock(level_and_flags, tag)

    def isInteractive(self) -> bool:
        if _USE_LINUX:
            return self.isScreenOn()
        return self._is_interactive

    def isScreenOn(self) -> bool:
        """屏幕是否点亮 — 在 Linux 上检查 DISPLAY 环境变量"""
        if _USE_LINUX:
            import os as _os_mod
            display = _os_mod.environ.get("DISPLAY", "")
            if display:
                return True
            # WAYLAND_DISPLAY 作为回退
            wayland = _os_mod.environ.get("WAYLAND_DISPLAY", "")
            return bool(wayland)
        return self._is_interactive

    def isPowerSaveMode(self) -> bool:
        """检查是否处于省电模式。读取 /sys 中的电源相关状态。"""
        try:
            # 检查 CPU 调速器是否为 powersave
            gov_path = "/sys/devices/system/cpu/cpu0/cpufreq/scaling_governor"
            if _os.path.exists(gov_path):
                with open(gov_path, "r") as f:
                    governor = f.read().strip()
                    if governor == "powersave":
                        return True
        except (OSError, IOError):
            pass
        return self._is_power_save

    def isDeviceIdleMode(self) -> bool:
        return self._is_device_idle

    def isIgnoringBatteryOptimizations(self, package_name: str) -> bool:
        return True

    def isSustainedPerformanceModeSupported(self) -> bool:
        return False

    def reboot(self, reason: Optional[str] = None):
        """重启设备。桌面端尝试执行 reboot 命令（通常需要 root）。"""
        try:
            import subprocess
            cmd = ["reboot"]
            if reason:
                cmd.append(reason)
            subprocess.run(cmd, capture_output=True, timeout=5)
        except Exception:
            sys.stderr.write(f"[PowerManager] reboot: {reason or ''}\n")

    def goToSleep(self, time_ms: int):
        """使设备进入睡眠。桌面端无实际效果。"""
        self._is_interactive = False
        sys.stderr.write(f"[PowerManager] goToSleep at {time_ms}\n")

    def wakeUp(self, time_ms: int):
        """唤醒设备。桌面端无实际效果。"""
        self._is_interactive = True
        sys.stderr.write(f"[PowerManager] wakeUp at {time_ms}\n")

    def userActivity(self, when: int, no_change_lights: bool = False):
        """报告用户活动事件。桌面端无实际效果。"""
        self._is_interactive = True


class PowerManagerWakeLock(Stub):
    """PowerManagerWakeLock — 唤醒锁"""

    def __init__(self, level_and_flags: int, tag: str):
        super().__init__()
        self._level_and_flags = level_and_flags
        self._tag = tag
        self._held = False
        self._work_source = None
        self._history_tag = None

    def acquire(self, timeout_ms: int = 0):
        self._held = True
        sys.stderr.write(f"[PowerManager] acquire wakelock: {self._tag}\n")

    def acquireLocked(self):
        self._held = True

    def release(self, flags: int = 0):
        self._held = False
        sys.stderr.write(f"[PowerManager] release wakelock: {self._tag}\n")

    def isHeld(self) -> bool:
        return self._held

    def setWorkSource(self, work_source: Any):
        self._work_source = work_source

    def setReferenceCounted(self, value: bool):
        pass

    def setHistoryTag(self, tag: str):
        self._history_tag = tag

    def toString(self) -> str:
        return f"WakeLock({self._tag}, held={self._held})"


# ═══════════════════════════════════════════════════════════════
# Vibrator / VibratorEffect
# ═══════════════════════════════════════════════════════════════

class Vibrator(Stub):
    """Vibrator — 振动器"""

    def __init__(self):
        super().__init__()
        self._ff_device: Optional[str] = None
        self._vibrating = False

    @staticmethod
    def _find_ff_device() -> Optional[str]:
        """查找 Linux 力反馈设备。"""
        try:
            for name in _os.listdir("/dev/input"):
                if name.startswith("event"):
                    path = f"/dev/input/{name}"
                    if _os.access(path, _os.W_OK):
                        return path
        except (OSError, IOError):
            pass
        # 检查 /sys/class/leds/vibrator
        try:
            if _os.path.exists("/sys/class/leds/vibrator"):
                return "/sys/class/leds/vibrator"
        except (OSError, IOError):
            pass
        return None

    def vibrate(self, *args):
        """振动。支持 vibrate(milliseconds) 和 vibrate(pattern, repeat)。"""
        if len(args) == 1 and isinstance(args[0], (int, float)):
            milliseconds = int(args[0])
            self._vibrating = True
            if self._ff_device is None:
                self._ff_device = self._find_ff_device()
            if self._ff_device and self._ff_device.startswith("/sys/class/leds"):
                try:
                    with open(_os.path.join(self._ff_device, "brightness"), "w") as f:
                        f.write("1")
                    import threading
                    def _stop():
                        _time.sleep(milliseconds / 1000.0)
                        try:
                            with open(_os.path.join(self._ff_device, "brightness"), "w") as f:
                                f.write("0")
                        except (OSError, IOError):
                            pass
                        self._vibrating = False
                    threading.Thread(target=_stop, daemon=True).start()
                    return
                except (OSError, IOError):
                    pass
            sys.stderr.write(f"[Vibrator] vibrate: {milliseconds}ms\n")
        elif len(args) == 2 and isinstance(args[0], list):
            pattern = args[0]
            repeat = args[1]
            if not pattern:
                return
            self._vibrating = True
            import threading
            def _run_pattern():
                idx = 0
                while True:
                    duration = pattern[idx % len(pattern)]
                    if idx % 2 == 0:
                        _time.sleep(duration / 1000.0)
                    else:
                        self.vibrate(duration)
                        _time.sleep(duration / 1000.0)
                    idx += 1
                    if repeat < 0 or idx >= len(pattern):
                        if repeat < 0:
                            continue
                        break
                self._vibrating = False
            threading.Thread(target=_run_pattern, daemon=True).start()
            sys.stderr.write(f"[Vibrator] vibrate pattern: {pattern}, repeat={repeat}\n")

    def cancel(self):
        """取消振动。"""
        self._vibrating = False
        if self._ff_device and self._ff_device.startswith("/sys/class/leds"):
            try:
                with open(_os.path.join(self._ff_device, "brightness"), "w") as f:
                    f.write("0")
            except (OSError, IOError):
                pass
        sys.stderr.write("[Vibrator] cancel\n")

    def hasVibrator(self) -> bool:
        """检查是否有振动器硬件。"""
        return self._find_ff_device() is not None

    def hasAmplitudeControl(self) -> bool:
        """检查是否支持振幅控制。"""
        try:
            if _os.path.exists("/sys/class/leds/vibrator/max_brightness"):
                with open("/sys/class/leds/vibrator/max_brightness", "r") as f:
                    return int(f.read().strip()) > 1
        except (OSError, IOError, ValueError):
            pass
        return False

    def areEffectsSupported(self, effect_ids: List[int]) -> List[int]:
        return [0] * len(effect_ids)

    def areAllEffectsSupported(self, effect_ids: List[int]) -> List[int]:
        return [0] * len(effect_ids)


class VibratorEffect(Stub):
    """VibratorEffect — 振动效果"""

    EFFECT_CLICK = 0
    EFFECT_DOUBLE_CLICK = 1
    EFFECT_TICK = 2
    EFFECT_THUD = 3
    EFFECT_POP = 4
    EFFECT_HEAVY_CLICK = 5

    def __init__(self, effect_id: int = 0):
        super().__init__()
        self._effect_id = effect_id
        self._timings: Optional[List[int]] = None
        self._amplitudes: Optional[List[int]] = None
        self._repeat = -1

    @staticmethod
    def createOneShot(milliseconds: int, amplitude: int) -> "VibratorEffect":
        """创建单次振动效果。"""
        effect = VibratorEffect(0)
        effect._timings = [0, milliseconds]
        effect._amplitudes = [0, amplitude]
        effect._repeat = -1
        return effect

    @staticmethod
    def createWaveform(pattern: List[int], repeat: int) -> "VibratorEffect":
        """创建波形振动效果。"""
        effect = VibratorEffect(0)
        effect._timings = list(pattern)
        effect._amplitudes = [255 if i % 2 == 1 else 0 for i in range(len(pattern))]
        effect._repeat = repeat
        return effect

    @staticmethod
    def createPredefined(effect_id: int) -> "VibratorEffect":
        """创建预定义振动效果。"""
        return VibratorEffect(effect_id)


# ═══════════════════════════════════════════════════════════════
# BatteryManager
# ═══════════════════════════════════════════════════════════════

class BatteryManager(Stub):
    """BatteryManager — 电池管理"""

    BATTERY_PROPERTY_CHARGE_COUNTER = 1
    BATTERY_PROPERTY_CURRENT_NOW = 2
    BATTERY_PROPERTY_CURRENT_AVERAGE = 3
    BATTERY_PROPERTY_CAPACITY = 4
    BATTERY_PROPERTY_STATUS = 5
    BATTERY_PROPERTY_ENERGY_COUNTER = 6
    BATTERY_PROPERTY_BATTERY_NOW = 7
    BATTERY_PROPERTY_BATTERY_LOW = 8

    BATTERY_STATUS_UNKNOWN = 1
    BATTERY_STATUS_CHARGING = 2
    BATTERY_STATUS_DISCHARGING = 3
    BATTERY_STATUS_NOT_CHARGING = 4
    BATTERY_STATUS_FULL = 5

    BATTERY_HEALTH_UNKNOWN = 1
    BATTERY_HEALTH_GOOD = 2
    BATTERY_HEALTH_OVERHEAT = 3
    BATTERY_HEALTH_DEAD = 4
    BATTERY_HEALTH_OVER_VOLTAGE = 5
    BATTERY_HEALTH_UNSPECIFIED_FAILURE = 6
    BATTERY_HEALTH_COLD = 7

    EXTRA_STATUS = "status"
    EXTRA_HEALTH = "health"
    EXTRA_PRESENT = "present"
    EXTRA_LEVEL = "level"
    EXTRA_SCALE = "scale"
    EXTRA_ICON_SMALL = "icon-small"
    EXTRA_PLUGGED = "plugged"
    EXTRA_VOLTAGE = "voltage"
    EXTRA_TEMPERATURE = "temperature"
    EXTRA_TECHNOLOGY = "technology"
    EXTRA_BATTERY_LOW = "battery_low"
    EXTRA_CHARGE_COUNTER = "charge_counter"

    BATTERY_PLUGGED_AC = 1
    BATTERY_PLUGGED_USB = 2
    BATTERY_PLUGGED_WIRELESS = 4

    def __init__(self):
        super().__init__()
        self._capacity = 85
        self._status = self.BATTERY_STATUS_CHARGING
        self._health = self.BATTERY_HEALTH_GOOD
        self._present = True
        self._level = 85
        self._plugged = self.BATTERY_PLUGGED_AC
        self._voltage = 4200
        self._temperature = 250
        self._technology = "Li-ion"
        self._charge_counter = 3500000
        self._current_now = -500000
        self._current_avg = -450000
        self._energy_counter = 15000000
        if _USE_LINUX:
            self._refresh_from_linux()

    _LINUX_STATUS_MAP = {
        "charging": 2,        # BATTERY_STATUS_CHARGING
        "discharging": 3,     # BATTERY_STATUS_DISCHARGING
        "not charging": 4,    # BATTERY_STATUS_NOT_CHARGING
        "full": 5,            # BATTERY_STATUS_FULL
        "unknown": 1,         # BATTERY_STATUS_UNKNOWN
    }

    def _get_linux_battery_info(self) -> dict:
        """从 LinuxPower 获取真实电池信息"""
        try:
            return LinuxPower.get_battery_info() or {}
        except Exception:
            return {}

    def _refresh_from_linux(self):
        """用 Linux 真实电池信息刷新内部状态"""
        info = self._get_linux_battery_info()
        if not info:
            return
        self._present = bool(info.get("present", False))
        level = info.get("level", -1)
        if self._present and isinstance(level, (int, float)) and level >= 0:
            self._level = int(level)
            self._capacity = int(level)
        else:
            # 无电池或电量未知
            self._level = -1
            self._capacity = -1
        status_str = str(info.get("status", "unknown")).lower()
        self._status = self._LINUX_STATUS_MAP.get(status_str, self.BATTERY_STATUS_UNKNOWN)
        voltage = info.get("voltage", 0)
        if isinstance(voltage, (int, float)):
            # LinuxPower 返回伏特, Android 期望毫伏
            self._voltage = int(voltage * 1000)
        temperature = info.get("temperature", 0)
        if isinstance(temperature, (int, float)):
            # LinuxPower 返回摄氏度, Android 期望十分之一度
            self._temperature = int(temperature * 10)
        self._technology = info.get("technology", "") or "Li-ion"
        # 充电状态决定 plugged
        if self._status == self.BATTERY_STATUS_CHARGING:
            self._plugged = self.BATTERY_PLUGGED_AC
        else:
            self._plugged = 0

    def getIntProperty(self, prop_id: int) -> int:
        if _USE_LINUX:
            self._refresh_from_linux()
        if prop_id == self.BATTERY_PROPERTY_CAPACITY:
            return self._capacity
        if prop_id == self.BATTERY_PROPERTY_STATUS:
            return self._status
        if prop_id == self.BATTERY_PROPERTY_CHARGE_COUNTER:
            return self._charge_counter
        if prop_id == self.BATTERY_PROPERTY_CURRENT_NOW:
            return self._current_now
        if prop_id == self.BATTERY_PROPERTY_CURRENT_AVERAGE:
            return self._current_avg
        if prop_id == self.BATTERY_PROPERTY_ENERGY_COUNTER:
            return self._energy_counter
        return -1

    def getLongProperty(self, prop_id: int) -> int:
        return self.getIntProperty(prop_id)

    def isCharging(self) -> bool:
        if _USE_LINUX:
            self._refresh_from_linux()
        return self._status == self.BATTERY_STATUS_CHARGING

    def isPresent(self) -> bool:
        """电池是否存在"""
        if _USE_LINUX:
            self._refresh_from_linux()
        return self._present

    def getLevel(self) -> int:
        """电池电量 (0-100, -1 表示未知)"""
        if _USE_LINUX:
            self._refresh_from_linux()
        return self._level

    def getVoltage(self) -> int:
        """电池电压 (毫伏)"""
        if _USE_LINUX:
            self._refresh_from_linux()
        return self._voltage

    def getTemperature(self) -> int:
        """电池温度 (十分之一摄氏度)"""
        if _USE_LINUX:
            self._refresh_from_linux()
        return self._temperature

    def getTechnology(self) -> str:
        """电池技术"""
        if _USE_LINUX:
            self._refresh_from_linux()
        return self._technology

    def getPlugType(self) -> int:
        """充电类型"""
        if _USE_LINUX:
            self._refresh_from_linux()
        return self._plugged

    def getStatus(self) -> int:
        """电池状态"""
        if _USE_LINUX:
            self._refresh_from_linux()
        return self._status

    def computeChargeTimeRemaining(self) -> int:
        if _USE_LINUX:
            # 无电池或已充满时返回 -1
            if not self._present or self._status == self.BATTERY_STATUS_FULL:
                return -1
        return 1800000  # 30 minutes

    def toString(self) -> str:
        return f"BatteryManager(level={self._level}%, charging={self.isCharging()})"


# ═══════════════════════════════════════════════════════════════
# 异常类
# ═══════════════════════════════════════════════════════════════

class NetworkOnMainThreadException(RuntimeError):
    """网络操作在主线程上执行"""
    pass


class TransactionTooLargeException(RuntimeError):
    """事务数据太大"""
    pass


class OperationCanceledException(RuntimeError):
    """操作被取消"""
    pass


# ═══════════════════════════════════════════════════════════════
# 其他工具类
# ═══════════════════════════════════════════════════════════════

class ConditionVariable(Stub):
    """ConditionVariable — 条件变量"""

    def __init__(self):
        super().__init__()
        self._cond = threading.Condition()
        self._open = False

    def block(self, timeout_ms: int = 0):
        with self._cond:
            if timeout_ms <= 0:
                while not self._open:
                    self._cond.wait()
            else:
                if not self._open:
                    self._cond.wait(timeout_ms / 1000.0)

    def open(self):
        with self._cond:
            self._open = True
            self._cond.notify_all()

    def close(self):
        with self._cond:
            self._open = False


class Semaphore(Stub):
    """Semaphore — 信号量"""

    def __init__(self, permits: int):
        super().__init__()
        self._sem = threading.Semaphore(permits)

    def acquire(self):
        self._sem.acquire()

    def tryAcquire(self) -> bool:
        return self._sem.acquire(blocking=False)

    def release(self):
        self._sem.release()

    def availablePermits(self) -> int:
        return self._sem._value


class TokenWatcher(Stub):
    """TokenWatcher — 令牌监视器"""

    def __init__(self, handler: Handler, tag: str):
        super().__init__()
        self._handler = handler
        self._tag = tag
        self._tokens: Dict[int, str] = {}

    def acquire(self, token: Any, tag: str):
        """获取令牌。"""
        self._tokens[id(token)] = tag
        if len(self._tokens) == 1:
            self.acquired()

    def release(self, token: Any):
        """释放令牌。"""
        self._tokens.pop(id(token), None)
        if not self._tokens:
            self.released()

    def isAcquired(self) -> bool:
        """检查是否有令牌被持有。"""
        return len(self._tokens) > 0

    def acquired(self):
        """当第一个令牌被获取时调用。子类可重写。"""
        pass

    def released(self):
        """当所有令牌被释放时调用。子类可重写。"""
        pass

    def dump(self):
        """转储令牌信息。"""
        for tid, tag in self._tokens.items():
            sys.stderr.write(f"[TokenWatcher] {self._tag}: token={tid}, tag={tag}\n")


class WorkSource(Stub):
    """WorkSource — 工作来源"""

    def __init__(self, uid: int = 0):
        super().__init__()
        self._uids: List[int] = [uid] if uid else []

    def add(self, uid: int):
        if uid not in self._uids:
            self._uids.append(uid)

    def remove(self, uid: int):
        if uid in self._uids:
            self._uids.remove(uid)

    def get(self, index: int) -> int:
        if 0 <= index < len(self._uids):
            return self._uids[index]
        return 0

    def size(self) -> int:
        return len(self._uids)

    def clear(self):
        self._uids.clear()

    def set(self, ws: "WorkSource"):
        if ws:
            self._uids = list(ws._uids)

    def toString(self) -> str:
        return f"WorkSource({self._uids})"


class UserHandle(Stub):
    """UserHandle — 用户句柄"""

    USER_ALL = -1
    USER_CURRENT = -2
    USER_CURRENT_OR_SELF = -3
    USER_NULL = -10000
    USER_OWNER = 0
    USER_SYSTEM = 0

    def __init__(self, user_id: int = 0):
        super().__init__()
        self._user_id = user_id

    @staticmethod
    def myUserId() -> int:
        return 0

    @staticmethod
    def getUserHandleForUid(uid: int) -> "UserHandle":
        return UserHandle(uid // 100000)

    @staticmethod
    def getUserId(uid: int) -> int:
        return uid // 100000

    @staticmethod
    def getUid(user_id: int, app_id: int) -> int:
        return user_id * 100000 + app_id

    @staticmethod
    def getAppId(uid: int) -> int:
        return uid % 100000

    def getIdentifier(self) -> int:
        return self._user_id

    def hashCode(self) -> int:
        return self._user_id

    def toString(self) -> str:
        return f"UserHandle({self._user_id})"


class DropBoxManager(Stub):
    """DropBoxManager — 系统日志收集"""

    IS_EMPTY = 0
    IS_TEXT = 1
    IS_GZIPPED = 2

    class Entry(Stub):
        def __init__(self, tag: str, milliseconds: int):
            super().__init__()
            self._tag = tag
            self._millis = milliseconds
            self._text = ""

        def getTag(self) -> str:
            return self._tag

        def getTimeMillis(self) -> int:
            return self._millis

        def getText(self, max_bytes: int) -> Optional[str]:
            return self._text

        def getFlags(self, max_bytes: int) -> int:
            return DropBoxManager.IS_TEXT

        def close(self):
            pass

    def __init__(self):
        super().__init__()
        self._entries: List[DropBoxManager.Entry] = []

    def addText(self, tag: str, text: str):
        entry = DropBoxManager.Entry(tag, int(_time.time() * 1000))
        entry._text = text
        self._entries.append(entry)

    def getNextEntry(self, tag: str, milliseconds: int) -> Optional[Entry]:
        for entry in self._entries:
            if entry.getTimeMillis() > milliseconds:
                if not tag or entry.getTag() == tag:
                    return entry
        return None

    def dump(self, pw: Any, prefix: str):
        pass


class FileObserver(Stub):
    """FileObserver — 文件监听"""

    ACCESS = 0x00000001
    MODIFY = 0x00000002
    ATTRIB = 0x00000004
    CLOSE_WRITE = 0x00000008
    CLOSE_NOWRITE = 0x00000010
    OPEN = 0x00000020
    MOVED_FROM = 0x00000040
    MOVED_TO = 0x00000080
    CREATE = 0x00000100
    DELETE = 0x00000200
    DELETE_SELF = 0x00000400
    MOVE_SELF = 0x00000800
    ALL_EVENTS = 0x00000FFF

    def __init__(self, path: str):
        super().__init__()
        self._path = path
        self._watching = False
        self._thread = None
        self._snapshot: Dict[str, float] = {}

    def startWatching(self):
        """开始监听文件变化。使用轮询方式检测文件系统事件。"""
        if self._watching:
            return
        self._watching = True
        self._snapshot = self._take_snapshot()
        self._thread = threading.Thread(target=self._poll_loop, daemon=True)
        self._thread.start()

    def _take_snapshot(self) -> Dict[str, float]:
        """获取当前目录的文件快照（文件名 -> 修改时间）。"""
        snapshot: Dict[str, float] = {}
        try:
            if _os.path.isdir(self._path):
                for name in _os.listdir(self._path):
                    full = _os.path.join(self._path, name)
                    try:
                        snapshot[name] = _os.path.getmtime(full)
                    except OSError:
                        pass
            elif _os.path.exists(self._path):
                try:
                    snapshot[self._path] = _os.path.getmtime(self._path)
                except OSError:
                    pass
        except OSError:
            pass
        return snapshot

    def _poll_loop(self):
        """轮询检测文件变化并触发事件。"""
        while self._watching:
            _time.sleep(0.5)
            try:
                current = self._take_snapshot()
                # 检测新建和修改
                for name, mtime in current.items():
                    if name not in self._snapshot:
                        self.onEvent(FileObserver.CREATE, name)
                    elif mtime != self._snapshot[name]:
                        self.onEvent(FileObserver.MODIFY, name)
                # 检测删除
                for name in list(self._snapshot.keys()):
                    if name not in current:
                        self.onEvent(FileObserver.DELETE, name)
                self._snapshot = current
            except Exception:
                pass

    def stopWatching(self):
        """停止监听。"""
        self._watching = False
        if self._thread and self._thread.is_alive():
            self._thread.join(timeout=1.0)
        self._thread = None

    def onEvent(self, event: int, path: Optional[str]):
        """文件事件回调，子类可重写。"""
        pass

    def finalize(self):
        """清理资源。"""
        self.stopWatching()


class FileUtils(Stub):
    """FileUtils — 文件工具"""

    @staticmethod
    def readFileToString(path):
        try:
            with open(path, "r", encoding="utf-8") as f:
                return f.read()
        except Exception:
            return ""

    @staticmethod
    def stringToFile(path, content):
        try:
            with open(path, "w", encoding="utf-8") as f:
                f.write(content)
            return True
        except Exception:
            return False

    @staticmethod
    def copyFile(src, dst):
        try:
            import shutil
            shutil.copy2(src, dst)
            return True
        except Exception:
            return False

    @staticmethod
    def deleteDirectory(path):
        try:
            import shutil
            shutil.rmtree(path)
            return True
        except Exception:
            return False

    @staticmethod
    def deleteContents(path):
        try:
            import shutil
            for entry in _os.listdir(path):
                full = _os.path.join(path, entry)
                if _os.path.isdir(full):
                    shutil.rmtree(full)
                else:
                    _os.remove(full)
            return True
        except Exception:
            return False

    @staticmethod
    def deleteOlderFiles(dir_: Any, min_count: int, min_age_ms: int):
        """删除目录中较旧的文件，但保留至少 min_count 个文件。
        min_age_ms 为文件的最小年龄（毫秒），低于此年龄的文件不会被删除。"""
        try:
            if not _os.path.isdir(dir_):
                return
            entries = []
            for name in _os.listdir(dir_):
                full = _os.path.join(dir_, name)
                if _os.path.isfile(full):
                    try:
                        mtime = _os.path.getmtime(full)
                        entries.append((full, mtime))
                    except OSError:
                        pass
            # 按修改时间排序（最新在前）
            entries.sort(key=lambda x: x[1], reverse=True)
            now = _time.time()
            min_age_s = min_age_ms / 1000.0
            # 保留最新的 min_count 个文件，删除超过 min_age 的旧文件
            for i, (path, mtime) in enumerate(entries):
                if i < min_count:
                    continue
                if (now - mtime) >= min_age_s:
                    try:
                        _os.remove(path)
                    except OSError:
                        pass
        except Exception:
            pass


class PatternMatcher(Stub):
    """PatternMatcher — 模式匹配器"""

    PATTERN_LITERAL = 0
    PATTERN_PREFIX = 1
    PATTERN_SIMPLE_GLOB = 2
    PATTERN_ADVANCED_GLOB = 3
    PATTERN_SUFFIX = 4

    def __init__(self, pattern: str, type_: int):
        super().__init__()
        self._pattern = pattern
        self._type = type_

    def getPath(self) -> str:
        return self._pattern

    def getType(self) -> int:
        return self._type

    def match(self, str_: str) -> bool:
        if self._type == self.PATTERN_LITERAL:
            return str_ == self._pattern
        if self._type == self.PATTERN_PREFIX:
            return str_.startswith(self._pattern)
        if self._type == self.PATTERN_SUFFIX:
            return str_.endswith(self._pattern)
        if self._type == self.PATTERN_SIMPLE_GLOB:
            import fnmatch
            return fnmatch.fnmatch(str_, self._pattern)
        return str_ == self._pattern

    def toString(self) -> str:
        return f"PatternMatcher({self._pattern}, type={self._type})"


# ═══════════════════════════════════════════════════════════════
# __all__
# ═══════════════════════════════════════════════════════════════

__all__ = [
    "Bundle",
    "BaseBundle",
    "PersistableBundle",
    "Handler",
    "Looper",
    "Message",
    "HandlerThread",
    "PowerManager",
    "PowerManagerWakeLock",
    "Vibrator",
    "VibratorEffect",
    "Build",
    "SystemClock",
    "Environment",
    "Parcel",
    "Parcelable",
    "Process",
    "AsyncTask",
    "CountDownTimer",
    "Messenger",
    "BatteryManager",
    "PatternMatcher",
    "ConditionVariable",
    "Semaphore",
    "TokenWatcher",
    "WorkSource",
    "UserHandle",
    "DropBoxManager",
    "FileObserver",
    "FileUtils",
    "NetworkOnMainThreadException",
    "TransactionTooLargeException",
    "OperationCanceledException",
]