|  |  |  | #!/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
 | 
					
						
							|  |  |  | 
 | 
					
						
							|  |  |  |   def test_a_quectel_cold_start(self):
 | 
					
						
							|  |  |  |     # delete assistance data to enforce cold start for GNSS
 | 
					
						
							|  |  |  |     # testing shows that this takes up to 20min
 | 
					
						
							|  |  |  | 
 | 
					
						
							|  |  |  |     _, 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*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):
 | 
					
						
							|  |  |  |     managed_processes['rawgpsd'].start()
 | 
					
						
							|  |  |  |     start_time = time.monotonic()
 | 
					
						
							|  |  |  |     glo = messaging.sub_sock("gpsLocation", timeout=0.1)
 | 
					
						
							|  |  |  | 
 | 
					
						
							|  |  |  |     timeout = 10*60 # 1 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()
 |