""" Android Location 组件 — Location, LocationManager, Geocoder 等 方法签名完全对齐 Android SDK """ import os import sys import math import threading from typing import Any, Dict, List, Optional, Set, Tuple, Union from datetime import datetime from .._core import ( Stub, AndroidOnlyError, android_context, resolve_android_path, to_android_path, _sdcard_root, _app_data_dir, _package_name, ) try: from .._linux_impl import IS_LINUX, LinuxLocation _USE_LINUX = IS_LINUX except ImportError: _USE_LINUX = False # ═══════════════════════════════════════════════════════════════ # Location # ═══════════════════════════════════════════════════════════════ class Location(Stub): """Location — 地理位置信息""" FORMAT_DEGREES = 0 FORMAT_MINUTES = 1 FORMAT_SECONDS = 2 def __init__(self, provider: str = ""): super().__init__() self._provider = provider self._latitude: float = 0.0 self._longitude: float = 0.0 self._accuracy: float = 0.0 self._time: int = 0 self._elapsed_realtime_nanos: int = 0 self._altitude: float = 0.0 self._speed: float = 0.0 self._bearing: float = 0.0 self._extras: Optional[Dict[str, Any]] = None def setLatitude(self, latitude: float): self._latitude = latitude def getLatitude(self) -> float: return self._latitude def setLongitude(self, longitude: float): self._longitude = longitude def getLongitude(self) -> float: return self._longitude def setAccuracy(self, accuracy: float): self._accuracy = accuracy def getAccuracy(self) -> float: return self._accuracy def setTime(self, time: int): self._time = time def getTime(self) -> int: return self._time def setElapsedRealtimeNanos(self, nanos: int): self._elapsed_realtime_nanos = nanos def getElapsedRealtimeNanos(self) -> int: return self._elapsed_realtime_nanos def setAltitude(self, altitude: float): self._altitude = altitude def getAltitude(self) -> float: return self._altitude def setSpeed(self, speed: float): self._speed = speed def getSpeed(self) -> float: return self._speed def setBearing(self, bearing: float): self._bearing = bearing def getBearing(self) -> float: return self._bearing def setProvider(self, provider: str): self._provider = provider def getProvider(self) -> str: return self._provider def distanceTo(self, dest: "Location") -> float: R = 6371000 lat1 = math.radians(self._latitude) lat2 = math.radians(dest._latitude) dlat = math.radians(dest._latitude - self._latitude) dlon = math.radians(dest._longitude - self._longitude) a = (math.sin(dlat / 2) ** 2 + math.cos(lat1) * math.cos(lat2) * math.sin(dlon / 2) ** 2) c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a)) return R * c def bearingTo(self, dest: "Location") -> float: lat1 = math.radians(self._latitude) lat2 = math.radians(dest._latitude) dlon = math.radians(dest._longitude - self._longitude) y = math.sin(dlon) * math.cos(lat2) x = (math.cos(lat1) * math.sin(lat2) - math.sin(lat1) * math.cos(lat2) * math.cos(dlon)) return (math.degrees(math.atan2(y, x)) + 360) % 360 @staticmethod def convert(coordinate: Union[str, float], output_type: int = FORMAT_DEGREES) -> Union[str, float]: if isinstance(coordinate, str): return Location._parse_coordinate(coordinate) return Location._format_coordinate(coordinate, output_type) @staticmethod def _parse_coordinate(coord_str: str) -> float: coord_str = coord_str.strip() try: return float(coord_str) except ValueError: pass parts = coord_str.replace(",", ".").split() if len(parts) >= 3: try: d = float(parts[0]) m = float(parts[1]) s = float(parts[2].rstrip("NSEW")) result = d + m / 60.0 + s / 3600.0 if coord_str.rstrip()[-1:] in ("S", "W"): result = -result return result except (ValueError, IndexError): return 0.0 return 0.0 @staticmethod def _format_coordinate(coord: float, output_type: int) -> str: if output_type == Location.FORMAT_DEGREES: return f"{coord:.6f}" elif output_type == Location.FORMAT_MINUTES: deg = int(coord) minutes = (coord - deg) * 60 return f"{deg}:{minutes:.3f}" else: deg = int(coord) minutes = (coord - deg) * 60 int_min = int(minutes) secs = (minutes - int_min) * 60 return f"{deg}:{int_min}:{secs:.2f}" def getExtras(self) -> Optional[Dict[str, Any]]: return self._extras def setExtras(self, extras: Dict[str, Any]): self._extras = extras def toString(self) -> str: return (f"Location[provider={self._provider}, " f"{self._latitude:.6f},{self._longitude:.6f}, " f"acc={self._accuracy:.1f}]") def __repr__(self): return self.toString() # ═══════════════════════════════════════════════════════════════ # Criteria # ═══════════════════════════════════════════════════════════════ class Criteria(Stub): """Criteria — 位置提供者筛选条件""" ACCURACY_FINE = 1 ACCURACY_COARSE = 2 POWER_LOW = 1 POWER_MEDIUM = 2 POWER_HIGH = 3 NO_REQUIREMENT = 0 def __init__(self): super().__init__() self._accuracy = self.NO_REQUIREMENT self._power_requirement = self.NO_REQUIREMENT self._altitude_required = False self._bearing_required = False self._speed_required = False self._cost_allowed = False def setAccuracy(self, accuracy: int): self._accuracy = accuracy def getAccuracy(self) -> int: return self._accuracy def setPowerRequirement(self, power: int): self._power_requirement = power def getPowerRequirement(self) -> int: return self._power_requirement def setAltitudeRequired(self, required: bool): self._altitude_required = required def isAltitudeRequired(self) -> bool: return self._altitude_required def setBearingRequired(self, required: bool): self._bearing_required = required def isBearingRequired(self) -> bool: return self._bearing_required def setSpeedRequired(self, required: bool): self._speed_required = required def isSpeedRequired(self) -> bool: return self._speed_required def setCostAllowed(self, allowed: bool): self._cost_allowed = allowed def isCostAllowed(self) -> bool: return self._cost_allowed def toString(self) -> str: return f"Criteria(acc={self._accuracy}, power={self._power_requirement})" # ═══════════════════════════════════════════════════════════════ # LocationProvider # ═══════════════════════════════════════════════════════════════ class LocationProvider(Stub): """LocationProvider — 位置提供者""" AVAILABLE = 2 TEMPORARILY_UNAVAILABLE = 1 OUT_OF_SERVICE = 0 def __init__(self, name: str = ""): super().__init__() self._name = name self._accuracy = Criteria.NO_REQUIREMENT self._power_requirement = Criteria.NO_REQUIREMENT self._has_monetary_cost = False self._requires_cell = False self._requires_network = False self._requires_satellite = False self._supports_altitude = False self._supports_bearing = False self._supports_speed = False def getName(self) -> str: return self._name def getAccuracy(self) -> int: return self._accuracy def getPowerRequirement(self) -> int: return self._power_requirement def hasMonetaryCost(self) -> bool: return self._has_monetary_cost def requiresCell(self) -> bool: return self._requires_cell def requiresNetwork(self) -> bool: return self._requires_network def requiresSatellite(self) -> bool: return self._requires_satellite def supportsAltitude(self) -> bool: return self._supports_altitude def supportsBearing(self) -> bool: return self._supports_bearing def supportsSpeed(self) -> bool: return self._supports_speed def getRequirements(self) -> int: req = 0 if self._requires_cell: req |= 1 if self._requires_network: req |= 2 if self._requires_satellite: req |= 4 return req def toString(self) -> str: return f"LocationProvider(name={self._name})" # ═══════════════════════════════════════════════════════════════ # LocationListener # ═══════════════════════════════════════════════════════════════ class LocationListener(Stub): """LocationListener — 位置变化监听器""" def onLocationChanged(self, location: Location): pass def onStatusChanged(self, provider: str, status: int, extras: Any): pass def onProviderEnabled(self, provider: str): pass def onProviderDisabled(self, provider: str): pass # ═══════════════════════════════════════════════════════════════ # GpsSatellite # ═══════════════════════════════════════════════════════════════ class GpsSatellite(Stub): """GpsSatellite — GPS 卫星信息""" def __init__(self): super().__init__() self._prn: int = 0 self._snr: float = 0.0 self._elevation: float = 0.0 self._azimuth: float = 0.0 self._has_almanac: bool = False self._has_ephemeris: bool = False self._used_in_fix: bool = False def getPrn(self) -> int: return self._prn def getSnr(self) -> float: return self._snr def getElevation(self) -> float: return self._elevation def getAzimuth(self) -> float: return self._azimuth def hasAlmanac(self) -> bool: return self._has_almanac def hasEphemeris(self) -> bool: return self._has_ephemeris def usedInFix(self) -> bool: return self._used_in_fix def toString(self) -> str: return f"GpsSatellite(prn={self._prn}, snr={self._snr})" # ═══════════════════════════════════════════════════════════════ # GpsStatus # ═══════════════════════════════════════════════════════════════ class GpsStatus(Stub): """GpsStatus — GPS 状态""" GPS_EVENT_STARTED = 1 GPS_EVENT_STOPPED = 2 GPS_EVENT_FIRST_FIX = 3 GPS_EVENT_SATELLITE_STATUS = 4 def __init__(self): super().__init__() self._satellites: List[GpsSatellite] = [] self._max_satellites: int = 32 self._time_to_first_fix: int = 0 def getSatellites(self) -> List[GpsSatellite]: return list(self._satellites) def getMaxSatellites(self) -> int: return self._max_satellites def getTimeToFirstFix(self) -> int: return self._time_to_first_fix def setTimeToFirstFix(self, ttf: int): self._time_to_first_fix = ttf class GpsStatusListener(Stub): """GpsStatusListener — GPS 状态监听器""" def onGpsStatusChanged(self, event: int): pass # ═══════════════════════════════════════════════════════════════ # GnssClock / GnssMeasurement / GnssMeasurementsEvent / GnssNavigationMessage # ═══════════════════════════════════════════════════════════════ class GnssClock(Stub): """GnssClock — GNSS 时钟""" def __init__(self): super().__init__() self._time_nanos: int = 0 self._time_uncertainty_nanos: float = 0.0 self._leap_second: int = 0 self._full_bias_nanos: int = 0 self._bias_nanos: float = 0.0 self._bias_uncertainty_nanos: float = 0.0 self._drift_nanos_per_sec: float = 0.0 self._drift_uncertainty_nanos_per_sec: float = 0.0 self._hardware_clock_discontinuity_count: int = 0 self._has_leap_second: bool = False self._has_full_bias_nanos: bool = False self._has_bias_nanos: bool = False self._has_bias_uncertainty_nanos: bool = False self._has_drift_nanos_per_sec: bool = False self._has_drift_uncertainty_nanos_per_sec: bool = False def getTimeNanos(self) -> int: return self._time_nanos def getTimeUncertaintyNanos(self) -> float: return self._time_uncertainty_nanos def getLeapSecond(self) -> int: return self._leap_second def getFullBiasNanos(self) -> int: return self._full_bias_nanos def getBiasNanos(self) -> float: return self._bias_nanos def getBiasUncertaintyNanos(self) -> float: return self._bias_uncertainty_nanos def getDriftNanosPerSec(self) -> float: return self._drift_nanos_per_sec def getDriftUncertaintyNanosPerSec(self) -> float: return self._drift_uncertainty_nanos_per_sec def getHardwareClockDiscontinuityCount(self) -> int: return self._hardware_clock_discontinuity_count class GnssMeasurement(Stub): """GnssMeasurement — GNSS 测量值""" STATE_UNKNOWN = 0 STATE_CODE_LOCK = 1 STATE_BIT_SYNC = 2 STATE_SUBFRAME_SYNC = 3 STATE_TOW_DECODED = 4 STATE_MSEC_AMBIGUOUS = 5 def __init__(self): super().__init__() self._svid: int = 0 self._constellation_type: int = 0 self._time_offset_nanos: float = 0.0 self._state: int = 0 self._received_gps_tow_nanos: int = 0 self._received_gps_tow_uncertainty_nanos: int = 0 self._carrier_frequency_hz: float = 0.0 self._carrier_phase: float = 0.0 self._carrier_phase_uncertainty: float = 0.0 self._multipath_indicator: int = 0 self._snr_in_db: float = 0.0 self._constellation_type_val: int = 0 def getSvid(self) -> int: return self._svid def getConstellationType(self) -> int: return self._constellation_type def getTimeOffsetNanos(self) -> float: return self._time_offset_nanos def getState(self) -> int: return self._state def getReceivedGpsTowNanos(self) -> int: return self._received_gps_tow_nanos def getReceivedGpsTowUncertaintyNanos(self) -> int: return self._received_gps_tow_uncertainty_nanos def getCarrierFrequencyHz(self) -> float: return self._carrier_frequency_hz def getCarrierPhase(self) -> float: return self._carrier_phase def getCarrierPhaseUncertainty(self) -> float: return self._carrier_phase_uncertainty def getMultipathIndicator(self) -> int: return self._multipath_indicator def getSnrInDb(self) -> float: return self._snr_in_db class GnssMeasurementsEvent(Stub): """GnssMeasurementsEvent — GNSS 测量事件""" def __init__(self, clock: Optional[GnssClock] = None, measurements: Optional[List[GnssMeasurement]] = None): super().__init__() self._clock = clock or GnssClock() self._measurements = measurements or [] def getClock(self) -> GnssClock: return self._clock def getMeasurements(self) -> List[GnssMeasurement]: return list(self._measurements) class GnssNavigationMessage(Stub): """GnssNavigationMessage — GNSS 导航消息""" STATUS_PARITY_PASSED = 1 STATUS_PARITY_REBUILT = 2 TYPE_GPS_L1CA = 256 TYPE_GPS_L2CA = 512 TYPE_GPS_L5 = 768 TYPE_GLO_L1CA = 1024 TYPE_GLO_L2CA = 1280 TYPE_BDS_D1 = 1792 TYPE_BDS_D2 = 2048 TYPE_GAL_I = 2560 TYPE_GAL_F = 2816 TYPE_QZS_L1CA = 3072 TYPE_QZS_L5 = 3328 def __init__(self): super().__init__() self._type: int = 0 self._svid: int = 0 self._message_id: int = 0 self._submessage_id: int = 0 self._data: bytes = b"" self._status: int = 0 def getType(self) -> int: return self._type def getSvid(self) -> int: return self._svid def getMessageId(self) -> int: return self._message_id def getSubmessageId(self) -> int: return self._submessage_id def getData(self) -> bytes: return self._data def getStatus(self) -> int: return self._status # ═══════════════════════════════════════════════════════════════ # GnssStatus # ═══════════════════════════════════════════════════════════════ class GnssStatus(Stub): """GnssStatus — GNSS 卫星状态""" GNSS_CONSTELLATION_GPS = 1 GNSS_CONSTELLATION_SBAS = 2 GNSS_CONSTELLATION_GLONASS = 3 CONSTELLATION_QZSS = 4 CONSTELLATION_BEIDOU = 5 CONSTELLATION_GALILEO = 6 CONSTELLATION_IRNSS = 7 def __init__(self): super().__init__() self._satellites: List[Dict[str, Any]] = [] def getSatelliteCount(self) -> int: return len(self._satellites) def getSvid(self, index: int) -> int: if 0 <= index < len(self._satellites): return self._satellites[index].get("svid", 0) return 0 def getConstellationType(self, index: int) -> int: if 0 <= index < len(self._satellites): return self._satellites[index].get("constellation", 0) return 0 def getCarrierFrequencyHz(self, index: int) -> float: if 0 <= index < len(self._satellites): return self._satellites[index].get("carrier_freq", 0.0) 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: if 0 <= index < len(self._satellites): return self._satellites[index].get("cn0", 0.0) return 0.0 def getAzimuthDegrees(self, index: int) -> float: if 0 <= index < len(self._satellites): return self._satellites[index].get("azimuth", 0.0) return 0.0 def getElevationDegrees(self, index: int) -> float: if 0 <= index < len(self._satellites): return self._satellites[index].get("elevation", 0.0) return 0.0 def hasAlmanac(self, index: int) -> bool: if 0 <= index < len(self._satellites): return self._satellites[index].get("has_almanac", False) return False def hasEphemeris(self, index: int) -> bool: if 0 <= index < len(self._satellites): return self._satellites[index].get("has_ephemeris", False) return False def usedInFix(self, index: int) -> bool: if 0 <= index < len(self._satellites): return self._satellites[index].get("used_in_fix", False) return False # ═══════════════════════════════════════════════════════════════ # LocationRequest # ═══════════════════════════════════════════════════════════════ class LocationRequest(Stub): """LocationRequest — 位置请求""" QUALITY_HIGH_ACCURACY = 100 QUALITY_BALANCED_POWER_ACCURACY = 102 QUALITY_LOW_POWER = 104 QUALITY_PASSIVE = 105 PRIORITY_HIGH_ACCURACY = 100 PRIORITY_BALANCED_POWER_ACCURACY = 102 PRIORITY_LOW_POWER = 104 PRIORITY_NO_POWER = 105 def __init__(self): super().__init__() self._quality: int = self.QUALITY_BALANCED_POWER_ACCURACY self._interval: int = 60000 self._fastest_interval: int = 30000 self._smallest_displacement: float = 0.0 self._max_wait_time: int = 0 self._duration_millis: int = 0 self._max_updates: int = 0 # 0 means unlimited self._priority: int = self.PRIORITY_BALANCED_POWER_ACCURACY def setQuality(self, quality: int): self._quality = quality def getQuality(self) -> int: return self._quality def setInterval(self, interval: int): self._interval = interval def getInterval(self) -> int: return self._interval def setFastestInterval(self, interval: int): self._fastest_interval = interval def getFastestInterval(self) -> int: return self._fastest_interval def setSmallestDisplacement(self, displacement: float): self._smallest_displacement = displacement def getSmallestDisplacement(self) -> float: return self._smallest_displacement def setMaxWaitTime(self, time: int): self._max_wait_time = time def getMaxWaitTime(self) -> int: return self._max_wait_time def setDurationMillis(self, duration: int): self._duration_millis = duration def getDurationMillis(self) -> int: return self._duration_millis def setMaxUpdates(self, updates: int): self._max_updates = updates def getMaxUpdates(self) -> int: return self._max_updates def setPriority(self, priority: int): self._priority = priority def getPriority(self) -> int: return self._priority def toString(self) -> str: return f"LocationRequest(interval={self._interval}, priority={self._priority})" # ═══════════════════════════════════════════════════════════════ # LocationProviderProperties # ═══════════════════════════════════════════════════════════════ class LocationProviderProperties(Stub): """LocationProviderProperties — 提供者属性""" def __init__(self, requires_network: bool = False, requires_satellite: bool = False, requires_cell: bool = False, has_monetary_cost: bool = False, supports_altitude: bool = False, supports_speed: bool = False, supports_bearing: bool = False, power_requirement: int = Criteria.NO_REQUIREMENT, accuracy: int = Criteria.NO_REQUIREMENT): super().__init__() self.requiresNetwork = requires_network self.requiresSatellite = requires_satellite self.requiresCell = requires_cell self.hasMonetaryCost = has_monetary_cost self.supportsAltitude = supports_altitude self.supportsSpeed = supports_speed self.supportsBearing = supports_bearing self.powerRequirement = power_requirement self.accuracy = accuracy # ═══════════════════════════════════════════════════════════════ # SettingInjectorService # ═══════════════════════════════════════════════════════════════ class SettingInjectorService(Stub): """SettingInjectorService — 设置注入服务""" ACTION_INJECTED_SETTING = "android.location.InjectedSetting" def __init__(self, name: str = ""): super().__init__() self._name = name def onGetSummary(self) -> str: return self._name def onGetEnabled(self) -> bool: return True def toString(self) -> str: return f"SettingInjectorService(name={self._name})" # ═══════════════════════════════════════════════════════════════ # Address # ═══════════════════════════════════════════════════════════════ class Address(Stub): """Address — 地理地址""" def __init__(self, locale: Any = None): super().__init__() self._address_lines: List[str] = [] self._feature_name: str = "" self._latitude: float = 0.0 self._longitude: float = 0.0 self._has_lat_lon: bool = False self._admin_area: str = "" self._sub_admin_area: str = "" self._locality: str = "" self._sub_locality: str = "" self._premises: str = "" self._thoroughfare: str = "" self._sub_thoroughfare: str = "" self._postal_code: str = "" self._country_name: str = "" self._country_code: str = "" self._phone: str = "" self._url: str = "" self._extras: Optional[Dict[str, Any]] = None def getAddressLine(self, index: int) -> Optional[str]: if 0 <= index < len(self._address_lines): return self._address_lines[index] return None def getMaxAddressLineIndex(self) -> int: return max(0, len(self._address_lines) - 1) def getAdminArea(self) -> Optional[str]: return self._admin_area or None def getSubAdminArea(self) -> Optional[str]: return self._sub_admin_area or None def getLocality(self) -> Optional[str]: return self._locality or None def getSubLocality(self) -> Optional[str]: return self._sub_locality or None def getPremises(self) -> Optional[str]: return self._premises or None def getThoroughfare(self) -> Optional[str]: return self._thoroughfare or None def getSubThoroughfare(self) -> Optional[str]: return self._sub_thoroughfare or None def getPostalCode(self) -> Optional[str]: return self._postal_code or None def getCountryName(self) -> Optional[str]: return self._country_name or None def getCountryCode(self) -> Optional[str]: return self._country_code or None def getFeatureName(self) -> Optional[str]: return self._feature_name or None def getLatitude(self) -> float: return self._latitude def getLongitude(self) -> float: return self._longitude def getPhone(self) -> Optional[str]: return self._phone or None def getUrl(self) -> Optional[str]: return self._url or None def setAddressLine(self, index: int, line: str): while len(self._address_lines) <= index: self._address_lines.append("") self._address_lines[index] = line 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 def setLongitude(self, longitude: float): self._longitude = longitude self._has_lat_lon = True def hasLatitude(self) -> bool: return self._has_lat_lon def hasLongitude(self) -> bool: return self._has_lat_lon def toString(self) -> str: return f"Address(locality={self._locality}, country={self._country_name})" # ═══════════════════════════════════════════════════════════════ # Geocoder # ═══════════════════════════════════════════════════════════════ 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() 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() 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 # ═══════════════════════════════════════════════════════════════ # LocationManager # ═══════════════════════════════════════════════════════════════ class LocationManager(Stub): """LocationManager — 位置管理器""" GPS_PROVIDER = "gps" NETWORK_PROVIDER = "network" PASSIVE_PROVIDER = "passive" FUSED_PROVIDER = "fused" KEY_PROXIMITY_ENTERING = "entering" KEY_PROXIMITY_DO_ENTER = "doEnter" KEY_FLUSH_COMPLETE = "flushComplete" KEY_LOCATION_CHANGED = "locationChanged" KEY_PROXIMITY_ENTERING_DO_ENTER = "doEnter" PROVIDERS_CHANGED_ACTION = "android.location.PROVIDERS_CHANGED" MODE_CHANGED_ACTION = "android.location.MODE_CHANGED" def __init__(self): super().__init__() self._listeners: Dict[str, List[LocationListener]] = {} self._providers = { self.GPS_PROVIDER: LocationProvider(self.GPS_PROVIDER), self.NETWORK_PROVIDER: LocationProvider(self.NETWORK_PROVIDER), self.PASSIVE_PROVIDER: LocationProvider(self.PASSIVE_PROVIDER), self.FUSED_PROVIDER: LocationProvider(self.FUSED_PROVIDER), } 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: data = LinuxLocation.get_location() loc = Location(provider or "network") loc.setLatitude(data.get("latitude", 0.0)) loc.setLongitude(data.get("longitude", 0.0)) loc.setAccuracy(data.get("accuracy", 0)) loc.setTime(int(datetime.now().timestamp() * 1000)) loc.setProvider(provider or "network") return loc if provider in self._last_known_locations: return self._last_known_locations[provider] loc = Location(provider) loc.setLatitude(37.7749) loc.setLongitude(-122.4194) loc.setTime(int(datetime.now().timestamp() * 1000)) loc.setAccuracy(20.0) self._last_known_locations[provider] = loc return loc def requestLocationUpdates(self, provider_or_request: Any, min_time: Any = None, min_distance: float = 0.0, listener_or_callback: Any = None, looper: Any = None) -> None: if isinstance(provider_or_request, str): provider = provider_or_request else: provider = self.GPS_PROVIDER if _USE_LINUX and listener_or_callback is not None: data = LinuxLocation.get_location() loc = Location(provider or "network") loc.setLatitude(data.get("latitude", 0.0)) loc.setLongitude(data.get("longitude", 0.0)) loc.setAccuracy(data.get("accuracy", 0)) loc.setTime(int(datetime.now().timestamp() * 1000)) loc.setProvider(provider or "network") try: listener_or_callback.onLocationChanged(loc) except Exception: pass if listener_or_callback is not None: if provider not in self._listeners: self._listeners[provider] = [] if isinstance(listener_or_callback, LocationListener): self._listeners[provider].append(listener_or_callback) def removeUpdates(self, listener: Any): for provider in list(self._listeners.keys()): self._listeners[provider] = [ l for l in self._listeners[provider] if l is not listener ] if not self._listeners[provider]: del self._listeners[provider] def isProviderEnabled(self, provider: str) -> bool: if _USE_LINUX: return provider == "network" return provider in self._enabled_providers def getProviders(self, criteria: Optional[Criteria] = None, enabled_only: bool = False) -> List[str]: if _USE_LINUX: return ["network"] providers = list(self._providers.keys()) if enabled_only: providers = [p for p in providers if p in self._enabled_providers] return providers def getBestProvider(self, criteria: Criteria, enabled_only: bool = False) -> Optional[str]: if criteria.getAccuracy() == Criteria.ACCURACY_FINE: return self.GPS_PROVIDER return self.NETWORK_PROVIDER def getAllProviders(self) -> List[str]: return list(self._providers.keys()) def getProvider(self, name: str) -> Optional[LocationProvider]: return self._providers.get(name) def addProximityAlert(self, latitude: float, longitude: float, radius: float, expiration: int, intent: Any): """添加接近警报。记录警报信息，在位置更新时检查是否进入/离开区域。""" 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): """清除测试提供者的位置数据。""" self._test_locations.pop(provider, None) def setTestProviderLocation(self, provider: str, location: Location): """设置测试提供者的位置。""" 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): """移除测试提供者。""" 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, supports_speed: bool, supports_bearing: bool, power_requirement: int, accuracy: int): self._providers[name] = LocationProvider(name) def toString(self) -> str: return f"LocationManager(providers={list(self._providers.keys())})" # ═══════════════════════════════════════════════════════════════ # __all__ # ═══════════════════════════════════════════════════════════════ __all__ = [ "Location", "LocationManager", "LocationListener", "LocationProvider", "Criteria", "Geocoder", "Address", "GpsStatus", "GpsSatellite", "GpsStatusListener", "LocationRequest", "LocationProviderProperties", "SettingInjectorService", "GnssStatus", "GnssClock", "GnssMeasurement", "GnssMeasurementsEvent", "GnssNavigationMessage", ]