feat: 深度完善 — 存根转真实实现、UI类方法补全、Java集合/IO增强

存根转真实实现:
- ContentResolver: query/insert/update/delete对file://URI真实文件操作, getType用mimetypes
- Context: getResources/getAssets/getString/getColor 委托真实Resources
- Bitmap.sameAs: PIL像素比对
- Canvas: quickReject/drawTextOnPath/drawPicture/drawDoubleRoundRect 真实实现
- Paint: measureText/breakText/getTextWidths/getFontMetrics/getTextBounds 用PIL字体
- PathMeasure: 真实路径展平/长度计算/坐标切线提取
- Path.op/Region.op: 布尔运算
- Camera: 3D变换矩阵计算
- SystemClock.setCurrentTimeMillis: date -s命令
- Environment.isExternalStorageRemovable: /proc/mounts检查
- Process.myUid: os.getuid()
- PowerManager.isPowerSaveMode: CPU governor检查
- Vibrator: Linux LED振动器设备支持
- FileObserver: 轮询线程检测文件变化
- SensorManager.getRotationMatrix: 完整旋转矩阵算法
- BatteryManager: LinuxPower真实电池数据
- Geocoder: Nominatim API逆地理编码
- LocationManager: addProximityAlert/setTestProviderLocation

UI类方法补全:
- TextView: setPaddingRelative/getTextColor/setBreakStrategy/setJustificationMode
- ImageView: setColorFilter/getColorFilter/clearColorFilter
- CheckBox/Switch: setOnCheckedChangeListener
- ListView: clearChoices
- ViewGroup: setLayoutTransition/setTouchscreenBlocksFocus/getChildVisibleRect
- MotionEvent: getHistoricalX/Y/Pressure/EventTime/addBatch/getEdgeFlags/setAction

Java集合增强:
- ArrayList: spliterator/stream
- HashSet: spliterator/stream/removeIf
- TreeMap: replaceAll/compute/computeIfAbsent/computeIfPresent/merge
- LinkedHashMap: clone (保留access_order)
- PriorityQueue: spliterator/clone

Java IO增强:
- BufferedWriter: append/check
- PrintWriter: append
- ByteArrayOutputStream: writeTo
- ByteArrayInputStream: markSupported
- BufferedInputStream: available/mark/reset/markSupported + 修复_fill缓冲区填充bug
- SequenceInputStream: 补全实现(Enumeration/多流/跨流聚合读取)

修复:
- ViewGroup.addView: 合并重复定义为智能参数分派
- view_ext.ViewGroup: 继承view.ViewGroup
- view_ext.KeyEvent: 继承view.KeyEvent
- hardware_sensor_ext.SensorManager: 继承hardware.SensorManager
- app_fragment: 使用java.lang.IllegalStateException
- content_intent.Intent: setData/getScheme正确从Uri提取scheme
- __init__.py: 补全Fragment/KeyEvent/MotionEvent/TrafficStats/RecyclerView/TreeMap/Vibrator等导出

统计: 221模块, 3662类, 30952方法, 182302行
This commit is contained in:
shuiyu1123 committed 2026-06-25 00:17:50 +00:00
1 parent 39b1da37c9
commit 4235f8692c
13 files changed
+2248 -163

No files matched your search

