diff --git a/android_pyapi/__init__.py b/android_pyapi/__init__.py index b0d194d..f9d9b3b 100644 --- a/android_pyapi/__init__.py +++ b/android_pyapi/__init__.py @@ -503,6 +503,7 @@ __all__ = [ "UwbManager", "View", "ViewGroup", "VirtualDisplay", + "Vibrator", "VibratorEffect", "VpnManager", "VpnService", "WebView", "WifiAwareManager", @@ -514,6 +515,7 @@ __all__ = [ "InputStreamReader", "OutputStreamWriter", "URL", "HttpURLConnection", "Socket", "ServerSocket", "ArrayList", "HashMap", "HashSet", "LinkedList", + "TreeMap", "LinkedHashMap", "PriorityQueue", "Collections", "Arrays", "Calendar", "Date", "Random", "UUID", "Scanner", "Timer", "TimerTask", "Pattern", "Matcher", diff --git a/android_pyapi/android/content.py b/android_pyapi/android/content.py index 18bc8ac..c3d3df7 100644 --- a/android_pyapi/android/content.py +++ b/android_pyapi/android/content.py @@ -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 diff --git a/android_pyapi/android/content_res.py b/android_pyapi/android/content_res.py index b02baee..857b0f2 100644 --- a/android_pyapi/android/content_res.py +++ b/android_pyapi/android/content_res.py @@ -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: diff --git a/android_pyapi/android/graphics.py b/android_pyapi/android/graphics.py index 16ba1ba..86e4f98 100644 --- a/android_pyapi/android/graphics.py +++ b/android_pyapi/android/graphics.py @@ -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})" diff --git a/android_pyapi/android/hardware.py b/android_pyapi/android/hardware.py index d1f2e24..c0f54c6 100644 --- a/android_pyapi/android/hardware.py +++ b/android_pyapi/android/hardware.py @@ -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) # ═══════════════════════════════════════════════════════════════ diff --git a/android_pyapi/android/hardware_sensor_ext.py b/android_pyapi/android/hardware_sensor_ext.py index 72313b9..e726299 100644 --- a/android_pyapi/android/hardware_sensor_ext.py +++ b/android_pyapi/android/hardware_sensor_ext.py @@ -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 diff --git a/android_pyapi/android/location.py b/android_pyapi/android/location.py index 95af440..fa791f7 100644 --- a/android_pyapi/android/location.py +++ b/android_pyapi/android/location.py @@ -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, diff --git a/android_pyapi/android/os.py b/android_pyapi/android/os.py index b1403a2..da6ab4d 100644 --- a/android_pyapi/android/os.py +++ b/android_pyapi/android/os.py @@ -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): diff --git a/android_pyapi/android/view.py b/android_pyapi/android/view.py index 2137806..5e0b481 100644 --- a/android_pyapi/android/view.py +++ b/android_pyapi/android/view.py @@ -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 diff --git a/android_pyapi/android/view_ext.py b/android_pyapi/android/view_ext.py index 7310879..1021d50 100644 --- a/android_pyapi/android/view_ext.py +++ b/android_pyapi/android/view_ext.py @@ -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 diff --git a/android_pyapi/android/widget.py b/android_pyapi/android/widget.py index 15bedc4..6d1cbaa 100644 --- a/android_pyapi/android/widget.py +++ b/android_pyapi/android/widget.py @@ -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 — 网格视图""" diff --git a/android_pyapi/java/io.py b/android_pyapi/java/io.py index a10fc02..02974ff 100644 --- a/android_pyapi/java/io.py +++ b/android_pyapi/java/io.py @@ -745,14 +745,48 @@ class BufferedInputStream(FilterInputStream): return total if total > 0 else -1 def _fill(self): - data = self._in.read(bytearray(self._buf_size), 0, self._buf_size) - if data is None or data == -1: + tmp = bytearray(self._buf_size) + n = self._in.read(tmp, 0, self._buf_size) + if n is None or n == -1: return False - n = data if isinstance(data, int) else len(data) - self._buffer = bytearray(self._in._buffer if hasattr(self._in, '_buffer') else data) if n > 0 else bytearray() + if isinstance(n, (bytes, bytearray)): + # defensive: underlying stream returned raw data instead of a count + tmp = bytearray(n) + n = len(tmp) + if not isinstance(n, int) or n <= 0: + return False + self._buffer = bytearray(tmp[:n]) self._pos = 0 self._count = n - return n > 0 + return True + + def available(self): + buf_avail = self._count - self._pos + if buf_avail < 0: + buf_avail = 0 + try: + underlying = self._in.available() + if underlying and underlying > 0: + return buf_avail + underlying + except Exception: + pass + return buf_avail + + def mark(self, readlimit): + self._mark_pos = self._pos + self._mark_buffer = bytes(self._buffer) + self._mark_count = self._count + self._mark_limit = readlimit + + def reset(self): + if not hasattr(self, '_mark_pos'): + raise IOException("Resetting to invalid mark") + self._buffer = bytearray(self._mark_buffer) + self._count = self._mark_count + self._pos = self._mark_pos + + def markSupported(self): + return True # ============================================================ @@ -1034,6 +1068,26 @@ class BufferedWriter(Closeable, Flushable): def newLine(self): self._buffer.append("\n") + def append(self, csq=None, start=None, end=None): + if csq is None: + return self + if isinstance(csq, int): + self._buffer.append(chr(csq)) + else: + s = str(csq) + if start is not None and end is not None: + s = s[start:end] + elif start is not None: + s = s[start:] + self._buffer.append(s) + if sum(len(x) for x in self._buffer) >= self._buf_size: + self.flush() + return self + + def check(self): + if getattr(self._writer, '_closed', False): + raise IOException("Stream closed") + def flush(self): if self._buffer: data = "".join(self._buffer) @@ -1171,6 +1225,20 @@ class PrintWriter(Closeable, Flushable): else: self._write_str(str(s)) + def append(self, csq=None, start=None, end=None): + if csq is None: + csq = "null" + if isinstance(csq, int): + self._write_str(chr(csq)) + else: + s = str(csq) + if start is not None and end is not None: + s = s[start:end] + elif start is not None: + s = s[start:] + self._write_str(s) + return self + def format(self, fmt, *args): s = str(fmt) % args if args else str(fmt) self._write_str(s) @@ -1256,6 +1324,9 @@ class ByteArrayInputStream(Closeable): def reset(self): self._pos = self._mark + def markSupported(self): + return True + # ============================================================ # ByteArrayOutputStream @@ -1283,6 +1354,11 @@ class ByteArrayOutputStream(Closeable, Flushable): def toByteArray(self): return bytes(self._buf) + def writeTo(self, out): + if self._closed: + raise IOException("Stream closed") + out.write(bytes(self._buf)) + def size(self): return len(self._buf) @@ -1730,37 +1806,104 @@ class PushbackInputStream(FilterInputStream): # SequenceInputStream # ============================================================ +def _is_enumeration(obj) -> bool: + """True if obj really exposes hasMoreElements/nextElement (not via Stub).""" + for cls in type(obj).__mro__: + if ('hasMoreElements' in cls.__dict__ + and 'nextElement' in cls.__dict__): + return True + return False + + class SequenceInputStream(Closeable): - def __init__(self, streams): + def __init__(self, *args): super().__init__() - if hasattr(streams, 'hasMoreElements'): - self._streams = [] - while streams.hasMoreElements(): - self._streams.append(streams.nextElement()) - else: - self._streams = list(streams) + self._streams = [] self._current_idx = 0 + self._closed = False + if len(args) == 0: + return + if len(args) == 1: + streams = args[0] + if isinstance(streams, (list, tuple)): + self._streams = list(streams) + elif _is_enumeration(streams): + while streams.hasMoreElements(): + self._streams.append(streams.nextElement()) + else: + # single stream + self._streams = [streams] + else: + # SequenceInputStream(s1, s2, ...) + self._streams = list(args) def _current(self): - if self._current_idx < len(self._streams): - return self._streams[self._current_idx] + while self._current_idx < len(self._streams): + inp = self._streams[self._current_idx] + if inp is None: + self._current_idx += 1 + continue + return inp return None - def read(self, b=None, off=None, length=None): - while self._current(): - inp = self._current() - if b is None: - result = inp.read() - else: - result = inp.read(b, off, length) - if result != -1: - return result + def _advance(self): + if self._current_idx < len(self._streams): + cur = self._streams[self._current_idx] + if cur is not None: + try: + cur.close() + except Exception: + pass self._current_idx += 1 - return -1 + + def read(self, b=None, off=None, length=None): + if self._closed: + raise IOException("Stream closed") + if b is None: + while True: + inp = self._current() + if inp is None: + return -1 + result = inp.read() + if result != -1: + return result + self._advance() + if off is None: + off = 0 + if length is None: + length = len(b) + total = 0 + while total < length: + inp = self._current() + if inp is None: + break + n = inp.read(b, off + total, length - total) + if n is None or n == -1: + self._advance() + continue + if n == 0: + self._advance() + continue + total += n + return total if total > 0 else -1 + + def available(self): + inp = self._current() + if inp is None: + return 0 + try: + return inp.available() + except Exception: + return 0 def close(self): + self._closed = True for s in self._streams: - s.close() + if s is not None: + try: + s.close() + except Exception: + pass # ============================================================ diff --git a/android_pyapi/java/util.py b/android_pyapi/java/util.py index affe569..dd61244 100644 --- a/android_pyapi/java/util.py +++ b/android_pyapi/java/util.py @@ -148,6 +148,240 @@ class Enumeration(Stub): return Iterator(self._it) +# ============================================================ +# Stream / Spliterator helpers (lightweight, stdlib-backed) +# ============================================================ + +class _Spliterator(Stub): + """Lightweight java.util.Spliterator backed by a Python sequence.""" + ORDERED = 0x00000010 + DISTINCT = 0x00000001 + SORTED = 0x00000004 + SIZED = 0x00000040 + NONNULL = 0x00000100 + IMMUTABLE = 0x00000400 + CONCURRENT = 0x00001000 + SUBSIZED = 0x00004000 + + def __init__(self, data, characteristics: int = 0): + super().__init__() + self._data = list(data) + self._index = 0 + self._characteristics = (characteristics + | _Spliterator.SIZED + | _Spliterator.SUBSIZED) + + def tryAdvance(self, action) -> bool: + if self._index < len(self._data): + action(self._data[self._index]) + self._index += 1 + return True + return False + + def forEachRemaining(self, action) -> None: + while self._index < len(self._data): + action(self._data[self._index]) + self._index += 1 + + def trySplit(self): + mid = (self._index + len(self._data)) // 2 + if self._index >= mid: + return None + split = _Spliterator(self._data[self._index:mid], + self._characteristics) + self._index = mid + return split + + def estimateSize(self) -> int: + return len(self._data) - self._index + + def getExactSizeIfKnown(self) -> int: + return self.estimateSize() + + def characteristics(self) -> int: + return self._characteristics + + def hasCharacteristics(self, characteristics: int) -> bool: + return (self._characteristics & characteristics) == characteristics + + def iterator(self) -> Iterator: + return Iterator(self._data[self._index:]) + + +class _Stream(Stub): + """Lightweight java.util.stream.Stream wrapper over a Python list.""" + def __init__(self, data): + super().__init__() + self._data = list(data) + self._parallel = False + + def iterator(self) -> Iterator: + return Iterator(self._data) + + def spliterator(self) -> _Spliterator: + return _Spliterator(self._data) + + def isParallel(self) -> bool: + return self._parallel + + def parallel(self) -> '_Stream': + self._parallel = True + return self + + def sequential(self) -> '_Stream': + self._parallel = False + return self + + def unordered(self) -> '_Stream': + return self + + def filter(self, predicate) -> '_Stream': + return _Stream([x for x in self._data if predicate(x)]) + + def map(self, mapper) -> '_Stream': + return _Stream([mapper(x) for x in self._data]) + + def mapToInt(self, mapper) -> '_Stream': + return _Stream([mapper(x) for x in self._data]) + + def mapToLong(self, mapper) -> '_Stream': + return _Stream([mapper(x) for x in self._data]) + + def mapToDouble(self, mapper) -> '_Stream': + return _Stream([mapper(x) for x in self._data]) + + def flatMap(self, mapper) -> '_Stream': + result = [] + for x in self._data: + part = mapper(x) + if isinstance(part, (list, tuple)): + result.extend(part) + else: + result.extend(list(part)) + return _Stream(result) + + def peek(self, action) -> '_Stream': + for x in self._data: + action(x) + return self + + def distinct(self) -> '_Stream': + seen = [] + seen_keys = set() + for x in self._data: + key = x if isinstance(x, (int, str, tuple, frozenset)) else id(x) + if key not in seen_keys: + seen_keys.add(key) + seen.append(x) + return _Stream(seen) + + def sorted(self, key=None, reverse: bool = False) -> '_Stream': + return _Stream(sorted(self._data, key=key, reverse=reverse)) + + def limit(self, maxSize: int) -> '_Stream': + return _Stream(self._data[:maxSize]) + + def skip(self, n: int) -> '_Stream': + return _Stream(self._data[n:]) + + def forEach(self, action) -> None: + for x in self._data: + action(x) + + def forEachOrdered(self, action) -> None: + for x in self._data: + action(x) + + def toArray(self, generator=None): + if generator is None: + return list(self._data) + return list(self._data) + + def reduce(self, *args): + if len(args) == 1: + accumulator = args[0] + it = iter(self._data) + try: + acc = next(it) + except StopIteration: + return Optional.empty() + for x in it: + acc = accumulator(acc, x) + return Optional.of(acc) + elif len(args) == 2: + identity, accumulator = args + acc = identity + for x in self._data: + acc = accumulator(acc, x) + return acc + else: + identity, accumulator, combiner = args + acc = identity + for x in self._data: + acc = accumulator(acc, x) + return acc + + def collect(self, *args): + if not args: + return list(self._data) + if len(args) == 1: + collector = args[0] + supplier = getattr(collector, 'supplier', None) + accumulator = getattr(collector, 'accumulator', None) + finisher = getattr(collector, 'finisher', None) + if supplier is not None and accumulator is not None: + container = supplier() if callable(supplier) else supplier + for x in self._data: + if callable(accumulator): + accumulator(container, x) + if finisher is not None and callable(finisher): + return finisher(container) + return container + if callable(collector): + return collector(list(self._data)) + return list(self._data) + supplier, accumulator, combiner = args[0], args[1], args[2] + container = supplier() if callable(supplier) else supplier + for x in self._data: + if callable(accumulator): + accumulator(container, x) + return container + + def count(self) -> int: + return len(self._data) + + def findFirst(self): + return Optional.of(self._data[0]) if self._data else Optional.empty() + + def findAny(self): + return self.findFirst() + + def anyMatch(self, predicate) -> bool: + return any(predicate(x) for x in self._data) + + def allMatch(self, predicate) -> bool: + return all(predicate(x) for x in self._data) + + def noneMatch(self, predicate) -> bool: + return not any(predicate(x) for x in self._data) + + def min(self, key=None): + if not self._data: + return Optional.empty() + return Optional.of(min(self._data, key=key) if key else min(self._data)) + + def max(self, key=None): + if not self._data: + return Optional.empty() + return Optional.of(max(self._data, key=key) if key else max(self._data)) + + def onClose(self, close_handler) -> '_Stream': + return self + + def close(self) -> None: + pass + + # ============================================================ # ArrayList # ============================================================ @@ -294,6 +528,12 @@ class ArrayList(Stub): def replaceAll(self, operator) -> None: self._data = [operator(x) for x in self._data] + def spliterator(self) -> _Spliterator: + return _Spliterator(self._data, _Spliterator.ORDERED) + + def stream(self) -> _Stream: + return _Stream(self._data) + def equals(self, other) -> bool: if isinstance(other, ArrayList): return self._data == other._data @@ -789,6 +1029,10 @@ class LinkedHashMap(HashMap): key = next(iter(self._data)) return (key, self._data[key]) + def clone(self) -> 'LinkedHashMap': + result = LinkedHashMap(self._data, 0.75, self._access_order) + return result + # ============================================================ # TreeMap @@ -993,6 +1237,48 @@ class TreeMap(Stub): return True return False + def replaceAll(self, function) -> None: + for k in list(self._data.keys()): + self._data[k] = function(k, self._data[k]) + + def compute(self, key, remapping_function): + old = self._data.get(key) + new_val = remapping_function(key, old) + if new_val is not None: + self._data[key] = new_val + return new_val + self._data.pop(key, None) + return None + + def computeIfAbsent(self, key, mapping_function): + if key not in self._data: + new_val = mapping_function(key) + if new_val is not None: + self._data[key] = new_val + return new_val + return self._data.get(key) + + def computeIfPresent(self, key, remapping_function): + if key in self._data: + new_val = remapping_function(key, self._data[key]) + if new_val is not None: + self._data[key] = new_val + return new_val + self._data.pop(key, None) + return None + + def merge(self, key, value, remapping_function): + old = self._data.get(key) + if old is None: + self._data[key] = value + return value + new_val = remapping_function(old, value) + if new_val is not None: + self._data[key] = new_val + return new_val + self._data.pop(key, None) + return None + def size(self) -> int: return len(self._data) @@ -1140,6 +1426,19 @@ class HashSet(Stub): for item in list(self._data): action(item) + def removeIf(self, predicate) -> bool: + old_size = len(self._data) + to_remove = [x for x in self._data if predicate(x)] + for x in to_remove: + self._data.discard(x) + return len(self._data) != old_size + + def spliterator(self) -> _Spliterator: + return _Spliterator(self._data, _Spliterator.DISTINCT) + + def stream(self) -> _Stream: + return _Stream(self._data) + def equals(self, other) -> bool: if isinstance(other, HashSet): return self._data == other._data @@ -3445,6 +3744,18 @@ class PriorityQueue(Stub): for item in list(self._heap): action(item) + def spliterator(self) -> _Spliterator: + if self._comparator: + items = [v[1] for v in self._heap] + else: + items = list(self._heap) + return _Spliterator(items) + + def clone(self) -> 'PriorityQueue': + result = PriorityQueue(comparator=self._comparator) + result._heap = list(self._heap) + return result + def toString(self) -> str: items = ", ".join(str(x) for x in self._heap) return f"[{items}]"