From 690022cea350ca9026c9f4784802217b70d4f522 Mon Sep 17 00:00:00 2001 From: shuiyu1123 Date: Wed, 24 Jun 2026 15:40:49 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20Linux=20=E7=9C=9F=E5=AE=9E=E5=AE=9E?= =?UTF-8?q?=E7=8E=B0=E5=B1=82=20=E2=80=94=20=E4=BB=8E=E5=AD=98=E6=A0=B9?= =?UTF-8?q?=E5=BA=93=E5=8F=98=E4=B8=BA=E5=8F=AF=E7=94=A8=E7=94=9F=E4=BA=A7?= =?UTF-8?q?=E5=8A=9B=E5=B7=A5=E5=85=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 _linux_impl.py (800+行): - LinuxNetwork: 真实IP/接口/DNS/网关/MAC (ip命令+/proc/net) - LinuxBluetooth: 真实蓝牙状态/扫描/配对/连接 (bluetoothctl) - LinuxSensors: 真实温度/CPU频率/风扇 (/sys/class/hwmon) - LinuxAudio: 真实音量控制/静音 (amixer) - LinuxClipboard: 真实剪贴板读写 (xclip+tkinter回退) - LinuxPower: 真实CPU/内存/电池 (psutil+/sys/class/power_supply) - LinuxLocation: 真实IP地理定位 (ip-api.com) - LinuxNotify: 真实桌面通知 (notify-send) - LinuxDisplay: 真实屏幕尺寸/截屏/鼠标位置 (xdotool+X11) - LinuxUIAutomation: 真实点击/输入/按键/滚动 (xdotool) - LinuxCamera: 真实摄像头拍照 (opencv/V4L2) - LinuxProcess: 真实进程列表/系统信息/应用管理 (psutil+/proc) - LinuxShell: 真实命令执行 (subprocess) 改造10+核心类使用Linux真实实现: - BluetoothAdapter/SensorManager/AudioManager - ConnectivityManager/WifiManager/TelephonyManager - LocationManager/PowerManager/BatteryManager/Build - ActivityManager/Activity/Toast/Display/ClipboardManager - NotificationManager 15项真实行为测试全部通过: - 真实IP: 10.60.218.244 - 真实位置: 北京 39.9042, 116.407 - 真实主机名: work-41-vci-g427c - 真实内核: 5.15.120.byteatom-ck.14 - 真实内存: 1.8GB/3.8GB - 真实进程列表 --- android_pyapi/_linux_impl.py | 1227 ++++++++++++++++++++++++++++ android_pyapi/android/app.py | 66 +- android_pyapi/android/app_tasks.py | 15 + android_pyapi/android/bluetooth.py | 50 ++ android_pyapi/android/content.py | 33 + android_pyapi/android/hardware.py | 82 ++ android_pyapi/android/location.py | 31 + android_pyapi/android/managers.py | 55 +- android_pyapi/android/media.py | 29 + android_pyapi/android/net.py | 95 +++ android_pyapi/android/net_ext.py | 54 ++ android_pyapi/android/net_wifi.py | 55 ++ android_pyapi/android/os.py | 163 ++++ android_pyapi/android/os_power.py | 75 +- android_pyapi/android/telephony.py | 31 + android_pyapi/android/view.py | 16 + android_pyapi/android/widget.py | 14 +- 17 files changed, 2083 insertions(+), 8 deletions(-) create mode 100644 android_pyapi/_linux_impl.py diff --git a/android_pyapi/_linux_impl.py b/android_pyapi/_linux_impl.py new file mode 100644 index 0000000..34a58f0 --- /dev/null +++ b/android_pyapi/_linux_impl.py @@ -0,0 +1,1227 @@ +""" +Linux 真实实现层 +================ +在 Linux 环境下，将 Android API 的存根行为替换为真实的系统调用。 +使 Android-pyApi 从"存根库"变为"可用的生产力工具"。 + +覆盖能力: + - 网络: ip 命令, /proc/net, socket + - 蓝牙: bluetoothctl / dbus + - 传感器: /sys/class/hwmon, /sys/class/thermal, sensors + - 音频: amixer / ALSA + - 剪贴板: xclip + - 电源/电池: /sys/class/power_supply, psutil + - 位置: IP 地理定位 (requests) + - 通知: notify-send + - Shell: subprocess + - 显示: xdotool, X11 + - UI自动化: xdotool + - 摄像头: opencv (V4L2) + - 进程管理: psutil, /proc + - 文件: 原生 Python os + - 媒体: ffmpeg/ffprobe +""" + +from __future__ import annotations + +import os +import sys +import re +import json +import time +import socket +import struct +import shutil +import subprocess +import platform +import hashlib +from pathlib import Path +from datetime import datetime +from typing import Any, Dict, List, Optional, Tuple + +# ═══════════════════════════════════════════════════════════════ +# 能力检测 +# ═══════════════════════════════════════════════════════════════ + +IS_LINUX = platform.system() == "Linux" + +def _which(name: str) -> Optional[str]: + return shutil.which(name) + +HAS_IP = _which("ip") is not None +HAS_BLUETOOTHCTL = _which("bluetoothctl") is not None +HAS_AMIXER = _which("amixer") is not None +HAS_XCLIP = _which("xclip") is not None +HAS_XDOTOOL = _which("xdotool") is not None +HAS_NOTIFY_SEND = _which("notify-send") is not None +HAS_SENSORS = _which("sensors") is not None +HAS_FFMPEG = _which("ffmpeg") is not None +HAS_FFPROBE = _which("ffprobe") is not None +HAS_XRANDR = _which("xrandr") is not None + +try: + import psutil + HAS_PSUTIL = True +except ImportError: + HAS_PSUTIL = False + +try: + import requests + HAS_REQUESTS = True +except ImportError: + HAS_REQUESTS = False + +try: + import cv2 + HAS_CV2 = True +except ImportError: + HAS_CV2 = False + +try: + import numpy as np + HAS_NUMPY = True +except ImportError: + HAS_NUMPY = False + +try: + from PIL import Image, ImageDraw, ImageGrab + HAS_PIL = True +except ImportError: + HAS_PIL = False + +try: + import dbus + HAS_DBUS = True +except ImportError: + HAS_DBUS = False + + +def _run(cmd: List[str] or str, timeout: int = 10, capture: bool = True) -> Tuple[int, str, str]: + """执行命令，返回 (exit_code, stdout, stderr)""" + try: + if isinstance(cmd, str): + result = subprocess.run(cmd, shell=True, capture_output=capture, + text=True, timeout=timeout) + else: + result = subprocess.run(cmd, capture_output=capture, + text=True, timeout=timeout) + return result.returncode, result.stdout.strip(), result.stderr.strip() + except subprocess.TimeoutExpired: + return -1, "", "timeout" + except Exception as e: + return -1, "", str(e) + + +def _read_file(path: str) -> Optional[str]: + try: + with open(path, "r") as f: + return f.read().strip() + except Exception: + return None + + +def _read_file_int(path: str) -> Optional[int]: + val = _read_file(path) + if val is not None: + try: + return int(val) + except ValueError: + return None + return None + + +def _list_dir(path: str) -> List[str]: + try: + return os.listdir(path) + except Exception: + return [] + + +# ═══════════════════════════════════════════════════════════════ +# 网络 — 真实实现 +# ═══════════════════════════════════════════════════════════════ + +class LinuxNetwork: + """真实网络信息 — 通过 ip 命令和 /proc/net""" + + @staticmethod + def get_interfaces() -> List[dict]: + """获取网络接口列表""" + if not HAS_IP: + return [] + code, out, _ = _run(["ip", "-j", "addr", "show"]) + if code == 0 and out: + try: + return json.loads(out) + except json.JSONDecodeError: + pass + # 回退到 /proc/net + interfaces = [] + for name in _list_dir("/sys/class/net/"): + state = _read_file(f"/sys/class/net/{name}/operstate") or "unknown" + mac = _read_file(f"/sys/class/net/{name}/address") or "" + interfaces.append({"ifname": name, "operstate": state, "address": mac}) + return interfaces + + @staticmethod + def get_ip_address(interface: str = None) -> str: + """获取IP地址""" + # 优先用 socket 获取外网IP + try: + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + s.connect(("8.8.8.8", 80)) + ip = s.getsockname()[0] + s.close() + return ip + except Exception: + pass + # 回退到 hostname + try: + return socket.gethostbyname(socket.gethostname()) + except Exception: + return "127.0.0.1" + + @staticmethod + def get_wifi_info() -> dict: + """获取WiFi信息""" + # 尝试 /proc/net/wireless + data = _read_file("/proc/net/wireless") + if data: + lines = data.strip().split("\n") + if len(lines) >= 3: + parts = lines[2].split() + if len(parts) >= 7: + return { + "ssid": "", + "interface": parts[0].rstrip(":"), + "status": parts[1], + "quality": float(parts[2]), + "signal": float(parts[3]), + "noise": float(parts[4]), + } + # 尝试 nmcli + code, out, _ = _run(["nmcli", "-t", "-f", "active,ssid,signal", "dev", "wifi"]) + if code == 0 and out: + for line in out.split("\n"): + if line.startswith("yes:"): + parts = line.split(":") + if len(parts) >= 3: + return {"ssid": parts[1], "signal": int(parts[2])} + return {"ssid": "", "signal": 0} + + @staticmethod + def get_connectivity() -> bool: + """检测网络连通性""" + # 方法1: 尝试 TCP 连接到公共 DNS + for host, port in [("8.8.8.8", 53), ("1.1.1.1", 53), ("114.114.114.114", 53)]: + try: + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.settimeout(3) + s.connect((host, port)) + s.close() + return True + except Exception: + continue + # 方法2: 尝试 HTTP 请求 + if HAS_REQUESTS: + try: + requests.get("http://www.baidu.com", timeout=3) + return True + except Exception: + pass + # 方法3: 检查是否有非 lo 接口且状态为 UP + for iface in LinuxNetwork.get_interfaces(): + name = iface.get("ifname", iface.get("name", "")) + state = iface.get("operstate", iface.get("state", "")) + if name and name != "lo" and state in ("up", "UP"): + return True + return False + + @staticmethod + def get_dns_servers() -> List[str]: + """获取DNS服务器""" + data = _read_file("/etc/resolv.conf") + if data: + dns = [] + for line in data.split("\n"): + line = line.strip() + if line.startswith("nameserver"): + parts = line.split() + if len(parts) >= 2: + dns.append(parts[1]) + return dns + return [] + + @staticmethod + def get_gateway() -> str: + """获取默认网关""" + data = _read_file("/proc/net/route") + if data: + for line in data.split("\n")[1:]: + parts = line.split() + if len(parts) >= 3 and parts[1] == "00000000": + # 网关是十六进制小端 + gw_hex = parts[2] + try: + gw_bytes = bytes.fromhex(gw_hex) + gw = socket.inet_ntoa(gw_bytes[::-1]) + return gw + except Exception: + pass + return "" + + @staticmethod + def get_mac_address(interface: str = None) -> str: + """获取MAC地址""" + if interface is None: + ifaces = _list_dir("/sys/class/net/") + if ifaces and "lo" in ifaces: + ifaces.remove("lo") + if ifaces: + interface = ifaces[0] + if interface: + return _read_file(f"/sys/class/net/{interface}/address") or "" + return "" + + @staticmethod + def get_rx_tx_bytes(interface: str = None) -> dict: + """获取网络收发字节数""" + if interface is None: + ifaces = _list_dir("/sys/class/net/") + if "lo" in ifaces: + ifaces.remove("lo") + if ifaces: + interface = ifaces[0] + if interface: + rx = _read_file_int(f"/sys/class/net/{interface}/statistics/rx_bytes") or 0 + tx = _read_file_int(f"/sys/class/net/{interface}/statistics/tx_bytes") or 0 + return {"rx_bytes": rx, "tx_bytes": tx, "interface": interface} + return {"rx_bytes": 0, "tx_bytes": 0, "interface": ""} + + +# ═══════════════════════════════════════════════════════════════ +# 蓝牙 — 真实实现 (bluetoothctl) +# ═══════════════════════════════════════════════════════════════ + +class LinuxBluetooth: + """真实蓝牙操作 — 通过 bluetoothctl""" + + STATE_OFF = 10 + STATE_TURNING_ON = 11 + STATE_ON = 12 + STATE_TURNING_OFF = 13 + + @staticmethod + def get_state() -> int: + if not HAS_BLUETOOTHCTL: + return LinuxBluetooth.STATE_OFF + code, out, _ = _run(["bluetoothctl", "show"]) + if "Powered: yes" in out: + return LinuxBluetooth.STATE_ON + return LinuxBluetooth.STATE_OFF + + @staticmethod + def enable() -> bool: + if not HAS_BLUETOOTHCTL: + return False + code, _, _ = _run(["bluetoothctl", "power", "on"]) + return code == 0 + + @staticmethod + def disable() -> bool: + if not HAS_BLUETOOTHCTL: + return False + code, _, _ = _run(["bluetoothctl", "power", "off"]) + return code == 0 + + @staticmethod + def get_bonded_devices() -> List[dict]: + if not HAS_BLUETOOTHCTL: + return [] + code, out, _ = _run(["bluetoothctl", "devices", "Paired"]) + devices = [] + for line in out.split("\n"): + if line.startswith("Device "): + parts = line.split(None, 2) + if len(parts) >= 3: + devices.append({ + "address": parts[1], + "name": parts[2], + "bonded": True, + }) + return devices + + @staticmethod + def scan(timeout: int = 10) -> List[dict]: + if not HAS_BLUETOOTHCTL: + return [] + # 启动扫描 + _run(["bluetoothctl", "scan", "on"], timeout=2) + time.sleep(min(timeout, 5)) + _run(["bluetoothctl", "scan", "off"], timeout=2) + # 获取设备列表 + code, out, _ = _run(["bluetoothctl", "devices"]) + devices = [] + for line in out.split("\n"): + if line.startswith("Device "): + parts = line.split(None, 2) + if len(parts) >= 3: + devices.append({ + "address": parts[1], + "name": parts[2], + }) + return devices + + @staticmethod + def pair(address: str) -> bool: + if not HAS_BLUETOOTHCTL: + return False + code, _, _ = _run(["bluetoothctl", "pair", address], timeout=30) + return code == 0 + + @staticmethod + def connect(address: str) -> bool: + if not HAS_BLUETOOTHCTL: + return False + code, _, _ = _run(["bluetoothctl", "connect", address], timeout=15) + return code == 0 + + @staticmethod + def disconnect(address: str) -> bool: + if not HAS_BLUETOOTHCTL: + return False + code, _, _ = _run(["bluetoothctl", "disconnect", address], timeout=10) + return code == 0 + + @staticmethod + def unpair(address: str) -> bool: + if not HAS_BLUETOOTHCTL: + return False + code, _, _ = _run(["bluetoothctl", "remove", address], timeout=10) + return code == 0 + + +# ═══════════════════════════════════════════════════════════════ +# 传感器 — 真实实现 (/sys/class/hwmon, /sys/class/thermal) +# ═══════════════════════════════════════════════════════════════ + +class LinuxSensors: + """真实传感器数据 — 通过 /sys 文件系统""" + + # Android Sensor Types + TYPE_ACCELEROMETER = 1 + TYPE_GYROSCOPE = 4 + TYPE_LIGHT = 5 + TYPE_PRESSURE = 6 + TYPE_PROXIMITY = 8 + TYPE_TEMPERATURE = 7 + TYPE_RELATIVE_HUMIDITY = 12 + TYPE_AMBIENT_TEMPERATURE = 13 + + @staticmethod + def get_temperatures() -> List[dict]: + """获取温度传感器读数""" + temps = [] + # /sys/class/thermal + for name in _list_dir("/sys/class/thermal/"): + if name.startswith("thermal_zone"): + temp = _read_file_int(f"/sys/class/thermal/{name}/temp") + typ = _read_file(f"/sys/class/thermal/{name}/type") or "" + if temp is not None: + temps.append({ + "name": name, + "type": typ, + "value": temp / 1000.0, # millicelsius -> celsius + "unit": "°C", + }) + # /sys/class/hwmon + for name in _list_dir("/sys/class/hwmon/"): + temp_files = [f for f in _list_dir(f"/sys/class/hwmon/{name}/") + if f.startswith("temp") and f.endswith("_input")] + for tf in temp_files: + temp = _read_file_int(f"/sys/class/hwmon/{name}/{tf}") + label_file = tf.replace("_input", "_label") + label = _read_file(f"/sys/class/hwmon/{name}/{label_file}") or tf + if temp is not None: + temps.append({ + "name": f"{name}/{label}", + "type": "hwmon", + "value": temp / 1000.0, + "unit": "°C", + }) + return temps + + @staticmethod + def get_fan_speeds() -> List[dict]: + """获取风扇转速""" + fans = [] + for name in _list_dir("/sys/class/hwmon/"): + fan_files = [f for f in _list_dir(f"/sys/class/hwmon/{name}/") + if f.startswith("fan") and f.endswith("_input")] + for ff in fan_files: + rpm = _read_file_int(f"/sys/class/hwmon/{name}/{ff}") + label_file = ff.replace("_input", "_label") + label = _read_file(f"/sys/class/hwmon/{name}/{label_file}") or ff + if rpm is not None: + fans.append({"name": label, "rpm": rpm}) + return fans + + @staticmethod + def get_cpu_frequency() -> List[dict]: + """获取CPU频率""" + freqs = [] + for name in _list_dir("/sys/devices/system/cpu/"): + if name.startswith("cpu") and name[3:].isdigit(): + cpu_id = name[3:] + cur = _read_file_int(f"/sys/devices/system/cpu/{name}/cpufreq/scaling_cur_freq") + if cur is not None: + freqs.append({ + "cpu": int(cpu_id), + "freq_mhz": cur / 1000.0, + }) + return freqs + + @staticmethod + def get_light_level() -> Optional[float]: + """获取环境光传感器 (如果有)""" + for name in _list_dir("/sys/class/hwmon/"): + for f in _list_dir(f"/sys/class/hwmon/{name}/"): + if "in" in f and f.endswith("_input"): + val = _read_file_int(f"/sys/class/hwmon/{name}/{f}") + if val is not None: + return val / 1000.0 + return None + + @staticmethod + def get_all_sensors() -> dict: + """获取所有传感器数据""" + return { + "temperatures": LinuxSensors.get_temperatures(), + "fans": LinuxSensors.get_fan_speeds(), + "cpu_freq": LinuxSensors.get_cpu_frequency(), + "light": LinuxSensors.get_light_level(), + } + + +# ═══════════════════════════════════════════════════════════════ +# 音频 — 真实实现 (amixer / ALSA) +# ═══════════════════════════════════════════════════════════════ + +class LinuxAudio: + """真实音频控制 — 通过 amixer""" + + @staticmethod + def get_volume() -> int: + """获取当前音量 (0-100)""" + if not HAS_AMIXER: + return 0 + code, out, _ = _run(["amixer", "sget", "Master"]) + if code == 0: + match = re.search(r'\[(\d+)%\]', out) + if match: + return int(match.group(1)) + return 0 + + @staticmethod + def set_volume(volume: int) -> bool: + """设置音量 (0-100)""" + if not HAS_AMIXER: + return False + volume = max(0, min(100, volume)) + code, _, _ = _run(["amixer", "sset", "Master", f"{volume}%"]) + return code == 0 + + @staticmethod + def get_mute() -> bool: + """获取静音状态""" + if not HAS_AMIXER: + return False + code, out, _ = _run(["amixer", "sget", "Master"]) + if code == 0: + return "[on]" not in out + return False + + @staticmethod + def set_mute(mute: bool) -> bool: + """设置静音""" + if not HAS_AMIXER: + return False + cmd = "mute" if mute else "unmute" + code, _, _ = _run(["amixer", "sset", "Master", cmd]) + return code == 0 + + @staticmethod + def get_audio_devices() -> List[dict]: + """获取音频设备列表""" + devices = [] + if HAS_AMIXER: + code, out, _ = _run(["amixer", "controls"]) + if code == 0: + for line in out.split("\n"): + devices.append({"name": line, "type": "alsa"}) + return devices + + +# ═══════════════════════════════════════════════════════════════ +# 剪贴板 — 真实实现 (xclip) +# ═══════════════════════════════════════════════════════════════ + +class LinuxClipboard: + """真实剪贴板 — 通过 xclip""" + + @staticmethod + def read() -> str: + if HAS_XCLIP: + code, out, _ = _run(["xclip", "-selection", "clipboard", "-o"]) + if code == 0: + return out + # 回退到 tkinter + try: + import tkinter + root = tkinter.Tk() + root.withdraw() + text = root.clipboard_get() + root.destroy() + return text + except Exception: + return "" + + @staticmethod + def write(text: str) -> bool: + if HAS_XCLIP: + try: + proc = subprocess.Popen( + ["xclip", "-selection", "clipboard"], + stdin=subprocess.PIPE, text=True + ) + proc.communicate(text) + return proc.returncode == 0 + except Exception: + pass + # 回退到 tkinter + try: + import tkinter + root = tkinter.Tk() + root.withdraw() + root.clipboard_clear() + root.clipboard_append(text) + root.update() + root.destroy() + return True + except Exception: + return False + + +# ═══════════════════════════════════════════════════════════════ +# 电源/电池 — 真实实现 (/sys/class/power_supply, psutil) +# ═══════════════════════════════════════════════════════════════ + +class LinuxPower: + """真实电源/电池信息""" + + BATTERY_STATUS_UNKNOWN = 1 + BATTERY_STATUS_CHARGING = 2 + BATTERY_STATUS_DISCHARGING = 3 + BATTERY_STATUS_NOT_CHARGING = 4 + BATTERY_STATUS_FULL = 5 + + @staticmethod + def get_battery_info() -> dict: + """获取电池信息""" + info = {"present": False, "level": -1, "status": "unknown", + "temperature": 0, "voltage": 0, "technology": ""} + # /sys/class/power_supply + for name in _list_dir("/sys/class/power_supply/"): + typ = _read_file(f"/sys/class/power_supply/{name}/type") or "" + if "Battery" in typ: + info["present"] = True + info["technology"] = _read_file(f"/sys/class/power_supply/{name}/technology") or "" + capacity = _read_file_int(f"/sys/class/power_supply/{name}/capacity") + if capacity is not None: + info["level"] = capacity + status = _read_file(f"/sys/class/power_supply/{name}/status") or "" + info["status"] = status.lower() + # 电压 (微伏) + volt = _read_file_int(f"/sys/class/power_supply/{name}/voltage_now") + if volt is not None: + info["voltage"] = volt / 1000000.0 + # 温度 (十分之一度) + temp = _read_file_int(f"/sys/class/power_supply/{name}/temp") + if temp is not None: + info["temperature"] = temp / 10.0 + return info + # psutil 回退 + if HAS_PSUTIL: + try: + battery = psutil.sensors_battery() + if battery: + info["present"] = True + info["level"] = battery.percent + info["status"] = "charging" if battery.power_plugged else "discharging" + except Exception: + pass + return info + + @staticmethod + def get_power_status() -> dict: + """获取电源状态""" + info = LinuxPower.get_battery_info() + return { + "battery_present": info["present"], + "battery_level": info["level"], + "battery_status": info["status"], + "is_charging": "charging" in info["status"], + "temperature": info["temperature"], + "voltage": info["voltage"], + } + + @staticmethod + def get_cpu_usage() -> float: + """获取CPU使用率""" + if HAS_PSUTIL: + return psutil.cpu_percent(interval=0.5) + # /proc/stat 回退 + def read_cpu(): + data = _read_file("/proc/stat") + if data: + parts = data.split("\n")[0].split()[1:] + return [int(x) for x in parts] + return [0] * 10 + c1 = read_cpu() + time.sleep(0.5) + c2 = read_cpu() + idle = c2[3] - c1[3] + total = sum(c2) - sum(c1) + if total > 0: + return (1 - idle / total) * 100 + return 0.0 + + @staticmethod + def get_memory_info() -> dict: + """获取内存信息""" + if HAS_PSUTIL: + mem = psutil.virtual_memory() + return { + "total": mem.total, + "available": mem.available, + "used": mem.used, + "free": mem.free, + "percent": mem.percent, + } + # /proc/meminfo 回退 + data = _read_file("/proc/meminfo") + info = {} + if data: + for line in data.split("\n"): + parts = line.split() + if len(parts) >= 2: + info[parts[0].rstrip(":")] = int(parts[1]) * 1024 + return { + "total": info.get("MemTotal", 0), + "available": info.get("MemAvailable", 0), + "used": info.get("MemTotal", 0) - info.get("MemAvailable", 0), + "free": info.get("MemFree", 0), + "percent": 0, + } + + +# ═══════════════════════════════════════════════════════════════ +# 位置 — 真实实现 (IP 地理定位) +# ═══════════════════════════════════════════════════════════════ + +class LinuxLocation: + """真实位置信息 — 通过 IP 地理定位""" + + _cached_location = None + _cache_time = 0 + + @staticmethod + def get_location() -> dict: + """通过 IP 获取地理位置""" + if not HAS_REQUESTS: + return {"latitude": 0.0, "longitude": 0.0, "accuracy": 0, "source": "none"} + + # 缓存 5 分钟 + now = time.time() + if LinuxLocation._cached_location and now - LinuxLocation._cache_time < 300: + return LinuxLocation._cached_location + + # 尝试多个 IP 定位服务 + services = [ + ("http://ip-api.com/json/", "ip-api"), + ("https://ipapi.co/json/", "ipapi"), + ("https://freeipapi.com/api/json", "freeipapi"), + ] + + for url, source in services: + try: + resp = requests.get(url, timeout=5) + if resp.status_code == 200: + data = resp.json() + lat = data.get("lat") or data.get("latitude") or 0.0 + lon = data.get("lon") or data.get("longitude") or 0.0 + location = { + "latitude": float(lat), + "longitude": float(lon), + "accuracy": 50000, # IP定位精度约 50km + "source": source, + "city": data.get("city", ""), + "region": data.get("regionName", data.get("region", "")), + "country": data.get("country", data.get("country_name", "")), + "country_code": data.get("countryCode", data.get("country", "")), + "timezone": data.get("timezone", ""), + "isp": data.get("isp", ""), + } + LinuxLocation._cached_location = location + LinuxLocation._cache_time = now + return location + except Exception: + continue + + return {"latitude": 0.0, "longitude": 0.0, "accuracy": 0, "source": "failed"} + + +# ═══════════════════════════════════════════════════════════════ +# 通知 — 真实实现 (notify-send) +# ═══════════════════════════════════════════════════════════════ + +class LinuxNotify: + """真实桌面通知 — 通过 notify-send""" + + @staticmethod + def send(title: str, text: str, icon: str = None, urgency: str = "normal", + timeout: int = -1) -> bool: + if not HAS_NOTIFY_SEND: + # 回退到 stderr + print(f"[{title}] {text}", file=sys.stderr) + return False + cmd = ["notify-send"] + cmd.extend(["-u", urgency]) + if timeout >= 0: + cmd.extend(["-t", str(timeout)]) + if icon: + cmd.extend(["-i", icon]) + cmd.extend([title, text]) + code, _, _ = _run(cmd) + return code == 0 + + +# ═══════════════════════════════════════════════════════════════ +# 显示/屏幕 — 真实实现 (xdotool, X11) +# ═══════════════════════════════════════════════════════════════ + +class LinuxDisplay: + """真实显示/屏幕信息""" + + @staticmethod + def get_screen_size() -> Tuple[int, int]: + """获取屏幕尺寸""" + if HAS_XDOTOOL: + code, out, _ = _run(["xdotool", "getdisplaygeometry"]) + if code == 0: + parts = out.split() + if len(parts) >= 2: + return int(parts[0]), int(parts[1]) + if HAS_XRANDR: + code, out, _ = _run(["xrandr", "--current"]) + if code == 0: + for line in out.split("\n"): + if " connected" in line and "*" in line: + match = re.search(r'(\d+)x(\d+)', line) + if match: + return int(match.group(1)), int(match.group(2)) + # 回退 + try: + import tkinter + root = tkinter.Tk() + w = root.winfo_screenwidth() + h = root.winfo_screenheight() + root.destroy() + return w, h + except Exception: + return 1920, 1080 + + @staticmethod + def screenshot(path: str = None) -> Optional[str]: + """截屏""" + if HAS_PIL: + try: + img = ImageGrab.grab() + if path: + img.save(path) + return path + else: + buf = f"/tmp/screenshot_{int(time.time())}.png" + img.save(buf) + return buf + except Exception: + pass + if HAS_XDOTOOL: + # 用 xdotool + import + if not path: + path = f"/tmp/screenshot_{int(time.time())}.png" + code, _, _ = _run(["import", "-window", "root", path]) + if code == 0: + return path + return None + + @staticmethod + def get_mouse_position() -> Tuple[int, int]: + """获取鼠标位置""" + if HAS_XDOTOOL: + code, out, _ = _run(["xdotool", "getmouselocation"]) + if code == 0: + parts = out.split() + x = int(parts[0].split(":")[1]) + y = int(parts[1].split(":")[1]) + return x, y + return 0, 0 + + +# ═══════════════════════════════════════════════════════════════ +# UI 自动化 — 真实实现 (xdotool) +# ═══════════════════════════════════════════════════════════════ + +class LinuxUIAutomation: + """真实 UI 自动化 — 通过 xdotool""" + + @staticmethod + def click(x: int, y: int) -> bool: + if not HAS_XDOTOOL: + return False + code, _, _ = _run(["xdotool", "mousemove", str(x), str(y), "click", "1"]) + return code == 0 + + @staticmethod + def double_click(x: int, y: int) -> bool: + if not HAS_XDOTOOL: + return False + code, _, _ = _run(["xdotool", "mousemove", str(x), str(y), "click", "--repeat", "2", "1"]) + return code == 0 + + @staticmethod + def right_click(x: int, y: int) -> bool: + if not HAS_XDOTOOL: + return False + code, _, _ = _run(["xdotool", "mousemove", str(x), str(y), "click", "3"]) + return code == 0 + + @staticmethod + def type_text(text: str) -> bool: + if not HAS_XDOTOOL: + return False + code, _, _ = _run(["xdotool", "type", "--clearmodifiers", text]) + return code == 0 + + @staticmethod + def press_key(key: str) -> bool: + if not HAS_XDOTOOL: + return False + # 转换常见键名 + key_map = { + "enter": "Return", "return": "Return", + "tab": "Tab", "escape": "Escape", "esc": "Escape", + "backspace": "BackSpace", "delete": "Delete", + "space": "space", "up": "Up", "down": "Down", + "left": "Left", "right": "Right", + "home": "Home", "end": "End", + "page_up": "Page_Up", "page_down": "Page_Down", + } + xdotool_key = key_map.get(key.lower(), key) + code, _, _ = _run(["xdotool", "key", xdotool_key]) + return code == 0 + + @staticmethod + def hotkey(*keys) -> bool: + if not HAS_XDOTOOL: + return False + key_map = { + "ctrl": "ctrl", "control": "ctrl", + "alt": "alt", "shift": "shift", + "super": "super", "win": "super", "meta": "super", + } + mapped = [key_map.get(k.lower(), k) for k in keys] + combo = "+".join(mapped) + code, _, _ = _run(["xdotool", "key", combo]) + return code == 0 + + @staticmethod + def scroll(amount: int) -> bool: + if not HAS_XDOTOOL: + return False + button = "4" if amount > 0 else "5" + count = abs(amount) + code, _, _ = _run(["xdotool", "click", "--repeat", str(count), button]) + return code == 0 + + @staticmethod + def find_window(title: str) -> Optional[int]: + """查找窗口""" + if not HAS_XDOTOOL: + return None + code, out, _ = _run(["xdotool", "search", "--name", title]) + if code == 0 and out: + return int(out.split("\n")[0]) + return None + + @staticmethod + def activate_window(window_id: int) -> bool: + if not HAS_XDOTOOL: + return False + code, _, _ = _run(["xdotool", "windowactivate", str(window_id)]) + return code == 0 + + @staticmethod + def close_window(window_id: int) -> bool: + if not HAS_XDOTOOL: + return False + code, _, _ = _run(["xdotool", "windowclose", str(window_id)]) + return code == 0 + + +# ═══════════════════════════════════════════════════════════════ +# 摄像头 — 真实实现 (opencv/V4L2) +# ═══════════════════════════════════════════════════════════════ + +class LinuxCamera: + """真实摄像头 — 通过 OpenCV/V4L2""" + + @staticmethod + def list_cameras() -> List[int]: + """列出可用摄像头""" + cameras = [] + for dev in _list_dir("/dev/"): + if dev.startswith("video"): + try: + idx = int(dev[5:]) + cameras.append(idx) + except ValueError: + pass + return cameras + + @staticmethod + def capture(camera_id: int = 0, output_path: str = None) -> Optional[str]: + """拍照""" + if not HAS_CV2: + return None + cap = cv2.VideoCapture(camera_id) + if not cap.isOpened(): + return None + ret, frame = cap.read() + cap.release() + if not ret: + return None + if not output_path: + output_path = f"/tmp/camera_{int(time.time())}.jpg" + cv2.imwrite(output_path, frame) + return output_path + + @staticmethod + def get_camera_info(camera_id: int = 0) -> dict: + """获取摄像头信息""" + info = {"id": camera_id, "available": False, "width": 0, "height": 0} + if HAS_CV2: + cap = cv2.VideoCapture(camera_id) + if cap.isOpened(): + info["available"] = True + info["width"] = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) + info["height"] = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) + info["fps"] = cap.get(cv2.CAP_PROP_FPS) + cap.release() + return info + + +# ═══════════════════════════════════════════════════════════════ +# 进程/应用管理 — 真实实现 (psutil, /proc) +# ═══════════════════════════════════════════════════════════════ + +class LinuxProcess: + """真实进程/应用管理""" + + @staticmethod + def list_processes() -> List[dict]: + """列出所有进程""" + if HAS_PSUTIL: + procs = [] + for p in psutil.process_iter(['pid', 'name', 'username', 'memory_percent', + 'cpu_percent', 'status', 'cmdline']): + try: + info = p.info + procs.append({ + "pid": info['pid'], + "name": info['name'], + "user": info['username'], + "memory": info.get('memory_percent', 0), + "cpu": info.get('cpu_percent', 0), + "status": info['status'], + "cmdline": " ".join(info.get('cmdline') or []), + }) + except (psutil.NoSuchProcess, psutil.AccessDenied): + pass + return procs + # /proc 回退 + procs = [] + for name in _list_dir("/proc/"): + if name.isdigit(): + pid = int(name) + comm = _read_file(f"/proc/{name}/comm") or "" + cmdline = _read_file(f"/proc/{name}/cmdline") or "" + status = _read_file(f"/proc/{name}/status") or "" + user = "" + for line in status.split("\n"): + if line.startswith("Uid:"): + uid = line.split()[1] + try: + import pwd + user = pwd.getpwuid(int(uid)).pw_name + except Exception: + user = uid + break + procs.append({ + "pid": pid, "name": comm, "user": user, + "cmdline": cmdline.replace("\x00", " "), + "status": "", "memory": 0, "cpu": 0, + }) + return procs + + @staticmethod + def get_running_apps() -> List[dict]: + """获取运行中的应用 (GUI 窗口)""" + if not HAS_XDOTOOL: + return [] + code, out, _ = _run(["xdotool", "search", "--onlyvisible", "--class", ""]) + apps = [] + if code == 0: + seen = set() + for wid in out.split("\n"): + if wid and wid not in seen: + seen.add(wid) + # 获取窗口名称 + code2, name, _ = _run(["xdotool", "getwindowname", wid]) + if name: + apps.append({"window_id": int(wid), "name": name}) + return apps + + @staticmethod + def kill_process(pid: int) -> bool: + """终止进程""" + if HAS_PSUTIL: + try: + psutil.Process(pid).terminate() + return True + except Exception: + return False + code, _, _ = _run(["kill", str(pid)]) + return code == 0 + + @staticmethod + def start_app(name: str) -> bool: + """启动应用""" + code, _, _ = _run([name], timeout=1) + # 非阻塞启动 + try: + subprocess.Popen([name], stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL) + return True + except Exception: + return False + + @staticmethod + def get_system_info() -> dict: + """获取系统信息""" + info = { + "platform": platform.system(), + "hostname": socket.gethostname(), + "kernel": platform.release(), + "arch": platform.machine(), + "python": platform.python_version(), + } + if HAS_PSUTIL: + info["cpu_count"] = psutil.cpu_count() + info["cpu_freq"] = psutil.cpu_freq().current if psutil.cpu_freq() else 0 + boot_time = psutil.boot_time() + info["uptime"] = time.time() - boot_time + mem = psutil.virtual_memory() + info["memory_total"] = mem.total + info["memory_used"] = mem.used + info["memory_percent"] = mem.percent + disk = psutil.disk_usage('/') + info["disk_total"] = disk.total + info["disk_used"] = disk.used + info["disk_percent"] = disk.percent + return info + + +# ═══════════════════════════════════════════════════════════════ +# Shell — 真实实现 +# ═══════════════════════════════════════════════════════════════ + +class LinuxShell: + """真实 Shell 执行""" + + @staticmethod + def exec(command: str, timeout: int = 30, root: bool = False) -> dict: + """执行 Shell 命令""" + if root and os.geteuid() != 0: + command = f"sudo {command}" + code, out, err = _run(command, timeout=timeout) + return { + "success": code == 0, + "output": out + ("\n" + err if err else ""), + "exit_code": code, + "stdout": out, + "stderr": err, + } + + @staticmethod + def exec_async(command: str) -> int: + """异步执行命令，返回 PID""" + proc = subprocess.Popen(command, shell=True, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL) + return proc.pid + + +# ═══════════════════════════════════════════════════════════════ +# 统一入口 +# ═══════════════════════════════════════════════════════════════ + +def get_linux_capabilities() -> dict: + """获取当前 Linux 环境的可用能力""" + return { + "is_linux": IS_LINUX, + "network": HAS_IP, + "bluetooth": HAS_BLUETOOTHCTL, + "audio": HAS_AMIXER, + "clipboard": HAS_XCLIP, + "ui_automation": HAS_XDOTOOL, + "notifications": HAS_NOTIFY_SEND, + "sensors": True, # /sys 总是可用 + "display": True, + "camera": HAS_CV2, + "ffmpeg": HAS_FFMPEG, + "ffprobe": HAS_FFPROBE, + "psutil": HAS_PSUTIL, + "requests": HAS_REQUESTS, + "PIL": HAS_PIL, + "opencv": HAS_CV2, + "numpy": HAS_NUMPY, + "dbus": HAS_DBUS, + "xrandr": HAS_XRANDR, + } + + +# 导出所有实现类 +__all__ = [ + "IS_LINUX", "get_linux_capabilities", + "LinuxNetwork", "LinuxBluetooth", "LinuxSensors", + "LinuxAudio", "LinuxClipboard", "LinuxPower", + "LinuxLocation", "LinuxNotify", "LinuxDisplay", + "LinuxUIAutomation", "LinuxCamera", "LinuxProcess", + "LinuxShell", + # 能力标志 + "HAS_IP", "HAS_BLUETOOTHCTL", "HAS_AMIXER", "HAS_XCLIP", + "HAS_XDOTOOL", "HAS_NOTIFY_SEND", "HAS_SENSORS", + "HAS_FFMPEG", "HAS_FFPROBE", "HAS_PSUTIL", "HAS_REQUESTS", + "HAS_CV2", "HAS_NUMPY", "HAS_PIL", "HAS_DBUS", "HAS_XRANDR", +] diff --git a/android_pyapi/android/app.py b/android_pyapi/android/app.py index b75c043..8817b8e 100644 --- a/android_pyapi/android/app.py +++ b/android_pyapi/android/app.py @@ -19,6 +19,12 @@ from .._core import ( _sdcard_root, _app_data_dir, _package_name, ) +try: + from .._linux_impl import IS_LINUX, LinuxNotify, LinuxShell, LinuxProcess, LinuxDisplay, LinuxUIAutomation, LinuxPower + _USE_LINUX = IS_LINUX +except ImportError: + _USE_LINUX = False + # ═══════════════════════════════════════════════════════════════ # Intent @@ -834,10 +840,16 @@ class NotificationManager(Stub): def notify(self, id_: int, notification: Notification): self._notifications[id_] = notification - sys.stderr.write( - f"[NotificationManager] notify id={id_}: " - f"title={notification._title} text={notification._text}\n" - ) + if _USE_LINUX: + title = getattr(notification, "_title", "") or "" + text = getattr(notification, "_text", "") or "" + if title or text: + LinuxNotify.send(title, text) + else: + sys.stderr.write( + f"[NotificationManager] notify id={id_}: " + f"title={notification._title} text={notification._text}\n" + ) def notifyAsPackage(self, package_name: str, tag: Optional[str], id_: int, notification: Notification): self.notify(id_, notification) @@ -1821,6 +1833,48 @@ class Activity(Stub): def getTheme(self): return getattr(self, '_theme', 0) + def getDeviceInfo(self): + """获取设备信息""" + if _USE_LINUX: + return LinuxProcess.get_system_info() + return {} + + def executeShell(self, command): + """执行 Shell 命令""" + if _USE_LINUX: + return LinuxShell.exec(command) + return {"success": False, "output": "Shell not available"} + + def screenshot(self, path=None): + """截屏""" + if _USE_LINUX: + return LinuxDisplay.screenshot(path) + return None + + def getScreenSize(self): + """获取屏幕尺寸""" + if _USE_LINUX: + return LinuxDisplay.get_screen_size() + return (1920, 1080) + + def click(self, x, y): + """点击屏幕坐标""" + if _USE_LINUX: + return LinuxUIAutomation.click(x, y) + return False + + def typeText(self, text): + """输入文本""" + if _USE_LINUX: + return LinuxUIAutomation.type_text(text) + return False + + def pressKey(self, key): + """按键""" + if _USE_LINUX: + return LinuxUIAutomation.press_key(key) + return False + # ═══════════════════════════════════════════════════════════════ # Service @@ -2009,9 +2063,13 @@ class ActivityManager(Stub): self._memory_info = self.MemoryInfo() def getMemoryInfo(self) -> "ActivityManager.MemoryInfo": + if _USE_LINUX: + return LinuxPower.get_memory_info() return self._memory_info def getRunningAppProcesses(self) -> Optional[List["ActivityManager.RunningAppProcessInfo"]]: + if _USE_LINUX: + return LinuxProcess.list_processes() return [] def killBackgroundProcesses(self, package_name: str): diff --git a/android_pyapi/android/app_tasks.py b/android_pyapi/android/app_tasks.py index 768dac9..eb8680a 100644 --- a/android_pyapi/android/app_tasks.py +++ b/android_pyapi/android/app_tasks.py @@ -13,6 +13,12 @@ from typing import Any, Callable, Dict, List, Optional, Tuple from .._core import Stub, AndroidOnlyError +try: + from .._linux_impl import IS_LINUX, LinuxProcess, LinuxPower + _USE_LINUX = IS_LINUX +except ImportError: + _USE_LINUX = False + __all__ = [ "TaskManager", "TaskInfo", "TaskStackListener", "TaskSnapshot", "TaskDescription", "ActivityManager", @@ -908,9 +914,18 @@ class ActivityManager(Stub): return list(self._services.values())[:maxNum] def getRunningAppProcesses(self) -> List[RunningAppProcessInfo]: + if _USE_LINUX: + return LinuxProcess.list_processes() return list(self._processes.values()) def getMemoryInfo(self, outInfo: MemoryInfo): + if _USE_LINUX: + mem = LinuxPower.get_memory_info() + outInfo.availMem = mem.get("available", 0) + outInfo.totalMem = mem.get("total", 0) + outInfo.threshold = int(mem.get("total", 0) * 0.1) + outInfo.lowMemory = mem.get("percent", 0) > 90.0 + return outInfo import psutil try: import psutil diff --git a/android_pyapi/android/bluetooth.py b/android_pyapi/android/bluetooth.py index 14c28c5..3793ce3 100644 --- a/android_pyapi/android/bluetooth.py +++ b/android_pyapi/android/bluetooth.py @@ -14,6 +14,13 @@ from .._core import ( _sdcard_root, _app_data_dir, _package_name, ) +# Linux 真实实现导入 +try: + from .._linux_impl import IS_LINUX, LinuxBluetooth, HAS_BLUETOOTHCTL + _USE_LINUX = IS_LINUX +except ImportError: + _USE_LINUX = False + # ═══════════════════════════════════════════════════════════════ # BluetoothProfile @@ -645,14 +652,20 @@ class BluetoothAdapter(Stub): return BluetoothAdapter._default_adapter def isEnabled(self) -> bool: + if _USE_LINUX: + return LinuxBluetooth.get_state() == LinuxBluetooth.STATE_ON return self._enabled def enable(self) -> bool: + if _USE_LINUX: + return LinuxBluetooth.enable() self._enabled = True self._state = self.STATE_ON return True def disable(self) -> bool: + if _USE_LINUX: + return LinuxBluetooth.disable() self._enabled = False self._state = self.STATE_OFF return True @@ -668,13 +681,36 @@ class BluetoothAdapter(Stub): return True def getBondedDevices(self) -> Set[BluetoothDevice]: + if _USE_LINUX: + devices = set() + for d in LinuxBluetooth.get_bonded_devices(): + dev = BluetoothDevice(d.get("address", "00:00:00:00:00:00")) + dev._name = d.get("name", "Unknown") + dev._bond_state = BluetoothDevice.BOND_BONDED + devices.add(dev) + return devices return set(self._bonded_devices) + def getBondedDeviceAddresses(self) -> Set[str]: + """获取已配对设备地址集合""" + if _USE_LINUX: + return {d.get("address", "") for d in LinuxBluetooth.get_bonded_devices()} + return {d.getAddress() for d in self._bonded_devices} + def startDiscovery(self) -> bool: + if _USE_LINUX: + try: + LinuxBluetooth.scan(timeout=5) + return True + except Exception: + return False self._discovering = True return True def cancelDiscovery(self) -> bool: + if _USE_LINUX: + # bluetoothctl scan off 在 LinuxBluetooth.scan 内部已处理 + return True self._discovering = False return True @@ -694,6 +730,8 @@ class BluetoothAdapter(Stub): pass def getState(self) -> int: + if _USE_LINUX: + return LinuxBluetooth.get_state() return self._state def listenUsingRfcommWithServiceRecord(self, name: str, uuid_str: str) -> BluetoothServerSocket: @@ -704,6 +742,18 @@ class BluetoothAdapter(Stub): return BluetoothServerSocket() def getRemoteDevice(self, address: str) -> Optional[BluetoothDevice]: + if _USE_LINUX: + # 从已配对设备中查找 + for d in LinuxBluetooth.get_bonded_devices(): + if d.get("address", "").upper() == address.upper(): + dev = BluetoothDevice(d.get("address", address)) + dev._name = d.get("name", "Unknown") + dev._bond_state = BluetoothDevice.BOND_BONDED + return dev + # 未配对但地址合法则返回一个设备对象 + if BluetoothAdapter.checkBluetoothAddress(address): + return BluetoothDevice(address) + return None if not BluetoothAdapter.checkBluetoothAddress(address): return None return BluetoothDevice(address) diff --git a/android_pyapi/android/content.py b/android_pyapi/android/content.py index 93281a5..f455ed4 100644 --- a/android_pyapi/android/content.py +++ b/android_pyapi/android/content.py @@ -16,6 +16,12 @@ from .._core import ( _sdcard_root, _app_data_dir, _package_name, ) +try: + from .._linux_impl import IS_LINUX, LinuxClipboard, LinuxNotify, LinuxLocation, LinuxProcess + _USE_LINUX = IS_LINUX +except ImportError: + _USE_LINUX = False + # ═══════════════════════════════════════════════════════════════ # Uri @@ -466,6 +472,7 @@ class Context(Stub): INPUT_METHOD_SERVICE = "input_method" CAMERA_SERVICE = "camera" BATTERY_SERVICE = "battery" + CLIPBOARD_SERVICE = "clipboard" # Mode 常量 MODE_PRIVATE = 0x0000 @@ -642,6 +649,8 @@ class Context(Stub): if name == Context.BATTERY_SERVICE: from .os import BatteryManager return BatteryManager() + if name == Context.CLIPBOARD_SERVICE: + return ClipboardManager() return self._services.get(name) def startService(self, intent: Any) -> Optional[Any]: @@ -1708,6 +1717,29 @@ class SyncResult(Stub): return f"SyncResult({self.stats.numEntries} entries)" +# ═══════════════════════════════════════════════════════════════ +# ClipboardManager +# ═══════════════════════════════════════════════════════════════ + +class ClipboardManager(Stub): + """ClipboardManager — 剪贴板管理器""" + + def getText(self): + if _USE_LINUX: + return LinuxClipboard.read() + return "" + + def setText(self, text): + if _USE_LINUX: + return LinuxClipboard.write(text) + return False + + def hasText(self): + if _USE_LINUX: + return len(LinuxClipboard.read()) > 0 + return False + + # ═══════════════════════════════════════════════════════════════ # 替换 _core.py 中的 _GlobalContext # ═══════════════════════════════════════════════════════════════ @@ -1748,4 +1780,5 @@ __all__ = [ "DialogInterface", "SyncAdapter", "SyncResult", + "ClipboardManager", ] \ No newline at end of file diff --git a/android_pyapi/android/hardware.py b/android_pyapi/android/hardware.py index b855a1a..d1f2e24 100644 --- a/android_pyapi/android/hardware.py +++ b/android_pyapi/android/hardware.py @@ -10,6 +10,13 @@ from typing import Any, Callable, Dict, List, Optional, Tuple from .._core import Stub, AndroidOnlyError, android_context, HAS_PIL +# Linux 真实实现导入 +try: + from .._linux_impl import IS_LINUX, LinuxBluetooth, LinuxSensors, HAS_BLUETOOTHCTL + _USE_LINUX = IS_LINUX +except ImportError: + _USE_LINUX = False + # ═══════════════════════════════════════════════════════════════ # Sensor @@ -251,14 +258,89 @@ class SensorManager(Stub): self._sensors[type_] = Sensor(type_) def getDefaultSensor(self, type_: int) -> Optional[Sensor]: + if _USE_LINUX: + sensor = self._build_linux_sensor(type_) + if sensor is not None: + return sensor return self._sensors.get(type_) + def _build_linux_sensor(self, type_: int) -> Optional[Sensor]: + """根据 Linux 真实传感器数据构造 Sensor 对象""" + try: + if type_ == Sensor.TYPE_AMBIENT_TEMPERATURE: + temps = LinuxSensors.get_temperatures() + if temps: + t = temps[0] + return Sensor(Sensor.TYPE_AMBIENT_TEMPERATURE, + name=t.get("name", "Linux Temperature"), + vendor="Linux", resolution=0.01, + max_range=125.0, power=0.0, min_delay=1000) + return None + if type_ == Sensor.TYPE_LIGHT: + light = LinuxSensors.get_light_level() + if light is not None: + return Sensor(Sensor.TYPE_LIGHT, name="Linux Light Sensor", + vendor="Linux", resolution=0.001, + max_range=40000.0, power=0.0, min_delay=1000) + return None + except Exception: + return None + return None + def getSensorList(self, type_: int) -> List[Sensor]: + if _USE_LINUX: + return self._get_linux_sensor_list(type_) if type_ == -1: # ALL return list(self._sensors.values()) sensor = self._sensors.get(type_) return [sensor] if sensor else [] + def _get_linux_sensor_list(self, type_: int) -> List[Sensor]: + """从 Linux 真实传感器数据生成传感器列表""" + sensors: List[Sensor] = [] + try: + if type_ == -1 or type_ == Sensor.TYPE_AMBIENT_TEMPERATURE: + for t in LinuxSensors.get_temperatures(): + sensors.append(Sensor( + Sensor.TYPE_AMBIENT_TEMPERATURE, + name=t.get("name", "Linux Temperature"), + vendor="Linux", resolution=0.01, + max_range=125.0, power=0.0, min_delay=1000)) + if type_ == -1 or type_ == Sensor.TYPE_LIGHT: + light = LinuxSensors.get_light_level() + if light is not None: + sensors.append(Sensor( + Sensor.TYPE_LIGHT, name="Linux Light Sensor", + vendor="Linux", resolution=0.001, + max_range=40000.0, power=0.0, min_delay=1000)) + # CPU 频率作为附加传感器 (映射到压力传感器类型以承载真实数据) + if type_ == -1 or type_ == Sensor.TYPE_PRESSURE: + for f in LinuxSensors.get_cpu_frequency(): + sensors.append(Sensor( + Sensor.TYPE_PRESSURE, + name=f"CPU{f.get('cpu', 0)} Frequency", + vendor="Linux", resolution=1.0, + max_range=10000.0, power=0.0, min_delay=1000)) + # 风扇转速作为附加传感器 (映射到相对湿度类型以承载真实数据) + if type_ == -1 or type_ == Sensor.TYPE_RELATIVE_HUMIDITY: + for fan in LinuxSensors.get_fan_speeds(): + sensors.append(Sensor( + Sensor.TYPE_RELATIVE_HUMIDITY, + name=f"Fan {fan.get('name', '')}", + vendor="Linux", resolution=1.0, + max_range=10000.0, power=0.0, min_delay=1000)) + except Exception: + pass + # 若未取到真实数据则回退到模拟传感器 + if not sensors: + if type_ == -1: + sensors = list(self._sensors.values()) + else: + sensor = self._sensors.get(type_) + if sensor: + sensors = [sensor] + return sensors + def getSensors(self) -> List[Sensor]: return list(self._sensors.values()) diff --git a/android_pyapi/android/location.py b/android_pyapi/android/location.py index 5ad2073..95af440 100644 --- a/android_pyapi/android/location.py +++ b/android_pyapi/android/location.py @@ -16,6 +16,12 @@ from .._core import ( _sdcard_root, _app_data_dir, _package_name, ) +try: + from .._linux_impl import IS_LINUX, LinuxLocation + _USE_LINUX = IS_LINUX +except ImportError: + _USE_LINUX = False + # ═══════════════════════════════════════════════════════════════ # Location @@ -952,6 +958,15 @@ class LocationManager(Stub): self.PASSIVE_PROVIDER, self.FUSED_PROVIDER} def getLastKnownLocation(self, provider: str) -> Optional[Location]: + if _USE_LINUX: + data = LinuxLocation.get_location() + loc = Location(provider or "network") + loc.setLatitude(data.get("latitude", 0.0)) + loc.setLongitude(data.get("longitude", 0.0)) + loc.setAccuracy(data.get("accuracy", 0)) + loc.setTime(int(datetime.now().timestamp() * 1000)) + loc.setProvider(provider or "network") + return loc if provider in self._last_known_locations: return self._last_known_locations[provider] loc = Location(provider) @@ -969,6 +984,18 @@ class LocationManager(Stub): provider = provider_or_request else: provider = self.GPS_PROVIDER + if _USE_LINUX and listener_or_callback is not None: + data = LinuxLocation.get_location() + loc = Location(provider or "network") + loc.setLatitude(data.get("latitude", 0.0)) + loc.setLongitude(data.get("longitude", 0.0)) + loc.setAccuracy(data.get("accuracy", 0)) + loc.setTime(int(datetime.now().timestamp() * 1000)) + loc.setProvider(provider or "network") + try: + listener_or_callback.onLocationChanged(loc) + except Exception: + pass if listener_or_callback is not None: if provider not in self._listeners: self._listeners[provider] = [] @@ -984,9 +1011,13 @@ class LocationManager(Stub): del self._listeners[provider] def isProviderEnabled(self, provider: str) -> bool: + if _USE_LINUX: + return provider == "network" return provider in self._enabled_providers def getProviders(self, criteria: Optional[Criteria] = None, enabled_only: bool = False) -> List[str]: + if _USE_LINUX: + return ["network"] providers = list(self._providers.keys()) if enabled_only: providers = [p for p in providers if p in self._enabled_providers] diff --git a/android_pyapi/android/managers.py b/android_pyapi/android/managers.py index cef34aa..624f9a2 100644 --- a/android_pyapi/android/managers.py +++ b/android_pyapi/android/managers.py @@ -20,6 +20,13 @@ from .._core import Stub, AndroidOnlyError, _app_data_dir, _sdcard_root from ..java.lang import RuntimeException +# Linux 真实实现导入 +try: + from .._linux_impl import IS_LINUX, LinuxNetwork, HAS_IP + _USE_LINUX = IS_LINUX +except ImportError: + _USE_LINUX = False + IllegalArgumentException = type("IllegalArgumentException", (RuntimeException,), {}) @@ -517,14 +524,58 @@ class ConnectivityManager(Stub): self._active_network: Optional[Any] = None self._all_networks: List[Any] = [] + def _get_linux_active_info(self): + """从 Linux 真实网络信息构造活跃 NetworkInfo""" + try: + from .net import NetworkInfo + interfaces = LinuxNetwork.get_interfaces() + # 有真实网络接口 (非环回) 且 operstate=up 即视为已连接 + connected = any( + (iface.get("ifname") or iface.get("name", "")) != "lo" + and iface.get("operstate", "").lower() == "up" + for iface in interfaces + ) + if not connected: + connected = LinuxNetwork.get_connectivity() + ip = LinuxNetwork.get_ip_address() + iface_name = "" + net_type = ConnectivityManager.TYPE_ETHERNET + for iface in interfaces: + name = iface.get("ifname") or iface.get("name", "") + state = iface.get("operstate", "") + if name and name != "lo" and state.lower() == "up": + iface_name = iface_name or name + if name and name.startswith(("wl", "wlan", "wlp")): + net_type = ConnectivityManager.TYPE_WIFI + iface_name = name + data = { + "connected": connected, + "ip": ip, + "interface": iface_name, + "type_name": "WIFI" if net_type == ConnectivityManager.TYPE_WIFI else "ETHERNET", + } + return NetworkInfo.from_linux_data(data, network_type=net_type) + except Exception: + return self._active_network_info + def getActiveNetworkInfo(self) -> Optional[Any]: + if _USE_LINUX: + return self._get_linux_active_info() return self._active_network_info def getNetworkInfo(self, networkType: int) -> Optional[Any]: + if _USE_LINUX: + info = self._get_linux_active_info() + if info is not None and info.getType() == networkType: + return info + return None return None - def getNetworkInfo(self, network: Any) -> Optional[Any]: - return None + def getConnectivityStatus(self) -> bool: + """返回当前网络连通性 (True=已连接)""" + if _USE_LINUX: + return LinuxNetwork.get_connectivity() + return False def getAllNetworkInfo(self) -> List[Any]: return self._all_networks diff --git a/android_pyapi/android/media.py b/android_pyapi/android/media.py index 102982f..54bff4b 100644 --- a/android_pyapi/android/media.py +++ b/android_pyapi/android/media.py @@ -13,6 +13,13 @@ from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union from .._core import Stub, AndroidOnlyError, android_context, HAS_PIL, HAS_PYMUPDF, resolve_android_path from .graphics import PointF +# Linux 真实实现导入 +try: + from .._linux_impl import IS_LINUX, LinuxAudio, HAS_AMIXER + _USE_LINUX = IS_LINUX +except ImportError: + _USE_LINUX = False + # ═══════════════════════════════════════════════════════════════ # AudioAttributes @@ -596,15 +603,37 @@ class AudioManager(Stub): self._parameters: Dict[str, str] = {} def getStreamVolume(self, streamType: int) -> int: + if _USE_LINUX: + # STREAM_MUSIC 及其它流类型映射到 Linux 主音量 + return LinuxAudio.get_volume() return self._stream_volumes.get(streamType, 0) def getStreamMaxVolume(self, streamType: int) -> int: + if _USE_LINUX: + return 100 return 15 def setStreamVolume(self, streamType: int, index: int, flags: int = 0): + if _USE_LINUX: + # STREAM_MUSIC 及其它流类型映射到 Linux 主音量 + LinuxAudio.set_volume(index) + return if streamType in self._stream_volumes: self._stream_volumes[streamType] = max(0, min(15, index)) + def isMicrophoneMute(self) -> bool: + """麦克风是否静音""" + if _USE_LINUX: + return LinuxAudio.get_mute() + return False + + def setMicrophoneMute(self, on: bool): + """设置麦克风静音""" + if _USE_LINUX: + LinuxAudio.set_mute(on) + return + # 存根: 无操作 + def adjustVolume(self, direction: int, flags: int = 0): pass diff --git a/android_pyapi/android/net.py b/android_pyapi/android/net.py index 3a6d6dc..0dd1397 100644 --- a/android_pyapi/android/net.py +++ b/android_pyapi/android/net.py @@ -17,6 +17,13 @@ from .._core import ( _sdcard_root, _app_data_dir, _package_name, ) +# Linux 真实实现导入 +try: + from .._linux_impl import IS_LINUX, LinuxNetwork, HAS_IP + _USE_LINUX = IS_LINUX +except ImportError: + _USE_LINUX = False + # ═══════════════════════════════════════════════════════════════ # Uri (re-export from content) @@ -379,6 +386,38 @@ class NetworkInfo(Stub): } return names.get(type_, "UNKNOWN") + @classmethod + def from_linux_data(cls, data: Optional[dict] = None, + network_type: int = 9) -> "NetworkInfo": + """从 Linux 真实网络数据构造 NetworkInfo + + data 字典可包含: ip, interface, connected, type_name 等 + """ + if data is None: + data = {} + connected = bool(data.get("connected", False)) + state = "CONNECTED" if connected else "DISCONNECTED" + detailed_state = "CONNECTED" if connected else "DISCONNECTED" + type_name = data.get("type_name") or cls._default_type_name_static(network_type) + extra_info = data.get("ip") or data.get("interface") + return cls( + type_=network_type, + state=state, + detailed_state=detailed_state, + type_name=type_name, + extra_info=extra_info, + is_available=True, + roaming=False, + ) + + @staticmethod + def _default_type_name_static(type_: int) -> str: + names = { + 0: "MOBILE", 1: "WIFI", 6: "WIMAX", 7: "BLUETOOTH", + 9: "ETHERNET", 17: "VPN", + } + return names.get(type_, "UNKNOWN") + def getType(self) -> int: return self._type @@ -1230,15 +1269,71 @@ class ConnectivityManager(Stub): self._network_caps: Dict[int, NetworkCapabilities] = {} self._link_props: Dict[int, LinkProperties] = {} + def _get_linux_active_info(self) -> Optional[NetworkInfo]: + """从 Linux 真实网络信息构造活跃 NetworkInfo""" + try: + interfaces = LinuxNetwork.get_interfaces() + # 有真实网络接口 (非环回) 且 operstate=up 即视为已连接 + connected = any( + (iface.get("ifname") or iface.get("name", "")) != "lo" + and iface.get("operstate", "").lower() == "up" + for iface in interfaces + ) + # 若有活跃接口但 get_connectivity 为 False (外网不可达), 仍视为已连接 + if not connected: + connected = LinuxNetwork.get_connectivity() + ip = LinuxNetwork.get_ip_address() + iface_name = "" + for iface in interfaces: + name = iface.get("ifname") or iface.get("name", "") + state = iface.get("operstate", "") + if name and name != "lo" and state.lower() == "up": + iface_name = name + break + # 判断网络类型 (无线接口 -> WIFI, 否则 ETHERNET) + net_type = self.TYPE_ETHERNET + for iface in interfaces: + name = iface.get("ifname") or iface.get("name", "") + if name and name.startswith(("wl", "wlan", "wlp")): + net_type = self.TYPE_WIFI + iface_name = iface_name or name + break + data = { + "connected": connected, + "ip": ip, + "interface": iface_name, + "type_name": "WIFI" if net_type == self.TYPE_WIFI else "ETHERNET", + } + return NetworkInfo.from_linux_data(data, network_type=net_type) + except Exception: + return self._active_network_info + def getActiveNetworkInfo(self) -> Optional[NetworkInfo]: + if _USE_LINUX: + return self._get_linux_active_info() return self._active_network_info def getNetworkInfo(self, network_type: int) -> Optional[NetworkInfo]: + if _USE_LINUX: + info = self._get_linux_active_info() + if info is not None and info.getType() == network_type: + return info + # 请求的类型与活跃网络不同时, 返回断开状态 + return NetworkInfo( + type_=network_type, state="DISCONNECTED", + detailed_state="DISCONNECTED", is_available=False, + ) network = self._networks.get(network_type) if network: return NetworkInfo(type_=network_type, state="CONNECTED", is_available=True) return self._active_network_info + def getConnectivityStatus(self) -> bool: + """返回当前网络连通性 (True=已连接)""" + if _USE_LINUX: + return LinuxNetwork.get_connectivity() + return True + def getNetworkInfoForUid(self, network: Network, uid: int) -> Optional[NetworkInfo]: return self._active_network_info diff --git a/android_pyapi/android/net_ext.py b/android_pyapi/android/net_ext.py index acaa2e4..1a2bd40 100644 --- a/android_pyapi/android/net_ext.py +++ b/android_pyapi/android/net_ext.py @@ -15,6 +15,13 @@ from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union from .._core import Stub, AndroidOnlyError +# Linux 真实实现导入 +try: + from .._linux_impl import IS_LINUX, LinuxNetwork, HAS_IP + _USE_LINUX = IS_LINUX +except ImportError: + _USE_LINUX = False + __all__ = [ "Network", "NetworkCapabilities", "NetworkRequest", "NetworkSpecifier", "MatchAllNetworkSpecifier", "NetworkAgent", "NetworkInfo", @@ -1291,17 +1298,64 @@ class ConnectivityManager(Stub): def getActiveNetwork(self) -> Optional[Network]: return self._active_network + def _get_linux_active_info(self): + """从 Linux 真实网络信息构造活跃 NetworkInfo""" + try: + from .net import NetworkInfo + interfaces = LinuxNetwork.get_interfaces() + # 有真实网络接口 (非环回) 且 operstate=up 即视为已连接 + connected = any( + (iface.get("ifname") or iface.get("name", "")) != "lo" + and iface.get("operstate", "").lower() == "up" + for iface in interfaces + ) + if not connected: + connected = LinuxNetwork.get_connectivity() + ip = LinuxNetwork.get_ip_address() + iface_name = "" + net_type = ConnectivityManager.TYPE_ETHERNET + for iface in interfaces: + name = iface.get("ifname") or iface.get("name", "") + state = iface.get("operstate", "") + if name and name != "lo" and state.lower() == "up": + iface_name = iface_name or name + if name and name.startswith(("wl", "wlan", "wlp")): + net_type = ConnectivityManager.TYPE_WIFI + iface_name = name + data = { + "connected": connected, + "ip": ip, + "interface": iface_name, + "type_name": "WIFI" if net_type == ConnectivityManager.TYPE_WIFI else "ETHERNET", + } + return NetworkInfo.from_linux_data(data, network_type=net_type) + except Exception: + return None + def getActiveNetworkInfo(self) -> Optional[NetworkInfo]: + if _USE_LINUX: + return self._get_linux_active_info() if self._active_network: return self._network_infos.get(self._active_network.getNetId()) return None def getNetworkInfo(self, networkType: int) -> Optional[NetworkInfo]: + if _USE_LINUX: + info = self._get_linux_active_info() + if info is not None and info.getType() == networkType: + return info + return None for info in self._network_infos.values(): if info.getType() == networkType: return info return None + def getConnectivityStatus(self) -> bool: + """返回当前网络连通性 (True=已连接)""" + if _USE_LINUX: + return LinuxNetwork.get_connectivity() + return self._active_network is not None + def getNetworkInfoForNetwork(self, network: Network) -> Optional[NetworkInfo]: return self._network_infos.get(network.getNetId()) diff --git a/android_pyapi/android/net_wifi.py b/android_pyapi/android/net_wifi.py index 9371da6..054da66 100644 --- a/android_pyapi/android/net_wifi.py +++ b/android_pyapi/android/net_wifi.py @@ -16,6 +16,13 @@ from .._core import ( _sdcard_root, _app_data_dir, _package_name, ) +# Linux 真实实现导入 +try: + from .._linux_impl import IS_LINUX, LinuxNetwork, HAS_IP + _USE_LINUX = IS_LINUX +except ImportError: + _USE_LINUX = False + # ═══════════════════════════════════════════════════════════════ # SupplicantState @@ -620,6 +627,8 @@ class WifiManager(Stub): self._p2p_channels: List[int] = [] def getConnectionInfo(self) -> Optional[WifiInfo]: + if _USE_LINUX: + return self._get_linux_connection_info() if self._connection_info is None: info = WifiInfo() info._ssid = '"mock_wifi"' @@ -632,6 +641,34 @@ class WifiManager(Stub): self._connection_info = info return self._connection_info + def getWifiInfo(self) -> Optional[WifiInfo]: + """getWifiInfo — getConnectionInfo 的别名""" + return self.getConnectionInfo() + + def _get_linux_connection_info(self) -> WifiInfo: + """从 LinuxNetwork.get_wifi_info() 构造 WifiInfo""" + info = WifiInfo() + try: + data = LinuxNetwork.get_wifi_info() or {} + ssid = data.get("ssid", "") + info._ssid = f'"{ssid}"' if ssid else "" + info._bssid = data.get("bssid", "00:00:00:00:00:00") + # signal 通常是 0-100, 转换为 RSSI (dBm, 约 -100..0) + signal = data.get("signal", 0) + if isinstance(signal, (int, float)) and signal > 0: + info._rssi = int(-100 + (signal / 100.0) * 60) + else: + info._rssi = int(signal) if signal else WifiInfo.INVALID_RSSI + iface = data.get("interface", "") + if iface: + info._frequency = 2412 # 默认 2.4GHz + info._link_speed = data.get("link_speed", 0) or 0 + info._network_id = 0 + info._supplicant_state = SupplicantState(SupplicantState.COMPLETED) + except Exception: + pass + return info + def getScanResults(self) -> List[ScanResult]: return list(self._scan_results) @@ -644,8 +681,26 @@ class WifiManager(Stub): return True def isWifiEnabled(self) -> bool: + if _USE_LINUX: + return self._linux_has_wireless_interface() return self._wifi_enabled + def _linux_has_wireless_interface(self) -> bool: + """检查是否存在无线网络接口""" + try: + for iface in LinuxNetwork.get_interfaces(): + name = iface.get("ifname") or iface.get("name", "") + if name and name.startswith(("wl", "wlan", "wlp")): + return True + # 回退: 检查 /sys/class/net 下的无线接口 + import os as _os + for name in _os.listdir("/sys/class/net/") if _os.path.isdir("/sys/class/net") else []: + if name.startswith(("wl", "wlan", "wlp")): + return True + except Exception: + pass + return False + def getWifiState(self) -> int: return self._wifi_state diff --git a/android_pyapi/android/os.py b/android_pyapi/android/os.py index ad67913..4f1a5fb 100644 --- a/android_pyapi/android/os.py +++ b/android_pyapi/android/os.py @@ -20,6 +20,13 @@ from .._core import ( 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 @@ -118,6 +125,50 @@ class Build(Stub): 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 @@ -1299,6 +1350,20 @@ class PowerManager(Stub): 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: @@ -1478,8 +1543,58 @@ class BatteryManager(Stub): 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: @@ -1498,9 +1613,57 @@ class BatteryManager(Stub): 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: diff --git a/android_pyapi/android/os_power.py b/android_pyapi/android/os_power.py index 22f913b..761596e 100644 --- a/android_pyapi/android/os_power.py +++ b/android_pyapi/android/os_power.py @@ -18,6 +18,13 @@ from .._core import Stub, AndroidOnlyError, _app_data_dir from ..java.lang import RuntimeException +# Linux 真实实现导入 +try: + from .._linux_impl import IS_LINUX, LinuxPower, LinuxProcess, HAS_PSUTIL as _HAS_PSUTIL + _USE_LINUX = IS_LINUX +except ImportError: + _USE_LINUX = False + IllegalArgumentException = type("IllegalArgumentException", (RuntimeException,), {}) @@ -505,6 +512,54 @@ class BatteryManager(Stub): self._discharge_time_remaining = -1 self._is_capacity_estimated = False self._capacity_percentage = 100.0 + if _USE_LINUX: + self._refresh_from_linux() + + _LINUX_STATUS_MAP = { + "charging": BATTERY_STATUS_CHARGING, + "discharging": BATTERY_STATUS_DISCHARGING, + "not charging": BATTERY_STATUS_NOT_CHARGING, + "full": BATTERY_STATUS_FULL, + "unknown": 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._battery_level = int(level) + self._capacity_percentage = float(level) + # BATTERY_PROPERTY_CAPACITY 在 Android 中返回百分比 (0-100) + self._capacity = int(level) + else: + # 无电池或电量未知 + self._battery_level = -1 + self._capacity_percentage = 0.0 + self._capacity = -1 + status_str = str(info.get("status", "unknown")).lower() + self._status = self._LINUX_STATUS_MAP.get(status_str, BATTERY_STATUS_UNKNOWN) + voltage = info.get("voltage", 0) + if isinstance(voltage, (int, float)): + self._voltage = int(voltage * 1000) # V -> mV + temperature = info.get("temperature", 0) + if isinstance(temperature, (int, float)): + self._temperature = temperature # 摄氏度 + self._technology = info.get("technology", "") or "Li-ion" + if self._status == BATTERY_STATUS_CHARGING: + self._plugged = BATTERY_PLUGGED_AC + else: + self._plugged = 0 @staticmethod def getInstance(context: Any) -> "BatteryManager": @@ -513,6 +568,8 @@ class BatteryManager(Stub): return BatteryManager._instance def getIntProperty(self, id: int) -> int: + if _USE_LINUX: + self._refresh_from_linux() properties = { BATTERY_PROPERTY_CAPACITY: self._capacity, BATTERY_PROPERTY_CHARGE_COUNTER: self._charge_counter, @@ -521,6 +578,9 @@ class BatteryManager(Stub): BATTERY_PROPERTY_ENERGY_COUNTER: self._energy_counter, BATTERY_PROPERTY_CHARGE_TIME_REMAINING: self._charge_time_remaining, BATTERY_PROPERTY_DISCHARGE_TIME_REMAINING: self._discharge_time_remaining, + BATTERY_PROPERTY_STATUS: self._status, + BATTERY_PROPERTY_TEMPERATURE: int(self._temperature), + BATTERY_PROPERTY_VOLTAGE: self._voltage, } return properties.get(id, 0) @@ -528,9 +588,13 @@ class BatteryManager(Stub): return self.getIntProperty(id) def isCharging(self) -> bool: + if _USE_LINUX: + self._refresh_from_linux() return self._status == BATTERY_STATUS_CHARGING or self._status == BATTERY_STATUS_FULL def getBatteryLevel(self) -> int: + if _USE_LINUX: + self._refresh_from_linux() return self._battery_level def getInt(intent: Any, key: str, defaultValue: int) -> int: @@ -687,9 +751,11 @@ class PowerManager(Stub): self._shutdown_requested = False self._thermal_status = THERMAL_STATUS_NONE self._power_save_state = PowerSaveState() - self._current_wake_up_reason = WAKE_REASON_UNKNOWN + self._current_wake_up_reason = PowerManager.WAKE_REASON_UNKNOWN def isInteractive(self) -> bool: + if _USE_LINUX: + return self.isScreenOn() return self._is_interactive def isPowerSaveMode(self) -> bool: @@ -705,6 +771,13 @@ class PowerManager(Stub): return True def isScreenOn(self) -> bool: + if _USE_LINUX: + import os as _os_mod + display = _os_mod.environ.get("DISPLAY", "") + if display: + return True + wayland = _os_mod.environ.get("WAYLAND_DISPLAY", "") + return bool(wayland) return self._is_screen_on def isWakeLockLevelSupported(self, level: int) -> bool: diff --git a/android_pyapi/android/telephony.py b/android_pyapi/android/telephony.py index 1ab18ce..184cc77 100644 --- a/android_pyapi/android/telephony.py +++ b/android_pyapi/android/telephony.py @@ -14,6 +14,12 @@ from .._core import ( _sdcard_root, _app_data_dir, _package_name, ) +try: + from .._linux_impl import IS_LINUX, LinuxNetwork, LinuxProcess, LinuxLocation + _USE_LINUX = IS_LINUX +except ImportError: + _USE_LINUX = False + # ═══════════════════════════════════════════════════════════════ # CellIdentity 基类集合 @@ -553,6 +559,9 @@ class TelephonyManager(Stub): self._context = context def getNetworkOperatorName(self) -> str: + if _USE_LINUX: + loc = LinuxLocation.get_location() + return loc.get("isp", "") or "" return "MockOperator" def getNetworkOperator(self) -> str: @@ -562,9 +571,15 @@ class TelephonyManager(Stub): return "us" def getNetworkType(self) -> int: + if _USE_LINUX: + if LinuxNetwork.get_connectivity(): + return self.NETWORK_TYPE_IWLAN + return self.NETWORK_TYPE_UNKNOWN return self.NETWORK_TYPE_LTE def getPhoneType(self) -> int: + if _USE_LINUX: + return self.PHONE_TYPE_NONE return self.PHONE_TYPE_GSM def getDataNetworkType(self) -> int: @@ -574,6 +589,8 @@ class TelephonyManager(Stub): return self.NETWORK_TYPE_LTE def getSimOperator(self) -> str: + if _USE_LINUX: + return "" return "310260" def getSimOperatorName(self) -> str: @@ -583,18 +600,29 @@ class TelephonyManager(Stub): return "us" def getSimSerialNumber(self) -> Optional[str]: + if _USE_LINUX: + return "" return "89014103211118510720" def getSubscriberId(self) -> Optional[str]: + if _USE_LINUX: + return "" return "310260123456789" def getDeviceId(self) -> Optional[str]: + if _USE_LINUX: + return LinuxNetwork.get_mac_address() or None return "000000000000000" def getDeviceSoftwareVersion(self) -> Optional[str]: + if _USE_LINUX: + info = LinuxProcess.get_system_info() + return info.get("kernel", "") return "00.00.00.00" def getLine1Number(self) -> Optional[str]: + if _USE_LINUX: + return "" return "15551234567" def getGroupIdLevel1(self) -> Optional[str]: @@ -619,6 +647,9 @@ class TelephonyManager(Stub): return self.SIM_STATE_READY def getCellLocation(self) -> Any: + # 桌面无基站，Linux 下返回 None + if _USE_LINUX: + return None return None def getAllCellInfo(self) -> Optional[List[CellInfo]]: diff --git a/android_pyapi/android/view.py b/android_pyapi/android/view.py index 2e127cc..797b504 100644 --- a/android_pyapi/android/view.py +++ b/android_pyapi/android/view.py @@ -10,6 +10,12 @@ from typing import Any, Callable, Dict, List, Optional, Tuple, Union from .._core import Stub, AndroidOnlyError, android_context, HAS_PYAUTOGUI +try: + from .._linux_impl import IS_LINUX, LinuxDisplay, LinuxUIAutomation, HAS_XDOTOOL + _USE_LINUX = IS_LINUX +except ImportError: + _USE_LINUX = False + # ═══════════════════════════════════════════════════════════════ # Gravity @@ -363,6 +369,12 @@ class View(Stub): def performClick(self) -> bool: return self.callOnClick() + def screenshot(self, path=None): + """截屏""" + if _USE_LINUX: + return LinuxDisplay.screenshot(path) + return None + def performLongClick(self) -> bool: if self._on_long_click_listener: self._on_long_click_listener(self) @@ -3260,9 +3272,13 @@ class Display(Stub): return self._display_id def getWidth(self) -> int: + if _USE_LINUX: + return LinuxDisplay.get_screen_size()[0] return self._width def getHeight(self) -> int: + if _USE_LINUX: + return LinuxDisplay.get_screen_size()[1] return self._height def getRotation(self) -> int: diff --git a/android_pyapi/android/widget.py b/android_pyapi/android/widget.py index 98c2544..f854a87 100644 --- a/android_pyapi/android/widget.py +++ b/android_pyapi/android/widget.py @@ -14,6 +14,12 @@ from .view import ( ViewGroup as VG, ) +try: + from .._linux_impl import IS_LINUX, LinuxNotify, LinuxClipboard + _USE_LINUX = IS_LINUX +except ImportError: + _USE_LINUX = False + # ═══════════════════════════════════════════════════════════════ # TextView @@ -3589,9 +3595,15 @@ class Toast(Stub): return toast def show(self): - pass + if _USE_LINUX: + LinuxNotify.send("Toast", self._text) + else: + print(f"[Toast] {self._text}") def cancel(self): + if _USE_LINUX: + # 桌面通知自动消失，无需操作 + return pass def setGravity(self, gravity: int, x_offset: int, y_offset: int):