code.panda.quecpython/demo.tracker-v2
1
0
Fork 0
mirror of https://gitee.com/qpy-solutions/tracker-v2.git synced 2025-05-19 02:58:25 +08:00
Code Issues Packages Projects Releases Wiki Activity

update: loc info get

Browse Source
This commit is contained in:
JackSun-qc 2022-03-29 15:09:33 +08:00
parent 9291493e1b
commit 8d98f5dd70
4 changed files with 26 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
code/location.py
View File

@ -81,19 +81,20 @@ class GPS(Singleton):
self.uart_obj.set_callback(gps_data_retrieve_cb) self.uart_obj.set_callback(gps_data_retrieve_cb)
gps_data_retrieve_queue = Queue(maxsize=8) gps_data_retrieve_queue = Queue(maxsize=8)
def gps_timer_callback(self, args): def first_gps_timer_callback(self, args):
if self.break_flag == 0: self.break_flag = 1
self.break_flag = 1
elif self.break_flag == 1: def second_gps_timer_callback(self, args):
self.break_flag = 2 global gps_data_retrieve_queue
if gps_data_retrieve_queue is not None: self.break_flag = 2
gps_data_retrieve_queue.put(0) if gps_data_retrieve_queue is not None:
gps_data_retrieve_queue.put(0)
def uart_read(self): def uart_read(self):
global gps_data_retrieve_queue global gps_data_retrieve_queue
while self.break_flag == 0: while self.break_flag == 0:
self.gps_timer.start(50, 1, self.gps_timer_callback) self.gps_timer.start(50, 1, self.first_gps_timer_callback)
nread = gps_data_retrieve_queue.get() nread = gps_data_retrieve_queue.get()
data = self.uart_obj.read(nread).decode() data = self.uart_obj.read(nread).decode()
self.gps_timer.stop() self.gps_timer.stop()
@ -103,7 +104,7 @@ class GPS(Singleton):
gga_data = '' gga_data = ''
vtg_data = '' vtg_data = ''
while self.break_flag == 1: while self.break_flag == 1:
self.gps_timer.start(3000, 1, self.gps_timer_callback) self.gps_timer.start(1500, 1, self.second_gps_timer_callback)
nread = gps_data_retrieve_queue.get() nread = gps_data_retrieve_queue.get()
if nread: if nread:
udata = self.uart_obj.read(nread).decode() udata = self.uart_obj.read(nread).decode()
@ -127,7 +128,7 @@ class GPS(Singleton):
quecgnss.gnssEnable(1) quecgnss.gnssEnable(1)
while self.break_flag == 0: while self.break_flag == 0:
self.gps_timer.start(50, 1, self.gps_timer_callback) self.gps_timer.start(50, 1, self.first_gps_timer_callback)
data = quecgnss.read(4096) data = quecgnss.read(4096)
self.gps_timer.stop() self.gps_timer.stop()
@ -137,7 +138,7 @@ class GPS(Singleton):
vtg_data = '' vtg_data = ''
count = 0 count = 0
while self.break_flag == 1: while self.break_flag == 1:
self.gps_timer.start(3000, 1, self.gps_timer_callback) self.gps_timer.start(1500, 1, self.second_gps_timer_callback)
gnss_data = quecgnss.read(4096) gnss_data = quecgnss.read(4096)
if gnss_data and gnss_data[1]: if gnss_data and gnss_data[1]:
udata = gnss_data[1].decode() if len(gnss_data) > 1 and gnss_data[1] else '' udata = gnss_data[1].decode() if len(gnss_data) > 1 and gnss_data[1] else ''
@ -172,7 +173,7 @@ class GPS(Singleton):
def read_location_GxRMC(self, gps_data): def read_location_GxRMC(self, gps_data):
rmc_re = ure.search( rmc_re = ure.search(
r"\$G[NP]RMC,[0-9]+\.[0-9]+,A,[0-9]+\.[0-9]+,[NS],[0-9]+\.[0-9]+,[EW],[0-9]+\.[0-9]+,[0-9]+\.[0-9]+,[0-9]+,,,[ADE],[SCUV]\*[0-9]+", r"\$G[NP]RMC,\d*\.*\d*,[AV]*,\d*\.*\d*,[NS]*,\d*\.*\d*,[EW]*,\d*\.*\d*,\d*\.*\d*,\d*,\d*\.*\d*,[EW]*,[ADEN]*,[SCUV]*\**\d*",
gps_data) gps_data)
if rmc_re: if rmc_re:
return rmc_re.group(0) return rmc_re.group(0)
@ -180,14 +181,14 @@ class GPS(Singleton):
def read_location_GxGGA(self, gps_data): def read_location_GxGGA(self, gps_data):
gga_re = ure.search( gga_re = ure.search(
r"\$G[BLPN]GGA,[0-9]+\.[0-9]+,[0-9]+\.[0-9]+,[NS],[0-9]+\.[0-9]+,[EW],[126],[0-9]+,[0-9]+\.[0-9]+,-*[0-9]+\.[0-9]+,M,-*[0-9]+\.[0-9]+,M,,\*[0-9]+", r"\$G[BLPN]GGA,\d*\.*\d*,\d*\.*\d*,[NS]*,\d*\.*\d*,[EW]*,[0126]*,\d*,\d*\.*\d*,-*\d*\.*\d*,M,-*\d*\.*\d*,M,\d*,\**\d*",
gps_data) gps_data)
if gga_re: if gga_re:
return gga_re.group(0) return gga_re.group(0)
return "" return ""
def read_location_GxVTG(self, gps_data): def read_location_GxVTG(self, gps_data):
vtg_re = ure.search(r"\$G[NP]VTG,[0-9]+\.[0-9]+,T,([0-9]+\.[0-9]+)??,M,[0-9]+\.[0-9]+,N,[0-9]+\.[0-9]+,K,[ADEN]\*\w*", gps_data) vtg_re = ure.search(r"\$G[NP]VTG,\d*\.*\d*,T,\d*\.*\d*,M,\d*\.*\d*,N,\d*\.*\d*,K,[ADEN]*\**\d*", gps_data)
if vtg_re: if vtg_re:
return vtg_re.group(0) return vtg_re.group(0)
return "" return ""
@ -195,7 +196,7 @@ class GPS(Singleton):
def read_location_GxVTG_speed(self, gps_data): def read_location_GxVTG_speed(self, gps_data):
vtg_data = self.read_location_GxVTG(gps_data) vtg_data = self.read_location_GxVTG(gps_data)
if vtg_data: if vtg_data:
speed_re = ure.search(r",N,[0-9]+\.[0-9]+,K,", vtg_data) speed_re = ure.search(r",N,\d+\.\d+,K,", vtg_data)
if speed_re: if speed_re:
return speed_re.group(0)[3:-3] return speed_re.group(0)[3:-3]
return "" return ""
@ -204,7 +205,6 @@ class GPS(Singleton):
res = {} res = {}
data = [] data = []
gps_data = self.read() gps_data = self.read()
log.debug('read_quecIot gps_data: %s' % gps_data)
r = self.read_location_GxRMC(gps_data) r = self.read_location_GxRMC(gps_data)
if r: if r:
data.append(r) data.append(r)

