from datetime import datetime import os import pathlib import re import shutil import tempfile from collections import defaultdict _EVENT_TIME_FORMAT = "%Y/%m/%d %H:%M:%S.%f" _EVENT_TIME_PATTERN = re.compile(r"(\d+(?:\/\d+){2}\s\d{2}(?::\d{2}){2}\.\d+)\sRead:") _GNSS_CLOCK_START_LOG_PATTERN = re.compile(r"^GnssClock:") _GNSS_CLOCK_TIME_NANOS_PATTERN = re.compile(f"^\s+TimeNanos\s*=\s*([-]?\d*)") _GNSS_CLOCK_FULL_BIAS_NANOS_PATTERN = re.compile(f"^\s+FullBiasNanos\s*=\s*([-]?\d*)") _GNSS_CLOCK_ELAPSED_TIME_NANOS_PATTERN = re.compile(f"^\s+elapsedRealtimeNanos\s*=\s*([-]?\d*)") class AdrInfo: """Represent one ADR value An ADR value is a decimal number range from 0 - 31 How to parse the ADR value: First, transform the decimal number to binary then we will get a 5 bit number The meaning of each bit is as follow: 0 0 0 0 0 HalfCycleReported HalfCycleResolved CycleSlip Reset Valid Special rule: For an ADR value in binary fits the pattern: * * 0 0 1, we call it a usable ADR More insight of ADR value: go/adrstates Attributes: is_valid: (bool) is_reset: (bool) is_cycle_slip: (bool) is_half_cycle_resolved: (bool) is_half_cycle_reported: (bool) is_usable: (bool) """ def __init__(self, adr_value: int, count: int): src = bin(int(adr_value)) self._valid = int(src[-1]) self._reset = int(src[-2]) self._cycle_slip = int(src[-3]) self._half_cycle_resolved = int(src[-4]) self._half_cycle_reported = int(src[-5]) self.count = count @property def is_usable(self): return self.is_valid and not self.is_reset and not self.is_cycle_slip @property def is_valid(self): return bool(self._valid) @property def is_reset(self): return bool(self._reset) @property def is_cycle_slip(self): return bool(self._cycle_slip) @property def is_half_cycle_resolved(self): return bool(self._half_cycle_resolved) @property def is_half_cycle_reported(self): return bool(self._half_cycle_reported) class AdrStatistic: """Represent the ADR statistic Attributes: usable_count: (int) valid_count: (int) reset_count: (int) cycle_slip_count: (int) half_cycle_resolved_count: (int) half_cycle_reported_count: (int) total_count: (int) usable_rate: (float) usable_count / total_count valid_rate: (float) valid_count / total_count """ def __init__(self): self.usable_count = 0 self.valid_count = 0 self.reset_count = 0 self.cycle_slip_count = 0 self.half_cycle_resolved_count = 0 self.half_cycle_reported_count = 0 self.total_count = 0 @property def usable_rate(self): denominator = max(1, self.total_count) result = self.usable_count / denominator return round(result, 3) @property def valid_rate(self): denominator = max(1, self.total_count) result = self.valid_count / denominator return round(result, 3) def add_adr_info(self, adr_info: AdrInfo): """Add ADR info record to increase the statistic Args: adr_info: AdrInfo object """ if adr_info.is_valid: self.valid_count += adr_info.count if adr_info.is_reset: self.reset_count += adr_info.count if adr_info.is_cycle_slip: self.cycle_slip_count += adr_info.count if adr_info.is_half_cycle_resolved: self.half_cycle_resolved_count += adr_info.count if adr_info.is_half_cycle_reported: self.half_cycle_reported_count += adr_info.count if adr_info.is_usable: self.usable_count += adr_info.count self.total_count += adr_info.count class GnssClockSubEvent: time_nanos: int full_bias_nanos: int elapsed_real_time_nanos: int def __init__(self, event_time): self.event_time = event_time def parse(self, line): if _GNSS_CLOCK_TIME_NANOS_PATTERN.search(line): self.time_nanos = int(_GNSS_CLOCK_TIME_NANOS_PATTERN.search(line).group(1)) elif _GNSS_CLOCK_FULL_BIAS_NANOS_PATTERN.search(line): self.full_bias_nanos = int( _GNSS_CLOCK_FULL_BIAS_NANOS_PATTERN.search(line).group(1)) elif _GNSS_CLOCK_ELAPSED_TIME_NANOS_PATTERN.search(line): self.elapsed_real_time_nanos = int( _GNSS_CLOCK_ELAPSED_TIME_NANOS_PATTERN.search(line).group(1)) def __repr__(self) -> str: return (f"event time: {self.event_time}, " f"timenanos: {self.time_nanos}, " f"full_bias: {self.full_bias_nanos}, " f"elapsed_realtime: {self.elapsed_real_time_nanos}") @property def gps_elapsed_realtime_diff(self): return self.time_nanos - self.full_bias_nanos - self.elapsed_real_time_nanos class GnssMeasurement: """Represent the content of measurement file generated by gps tool""" FILE_PATTERN = "/storage/emulated/0/Android/data/com.android.gpstool/files/MEAS*.txt" def __init__(self, ad): self.ad = ad def _generate_local_temp_path(self, file_name="file.txt"): """Generate a file path for temporarily usage Returns: string: local file path """ folder = tempfile.mkdtemp() file_path = os.path.join(folder, file_name) return file_path def _get_latest_measurement_file_path(self): """Get the latest measurement file path on device Returns: string: file path on device """ command = f"ls -tr {self.FILE_PATTERN} | tail -1" result = self.ad.adb.shell(command) return result def get_latest_measurement_file(self): """Pull the latest measurement file from device to local Returns: string: local file path to the measurement file Raise: FileNotFoundError: can't get measurement file from device """ self.ad.log.info("Get measurement file from device") dest = self._generate_local_temp_path(file_name="measurement.txt") src = self._get_latest_measurement_file_path() if not src: raise FileNotFoundError(f"Can not find measurement file: pattern {self.FILE_PATTERN}") self.ad.pull_files(src, dest) return dest def _get_adr_src_value(self): """Get ADR value from measurement file Returns: dict: {: count, : count...} """ try: file_path = self.get_latest_measurement_file() adr_src = defaultdict(int) adr_src_regex = re.compile("=\s(\d*)") with open(file_path) as f: for line in f: if "AccumulatedDeltaRangeState" in line: value = re.search(adr_src_regex, line) if not value: self.ad.log.warn("Can't get ADR value %s" % line) continue key = value.group(1) adr_src[key] += 1 return adr_src finally: folder = pathlib.PurePosixPath(file_path).parent shutil.rmtree(folder, ignore_errors=True) def get_adr_static(self): """Get ADR statistic Summarize ADR value from measurement file Returns: AdrStatistic object """ self.ad.log.info("Get ADR statistic") adr_src = self._get_adr_src_value() adr_static = AdrStatistic() for key, value in adr_src.items(): self.ad.log.debug("ADR value: %s - count: %s" % (key, value)) adr_info = AdrInfo(key, value) adr_static.add_adr_info(adr_info) return adr_static def get_gnss_clock_info(self): sub_events = [] event_time = None with tempfile.TemporaryDirectory() as folder: local_measurement_file = os.path.join(folder, "measurement_file") self.ad.pull_files(self._get_latest_measurement_file_path(), local_measurement_file) with open(local_measurement_file) as f: for line in f: if re.search(_EVENT_TIME_PATTERN, line): event_time = re.search(_EVENT_TIME_PATTERN, line).group(1) event_time = datetime.strptime(event_time, _EVENT_TIME_FORMAT) elif re.search(_GNSS_CLOCK_START_LOG_PATTERN, line): sub_events.append(GnssClockSubEvent(event_time)) elif line.startswith(" "): sub_events[-1].parse(line) return sub_events