+285 -27
View File
@@ -4,6 +4,7 @@ Android Content 组件 — Context, ContentResolver, Uri, SharedPreferences 等
"""
import json
import mimetypes
import os
import sys
import threading
@@ -372,34 +373,134 @@ class ContentResolver(Stub):
super().__init__()
self._observers: Dict[str, List[Any]] = {}
@staticmethod
def _uri_to_file_path(uri) -> Optional[str]:
"""将 file:// 或纯路径 Uri 转换为本地文件系统路径"""
try:
uri_str = str(uri)
except Exception:
return None
# 处理字符串输入
if isinstance(uri, str):
if uri.startswith("file://"):
return uri[7:]
if "://" not in uri:
return uri if uri else None
return None
scheme = getattr(uri, "_scheme", "") or ""
path = getattr(uri, "_path", "") or ""
if scheme == "file":
return path if path else None
if scheme == "" and path:
return path
return None
def query(self, uri: Uri, projection: Optional[List[str]], selection: Optional[str],
selection_args: Optional[List[str]], sort_order: Optional[str],
cancellation_signal: Any = None) -> Any:
"""Query content provider"""
sys.stderr.write(f"[ContentResolver] query: uri={uri}\n")
"""Query content provider. 对 file:// URI 返回基于目录列表的 Cursor。"""
path = self._uri_to_file_path(uri)
if path is not None and os.path.isdir(path):
try:
from .database import Cursor, MatrixCursor
cols = projection or ["_id", "name", "size", "last_modified"]
cursor = MatrixCursor(cols)
for idx, name in enumerate(sorted(os.listdir(path))):
full = os.path.join(path, name)
try:
st = os.stat(full)
size = st.st_size
mtime = int(st.st_mtime * 1000)
except OSError:
size = 0
mtime = 0
cursor.addRow([idx, name, size, mtime])
return cursor
except Exception:
pass
return None
def insert(self, uri: Uri, values: ContentValues) -> Optional[Uri]:
sys.stderr.write(f"[ContentResolver] insert: uri={uri}\n")
"""Insert. 对 file:// URI 将 ContentValues 写为 JSON 文件。"""
path = self._uri_to_file_path(uri)
if path is not None and isinstance(values, ContentValues):
try:
os.makedirs(path if os.path.isdir(path) else os.path.dirname(path) or ".", exist_ok=True)
target = path if not os.path.isdir(path) else os.path.join(path, "insert.json")
data = values._values if hasattr(values, "_values") else {}
with open(target, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, default=str)
return Uri.withAppendedPath(uri, os.path.basename(target))
except OSError:
pass
return uri
def update(self, uri: Uri, values: ContentValues, selection: Optional[str],
selection_args: Optional[List[str]]) -> int:
sys.stderr.write(f"[ContentResolver] update: uri={uri}\n")
"""Update. 对 file:// URI 更新对应 JSON 文件。"""
path = self._uri_to_file_path(uri)
if path is not None and isinstance(values, ContentValues) and os.path.isfile(path):
try:
data = values._values if hasattr(values, "_values") else {}
with open(path, "r", encoding="utf-8") as f:
existing = json.load(f)
if isinstance(existing, dict):
existing.update(data)
with open(path, "w", encoding="utf-8") as f:
json.dump(existing, f, ensure_ascii=False, default=str)
return 1
except (OSError, ValueError):
pass
return 0
def delete(self, uri: Uri, selection: Optional[str],
selection_args: Optional[List[str]]) -> int:
sys.stderr.write(f"[ContentResolver] delete: uri={uri}\n")
"""Delete. 对 file:// URI 删除对应文件或目录。"""
path = self._uri_to_file_path(uri)
if path is not None:
try:
if os.path.isfile(path):
os.remove(path)
return 1
if os.path.isdir(path):
import shutil
shutil.rmtree(path)
return 1
except OSError:
pass
return 0
def applyBatch(self, authority: str, operations: List[Any]) -> List[Any]:
sys.stderr.write(f"[ContentResolver] applyBatch: authority={authority}\n")
return []
"""Apply batch operations, returning per-op results."""
results: List[Any] = []
for op in operations:
try:
op_type = getattr(op, "_type", None)
uri = getattr(op, "_uri", None)
values = getattr(op, "_values", None)
if op_type == 1 and uri is not None: # INSERT
results.append(ContentProviderResult(uri=self.insert(uri, values or ContentValues())))
elif op_type == 2 and uri is not None: # UPDATE
count = self.update(uri, values or ContentValues(),
getattr(op, "_selection", None),
getattr(op, "_selection_args", None))
results.append(ContentProviderResult(count=count))
elif op_type == 3 and uri is not None: # DELETE
count = self.delete(uri, getattr(op, "_selection", None),
getattr(op, "_selection_args", None))
results.append(ContentProviderResult(count=count))
else:
results.append(ContentProviderResult(count=0))
except Exception:
results.append(ContentProviderResult(count=0))
return results
def bulkInsert(self, uri: Uri, values: List[ContentValues]) -> int:
sys.stderr.write(f"[ContentResolver] bulkInsert: uri={uri}\n")
return len(values) if values else 0
count = 0
for v in values or []:
if self.insert(uri, v) is not None:
count += 1
return count
def call(self, uri: Uri, method: str, arg: Optional[str], extras: Optional[ContentValues]) -> Optional[ContentValues]:
return None
@@ -426,7 +527,17 @@ class ContentResolver(Stub):
if not self._observers[uri_str]:
del self._observers[uri_str]
def getType(self, uri: Uri) -> Optional[str]:
def getType(self, uri) -> Optional[str]:
"""返回 URI 指向资源的 MIME 类型。"""
path = self._uri_to_file_path(uri)
if path:
mime, _ = mimetypes.guess_type(path)
return mime or "application/octet-stream"
scheme = getattr(uri, "_scheme", "") or ""
if isinstance(uri, str) and "://" in uri:
scheme = uri.split("://")[0]
if scheme == "content":
return "vnd.android.cursor.dir/vnd.android.content"
return None
def acquireContentProviderClient(self, uri: Uri) -> Optional["ContentProviderClient"]:
@@ -436,12 +547,40 @@ class ContentResolver(Stub):
return ContentProviderClient(uri)
def getStreamTypes(self, uri: Uri, mime_type_filter: str) -> Optional[List[str]]:
return None
"""返回 URI 支持的流 MIME 类型列表。"""
mime = self.getType(uri)
if mime is None:
return None
if mime_type_filter and mime_type_filter.endswith("/*"):
prefix = mime_type_filter[:-1]
if mime.startswith(prefix):
return [mime]
return None
if mime == mime_type_filter:
return [mime]
return [mime] if mime else None
def openInputStream(self, uri: Uri) -> Any:
"""打开输入流。对 file:// URI 返回真实文件流。"""
path = self._uri_to_file_path(uri)
if path and os.path.isfile(path):
try:
return open(path, "rb")
except OSError:
return None
return None
def openOutputStream(self, uri: Uri) -> Any:
"""打开输出流。对 file:// URI 返回真实文件流。"""
path = self._uri_to_file_path(uri)
if path:
try:
d = os.path.dirname(path)
if d:
os.makedirs(d, exist_ok=True)
return open(path, "wb")
except OSError:
return None
return None
@@ -594,10 +733,16 @@ class Context(Stub):
return PackageManager()
def getResources(self):
return None
from .content_res import Resources, AssetManager
if not hasattr(self, "_resources") or self._resources is None:
self._resources = Resources(self.getAssets())
return self._resources
def getAssets(self):
return None
from .content_res import AssetManager
if not hasattr(self, "_assets") or self._assets is None:
self._assets = AssetManager()
return self._assets
def getMainLooper(self):
from .os import Looper
@@ -668,15 +813,34 @@ class Context(Stub):
pass
def getString(self, res_id: int, *args) -> str:
"""返回资源字符串。优先委托给 Resources,否则返回合理默认值。"""
try:
res = self.getResources()
val = res.getString(res_id, *args)
if val is not None:
return val
except Exception:
pass
return str(res_id)
def getColor(self, color_id: int) -> int:
return 0
"""返回资源颜色。优先委托给 Resources,否则返回不透明黑色。"""
try:
res = self.getResources()
return res.getColor(color_id)
except Exception:
return 0xFF000000
def getDrawable(self, drawable_id: int):
return None
"""返回 Drawable。优先委托给 Resources。"""
try:
res = self.getResources()
return res.getDrawable(drawable_id)
except Exception:
return None
def getWallpaper(self):
"""返回壁纸 Drawable。桌面端返回 None(无系统壁纸)。"""
return None
def getWallpaperDesiredMinimumWidth(self) -> int:
@@ -795,18 +959,55 @@ class Context(Stub):
self._theme = resid
def getText(self, resId):
"""返回资源文本。优先委托给 Resources。"""
try:
res = self.getResources()
val = res.getText(resId)
if val is not None:
return val
except Exception:
pass
return str(resId)
def getQuantityString(self, resId, quantity, *formatArgs):
return str(resId)
"""返回复数字符串。优先委托给 Resources。"""
try:
res = self.getResources()
return res.getQuantityString(resId, quantity, *formatArgs)
except Exception:
return str(resId)
def getStringArray(self, resId):
"""返回字符串数组。优先委托给 Resources。"""
try:
res = self.getResources()
val = res.getStringArray(resId)
if val is not None:
return val
except Exception:
pass
return []
def getTextArray(self, resId):
"""返回文本数组。优先委托给 Resources。"""
try:
res = self.getResources()
val = res.getTextArray(resId)
if val is not None:
return val
except Exception:
pass
return []
def getIntArray(self, resId):
"""返回整型数组。优先委托给 Resources。"""
try:
res = self.getResources()
val = res.getIntArray(resId)
if val is not None:
return val
except Exception:
pass
return []
# ─── Context 派生 / 回调 (扩展) ───────────────────────────
@@ -1386,29 +1587,55 @@ class ContentProvider(Stub):
def query(self, uri: Uri, projection: Optional[List[str]], selection: Optional[str],
selection_args: Optional[List[str]], sort_order: Optional[str]) -> Optional[Any]:
"""默认实现:委托给关联 Context 的 ContentResolver。"""
if self._context is not None:
return self._context.getContentResolver().query(
uri, projection, selection, selection_args, sort_order)
return None
def insert(self, uri: Uri, values: ContentValues) -> Optional[Uri]:
if self._context is not None:
return self._context.getContentResolver().insert(uri, values)
return None
def update(self, uri: Uri, values: ContentValues, selection: Optional[str],
selection_args: Optional[List[str]]) -> int:
if self._context is not None:
return self._context.getContentResolver().update(uri, values, selection, selection_args)
return 0
def delete(self, uri: Uri, selection: Optional[str], selection_args: Optional[List[str]]) -> int:
if self._context is not None:
return self._context.getContentResolver().delete(uri, selection, selection_args)
return 0
def getType(self, uri: Uri) -> Optional[str]:
"""返回 URI 的 MIME 类型。"""
path = ContentResolver._uri_to_file_path(uri)
if path:
mime, _ = mimetypes.guess_type(path)
return mime or "application/octet-stream"
return None
def getStreamTypes(self, uri: Uri, mime_type_filter: str) -> Optional[List[str]]:
return None
mime = self.getType(uri)
if mime is None:
return None
return [mime]
def openFile(self, uri: Uri, mode: str) -> Any:
"""打开文件。对 file:// URI 返回真实文件对象。"""
path = ContentResolver._uri_to_file_path(uri)
if path and os.path.exists(path):
try:
fmode = "rb" if "r" in mode else "wb"
return open(path, fmode)
except OSError:
return None
return None
def openAssetFile(self, uri: Uri, mode: str) -> Any:
return None
return self.openFile(uri, mode)
def attachInfo(self, context: Context, info: Any):
self._context = context
@@ -1444,29 +1671,44 @@ class ContentProviderClient(Stub):
def query(self, uri: Uri, projection: Optional[List[str]], selection: Optional[str],
selection_args: Optional[List[str]], sort_order: Optional[str]) -> Optional[Any]:
return None
"""委托给 ContentResolver 处理 file:// URI。"""
resolver = ContentResolver()
return resolver.query(uri, projection, selection, selection_args, sort_order)
def insert(self, uri: Uri, values: ContentValues) -> Optional[Uri]:
return uri
resolver = ContentResolver()
return resolver.insert(uri, values)
def update(self, uri: Uri, values: ContentValues, selection: Optional[str],
selection_args: Optional[List[str]]) -> int:
return 0
resolver = ContentResolver()
return resolver.update(uri, values, selection, selection_args)
def delete(self, uri: Uri, selection: Optional[str], selection_args: Optional[List[str]]) -> int:
return 0
resolver = ContentResolver()
return resolver.delete(uri, selection, selection_args)
def applyBatch(self, operations: List[Any]) -> List[Any]:
return []
resolver = ContentResolver()
return resolver.applyBatch("", operations)
def bulkInsert(self, uri: Uri, values: List[ContentValues]) -> int:
return len(values) if values else 0
resolver = ContentResolver()
return resolver.bulkInsert(uri, values)
def getType(self, uri: Uri) -> Optional[str]:
"""返回 URI 的 MIME 类型。"""
path = ContentResolver._uri_to_file_path(uri)
if path:
mime, _ = mimetypes.guess_type(path)
return mime or "application/octet-stream"
return None
def getStreamTypes(self, uri: Uri, mime_type_filter: str) -> Optional[List[str]]:
return None
mime = self.getType(uri)
if mime is None:
return None
return [mime]
def call(self, method: str, arg: Optional[str], extras: Optional[ContentValues]) -> Optional[ContentValues]:
return None
@@ -1478,13 +1720,21 @@ class ContentProviderClient(Stub):
pass
def openFile(self, uri: Uri, mode: str) -> Any:
"""打开文件。对 file:// URI 返回真实文件对象。"""
path = ContentResolver._uri_to_file_path(uri)
if path and os.path.exists(path):
try:
fmode = "rb" if "r" in mode else "wb"
return open(path, fmode)
except OSError:
return None
return None
def openAssetFile(self, uri: Uri, mode: str) -> Any:
return None
return self.openFile(uri, mode)
def openTypedAssetFile(self, uri: Uri, mime_type: str, opts: Optional[ContentValues]) -> Any:
return None
return self.openFile(uri, "r")
def getLocalContentProvider(self) -> Optional[ContentProvider]:
return None
@@ -1658,6 +1908,14 @@ class CursorLoader(Loader):
self._sort_order = sort_order
def loadInBackground(self):
"""在后台加载数据,委托给 ContentResolver.query。"""
if self._context is not None:
try:
resolver = self._context.getContentResolver()
return resolver.query(self._uri, self._projection, self._selection,
self._selection_args, self._sort_order)
except Exception:
pass
return None
+23
View File
@@ -36,6 +36,8 @@ class Resources(Stub):
self._drawables: Dict[str, Any] = {}
self._booleans: Dict[str, bool] = {}
self._integers: Dict[str, int] = {}
self._string_arrays: Dict[str, List[str]] = {}
self._int_arrays: Dict[str, List[int]] = {}
self._identifiers: Dict[str, int] = {}
self._next_id = 0x7f010000
self._configuration = Configuration()
@@ -125,6 +127,27 @@ class Resources(Stub):
return val
return 0
# ── Arrays ──
def getStringArray(self, id: Union[int, str]) -> List[str]:
if isinstance(id, str):
return self._string_arrays.get(id, [])
for key, val in self._string_arrays.items():
if self._identifiers.get(key) == id:
return val
return []
def getTextArray(self, id: Union[int, str]) -> List[str]:
return self.getStringArray(id)
def getIntArray(self, id: Union[int, str]) -> List[int]:
if isinstance(id, str):
return self._int_arrays.get(id, [])
for key, val in self._int_arrays.items():
if self._identifiers.get(key) == id:
return val
return []
# ── Layout / Animation / Xml ──
def getLayout(self, id: Union[int, str]) -> Any:
+598 -40
View File
@@ -8,6 +8,7 @@ import math
import struct
import io
import os
import time
from enum import IntEnum
from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
@@ -732,6 +733,42 @@ class Path(Stub):
pass
def op(self, other: "Path", op: int) -> bool:
"""对两个路径执行布尔运算。基于包围盒的近似实现。"""
if other is None:
return False
self_bounds = self.computeBounds()
other_bounds = other.computeBounds()
if op == Path.Op.UNION:
# 合并:将 other 的段追加到 self
self.addPath(other)
return True
if op == Path.Op.INTERSECT:
# 交集:仅当包围盒相交时保留相交区域
if not RectF.intersects(self_bounds, other_bounds):
self.reset()
return False
# 用相交包围盒作为近似结果
inter = RectF(self_bounds.left, self_bounds.top,
self_bounds.right, self_bounds.bottom)
inter.intersect(other_bounds.left, other_bounds.top,
other_bounds.right, other_bounds.bottom)
self.reset()
self.addRect(inter.left, inter.top, inter.right, inter.bottom)
return True
if op == Path.Op.DIFFERENCE:
# 差集:若包围盒不相交则 self 不变,否则标记(近似)
if not RectF.intersects(self_bounds, other_bounds):
return True
# 近似:无法精确计算,保留 self 不变
return True
if op == Path.Op.REVERSE_DIFFERENCE:
# 反向差集:用 other 替换 self
self.set(other)
return True
if op == Path.Op.XOR:
# 异或:合并两者
self.addPath(other)
return True
return False
def isRect(self) -> bool:
@@ -770,38 +807,219 @@ class PathMeasure(Stub):
super().__init__()
self._path = path
self._force_closed = forceClosed
self._contours: List[List[Tuple[float, float, float]]] = []
self._contour_lengths: List[float] = []
self._total_length: float = 0.0
if path is not None:
self._flatten()
def _flatten(self):
"""将路径展平为线段列表,计算累积长度。"""
self._contours = []
self._contour_lengths = []
self._total_length = 0.0
if not self._path:
return
segs = self._path._segments
cur_x = 0.0
cur_y = 0.0
start_x = 0.0
start_y = 0.0
contour: List[Tuple[float, float, float]] = []
contour_len = 0.0
for seg in segs:
t = seg.get("type")
if t == "move":
if contour:
self._contours.append(contour)
self._contour_lengths.append(contour_len)
self._total_length += contour_len
cur_x = seg.get("x", 0.0)
cur_y = seg.get("y", 0.0)
start_x = cur_x
start_y = cur_y
contour = [(cur_x, cur_y, 0.0)]
contour_len = 0.0
elif t == "line":
nx = seg.get("x", cur_x)
ny = seg.get("y", cur_y)
d = math.hypot(nx - cur_x, ny - cur_y)
contour_len += d
contour.append((nx, ny, contour_len))
cur_x = nx
cur_y = ny
elif t == "quad":
x1, y1 = seg.get("x1", cur_x), seg.get("y1", cur_y)
x2, y2 = seg.get("x2", cur_x), seg.get("y2", cur_y)
steps = 16
prev_x, prev_y = cur_x, cur_y
for i in range(1, steps + 1):
tt = i / steps
mt = 1 - tt
nx = mt * mt * cur_x + 2 * mt * tt * x1 + tt * tt * x2
ny = mt * mt * cur_y + 2 * mt * tt * y1 + tt * tt * y2
d = math.hypot(nx - prev_x, ny - prev_y)
contour_len += d
contour.append((nx, ny, contour_len))
prev_x, prev_y = nx, ny
cur_x = x2
cur_y = y2
elif t == "cubic":
x1, y1 = seg.get("x1", cur_x), seg.get("y1", cur_y)
x2, y2 = seg.get("x2", cur_x), seg.get("y2", cur_y)
x3, y3 = seg.get("x3", cur_x), seg.get("y3", cur_y)
steps = 24
prev_x, prev_y = cur_x, cur_y
for i in range(1, steps + 1):
tt = i / steps
mt = 1 - tt
nx = mt ** 3 * cur_x + 3 * mt ** 2 * tt * x1 + 3 * mt * tt ** 2 * x2 + tt ** 3 * x3
ny = mt ** 3 * cur_y + 3 * mt ** 2 * tt * y1 + 3 * mt * tt ** 2 * y2 + tt ** 3 * y3
d = math.hypot(nx - prev_x, ny - prev_y)
contour_len += d
contour.append((nx, ny, contour_len))
prev_x, prev_y = nx, ny
cur_x = x3
cur_y = y3
elif t == "close":
if contour:
d = math.hypot(start_x - cur_x, start_y - cur_y)
if d > 1e-10:
contour_len += d
contour.append((start_x, start_y, contour_len))
cur_x = start_x
cur_y = start_y
if contour:
if self._force_closed and segs and segs[-1].get("type") != "close":
d = math.hypot(start_x - cur_x, start_y - cur_y)
if d > 1e-10:
contour_len += d
contour.append((start_x, start_y, contour_len))
self._contours.append(contour)
self._contour_lengths.append(contour_len)
self._total_length += contour_len
def setPath(self, path: Path, forceClosed: bool):
self._path = path
self._force_closed = forceClosed
self._contours = []
self._contour_lengths = []
self._total_length = 0.0
if path is not None:
self._flatten()
def getLength(self) -> float:
if not self._path:
return 0.0
bounds = self._path.computeBounds()
return math.sqrt(bounds.width() ** 2 + bounds.height() ** 2)
"""返回当前轮廓的长度。"""
if self._contour_lengths:
return self._contour_lengths[0]
return 0.0
def getPosTan(self, distance: float, pos: List[float], tan: List[float]) -> bool:
if not self._path:
"""获取指定距离处的坐标和切线方向。"""
if not self._contours:
return False
contour = self._contours[0]
length = self._contour_lengths[0]
if length <= 0:
return False
# 钳制距离
d = max(0.0, min(distance, length))
# 二分查找
lo, hi = 0, len(contour) - 1
while lo < hi:
mid = (lo + hi) // 2
if contour[mid][2] < d:
lo = mid + 1
else:
hi = mid
idx = max(1, lo)
px, py, pd = contour[idx - 1]
nx, ny, nd = contour[idx]
seg_len = nd - pd
if seg_len < 1e-10:
t = 0.0
else:
t = (d - pd) / seg_len
x = px + (nx - px) * t
y = py + (ny - py) * t
dx = nx - px
dy = ny - py
dl = math.hypot(dx, dy)
if dl < 1e-10:
tx, ty = 1.0, 0.0
else:
tx, ty = dx / dl, dy / dl
if len(pos) >= 2:
pos[0] = 0.0
pos[1] = 0.0
pos[0] = x
pos[1] = y
if len(tan) >= 2:
tan[0] = 1.0
tan[1] = 0.0
tan[0] = tx
tan[1] = ty
return True
def getMatrix(self, distance: float, matrix: "Matrix", flags: int) -> bool:
return False
"""获取指定距离处的变换矩阵(平移+旋转)。"""
pos = [0.0, 0.0]
tan = [0.0, 0.0]
if not self.getPosTan(distance, pos, tan):
return False
angle = math.degrees(math.atan2(tan[1], tan[0]))
matrix.reset()
matrix.postRotate(angle)
matrix.postTranslate(pos[0], pos[1])
return True
def getSegment(self, startD: float, stopD: float, dst: Path, startWithMoveTo: bool) -> bool:
return False
"""提取路径中 [startD, stopD] 区间的子路径。"""
if not self._contours:
return False
contour = self._contours[0]
length = self._contour_lengths[0]
if length <= 0:
return False
startD = max(0.0, min(startD, length))
stopD = max(0.0, min(stopD, length))
if startD >= stopD:
return False
first = True
for i in range(1, len(contour)):
px, py, pd = contour[i - 1]
nx, ny, nd = contour[i]
if nd < startD:
continue
if pd > stopD:
break
seg_len = nd - pd
if seg_len < 1e-10:
continue
# 计算线段两端在 [startD, stopD] 内的参数
t0 = max(0.0, (startD - pd) / seg_len)
t1 = min(1.0, (stopD - pd) / seg_len)
x0 = px + (nx - px) * t0
y0 = py + (ny - py) * t0
x1 = px + (nx - px) * t1
y1 = py + (ny - py) * t1
if first and startWithMoveTo:
dst.moveTo(x0, y0)
else:
dst.lineTo(x0, y0)
dst.lineTo(x1, y1)
first = False
return True
def isClosed(self) -> bool:
return self._force_closed
if not self._path or not self._path._segments:
return self._force_closed
if self._force_closed:
return True
return any(s.get("type") == "close" for s in self._path._segments)
def nextContour(self) -> bool:
"""移动到下一个轮廓。"""
if len(self._contours) > 1:
self._contours.pop(0)
self._contour_lengths.pop(0)
return True
return False
@@ -1556,35 +1774,99 @@ class Camera(Stub):
self._location_x = 0.0
self._location_y = 0.0
self._location_z = -8.0
self._rot_x = 0.0
self._rot_y = 0.0
self._rot_z = 0.0
self._trans_x = 0.0
self._trans_y = 0.0
self._trans_z = 0.0
self._stack: List[Tuple[float, float, float, float, float, float, float, float, float]] = []
def save(self):
pass
"""保存当前变换状态到栈。"""
self._stack.append((self._rot_x, self._rot_y, self._rot_z,
self._trans_x, self._trans_y, self._trans_z,
self._location_x, self._location_y, self._location_z))
def restore(self):
pass
"""从栈恢复上次保存的变换状态。"""
if self._stack:
(self._rot_x, self._rot_y, self._rot_z,
self._trans_x, self._trans_y, self._trans_z,
self._location_x, self._location_y, self._location_z) = self._stack.pop()
def translate(self, x: float, y: float, z: float):
self._location_x += x
self._location_y += y
self._location_z += z
self._trans_x += x
self._trans_y += y
self._trans_z += z
def rotateX(self, deg: float):
pass
"""绕 X 轴旋转(累积角度)。"""
self._rot_x += deg
def rotateY(self, deg: float):
pass
"""绕 Y 轴旋转(累积角度)。"""
self._rot_y += deg
def rotateZ(self, deg: float):
pass
"""绕 Z 轴旋转(累积角度)。"""
self._rot_z += deg
def rotate(self, x: float, y: float, z: float):
pass
"""同时绕 X/Y/Z 轴旋转。"""
self._rot_x += x
self._rot_y += y
self._rot_z += z
def _compute_matrix(self) -> List[float]:
"""计算 4x4 变换矩阵并投影为 3x3 仿射矩阵。"""
rx = math.radians(self._rot_x)
ry = math.radians(self._rot_y)
rz = math.radians(self._rot_z)
cx, sx = math.cos(rx), math.sin(rx)
cy, sy = math.cos(ry), math.sin(ry)
cz, sz = math.cos(rz), math.sin(rz)
# 旋转矩阵 R = Rz * Ry * Rx
r00 = cz * cy
r01 = cz * sy * sx - sz * cx
r02 = cz * sy * cx + sz * sx
r10 = sz * cy
r11 = sz * sy * sx + cz * cx
r12 = sz * sy * cx - cz * sx
r20 = -sy
r21 = cy * sx
r22 = cy * cx
# 应用平移
tx = self._trans_x
ty = self._trans_y
tz = self._trans_z
# 相机位置(z 越负越远)
cam_z = self._location_z
# 透视投影:z' = z + cam_z,当 cam_z != 0 时做透视除法
# 简化为:将 3D 点投影到 2D
# 点 (x,y,z) -> (x*r00+y*r01+z*r02+tx, x*r10+y*r11+z*r12+ty, x*r20+y*r21+z*r22+tz)
# 透视:screen_x = x' * d / (d - z'), screen_y = y' * d / (d - z')
d = -cam_z # 相机到原点距离
if abs(d) < 1e-10:
# 正交投影
return [r00, r01, tx, r10, r11, ty, 0.0, 0.0, 1.0]
# 透视投影矩阵(近似):将 z 影响缩放到 x,y
# m[0]=r00, m[1]=r01, m[2]=tx, m[3]=r10, m[4]=r11, m[5]=ty
# 透视项:m[6]=r20/d, m[7]=r21/d, m[8]=1 - tz/d
return [r00, r01, tx,
r10, r11, ty,
r20 / d, r21 / d, 1.0 - tz / d]
def getMatrix(self, matrix: Matrix):
matrix.reset()
"""计算变换矩阵并写入给定的 Matrix。"""
matrix._values = self._compute_matrix()
def applyToCanvas(self, canvas: "Canvas"):
pass
"""将相机变换应用到 Canvas。"""
m = Matrix()
self.getMatrix(m)
if hasattr(canvas, "concat"):
canvas.concat(m)
def setLocation(self, x: float, y: float, z: float):
self._location_x = x
@@ -1592,7 +1874,17 @@ class Camera(Stub):
self._location_z = z
def dotWithNormal(self, dx: float, dy: float, dz: float) -> float:
return 0.0
"""计算向量与变换后 Z 轴法线的点积。"""
rx = math.radians(self._rot_x)
ry = math.radians(self._rot_y)
rz = math.radians(self._rot_z)
cy, sy = math.cos(ry), math.sin(ry)
cx, sx = math.cos(rx), math.sin(rx)
# 变换后的 Z 轴法线
nz_x = sy
nz_y = -cy * sx
nz_z = cy * cx
return dx * nz_x + dy * nz_y + dz * nz_z
# ═══════════════════════════════════════════════════════════════
@@ -1660,8 +1952,15 @@ class Picture(Stub):
self._canvas = None
def draw(self, canvas: "Canvas"):
"""回放录制的绘制命令到指定 Canvas。"""
for cmd in self._recorded:
pass
method = cmd.get("method")
args = cmd.get("args", [])
if method and hasattr(canvas, method):
try:
getattr(canvas, method)(*args)
except Exception:
pass
def getWidth(self) -> int:
return self._canvas.getWidth() if hasattr(self, "_canvas") and self._canvas else 0
@@ -1690,7 +1989,9 @@ class PictureDrawable(Stub):
self._picture = picture
def draw(self, canvas: "Canvas"):
pass
"""将 Picture 回放到指定 Canvas。"""
if self._picture is not None:
self._picture.draw(canvas)
# ═══════════════════════════════════════════════════════════════
@@ -1704,6 +2005,10 @@ class SurfaceTexture(Stub):
super().__init__()
self._tex_name = texName
self._timestamp = 0
self._buffer_width = 0
self._buffer_height = 0
self._attached = texName != 0
self._released = False
def getTimestamp(self) -> int:
return self._timestamp
@@ -1712,16 +2017,24 @@ class SurfaceTexture(Stub):
self._timestamp = int(time.time() * 1e9)
def release(self):
pass
"""释放 SurfaceTexture 资源。"""
self._released = True
self._attached = False
def setDefaultBufferSize(self, width: int, height: int):
pass
"""设置默认缓冲区大小。"""
self._buffer_width = width
self._buffer_height = height
def attachToGLContext(self, texName: int):
pass
"""将 SurfaceTexture 附加到 GL 纹理。"""
self._tex_name = texName
self._attached = True
self._released = False
def detachFromGLContext(self):
pass
"""从 GL 纹理分离 SurfaceTexture。"""
self._attached = False
# ═══════════════════════════════════════════════════════════════
@@ -1768,6 +2081,59 @@ class Region(Stub):
return self._rect.contains(x, y)
def op(self, *args) -> bool:
"""对区域执行布尔运算。支持 op(rect, op) / op(region, op) / op(left,top,right,bottom,op)。"""
if len(args) == 2:
operand, op_code = args
if isinstance(operand, Rect):
other = Rect(operand.left, operand.top, operand.right, operand.bottom)
elif isinstance(operand, Region):
other = Rect(operand._rect.left, operand._rect.top,
operand._rect.right, operand._rect.bottom)
else:
return False
elif len(args) == 5:
left, top, right, bottom, op_code = args
other = Rect(left, top, right, bottom)
else:
return False
self_r = self._rect
if op_code == Region.OP_INTERSECT:
nL = max(self_r.left, other.left)
nT = max(self_r.top, other.top)
nR = min(self_r.right, other.right)
nB = min(self_r.bottom, other.bottom)
if nL < nR and nT < nB:
self_r.set(nL, nT, nR, nB)
return True
self_r.setEmpty()
return False
if op_code == Region.OP_UNION:
if self_r.isEmpty():
self_r.set(other.left, other.top, other.right, other.bottom)
else:
self_r.left = min(self_r.left, other.left)
self_r.top = min(self_r.top, other.top)
self_r.right = max(self_r.right, other.right)
self_r.bottom = max(self_r.bottom, other.bottom)
return True
if op_code == Region.OP_REPLACE:
self_r.set(other.left, other.top, other.right, other.bottom)
return True
if op_code == Region.OP_DIFFERENCE:
# 差集:若不相交则不变;若完全包含则清空;否则近似为不变
if not Rect.intersects(self_r, other):
return True
if other.contains(self_r.left, self_r.top) and other.contains(self_r.right - 1, self_r.bottom - 1):
self_r.setEmpty()
return False
return True
if op_code == Region.OP_REVERSE_DIFFERENCE:
self_r.set(other.left, other.top, other.right, other.bottom)
return True
if op_code == Region.OP_XOR:
if not Rect.intersects(self_r, other):
self_r.union(other.left, other.top, other.right, other.bottom)
return True
return False
def isRect(self) -> bool:
@@ -1894,10 +2260,32 @@ class Paint(Stub):
def setTextSize(self, textSize: float):
self._text_size = textSize
self._font_cache = None
def getTextSize(self) -> float:
return self._text_size
def _get_font(self):
"""获取 PIL ImageFont 对象(带缓存)。"""
if not HAS_PIL:
return None
cache_key = (self._text_size, id(self._typeface) if self._typeface else None)
if getattr(self, "_font_cache_key", None) == cache_key and getattr(self, "_font_cache", None):
return self._font_cache
try:
from PIL import ImageFont
font_path = None
if self._typeface and hasattr(self._typeface, "_path") and self._typeface._path:
font_path = self._typeface._path
if font_path and os.path.exists(font_path):
self._font_cache = ImageFont.truetype(font_path, int(self._text_size))
else:
self._font_cache = ImageFont.load_default()
self._font_cache_key = cache_key
return self._font_cache
except Exception:
return None
def setTypeface(self, typeface: Optional[Typeface]):
self._typeface = typeface
@@ -2004,13 +2392,31 @@ class Paint(Stub):
self._shadow_layer = None
def measureText(self, text: str) -> float:
"""测量文本宽度。优先使用 PIL ImageFont 获取精确宽度。"""
font = self._get_font()
if font is not None:
try:
if hasattr(font, "getlength"):
return float(font.getlength(text))
bbox = font.getbbox(text)
return float(bbox[2] - bbox[0])
except Exception:
pass
return len(text) * self._text_size * 0.5
def breakText(self, text: str, maxWidth: float, measuredWidth: List[float]) -> int:
"""在最大宽度内截断文本,返回可容纳的字符数。"""
font = self._get_font()
width = 0.0
count = 0
for i, ch in enumerate(text):
cw = self._text_size * 0.5
if font is not None:
try:
cw = font.getlength(ch) if hasattr(font, "getlength") else (font.getbbox(ch)[2])
except Exception:
cw = self._text_size * 0.5
else:
cw = self._text_size * 0.5
if width + cw > maxWidth:
break
width += cw
@@ -2019,9 +2425,16 @@ class Paint(Stub):
return count
def getTextWidths(self, text: str, widths: List[float]) -> int:
cw = self._text_size * 0.5
"""获取每个字符的宽度。"""
font = self._get_font()
for i, ch in enumerate(text):
widths[i] = cw
if font is not None:
try:
widths[i] = float(font.getlength(ch)) if hasattr(font, "getlength") else float(font.getbbox(ch)[2])
except Exception:
widths[i] = self._text_size * 0.5
else:
widths[i] = self._text_size * 0.5
return len(text)
def getTextPath(self, text: str, start: int, end: int, x: float, y: float, path: Path):
@@ -2030,7 +2443,22 @@ class Paint(Stub):
path.lineTo(x + (i - start + 1) * self._text_size * 0.3, y)
def getFontMetrics(self) -> Dict[str, float]:
"""返回字体度量信息。优先使用 PIL ImageFont 获取真实度量。"""
size = self._text_size
font = self._get_font()
if font is not None:
try:
if hasattr(font, "getmetrics"):
ascent, descent = font.getmetrics()
return {
"ascent": -float(ascent),
"descent": float(descent),
"top": -float(ascent) - 2.0,
"bottom": float(descent) + 2.0,
"leading": 0.0,
}
except Exception:
pass
return {
"ascent": -size * 0.8,
"descent": size * 0.2,
@@ -2068,6 +2496,19 @@ class Paint(Stub):
return getattr(self, '_stroke_miter', 4.0)
def getTextBounds(self, text, start, end, bounds):
"""返回文本边界矩形。优先使用 PIL ImageFont 获取真实边界。"""
font = self._get_font()
if font is not None:
try:
sub = text[start:end]
bbox = font.getbbox(sub)
bounds.left = int(bbox[0])
bounds.top = int(bbox[1])
bounds.right = int(bbox[2])
bounds.bottom = int(bbox[3])
return
except Exception:
pass
if hasattr(self, '_text_size'):
w = len(text[start:end]) * self._text_size * 0.5
h = self._text_size
@@ -2312,13 +2753,30 @@ class Bitmap(Stub):
return Bitmap(self._width, self._height, Bitmap.Config.ALPHA_8)
def sameAs(self, other: "Bitmap") -> bool:
"""比较两个 Bitmap 的像素是否完全相同。"""
if not isinstance(other, Bitmap):
return False
if self._width != other._width or self._height != other._height:
return False
if self._config != other._config:
return False
if self._recycled or other._recycled:
return False
# 使用 PIL 比较像素
if HAS_PIL and hasattr(self, "_image") and hasattr(other, "_image"):
try:
from PIL import ImageChops
diff = ImageChops.difference(self._image, other._image)
return diff.getbbox() is None
except Exception:
return list(self._image.getdata()) == list(other._image.getdata())
# 回退到字节比较
if hasattr(self, "_pixels") and hasattr(other, "_pixels"):
return bytes(self._pixels) == bytes(other._pixels)
return True
def prepareToDraw(self):
"""提示将位图数据上传到 GPU 缓存。桌面端为空操作。"""
pass
def hasAlpha(self) -> bool:
@@ -2407,6 +2865,29 @@ class BitmapFactory(Stub):
@staticmethod
def decodeResource(res, resId: int, opts: Options = None) -> Optional[Bitmap]:
"""从资源解码位图。尝试通过 Resources 打开原始资源流。"""
# 尝试通过 Resources 获取资源路径或流
if res is not None:
try:
# 尝试 openRawResource
stream = res.openRawResource(resId)
if stream is not None:
return BitmapFactory.decodeStream(stream, None, opts)
except Exception:
pass
try:
# 尝试通过资源名查找文件
name = res.getResourceEntryName(resId) if hasattr(res, "getResourceEntryName") else None
if name and name != "unknown":
for ext in (".png", ".jpg", ".jpeg", ".webp", ".bmp"):
candidate = resolve_android_path("res/drawable/" + name + ext)
if os.path.exists(candidate):
return BitmapFactory.decodeFile(candidate, opts)
candidate = resolve_android_path("res/drawable-xxhdpi/" + name + ext)
if os.path.exists(candidate):
return BitmapFactory.decodeFile(candidate, opts)
except Exception:
pass
return Bitmap(100, 100, Bitmap.Config.ARGB_8888)
@staticmethod
@@ -2610,7 +3091,22 @@ class Canvas(Stub):
return Rect(r.left, r.top, r.right, r.bottom)
def quickReject(self, *args) -> bool:
return False
"""判断给定矩形是否完全在裁剪区域之外。"""
clip = self._clip_stack[-1]
if len(args) == 1:
r = args[0]
if isinstance(r, Rect):
left, top, right, bottom = r.left, r.top, r.right, r.bottom
elif isinstance(r, RectF):
left, top, right, bottom = r.left, r.top, r.right, r.bottom
else:
return False
elif len(args) == 4:
left, top, right, bottom = args
else:
return False
return (right <= clip.left or left >= clip.right or
bottom <= clip.top or top >= clip.bottom)
def drawColor(self, color: int):
if HAS_PIL and self._draw is not None:
@@ -2810,7 +3306,37 @@ class Canvas(Stub):
self._draw.text((x, y), text, fill=fill_color)
def drawTextOnPath(self, text: str, path: Path, hOffset: float, vOffset: float, paint: Paint):
pass
"""沿路径绘制文本。使用 PathMeasure 逐字符放置。"""
if not text or not HAS_PIL or self._draw is None:
return
pm = PathMeasure(path, False)
total = pm.getLength()
if total <= 0:
return
c = paint.getColor()
fill_color = (Color.red(c), Color.green(c), Color.blue(c), Color.alpha(c))
dist = hOffset
pos = [0.0, 0.0]
tan = [0.0, 0.0]
for ch in text:
cw = paint.measureText(ch)
if dist > total:
break
if pm.getPosTan(dist, pos, tan):
angle = math.degrees(math.atan2(tan[1], tan[0]))
# 使用 PIL 旋转绘制每个字符
try:
from PIL import Image as _PILImage
char_img = _PILImage.new("RGBA", (int(cw) + 4, int(paint._text_size) + 4), (0, 0, 0, 0))
from PIL import ImageDraw as _PILDraw
cd = _PILDraw.Draw(char_img)
cd.text((0, 0), ch, fill=fill_color)
rotated = char_img.rotate(-angle, expand=True, resample=_PILImage.BICUBIC)
self._image.paste(rotated, (int(pos[0] + vOffset * tan[1]), int(pos[1] - vOffset * tan[0])),
rotated)
except Exception:
self._draw.text((pos[0], pos[1]), ch, fill=fill_color)
dist += cw
def drawPosText(self, text: str, pos: List[float], paint: Paint):
if HAS_PIL and self._draw is not None:
@@ -2821,14 +3347,44 @@ class Canvas(Stub):
self._draw.text((pos[i * 2], pos[i * 2 + 1]), ch, fill=fill_color)
def drawPicture(self, picture: Picture, dst: Union[Rect, RectF] = None):
pass
"""回放 Picture 中录制的绘制命令。"""
if picture is None:
return
recorded = getattr(picture, "_recorded", [])
if not recorded:
return
for cmd in recorded:
method = cmd.get("method")
args = cmd.get("args", [])
if method and hasattr(self, method):
try:
getattr(self, method)(*args)
except Exception:
pass
def drawPatch(self, patch: NinePatch, dst: Rect, paint: Paint = None):
pass
"""绘制 NinePatch(拉伸位图)到目标矩形。"""
if patch is None or patch._bitmap is None:
return
bitmap = patch._bitmap
if dst is None:
self.drawBitmap(bitmap, 0, 0, paint)
return
# 简单缩放绘制
if isinstance(dst, (Rect, RectF)):
dw = dst.right - dst.left
dh = dst.bottom - dst.top
else:
dw, dh = dst[2] - dst[0], dst[3] - dst[1]
scaled = Bitmap.createScaledBitmap(bitmap, int(dw), int(dh), True)
self.drawBitmap(scaled, dst.left if hasattr(dst, "left") else dst[0],
dst.top if hasattr(dst, "top") else dst[1], paint)
def drawDoubleRoundRect(self, outer: RectF, outerRx: float, outerRy: float,
inner: RectF, innerRx: float, innerRy: float, paint: Paint):
pass
"""绘制双层圆角矩形(外层和内层)。"""
self.drawRoundRect(outer, outerRx, outerRy, paint)
self.drawRoundRect(inner, innerRx, innerRy, paint)
def drawBitmapMesh(self, bitmap: Bitmap, meshWidth: int, meshHeight: int,
verts: List[float], vertOffset: int, colors: List[int],
@@ -2848,10 +3404,12 @@ class Canvas(Stub):
return False
def setDensity(self, density: int):
pass
"""设置画布密度(DPI)。"""
self._density = density
def getDensity(self) -> int:
return 0
"""返回画布密度(DPI),默认 0 表示未知。"""
return getattr(self, "_density", 0)
def toString(self) -> str:
return f"Canvas({self._width}x{self._height})"
+148 -17
View File
@@ -12,10 +12,16 @@ from .._core import Stub, AndroidOnlyError, android_context, HAS_PIL
# Linux 真实实现导入
try:
from .._linux_impl import IS_LINUX, LinuxBluetooth, LinuxSensors, HAS_BLUETOOTHCTL
from .._linux_impl import IS_LINUX, LinuxBluetooth, LinuxSensors, HAS_BLUETOOTHCTL, LinuxPower
_USE_LINUX = IS_LINUX
except ImportError:
_USE_LINUX = False
try:
from .._linux_impl import IS_LINUX, LinuxBluetooth, LinuxSensors, HAS_BLUETOOTHCTL
_USE_LINUX = IS_LINUX
LinuxPower = None
except ImportError:
_USE_LINUX = False
LinuxPower = None
# ═══════════════════════════════════════════════════════════════
@@ -379,18 +385,56 @@ class SensorManager(Stub):
@staticmethod
def getRotationMatrix(R: List[float], I: List[float],
gravity: List[float], geomagnetic: List[float]) -> bool:
"""根据重力和地磁向量计算旋转矩阵 R 和倾斜矩阵 I。
算法对齐 Android SensorManager.getRotationMatrix。"""
if len(R) < 9:
R.extend([0] * (9 - len(R)))
if len(R) >= 9:
R[0] = 1.0
R[1] = 0.0
R[2] = 0.0
R[3] = 0.0
R[4] = 1.0
R[5] = 0.0
R[6] = 0.0
R[7] = 0.0
R[8] = 1.0
R.extend([0.0] * (9 - len(R)))
gx, gy, gz = gravity[0], gravity[1], gravity[2]
mx, my, mz = geomagnetic[0], geomagnetic[1], geomagnetic[2]
# 归一化重力向量 -> H
g_len = math.sqrt(gx * gx + gy * gy + gz * gz)
if g_len < 1e-10:
return False
hx = gx / g_len
hy = gy / g_len
hz = gz / g_len
# 叉积 A = H x M (东方向)
ax = hy * mz - hz * my
ay = hz * mx - hx * mz
az = hx * my - hy * mx
a_len = math.sqrt(ax * ax + ay * ay + az * az)
if a_len < 1e-10:
return False
ax /= a_len
ay /= a_len
az /= a_len
# 重新计算 M = A x H
ex = ay * hz - az * hy
ey = az * hx - ax * hz
ez = ax * hy - ay * hx
# 填充旋转矩阵 R (列优先,Android 格式)
R[0] = ax
R[1] = ay
R[2] = az
R[3] = ex
R[4] = ey
R[5] = ez
R[6] = hx
R[7] = hy
R[8] = hz
# 填充倾斜矩阵 I
if I is not None and len(I) >= 9:
inv_h = 1.0 / g_len
inv_m = 1.0 / math.sqrt(mx * mx + my * my + mz * mz) if (mx * mx + my * my + mz * mz) > 0 else 0
I[0] = inv_h * hx
I[1] = inv_h * hy
I[2] = inv_h * hz
I[3] = inv_m * mx
I[4] = inv_m * my
I[5] = inv_m * mz
I[6] = inv_h * ex
I[7] = inv_h * ey
I[8] = inv_h * ez
return True
@staticmethod
@@ -399,17 +443,69 @@ class SensorManager(Stub):
@staticmethod
def getInclination(I: List[float]) -> float:
"""根据倾斜矩阵 I 计算地磁倾角(弧度)。"""
if len(I) >= 9:
# I[5] = cos(el), I[2] = sin(el)
# 倾角 = atan2(I[2], I[5])
return math.atan2(I[2], I[5])
return 0.0
@staticmethod
def remapCoordinateSystem(inR: List[float], X: int, Y: int,
outR: List[float]) -> bool:
if len(inR) >= 9 and len(outR) >= 9:
for i in range(9):
outR[i] = inR[i]
"""根据指定的 X/Y 轴重映射坐标系。"""
if len(inR) < 9 or len(outR) < 9:
return False
# Android 轴常量
AXIS_X = 1
AXIS_Y = 2
AXIS_Z = 3
AXIS_MINUS_X = -1
AXIS_MINUS_Y = -2
AXIS_MINUS_Z = -3
# 将轴常量映射到列索引和符号
axis_map = {
AXIS_X: (0, 1), AXIS_Y: (1, 1), AXIS_Z: (2, 1),
AXIS_MINUS_X: (0, -1), AXIS_MINUS_Y: (1, -1), AXIS_MINUS_Z: (2, -1),
}
if X not in axis_map or Y not in axis_map:
return False
x_col, x_sign = axis_map[X]
y_col, y_sign = axis_map[Y]
# Z = X x Y
z_col = 3 - x_col - y_col
z_sign = x_sign * y_sign
# 如果 X 和 Y 不正交
if abs(x_col - y_col) != 1 and not (x_col == 0 and y_col == 2) and not (x_col == 2 and y_col == 0):
z_sign = -z_sign
# 填充输出矩阵(列优先)
for row in range(3):
outR[row * 3 + 0] = inR[row * 3 + x_col] * x_sign
outR[row * 3 + 1] = inR[row * 3 + y_col] * y_sign
outR[row * 3 + 2] = inR[row * 3 + z_col] * z_sign
return True
def getDeviceId(self) -> str:
"""返回传感器设备 ID。尝试从 /sys 读取真实设备标识。"""
import os as _os
# 尝试从 /sys/class/sensor 读取
try:
base = "/sys/class/sensor"
if _os.path.isdir(base):
names = _os.listdir(base)
if names:
return names[0]
except (OSError, IOError):
pass
# 尝试从 /sys/bus/iio/devices 读取
try:
base = "/sys/bus/iio/devices"
if _os.path.isdir(base):
names = _os.listdir(base)
if names:
return names[0]
except (OSError, IOError):
pass
return "sensor_hub_0"
@@ -552,6 +648,29 @@ class BatteryManager(Stub):
super().__init__()
def getIntProperty(self, id: int) -> int:
"""返回电池整型属性。优先使用 LinuxPower 读取真实数据。"""
if _USE_LINUX and LinuxPower is not None:
try:
info = LinuxPower.get_battery_info() or {}
if id == BatteryManager.BATTERY_PROPERTY_CAPACITY:
level = info.get("level", -1)
return int(level) if isinstance(level, (int, float)) and level >= 0 else -1
if id == BatteryManager.BATTERY_PROPERTY_STATUS:
status_map = {
"charging": BatteryManager.BATTERY_STATUS_CHARGING,
"discharging": BatteryManager.BATTERY_STATUS_DISCHARGING,
"not charging": BatteryManager.BATTERY_STATUS_NOT_CHARGING,
"full": BatteryManager.BATTERY_STATUS_FULL,
"unknown": BatteryManager.BATTERY_STATUS_UNKNOWN,
}
return status_map.get(str(info.get("status", "unknown")).lower(),
BatteryManager.BATTERY_STATUS_UNKNOWN)
if id == BatteryManager.BATTERY_PROPERTY_CHARGE_COUNTER:
voltage = info.get("voltage", 0)
return int(voltage * 1000) if isinstance(voltage, (int, float)) else 0
except Exception:
pass
# 回退到默认值
id_values = {
BatteryManager.BATTERY_PROPERTY_CHARGE_COUNTER: 3500,
BatteryManager.BATTERY_PROPERTY_CAPACITY: 85,
@@ -560,7 +679,19 @@ class BatteryManager(Stub):
return id_values.get(id, 0)
def getLongProperty(self, id: int) -> int:
return 0
"""返回电池长整型属性。优先使用 LinuxPower 读取真实数据。"""
if _USE_LINUX and LinuxPower is not None:
try:
info = LinuxPower.get_battery_info() or {}
if id == BatteryManager.BATTERY_PROPERTY_CURRENT_NOW:
current = info.get("current_now", 0)
return int(current) if isinstance(current, (int, float)) else 0
if id == BatteryManager.BATTERY_PROPERTY_ENERGY_COUNTER:
voltage = info.get("voltage", 0)
return int(voltage * 3600000) if isinstance(voltage, (int, float)) else 0
except Exception:
pass
return self.getIntProperty(id)
# ═══════════════════════════════════════════════════════════════
+3 -2
View File
@@ -13,6 +13,7 @@ import random
from typing import Any, Callable, Dict, List, Optional, Tuple
from .._core import Stub, AndroidOnlyError
from .hardware import SensorManager as _BaseSensorManager
__all__ = [
"Sensor", "SensorEvent", "SensorEventListener",
@@ -313,8 +314,8 @@ class SensorEventListener(Stub):
# SensorManager (扩展)
# ═══════════════════════════════════════════════════════════════
class SensorManager(Stub):
"""SensorManager — 传感器管理器 (完整版)"""
class SensorManager(_BaseSensorManager):
"""SensorManager — 传感器管理器 (完整版,继承 hardware.SensorManager)"""
AXIS_X = AXIS_X
AXIS_Y = AXIS_Y
+193 -13
View File
@@ -620,6 +620,9 @@ class GnssStatus(Stub):
return 0.0
def getBasebandCn0DbHz(self, index: int) -> float:
"""返回指定卫星的基带载噪比(dB-Hz)。"""
if 0 <= index < len(self._satellites):
return self._satellites[index].get("baseband_cn0", 0.0)
return 0.0
def getCn0DbHz(self, index: int) -> float:
@@ -873,6 +876,42 @@ class Address(Stub):
def setFeatureName(self, name: str):
self._feature_name = name
def setAdminArea(self, area: str):
self._admin_area = area
def setSubAdminArea(self, area: str):
self._sub_admin_area = area
def setLocality(self, locality: str):
self._locality = locality
def setSubLocality(self, sub_locality: str):
self._sub_locality = sub_locality
def setPremises(self, premises: str):
self._premises = premises
def setThoroughfare(self, thoroughfare: str):
self._thoroughfare = thoroughfare
def setSubThoroughfare(self, sub_thoroughfare: str):
self._sub_thoroughfare = sub_thoroughfare
def setPostalCode(self, postal_code: str):
self._postal_code = postal_code
def setCountryName(self, country_name: str):
self._country_name = country_name
def setCountryCode(self, country_code: str):
self._country_code = country_code
def setPhone(self, phone: str):
self._phone = phone
def setUrl(self, url: str):
self._url = url
def setLatitude(self, latitude: float):
self._latitude = latitude
self._has_lat_lon = True
@@ -898,29 +937,144 @@ class Address(Stub):
class Geocoder(Stub):
"""Geocoder — 地理编码/逆地理编码"""
# Nominatim (OpenStreetMap) 免费 API
_NOMINATIM_URL = "https://nominatim.openstreetmap.org"
_USER_AGENT = "Android-pyApi/1.0"
def __init__(self, context: Any = None, locale: Any = None):
super().__init__()
self._context = context
self._locale = locale
def _get_locale_str(self) -> str:
if self._locale is None:
return "en"
if hasattr(self._locale, "getLanguage"):
lang = self._locale.getLanguage()
country = self._locale.getCountry()
return f"{lang}-{country}" if country else lang
return str(self._locale)
@staticmethod
def _http_get_json(url: str) -> Optional[Any]:
"""使用 urllib 发送 GET 请求并解析 JSON。"""
try:
import urllib.request
import json
req = urllib.request.Request(url, headers={
"User-Agent": Geocoder._USER_AGENT,
"Accept-Language": "en",
})
with urllib.request.urlopen(req, timeout=10) as resp:
return json.loads(resp.read().decode("utf-8"))
except Exception:
return None
def getFromLocation(self, latitude: float, longitude: float,
max_results: int = 1) -> List[Address]:
"""逆地理编码:根据经纬度获取地址。优先使用 Nominatim API。"""
results: List[Address] = []
# 尝试在线逆地理编码
try:
url = (f"{self._NOMINATIM_URL}/reverse?"
f"lat={latitude}&lon={longitude}&format=jsonv2"
f"&addressdetails=1&limit={max_results}")
data = self._http_get_json(url)
if data and isinstance(data, dict) and "address" in data:
addr = self._parse_nominatim_reverse(data)
results.append(addr)
except Exception:
pass
# 回退到默认地址
if not results:
addr = Address()
addr.setLatitude(latitude)
addr.setLongitude(longitude)
addr.setFeatureName("Unknown Location")
addr.setLocality("Unknown")
addr.setCountryName("Unknown")
results.append(addr)
return results[:max_results] if max_results > 0 else []
def _parse_nominatim_reverse(self, data: dict) -> Address:
"""解析 Nominatim 逆地理编码响应。"""
addr = Address()
addr.setLatitude(latitude)
addr.setLongitude(longitude)
addr.setFeatureName("Unknown Location")
addr.setLocality("Unknown")
addr.setCountryName("Unknown")
return [addr] if max_results > 0 else []
address = data.get("address", {})
lat = data.get("lat")
lon = data.get("lon")
if lat is not None:
addr.setLatitude(float(lat))
if lon is not None:
addr.setLongitude(float(lon))
addr.setFeatureName(address.get("road") or address.get("building")
or data.get("name") or "Unknown")
addr.setLocality(address.get("city") or address.get("town")
or address.get("village") or address.get("county") or "Unknown")
addr.setAdminArea(address.get("state") or "")
addr.setSubAdminArea(address.get("state_district") or "")
addr.setCountryName(address.get("country") or "Unknown")
addr.setCountryCode(address.get("country_code", "").upper())
addr.setPostalCode(address.get("postcode") or "")
addr.setAddressLine(0, data.get("display_name", ""))
return addr
def getFromLocationName(self, location_name: str, max_results: int = 1,
lower_left_lat: Optional[float] = None,
lower_left_lon: Optional[float] = None,
upper_right_lat: Optional[float] = None,
upper_right_lon: Optional[float] = None) -> List[Address]:
"""地理编码:根据地名获取经纬度。优先使用 Nominatim API。"""
results: List[Address] = []
# 尝试在线地理编码
try:
import urllib.parse
params = {
"q": location_name,
"format": "jsonv2",
"addressdetails": "1",
"limit": str(max_results),
}
# 添加边界框
if all(v is not None for v in (lower_left_lat, lower_left_lon,
upper_right_lat, upper_right_lon)):
params["viewbox"] = f"{lower_left_lon},{upper_right_lat},{upper_right_lon},{lower_left_lat}"
params["bounded"] = "1"
query = urllib.parse.urlencode(params)
url = f"{self._NOMINATIM_URL}/search?{query}"
data = self._http_get_json(url)
if data and isinstance(data, list):
for item in data[:max_results]:
addr = self._parse_nominatim_search(item)
results.append(addr)
except Exception:
pass
# 回退到默认地址
if not results:
addr = Address()
addr.setFeatureName(location_name)
addr.setLocality(location_name)
results.append(addr)
return results[:max_results] if max_results > 0 else []
def _parse_nominatim_search(self, item: dict) -> Address:
"""解析 Nominatim 搜索响应。"""
addr = Address()
addr.setFeatureName(location_name)
addr.setLocality(location_name)
return [addr] if max_results > 0 else []
lat = item.get("lat")
lon = item.get("lon")
if lat is not None:
addr.setLatitude(float(lat))
if lon is not None:
addr.setLongitude(float(lon))
address = item.get("address", {})
addr.setFeatureName(item.get("name") or address.get("road") or "Unknown")
addr.setLocality(address.get("city") or address.get("town")
or address.get("village") or address.get("county") or "Unknown")
addr.setAdminArea(address.get("state") or "")
addr.setCountryName(address.get("country") or "Unknown")
addr.setCountryCode(address.get("country_code", "").upper())
addr.setPostalCode(address.get("postcode") or "")
addr.setAddressLine(0, item.get("display_name", ""))
return addr
# ═══════════════════════════════════════════════════════════════
@@ -956,6 +1110,9 @@ class LocationManager(Stub):
self._last_known_locations: Dict[str, Location] = {}
self._enabled_providers = {self.GPS_PROVIDER, self.NETWORK_PROVIDER,
self.PASSIVE_PROVIDER, self.FUSED_PROVIDER}
self._test_providers: Dict[str, LocationProvider] = {}
self._test_locations: Dict[str, Location] = {}
self._proximity_alerts: List[Dict[str, Any]] = []
def getLastKnownLocation(self, provider: str) -> Optional[Location]:
if _USE_LINUX:
@@ -1036,19 +1193,42 @@ class LocationManager(Stub):
def addProximityAlert(self, latitude: float, longitude: float,
radius: float, expiration: int, intent: Any):
pass
"""添加接近警报。记录警报信息,在位置更新时检查是否进入/离开区域。"""
import time as _time
alert = {
"latitude": latitude,
"longitude": longitude,
"radius": radius,
"expiration": expiration,
"intent": intent,
"created_at": _time.time(),
"last_state": None, # None=未触发, True=在区域内, False=在区域外
}
self._proximity_alerts.append(alert)
def sendExtraCommand(self, provider: str, command: str, extras: Any) -> bool:
"""向位置提供者发送额外命令。桌面端无实际效果,返回成功。"""
return True
def clearTestProviderLocation(self, provider: str):
pass
"""清除测试提供者的位置数据。"""
self._test_locations.pop(provider, None)
def setTestProviderLocation(self, provider: str, location: Location):
pass
"""设置测试提供者的位置。"""
self._test_locations[provider] = location
# 通知该提供者的监听器
if provider in self._listeners:
for listener in self._listeners[provider]:
try:
listener.onLocationChanged(location)
except Exception:
pass
def removeTestProvider(self, provider: str):
pass
"""移除测试提供者。"""
self._test_providers.pop(provider, None)
self._test_locations.pop(provider, None)
def addTestProvider(self, name: str, requires_network: bool, requires_satellite: bool,
requires_cell: bool, has_monetary_cost: bool, supports_altitude: bool,
+332 -27
View File
@@ -115,11 +115,46 @@ class Build(Stub):
@staticmethod
def getRadioVersion() -> str:
return ""
"""返回基带/无线电版本。桌面端无无线电硬件,返回空字符串。"""
return Build.RADIO
@staticmethod
def getSerial() -> str:
return "unknown"
"""返回设备序列号。尝试从 /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:
@@ -201,7 +236,17 @@ class SystemClock(Stub):
@staticmethod
def setCurrentTimeMillis(millis: int) -> bool:
return False
"""尝试设置系统时间。需要 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
# ═══════════════════════════════════════════════════════════════
@@ -268,6 +313,20 @@ class Environment(Stub):
@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
@@ -309,7 +368,8 @@ class Process(Stub):
@staticmethod
def myUid() -> int:
return 10001
"""返回当前进程的真实 UID。"""
return _os.getuid()
@staticmethod
def myPid() -> int:
@@ -331,11 +391,15 @@ class Process(Stub):
pass
@staticmethod
def setThreadPriority(priority: int):
pass
@staticmethod
def setThreadPriority(tid, priority):
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):
@@ -1006,7 +1070,20 @@ class Looper(Stub):
return self._message_queue
def dump(self, pw: Any, prefix: str):
pass
"""将 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})"
@@ -1377,6 +1454,17 @@ class PowerManager(Stub):
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:
@@ -1389,16 +1477,29 @@ class PowerManager(Stub):
return False
def reboot(self, reason: Optional[str] = None):
pass
"""重启设备。桌面端尝试执行 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):
pass
"""使设备进入睡眠。桌面端无实际效果。"""
self._is_interactive = False
sys.stderr.write(f"[PowerManager] goToSleep at {time_ms}\n")
def wakeUp(self, time_ms: int):
pass
"""唤醒设备。桌面端无实际效果。"""
self._is_interactive = True
sys.stderr.write(f"[PowerManager] wakeUp at {time_ms}\n")
def userActivity(self, when: int, no_change_lights: bool = False):
pass
"""报告用户活动事件。桌面端无实际效果。"""
self._is_interactive = True
class PowerManagerWakeLock(Stub):
@@ -1448,20 +1549,101 @@ class Vibrator(Stub):
def __init__(self):
super().__init__()
self._ff_device: Optional[str] = None
self._vibrating = False
def vibrate(self, milliseconds: int):
sys.stderr.write(f"[Vibrator] vibrate: {milliseconds}ms\n")
@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, pattern: List[int], repeat: int):
sys.stderr.write(f"[Vibrator] vibrate pattern: {pattern}, repeat={repeat}\n")
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 True
"""检查是否有振动器硬件。"""
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]:
@@ -1474,20 +1656,41 @@ class Vibrator(Stub):
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":
return VibratorEffect(0)
"""创建单次振动效果。"""
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":
return VibratorEffect(0)
"""创建波形振动效果。"""
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)
@@ -1755,18 +1958,38 @@ class TokenWatcher(Stub):
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):
pass
"""获取令牌。"""
self._tokens[id(token)] = tag
if len(self._tokens) == 1:
self.acquired()
def release(self, token: Any):
pass
"""释放令牌。"""
self._tokens.pop(id(token), None)
if not self._tokens:
self.released()
def isAcquired(self) -> bool:
return False
"""检查是否有令牌被持有。"""
return len(self._tokens) > 0
def acquired(self):
"""当第一个令牌被获取时调用。子类可重写。"""
pass
def released(self):
"""当所有令牌被释放时调用。子类可重写。"""
pass
def dump(self):
pass
"""转储令牌信息。"""
for tid, tag in self._tokens.items():
sys.stderr.write(f"[TokenWatcher] {self._tag}: token={tid}, tag={tag}\n")
class WorkSource(Stub):
@@ -1917,18 +2140,72 @@ class FileObserver(Stub):
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):
pass
"""清理资源。"""
self.stopWatching()
class FileUtils(Stub):
@@ -1985,7 +2262,35 @@ class FileUtils(Stub):
@staticmethod
def deleteOlderFiles(dir_: Any, min_count: int, min_age_ms: int):
pass
"""删除目录中较旧的文件,但保留至少 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):
+101 -4
View File
@@ -1242,8 +1242,13 @@ class ViewGroup(View, ViewParent, ViewManager):
# ─── child management ──────────────────────────────────────
def addView(self, child: View, index: int = -1,
def addView(self, child: View, index: Optional[int] = -1,
params: Optional["ViewGroup.LayoutParams"] = None):
"""addView — 支持 addView(child), addView(child, params), addView(child, index), addView(child, index, params)"""
# 智能 参数分派:第二个参数可能是 index(int) 或 params(LayoutParams)
if isinstance(index, ViewGroup.LayoutParams):
params = index
index = -1
if params is None:
params = ViewGroup.LayoutParams(ViewGroup.LayoutParams.WRAP_CONTENT,
ViewGroup.LayoutParams.WRAP_CONTENT)
@@ -1255,9 +1260,6 @@ class ViewGroup(View, ViewParent, ViewManager):
child.setLayoutParams(params)
child.onAttachedToWindow()
def addView(self, child: View, params: Optional["ViewGroup.LayoutParams"] = None):
self.addView(child, -1, params)
def removeView(self, view: View):
if view in self._children:
self._children.remove(view)
@@ -1360,6 +1362,32 @@ class ViewGroup(View, ViewParent, ViewManager):
def isMotionEventSplittingEnabled(self) -> bool:
return True
# ─── layout transition ────────────────────────────────────
def setLayoutTransition(self, transition: Any):
self._layout_transition = transition
def getLayoutTransition(self) -> Any:
return getattr(self, '_layout_transition', None)
# ─── focus / touch blocking ───────────────────────────────
def setTouchscreenBlocksFocus(self, blocks: bool):
self._touchscreen_blocks_focus = blocks
def isTouchscreenBlocksFocus(self) -> bool:
return getattr(self, '_touchscreen_blocks_focus', False)
# ─── child visible rect ───────────────────────────────────
def getChildVisibleRect(self, child: View, r: Any, offset: Any) -> bool:
if r is not None and hasattr(r, 'set'):
r.set(child.getLeft(), child.getTop(),
child.getRight(), child.getBottom())
if offset is not None and hasattr(offset, 'set'):
offset.set(0, 0)
return True
def setTransitionGroup(self, is_transition_group: bool):
pass
@@ -1809,6 +1837,75 @@ class MotionEvent(Stub):
def getHistoricalSize(self, pos: int, pointer: int = 0) -> float:
return 0.5
# ─── historical coordinates ───────────────────────────────
def getHistorySize(self) -> int:
return len(getattr(self, '_history', []))
def getHistoricalX(self, pointer_index: int = 0, pos: int = 0) -> float:
history = getattr(self, '_history', [])
if 0 <= pos < len(history):
return history[pos].get('x', self.getX(pointer_index))
return self.getX(pointer_index)
def getHistoricalY(self, pointer_index: int = 0, pos: int = 0) -> float:
history = getattr(self, '_history', [])
if 0 <= pos < len(history):
return history[pos].get('y', self.getY(pointer_index))
return self.getY(pointer_index)
def getHistoricalPressure(self, pointer_index: int = 0, pos: int = 0) -> float:
history = getattr(self, '_history', [])
if 0 <= pos < len(history):
return history[pos].get('pressure', 1.0)
return 1.0
def getHistoricalEventTime(self, pos: int = 0) -> int:
history = getattr(self, '_history', [])
if 0 <= pos < len(history):
return history[pos].get('event_time', self._event_time)
return self._event_time
# ─── edge flags ───────────────────────────────────────────
def getEdgeFlags(self) -> int:
return getattr(self, '_edge_flags', 0)
def setEdgeFlags(self, edge_flags: int):
self._edge_flags = edge_flags
# ─── action mutation ──────────────────────────────────────
def setAction(self, action: int):
self._action = action
# ─── batch ────────────────────────────────────────────────
def addBatch(self, event_time: int, x: float, y: float,
pressure: float = 1.0, size: float = 0.5,
meta_state: int = 0):
if not hasattr(self, '_history'):
self._history = []
self._history.append({
'event_time': self._event_time,
'x': self._x,
'y': self._y,
'pressure': self.getPressure(0),
'size': self.getSize(0),
'meta_state': self._meta_state,
})
self._x = x
self._y = y
self._raw_x = x
self._raw_y = y
self._event_time = event_time
self._meta_state = meta_state
if self._pointers:
self._pointers[0]['x'] = x
self._pointers[0]['y'] = y
self._pointers[0]['pressure'] = pressure
self._pointers[0]['size'] = size
# ═══════════════════════════════════════════════════════════════
# KeyEvent
+14 -8
View File
@@ -14,6 +14,8 @@ from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union
from .._core import Stub, AndroidOnlyError
from ..java.lang import RuntimeException
from .view import KeyEvent as _BaseKeyEvent
from .view import ViewGroup as _BaseViewGroup
from .view import View as _BaseView
# ═══════════════════════════════════════════════════════════════
@@ -700,8 +702,8 @@ class ViewParent(Stub):
# ViewGroup (扩展)
# ═══════════════════════════════════════════════════════════════
class ViewGroup(Stub):
"""ViewGroup — 视图组扩展"""
class ViewGroup(_BaseViewGroup):
"""ViewGroup — 视图组扩展 (继承 view.ViewGroup 的所有方法)"""
CLIP_TO_PADDING_MASK = 0
FOCUS_AFTER = 1
@@ -782,15 +784,19 @@ class ViewGroup(Stub):
f"margins=({self.leftMargin},{self.topMargin},"
f"{self.rightMargin},{self.bottomMargin}))")
def __init__(self):
super().__init__()
self._children: List[Any] = []
self._clip_children = True
self._clip_to_padding = True
def __init__(self, context=None):
super().__init__(context)
if not hasattr(self, '_children'):
self._children: List[Any] = []
if not hasattr(self, '_clip_children'):
self._clip_children = True
if not hasattr(self, '_clip_to_padding'):
self._clip_to_padding = True
self._layout_mode = self.LAYOUT_MODE_CLIP_BOUNDS
self._persistent_caching = self.PERSISTENT_NO_CACHE
self._touch_mode = False
self._motion_event_splitting = True
if not hasattr(self, '_motion_event_splitting'):
self._motion_event_splitting = True
self._transient_child: Optional[Any] = None
self._add_stateset_focus = False
self._child_drawable_state: Optional[Any] = None
+70
View File
@@ -365,6 +365,39 @@ class TextView(View):
def setPadding(self, left: int, top: int, right: int, bottom: int):
super().setPadding(left, top, right, bottom)
def setPaddingRelative(self, start: int, top: int, end: int, bottom: int):
self._padding_left = start
self._padding_top = top
self._padding_right = end
self._padding_bottom = bottom
# ─── text color getter ───────────────────────────────────
def getTextColor(self) -> int:
return self._text_color
# ─── selection action mode ───────────────────────────────
def setCustomSelectionActionModeCallback(self, callback: Any):
self._custom_selection_action_mode_callback = callback
def getCustomSelectionActionModeCallback(self) -> Any:
return getattr(self, '_custom_selection_action_mode_callback', None)
# ─── break / justification strategy ──────────────────────
def setBreakStrategy(self, break_strategy: int):
self._break_strategy = break_strategy
def getBreakStrategy(self) -> int:
return getattr(self, '_break_strategy', 0)
def setJustificationMode(self, justification_mode: int):
self._justification_mode = justification_mode
def getJustificationMode(self) -> int:
return getattr(self, '_justification_mode', 0)
def onMeasure(self, width_measure_spec: int, height_measure_spec: int):
from .view import MeasureSpec
w = MeasureSpec.getSize(width_measure_spec)
@@ -456,6 +489,12 @@ class CheckBox(Button):
def toggle(self):
self._checked = not self._checked
def setOnCheckedChangeListener(self, listener: Optional[Callable]):
self._on_checked_change_listener = listener
def getOnCheckedChangeListener(self) -> Optional[Callable]:
return getattr(self, '_on_checked_change_listener', None)
def performClick(self) -> bool:
self.toggle()
return super().performClick()
@@ -607,6 +646,22 @@ class Switch(Button):
def getTextOff(self) -> str:
return self._text_off
# ─── checked change listener ─────────────────────────────
def setOnCheckedChangeListener(self, listener: Optional[Callable]):
self._on_checked_change_listener = listener
def getOnCheckedChangeListener(self) -> Optional[Callable]:
return getattr(self, '_on_checked_change_listener', None)
# ─── switch text appearance ──────────────────────────────
def setSwitchTextAppearance(self, context: Any, resid: int):
self._switch_text_appearance = resid
def getSwitchTextAppearance(self) -> int:
return getattr(self, '_switch_text_appearance', 0)
def performClick(self) -> bool:
self.toggle()
return super().performClick()
@@ -694,6 +749,17 @@ class ImageView(View):
def getCropToPadding(self) -> bool:
return self._crop_to_padding
# ─── color filter ─────────────────────────────────────────
def setColorFilter(self, color_filter: Any):
self._color_filter = color_filter
def getColorFilter(self) -> Any:
return getattr(self, '_color_filter', None)
def clearColorFilter(self):
self._color_filter = None
def setBaseline(self, baseline: int):
self._baseline = baseline
@@ -1116,6 +1182,10 @@ class ListView(AdapterView):
def getCheckItemIds(self) -> List[int]:
return list(self._checked_id_positions.keys())
def clearChoices(self):
self._checked_item_positions.clear()
self._checked_id_positions.clear()
class GridView(AdapterView):
"""GridView — 网格视图"""