5
code/logging.py
View File

@ -13,7 +13,7 @@
# limitations under the License. # limitations under the License.
import utime import utime
import usr.settings as settings from usr.settings import settings
def asyncLog(name, level, *message, timeout=None, await_connection=True): def asyncLog(name, level, *message, timeout=None, await_connection=True):
@ -27,7 +27,8 @@ def asyncLog(name, level, *message, timeout=None, await_connection=True):
def log(name, level, *message, local_only=False, return_only=False, timeout=None): def log(name, level, *message, local_only=False, return_only=False, timeout=None):
if not settings.settings.get().get('sys', {}).get('sw_log', True): current_settings = settings.get()
if not current_settings.get('sys', {}).get('sw_log', True):
return return
if hasattr(utime, "strftime"): if hasattr(utime, "strftime"):

13
code/settings.py
View File

@ -80,17 +80,8 @@ _settings_lock = _thread.allocate_lock()
def settings_lock(func_name): def settings_lock(func_name):
def settings_lock_fun(func): def settings_lock_fun(func):
def wrapperd_fun(*args, **kwargs): def wrapperd_fun(*args, **kwargs):
if not _settings_lock.locked(): with _settings_lock:
if _settings_lock.acquire(): return func(*args, **kwargs)
source_fun = func(*args, **kwargs)
_settings_lock.release()
return source_fun
else:
print('_settings_lock acquire falied. func: %s, args: %s' % (func_name, args))
return False
else:
print('_settings_lock is locked. func: %s, args: %s' % (func_name, args))
return False
return wrapperd_fun return wrapperd_fun
return settings_lock_fun return settings_lock_fun

8
code/tracker.py
View File

@ -31,7 +31,7 @@ from usr.common import numiter
from usr.common import Singleton from usr.common import Singleton
from usr.mpower import PowerManage from usr.mpower import PowerManage
from usr.logging import getLogger from usr.logging import getLogger
from usr.location import GPS # from usr.location import GPS
from usr.location import Location from usr.location import Location
from usr.timer import LEDTimer from usr.timer import LEDTimer
@ -298,14 +298,15 @@ class SelfCheck(object):
def gps_check(self): def gps_check(self):
# return True if OK # return True if OK
gps = GPS(settings.default_values_sys._gps_cfg) # gps = GPS(settings.default_values_sys._gps_cfg)
locator = Location()
retry = 0 retry = 0
gps_data = None gps_data = None
sleep_time = 1 sleep_time = 1
while retry < 5: while retry < 5:
gps_data = gps.read() gps_data = locator.read()
if gps_data: if gps_data:
break break
else: else:
@ -313,6 +314,7 @@ class SelfCheck(object):
utime.sleep(sleep_time) utime.sleep(sleep_time)
sleep_time *= 2 sleep_time *= 2
del locator
if gps_data: if gps_data:
return True return True