""" Android Location Geofence 组件 — Geofence, GeofencingEvent, GeofenceRequest 等 方法签名完全对齐 Android SDK """ from __future__ import annotations import math import time import threading from typing import Any, Callable, Dict, List, Optional, Set, Tuple from .._core import Stub, AndroidOnlyError from ..java.lang import RuntimeException # ═══════════════════════════════════════════════════════════════ # Geofence 常量 # ═══════════════════════════════════════════════════════════════ GEOFENCE_TRANSITION_ENTER = 1 GEOFENCE_TRANSITION_EXIT = 2 GEOFENCE_TRANSITION_DWELL = 4 GEOFENCE_ENTER = GEOFENCE_TRANSITION_ENTER GEOFENCE_EXIT = GEOFENCE_TRANSITION_EXIT GEOFENCE_DWELL = GEOFENCE_TRANSITION_DWELL GEOFENCE_RADIUS_INACCURACY = 1 GEOFENCE_TRANSITION_UNKNOWN = -1 MONITORING_TYPE_PASSIVE = 0 MONITORING_TYPE_ACTIVE = 1 MONITORING_TYPE_PASSIVE_INTERVAL = 2 MONITORING_TYPE_LOCATION = 3 MONITORING_TYPE_FUSED_LOCATION = 4 MONITORING_TYPE_GEOFENCE = MONITORING_TYPE_ACTIVE GEOFENCE_REMOVE_BY_LIST = 1 GEOFENCE_REMOVE_BY_SIGNATURE = 2 GEOFENCE_REMOVE_ALL = 3 MONITORING_TYPE_GEOFENCE_ENTER = 1 MONITORING_TYPE_GEOFENCE_EXIT = 2 MONITORING_TYPE_GEOFENCE_DWELL = 3 GEOFENCE_AVAILABILITY_UNKNOWN = 0 GEOFENCE_AVAILABILITY_AVAILABLE = 1 GEOFENCE_AVAILABILITY_UNAVAILABLE = 2 GEOFENCE_ENTER_INIT = 0 GEOFENCE_ENTER_FINAL = 1 GEOFENCE_EXIT_INIT = 2 GEOFENCE_EXIT_FINAL = 3 # ═══════════════════════════════════════════════════════════════ # Geofence # ═══════════════════════════════════════════════════════════════ class Geofence(Stub): """Geofence — 地理围栏""" def __init__(self, request_id: str = "", latitude: float = 0.0, longitude: float = 0.0, radius: float = 100.0, expiration_duration: int = -1, transition_types: int = 0): super().__init__() self._request_id = request_id self._latitude = latitude self._longitude = longitude self._radius = max(0.0, radius) self._expiration = expiration_duration self._transitions = transition_types self._monitoring_type = MONITORING_TYPE_PASSIVE self._is_monitoring = False def getRequestId(self) -> str: return self._request_id def getLatitude(self) -> float: return self._latitude def getLongitude(self) -> float: return self._longitude def getRadius(self) -> float: return self._radius def getExpiration(self) -> int: return self._expiration def getTransitions(self) -> int: return self._transitions def setMonitoringType(self, monitor_type: int) -> None: self._monitoring_type = monitor_type def getMonitoringType(self) -> int: return self._monitoring_type def isExpired(self) -> bool: if self._expiration < 0: return False return int(time.time() * 1000) > self._expiration def isMonitoring(self) -> bool: return self._is_monitoring def setMonitoring(self, monitoring: bool) -> None: self._is_monitoring = monitoring def containsLocation(self, latitude: float, longitude: float) -> bool: """检查指定位置是否在此地理围栏内""" dx = latitude - self._latitude dy = longitude - self._longitude distance = math.sqrt(dx * dx + dy * dy) * 111319.9 return distance <= self._radius def toString(self) -> str: return (f"Geofence(id={self._request_id}, " f"lat={self._latitude:.4f}, lng={self._longitude:.4f}, " f"radius={self._radius:.1f}m)") class GeofenceBuilder(Stub): """GeofenceBuilder — 地理围栏构建器""" def __init__(self): super().__init__() self._request_id: str = "" self._latitude: float = 0.0 self._longitude: float = 0.0 self._radius: float = 100.0 self._expiration_duration: int = -1 self._transition_types: int = 0 self._monitoring_type: int = MONITORING_TYPE_PASSIVE self._loitering_delay: int = 0 self._notification_responsiveness: int = 0 self._last_transition: int = GEOFENCE_TRANSITION_UNKNOWN def setRequestId(self, request_id: str) -> "GeofenceBuilder": self._request_id = request_id return self def setLatitude(self, latitude: float) -> "GeofenceBuilder": self._latitude = latitude return self def setLongitude(self, longitude: float) -> "GeofenceBuilder": self._longitude = longitude return self def setRadius(self, radius: float) -> "GeofenceBuilder": self._radius = max(0.0, radius) return self def setExpirationDuration(self, duration_ms: int) -> "GeofenceBuilder": self._expiration_duration = duration_ms return self def setTransitionTypes(self, types: int) -> "GeofenceBuilder": self._transition_types = types return self def setMonitoringType(self, monitor_type: int) -> "GeofenceBuilder": self._monitoring_type = monitor_type return self def setLoiteringDelay(self, delay_ms: int) -> "GeofenceBuilder": self._loitering_delay = delay_ms return self def setNotificationResponsiveness(self, ms: int) -> "GeofenceBuilder": self._notification_responsiveness = ms return self def build(self) -> Geofence: geofence = Geofence( self._request_id, self._latitude, self._longitude, self._radius, self._expiration_duration, self._transition_types ) geofence.setMonitoringType(self._monitoring_type) return geofence def toString(self) -> str: return (f"GeofenceBuilder(id={self._request_id}, " f"lat={self._latitude:.4f}, lng={self._longitude:.4f})") # ═══════════════════════════════════════════════════════════════ # GeofencingEvent # ═══════════════════════════════════════════════════════════════ class GeofencingEvent(Stub): """GeofencingEvent — 地理围栏事件""" def __init__(self): super().__init__() self._triggering_geofences: List[Geofence] = [] self._triggering_locations: List[Tuple[float, float]] = [] self._transition_type: int = GEOFENCE_TRANSITION_UNKNOWN self._error_code: int = 0 self._initial_trigger: bool = False self._event_time: int = 0 def getTriggeringGeofences(self) -> List[Geofence]: return list(self._triggering_geofences) def setTriggeringGeofences(self, geofences: List[Geofence]) -> None: self._triggering_geofences = list(geofences) def getTriggeringLocations(self) -> List[Tuple[float, float]]: return list(self._triggering_locations) def addTriggeringLocation(self, latitude: float, longitude: float) -> None: self._triggering_locations.append((latitude, longitude)) def getTransitionType(self) -> int: return self._transition_type def setTransitionType(self, transition: int) -> None: self._transition_type = transition def getErrorCode(self) -> int: return self._error_code def setErrorCode(self, code: int) -> None: self._error_code = code def isInitialTrigger(self) -> bool: return self._initial_trigger def setInitialTrigger(self, initial: bool) -> None: self._initial_trigger = initial def getEventTime(self) -> int: return self._event_time def setEventTime(self, time_ms: int) -> None: self._event_time = time_ms def hasError(self) -> bool: return self._error_code != 0 def toString(self) -> str: types = {1: "ENTER", 2: "EXIT", 4: "DWELL"} return (f"GeofencingEvent(transition={types.get(self._transition_type, 'UNKNOWN')}, " f"geofences={len(self._triggering_geofences)}, " f"error={self._error_code})") # ═══════════════════════════════════════════════════════════════ # GeofenceRequest # ═══════════════════════════════════════════════════════════════ class GeofenceRequest(Stub): """GeofenceRequest — 地理围栏请求""" INITIAL_TRIGGER_ENTER = 1 INITIAL_TRIGGER_EXIT = 2 INITIAL_TRIGGER_DWELL = 4 def __init__(self): super().__init__() self._geofences: List[Geofence] = [] self._initial_triggers: int = 0 self._expiration_duration: int = -1 self._loitering_delay: int = 0 self._notification_responsiveness: int = 0 def getGeofences(self) -> List[Geofence]: return list(self._geofences) def addGeofence(self, geofence: Geofence) -> None: self._geofences.append(geofence) def setInitialTriggers(self, triggers: int) -> None: self._initial_triggers = triggers def getInitialTriggers(self) -> int: return self._initial_triggers def setExpirationDuration(self, duration: int) -> None: self._expiration_duration = duration def getExpirationDuration(self) -> int: return self._expiration_duration def setLoiteringDelay(self, delay: int) -> None: self._loitering_delay = delay def getLoiteringDelay(self) -> int: return self._loitering_delay def setNotificationResponsiveness(self, ms: int) -> None: self._notification_responsiveness = ms def getNotificationResponsiveness(self) -> int: return self._notification_responsiveness def toString(self) -> str: return (f"GeofenceRequest(geofences={len(self._geofences)}, " f"triggers={self._initial_triggers})") # ═══════════════════════════════════════════════════════════════ # GeofenceHardware # ═══════════════════════════════════════════════════════════════ class GeofenceHardware(Stub): """GeofenceHardware — 硬件地理围栏""" GEOFENCE_ENTERED = 0 GEOFENCE_EXITED = 1 GEOFENCE_UNCERTAIN = 2 MONITORING_TYPE_GEOFENCE = MONITORING_TYPE_GEOFENCE def __init__(self): super().__init__() self._geofences: Dict[int, Geofence] = {} self._next_id: int = 1 self._monitor_callback: Optional["GeofenceHardwareMonitorCallback"] = None self._lock = threading.Lock() def addGeofence(self, fence_id: int, geofence: Geofence) -> bool: with self._lock: if fence_id < 0: fence_id = self._next_id self._next_id += 1 self._geofences[fence_id] = geofence return True def removeGeofence(self, fence_id: int) -> bool: with self._lock: return self._geofences.pop(fence_id, None) is not None def pauseGeofence(self, fence_id: int) -> bool: return fence_id in self._geofences def resumeGeofence(self, fence_id: int, monitor_transition: int) -> bool: return fence_id in self._geofences def getMonitoringTypes(self) -> List[int]: return [MONITORING_TYPE_GEOFENCE] def getState(self, monitor_type: int) -> int: return GEOFENCE_AVAILABILITY_AVAILABLE def setMonitorCallback(self, callback: Optional["GeofenceHardwareMonitorCallback"]) -> None: self._monitor_callback = callback def toString(self) -> str: return f"GeofenceHardware(geofences={len(self._geofences)})" # ═══════════════════════════════════════════════════════════════ # GeofenceHardwareRequest / Float / Circle # ═══════════════════════════════════════════════════════════════ class GeofenceHardwareRequest(Stub): """GeofenceHardwareRequest — 硬件地理围栏请求""" GEOFENCE_TYPE_CIRCLE = 0 GEOFENCE_TYPE_FLOAT = 1 def __init__(self, latitude: float = 0.0, longitude: float = 0.0, radius: float = 100.0, geofence_type: int = GEOFENCE_TYPE_CIRCLE): super().__init__() self._latitude = latitude self._longitude = longitude self._radius = radius self._geofence_type = geofence_type self._last_transition: int = GEOFENCE_TRANSITION_UNKNOWN self._monitor_transition: int = GEOFENCE_TRANSITION_ENTER self._unknown_timer: int = 30000 self._monitor_notification: int = 5 self._source_technologies: int = 0 def getLatitude(self) -> float: return self._latitude def getLongitude(self) -> float: return self._longitude def getRadius(self) -> float: return self._radius def getGeofenceType(self) -> int: return self._geofence_type def setLastTransition(self, transition: int) -> None: self._last_transition = transition def getLastTransition(self) -> int: return self._last_transition def setMonitorTransitions(self, transitions: int) -> None: self._monitor_transition = transitions def getMonitorTransitions(self) -> int: return self._monitor_transition def setUnknownTimer(self, timer_ms: int) -> None: self._unknown_timer = timer_ms def getUnknownTimer(self) -> int: return self._unknown_timer def setNotificationResponsiveness(self, ms: int) -> None: self._monitor_notification = ms def getNotificationResponsiveness(self) -> int: return self._monitor_notification def setSourceTechnologies(self, technologies: int) -> None: self._source_technologies = technologies def getSourceTechnologies(self) -> int: return self._source_technologies def toString(self) -> str: return (f"GeofenceHardwareRequest(lat={self._latitude:.4f}, " f"lng={self._longitude:.4f}, r={self._radius:.1f})") class GeofenceHardwareRequestFloat(GeofenceHardwareRequest): """GeofenceHardwareRequestFloat — 浮点型硬件围栏请求""" def __init__(self, latitude: float = 0.0, longitude: float = 0.0, radius: float = 100.0): super().__init__(latitude, longitude, radius, GeofenceHardwareRequest.GEOFENCE_TYPE_FLOAT) def toString(self) -> str: return (f"GeofenceHardwareRequestFloat(lat={self._latitude:.4f}, " f"lng={self._longitude:.4f}, r={self._radius:.1f})") class GeofenceHardwareRequestCircle(GeofenceHardwareRequest): """GeofenceHardwareRequestCircle — 圆形硬件围栏请求""" def __init__(self, latitude: float = 0.0, longitude: float = 0.0, radius: float = 100.0): super().__init__(latitude, longitude, radius, GeofenceHardwareRequest.GEOFENCE_TYPE_CIRCLE) def toString(self) -> str: return (f"GeofenceHardwareRequestCircle(lat={self._latitude:.4f}, " f"lng={self._longitude:.4f}, r={self._radius:.1f})") # ═══════════════════════════════════════════════════════════════ # GeofenceHardwareMonitorCallback # ═══════════════════════════════════════════════════════════════ class GeofenceHardwareMonitorCallback(Stub): """GeofenceHardwareMonitorCallback — 硬件围栏监控回调""" def onMonitoringSystemChange(self, monitor_type: int, available: bool, location: Optional[Any]) -> None: pass def onGeofenceTransition(self, fence_id: int, transition: int, location: Any, timestamp: int, monitoring_type: int) -> None: pass def onGeofenceExpiration(self, fence_id: int, monitoring_type: int) -> None: pass # ═══════════════════════════════════════════════════════════════ # GeofenceManager # ═══════════════════════════════════════════════════════════════ class GeofenceManager(Stub): """GeofenceManager — 地理围栏管理器""" def __init__(self): super().__init__() self._registered_geofences: Dict[str, Geofence] = {} self._listeners: Dict[str, Callable] = {} self._lock = threading.Lock() self._monitoring = False def addGeofence(self, geofence: Geofence, listener: Callable) -> None: with self._lock: self._registered_geofences[geofence.getRequestId()] = geofence self._listeners[geofence.getRequestId()] = listener geofence.setMonitoring(True) def removeGeofence(self, request_id: str) -> None: with self._lock: geofence = self._registered_geofences.pop(request_id, None) self._listeners.pop(request_id, None) if geofence: geofence.setMonitoring(False) def removeAllGeofences(self) -> None: with self._lock: for geofence in self._registered_geofences.values(): geofence.setMonitoring(False) self._registered_geofences.clear() self._listeners.clear() def getRegisteredGeofences(self) -> List[Geofence]: with self._lock: return list(self._registered_geofences.values()) def isMonitoring(self) -> bool: return self._monitoring def setMonitoring(self, monitoring: bool) -> None: self._monitoring = monitoring def triggerTransition(self, geofence_id: str, transition: int, latitude: float, longitude: float) -> None: with self._lock: if geofence_id in self._listeners: event = GeofencingEvent() event.setTransitionType(transition) event.addTriggeringLocation(latitude, longitude) if geofence_id in self._registered_geofences: event.setTriggeringGeofences( [self._registered_geofences[geofence_id]] ) self._listeners[geofence_id](event) def toString(self) -> str: return (f"GeofenceManager(geofences={len(self._registered_geofences)}, " f"monitoring={self._monitoring})")