You can not select more than 25 topics
			Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
		
		
		
		
		
			
		
			
				
					
					
						
							80 lines
						
					
					
						
							2.0 KiB
						
					
					
				
			
		
		
	
	
							80 lines
						
					
					
						
							2.0 KiB
						
					
					
				#!/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()
 | 
						|
 |