mawei/StarPilot
1
0
Fork 0
mirror of https://github.com/firestar5683/StarPilot.git synced 2026-07-01 03:22:07 +08:00
Code Issues Packages Projects Releases Wiki Activity

locationd: speedup unit tests (#29262)

Browse Source 
old-commit-hash: b420e6fb134cdfc4dd46802d6f8c6e170423a296
This commit is contained in:
Adeeb Shihadeh
2023-08-06 21:37:04 -07:00
committed by GitHub
parent a7ec9933c8
commit e31cf1fe9e
1 changed files with 12 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
selfdrive/locationd/test/test_locationd.py
+12 -23
View File
@@ -14,7 +14,6 @@ from selfdrive.manager.process_config import managed_processes
class TestLocationdProc(unittest.TestCase):
MAX_WAITS = 1000
LLD_MSGS = ['gpsLocationExternal', 'cameraOdometry', 'carState', 'liveCalibration',
'accelerometer', 'gyroscope', 'magnetometer']
@@ -23,30 +22,20 @@ class TestLocationdProc(unittest.TestCase):
self.pm = messaging.PubMaster(self.LLD_MSGS)
Params().put_bool("UbloxAvailable", True)
self.params = Params()
self.params.put_bool("UbloxAvailable", True)
managed_processes['locationd'].prepare()
managed_processes['locationd'].start()
time.sleep(1)
def tearDown(self):
managed_processes['locationd'].stop()
def send_msg(self, msg):
self.pm.send(msg.which(), msg)
waits_left = self.MAX_WAITS
while waits_left and not self.pm.all_readers_updated(msg.which()):
time.sleep(0)
waits_left -= 1
time.sleep(0.0001)
def get_fake_msg(self, name, t):
def get_msg(self, name, t):
try:
msg = messaging.new_message(name)
except capnp.lib.capnp.KjException:
msg = messaging.new_message(name, 0)
if name == "gpsLocationExternal":
msg.gpsLocationExternal.flags = 1
msg.gpsLocationExternal.verticalAccuracy = 1.0
@@ -74,27 +63,27 @@ class TestLocationdProc(unittest.TestCase):
return msg
def test_params_gps(self):
# first reset params
Params().remove('LastGPSPosition')
self.params.remove('LastGPSPosition')
self.x = -2710700 + (random.random() * 1e5)
self.y = -4280600 + (random.random() * 1e5)
self.z = 3850300 + (random.random() * 1e5)
self.lat, self.lon, self.alt = ecef2geodetic([self.x, self.y, self.z])
self.fake_duration = 90 # secs
# get fake messages at the correct frequency, listed in services.py
fake_msgs = []
for sec in range(self.fake_duration):
msgs = []
for sec in range(65):
for name in self.LLD_MSGS:
for j in range(int(service_list[name].frequency)):
fake_msgs.append(self.get_fake_msg(name, int((sec + j / service_list[name].frequency) * 1e9)))
msgs.append(self.get_msg(name, int((sec + j / service_list[name].frequency) * 1e9)))
for fake_msg in sorted(fake_msgs, key=lambda x: x.logMonoTime):
self.send_msg(fake_msg)
for msg in sorted(msgs, key=lambda x: x.logMonoTime):
self.pm.send(msg.which(), msg)
if msg.which() == "cameraOdometry":
self.pm.wait_for_readers_to_update(msg.which(), 0.1)
time.sleep(1) # wait for async params write
lastGPS = json.loads(Params().get('LastGPSPosition'))
lastGPS = json.loads(self.params.get('LastGPSPosition'))
self.assertAlmostEqual(lastGPS['latitude'], self.lat, places=3)
self.assertAlmostEqual(lastGPS['longitude'], self.lon, places=3)
self.assertAlmostEqual(lastGPS['altitude'], self.alt, places=3)