Improve GPS tests, add qcom tests (#26060)
* first ignore
* init gps test
* make LimeGPS git clone
* gps test v1
* add static signal gen script
* remove LD_PRELOAD by using rpath, update values after testing
* cleanUp
* update fuzzy tests
* finalize qcom gps tests
* add downloader
* finalize unit tests
* inc limeGPS startup time
* loosen init time
* add ublox warmstart test
* improve location tests
Co-authored-by: Kurt Nistelberger <kurt.nistelberger@gmail.com>
old-commit-hash: 9c92814585
taco
Kurt Nistelberger
3 years ago
committed by
GitHub
parent
5b90a8ce15
commit
9c791ad9b2
7 changed files with 327 additions and 121 deletions
@ -1,83 +0,0 @@ |
||||
#!/usr/bin/env python3 |
||||
import os |
||||
import sys |
||||
import random |
||||
import datetime as dt |
||||
import subprocess as sp |
||||
from typing import Tuple |
||||
|
||||
from laika.downloader import download_nav |
||||
from laika.gps_time import GPSTime |
||||
from laika.helpers import ConstellationId |
||||
|
||||
cache_dir = '/tmp/gpstest/' |
||||
|
||||
|
||||
def download_rinex(): |
||||
# TODO: check if there is a better way to get the full brdc file for LimeGPS |
||||
gps_time = GPSTime.from_datetime(dt.datetime.utcnow()) |
||||
utc_time = dt.datetime.utcnow() - dt.timedelta(1) |
||||
gps_time = GPSTime.from_datetime(dt.datetime(utc_time.year, utc_time.month, utc_time.day)) |
||||
return download_nav(gps_time, cache_dir, ConstellationId.GPS) |
||||
|
||||
|
||||
def get_random_coords(lat, lon) -> Tuple[int, int]: |
||||
# jump around the world |
||||
# max values, lat: -90 to 90, lon: -180 to 180 |
||||
|
||||
lat_add = random.random()*20 + 10 |
||||
lon_add = random.random()*20 + 20 |
||||
|
||||
lat = ((lat + lat_add + 90) % 180) - 90 |
||||
lon = ((lon + lon_add + 180) % 360) - 180 |
||||
return round(lat, 5), round(lon, 5) |
||||
|
||||
|
||||
def check_availability() -> bool: |
||||
cmd = ["LimeSuite/builddir/LimeUtil/LimeUtil", "--find"] |
||||
output = sp.check_output(cmd) |
||||
|
||||
if output.strip() == b"": |
||||
return False |
||||
|
||||
print(f"Device: {output.strip().decode('utf-8')}") |
||||
return True |
||||
|
||||
|
||||
def main(): |
||||
if not os.path.exists('LimeGPS'): |
||||
print("LimeGPS not found run 'setup.sh' first") |
||||
return |
||||
|
||||
if not os.path.exists('LimeSuite'): |
||||
print("LimeSuite not found run 'setup.sh' first") |
||||
return |
||||
|
||||
if not check_availability(): |
||||
print("No limeSDR device found!") |
||||
return |
||||
|
||||
rinex_file = download_rinex() |
||||
lat, lon = get_random_coords(47.2020, 15.7403) |
||||
|
||||
if len(sys.argv) == 3: |
||||
lat = float(sys.argv[1]) |
||||
lon = float(sys.argv[2]) |
||||
|
||||
try: |
||||
print(f"starting LimeGPS, Location: {lat},{lon}") |
||||
cmd = ["LimeGPS/LimeGPS", "-e", rinex_file, "-l", f"{lat},{lon},100"] |
||||
sp.check_output(cmd, stderr=sp.PIPE) |
||||
except KeyboardInterrupt: |
||||
print("stopping LimeGPS") |
||||
except Exception as e: |
||||
out_stderr = e.stderr.decode('utf-8')# pylint:disable=no-member |
||||
if "Device is busy." in out_stderr: |
||||
print("GPS simulation is already running, Device is busy!") |
||||
return |
||||
|
||||
print(f"LimeGPS crashed: {str(e)}") |
||||
print(f"stderr:\n{e.stderr.decode('utf-8')}")# pylint:disable=no-member |
||||
|
||||
if __name__ == "__main__": |
||||
main() |
||||
@ -0,0 +1,16 @@ |
||||
#!/bin/bash |
||||
|
||||
# NOTE: can only run inside limeGPS test box! |
||||
|
||||
# run limeGPS with random static location |
||||
timeout 300 ./simulate_gps_signal.py & |
||||
gps_PID=$? |
||||
|
||||
echo "starting limeGPS..." |
||||
sleep 10 |
||||
|
||||
# run unit tests (skipped when module not present) |
||||
python -m unittest test_gps.py |
||||
python -m unittest test_gps_qcom.py |
||||
|
||||
kill $gps_PID |
||||
@ -0,0 +1,102 @@ |
||||
#!/usr/bin/env python3 |
||||
import os |
||||
import random |
||||
import argparse |
||||
import datetime as dt |
||||
import subprocess as sp |
||||
from typing import Tuple |
||||
|
||||
from laika.downloader import download_nav |
||||
from laika.gps_time import GPSTime |
||||
from laika.helpers import ConstellationId |
||||
|
||||
cache_dir = '/tmp/gpstest/' |
||||
|
||||
|
||||
def download_rinex(): |
||||
# TODO: check if there is a better way to get the full brdc file for LimeGPS |
||||
gps_time = GPSTime.from_datetime(dt.datetime.utcnow()) |
||||
utc_time = dt.datetime.utcnow() - dt.timedelta(1) |
||||
gps_time = GPSTime.from_datetime(dt.datetime(utc_time.year, utc_time.month, utc_time.day)) |
||||
return download_nav(gps_time, cache_dir, ConstellationId.GPS) |
||||
|
||||
def get_coords(lat, lon, s1, s2, o1=0, o2=0) -> Tuple[int, int]: |
||||
lat_add = random.random()*s1 + o1 |
||||
lon_add = random.random()*s2 + o2 |
||||
|
||||
lat = ((lat + lat_add + 90) % 180) - 90 |
||||
lon = ((lon + lon_add + 180) % 360) - 180 |
||||
return round(lat, 5), round(lon, 5) |
||||
|
||||
def get_continuous_coords(lat, lon) -> Tuple[int, int]: |
||||
# continuously move around the world |
||||
return get_coords(lat, lon, 0.01, 0.01) |
||||
|
||||
def get_random_coords(lat, lon) -> Tuple[int, int]: |
||||
# jump around the world |
||||
return get_coords(lat, lon, 20, 20, 10, 20) |
||||
|
||||
def check_availability() -> bool: |
||||
cmd = ["LimeSuite/builddir/LimeUtil/LimeUtil", "--find"] |
||||
output = sp.check_output(cmd) |
||||
|
||||
if output.strip() == b"": |
||||
return False |
||||
|
||||
print(f"Device: {output.strip().decode('utf-8')}") |
||||
return True |
||||
|
||||
def main(lat, lon, jump_sim, contin_sim): |
||||
if not os.path.exists('LimeGPS'): |
||||
print("LimeGPS not found run 'setup.sh' first") |
||||
return |
||||
|
||||
if not os.path.exists('LimeSuite'): |
||||
print("LimeSuite not found run 'setup.sh' first") |
||||
return |
||||
|
||||
if not check_availability(): |
||||
print("No limeSDR device found!") |
||||
return |
||||
|
||||
rinex_file = download_rinex() |
||||
|
||||
if lat == 0 and lon == 0: |
||||
lat, lon = get_random_coords(47.2020, 15.7403) |
||||
|
||||
timeout = None |
||||
if jump_sim: |
||||
timeout = 30 |
||||
|
||||
while True: |
||||
try: |
||||
print(f"starting LimeGPS, Location: {lat},{lon}") |
||||
cmd = ["LimeGPS/LimeGPS", "-e", rinex_file, "-l", f"{lat},{lon},100"] |
||||
sp.check_output(cmd, stderr=sp.PIPE, timeout=timeout) |
||||
except KeyboardInterrupt: |
||||
print("stopping LimeGPS") |
||||
return |
||||
except sp.TimeoutExpired: |
||||
print("LimeGPS timeout reached!") |
||||
except Exception as e: |
||||
out_stderr = e.stderr.decode('utf-8')# pylint:disable=no-member |
||||
if "Device is busy." in out_stderr: |
||||
print("GPS simulation is already running, Device is busy!") |
||||
return |
||||
|
||||
print(f"LimeGPS crashed: {str(e)}") |
||||
print(f"stderr:\n{e.stderr.decode('utf-8')}")# pylint:disable=no-member |
||||
|
||||
if contin_sim: |
||||
lat, lon = get_continuous_coords(lat, lon) |
||||
else: |
||||
lat, lon = get_random_coords(lat, lon) |
||||
|
||||
if __name__ == "__main__": |
||||
parser = argparse.ArgumentParser(description="Simulate static [or random jumping] GPS signal.") |
||||
parser.add_argument("lat", type=float, nargs='?', default=0) |
||||
parser.add_argument("lon", type=float, nargs='?', default=0) |
||||
parser.add_argument("--jump", action="store_true", help="signal that jumps around the world") |
||||
parser.add_argument("--contin", action="store_true", help="continuously/slowly moving around the world") |
||||
args = parser.parse_args() |
||||
main(args.lat, args.lon, args.jump, args.contin) |
||||
@ -0,0 +1,109 @@ |
||||
#!/usr/bin/env python3 |
||||
import time |
||||
import unittest |
||||
import subprocess as sp |
||||
|
||||
from common.params import Params |
||||
from system.hardware import TICI |
||||
import cereal.messaging as messaging |
||||
from selfdrive.manager.process_config import managed_processes |
||||
|
||||
|
||||
def exec_mmcli(cmd): |
||||
cmd = "mmcli -m 0 " + cmd |
||||
p = sp.Popen(cmd, shell=True, stdout=sp.PIPE, stderr=sp.PIPE) |
||||
return p.communicate() |
||||
|
||||
|
||||
def wait_for_location(socket, timeout): |
||||
while True: |
||||
events = messaging.drain_sock(socket) |
||||
for event in events: |
||||
if event.gpsLocation.flags % 2: |
||||
return False |
||||
|
||||
timeout -= 1 |
||||
if timeout <= 0: |
||||
return True |
||||
|
||||
time.sleep(0.1) |
||||
continue |
||||
|
||||
|
||||
class TestGPS(unittest.TestCase): |
||||
@classmethod |
||||
def setUpClass(cls): |
||||
if not TICI: |
||||
raise unittest.SkipTest |
||||
|
||||
ublox_available = Params().get_bool("UbloxAvailable") |
||||
if ublox_available: |
||||
raise unittest.SkipTest |
||||
|
||||
@unittest.skip("Skip cold start test due to time") |
||||
def test_quectel_cold_start(self): |
||||
# delete assistance data to enforce cold start for GNSS |
||||
# testing shows that this takes up to 20min |
||||
|
||||
# invalidate supl setting, cannot be reset |
||||
_, err = exec_mmcli("--location-set-supl-server=unittest:1") |
||||
|
||||
_, err = exec_mmcli("--command='AT+QGPSDEL=0'") |
||||
assert len(err) == 0, f"GPSDEL failed: {err}" |
||||
|
||||
managed_processes['rawgpsd'].start() |
||||
start_time = time.monotonic() |
||||
glo = messaging.sub_sock("gpsLocation", timeout=0.1) |
||||
|
||||
timeout = 10*60*25 # 25 minute |
||||
timedout = wait_for_location(glo, timeout) |
||||
managed_processes['rawgpsd'].stop() |
||||
|
||||
assert timedout is False, "Waiting for location timed out (25min)!" |
||||
|
||||
duration = time.monotonic() - start_time |
||||
assert duration < 50, f"Received GPS location {duration}!" |
||||
|
||||
|
||||
def test_a_quectel_cold_start_AGPS(self): |
||||
_, err = exec_mmcli("--command='AT+QGPSDEL=0'") |
||||
assert len(err) == 0, f"GPSDEL failed: {err}" |
||||
|
||||
# setup AGPS |
||||
exec_mmcli("--location-set-supl-server=supl.google.com:7276") |
||||
|
||||
managed_processes['rawgpsd'].start() |
||||
start_time = time.monotonic() |
||||
glo = messaging.sub_sock("gpsLocation", timeout=0.1) |
||||
|
||||
timeout = 10*60*3 # 3 minute |
||||
timedout = wait_for_location(glo, timeout) |
||||
managed_processes['rawgpsd'].stop() |
||||
|
||||
assert timedout is False, "Waiting for location timed out (3min)!" |
||||
|
||||
duration = time.monotonic() - start_time |
||||
assert duration < 60, f"Received GPS location {duration}!" |
||||
|
||||
|
||||
def test_b_quectel_startup(self): |
||||
|
||||
# setup AGPS |
||||
exec_mmcli("--location-set-supl-server=supl.google.com:7276") |
||||
|
||||
managed_processes['rawgpsd'].start() |
||||
start_time = time.monotonic() |
||||
glo = messaging.sub_sock("gpsLocation", timeout=0.1) |
||||
|
||||
timeout = 10*60*3 # 3 minute |
||||
timedout = wait_for_location(glo, timeout) |
||||
managed_processes['rawgpsd'].stop() |
||||
|
||||
assert timedout is False, "Waiting for location timed out (3min)!" |
||||
|
||||
duration = time.monotonic() - start_time |
||||
assert duration < 60, f"Received GPS location {duration}!" |
||||
|
||||
|
||||
if __name__ == "__main__": |
||||
unittest.main() |
||||
Loading…
Reference in new issue