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: tracker ostime to rtc

Browse Source
This commit is contained in:
JackSun-qc 2022-03-24 13:30:00 +08:00
parent 3eee47be2b
commit 90a79fd0dc
11 changed files with 1644 additions and 145 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1569
ali_cloud_module.json Normal file
View File

File diff suppressed because it is too large Load Diff

57
code/location.py
View File

@ -1,6 +1,5 @@
import ure
# import _thread
import osTimer
import cellLocator
import usr.settings as settings
@ -36,38 +35,17 @@ def gps_data_retrieve_cb(para_list):
'''
global gps_data_retrieve_queue
toRead = para_list[2]
log.debug('gps_data_retrieve_cb para_list: %s' % str(para_list))
if toRead:
if gps_data_retrieve_queue.size() >= 8:
gps_data_retrieve_queue.get()
gps_data_retrieve_queue.put(toRead)
def gps_data_retrieve_thread(argv):
'''
GPS data retrieve thread
Receive a message from queue of data length.
Then read the corresponding length of data from UART into self.gps_data.
So self.gps_data will be updated immediately once the data comes to UART that
the self.gps_data could keep the latest data.
'''
global gps_data_retrieve_queue
self = argv
while True:
current_settings = settings.settings.get()
if current_settings['sys']['gps_mode'] & settings.default_values_sys._gps_mode.external:
self.gps_data = self.uart_read().decode()
elif current_settings['sys']['gps_mode'] & settings.default_values_sys._gps_mode.internal:
self.gps_data = self.quecgnss_read()
class GPS(Singleton):
def __init__(self, gps_cfg):
self.gps_data = ''
self.gps_cfg = gps_cfg
self.gps_timer = osTimer()
self.gps_over_timer = osTimer()
self.break_flag = 0
current_settings = settings.settings.get()
if current_settings['sys']['gps_mode'] & settings.default_values_sys._gps_mode.external:
@ -87,43 +65,45 @@ class GPS(Singleton):
)
self.uart_obj.set_callback(gps_data_retrieve_cb)
gps_data_retrieve_queue = Queue(maxsize=8)
# _thread.start_new_thread(gps_data_retrieve_thread, (self,))
def gps_timer_callback(self, args):
self.break_flag = 1
def gps_over_timer_cb(self, args):
global gps_data_retrieve_queue
gps_data_retrieve_queue.put(0)
def uart_read(self):
log.debug('start uart_read')
global gps_data_retrieve_queue
while self.break_flag == 0:
self.gps_timer.start(200, 1, self.gps_timer_callback)
self.gps_over_timer.start(20000, 1, self.gps_over_timer_cb)
nread = gps_data_retrieve_queue.get()
data = self.uart_obj.read(nread).decode()
self.gps_timer.stop()
self.gps_over_timer.stop()
log.debug('uart_read nread')
while True:
nread = gps_data_retrieve_queue.get()
data = self.uart_obj.read(nread).decode()
rmc_data = self.read_location_GxRMC(data)
if not rmc_data:
continue
gga_data = self.read_location_GxGGA(data)
if not gga_data:
continue
vtg_data = self.read_location_GxVTG(data)
if not vtg_data:
continue
break
self.break_flag = 0
return self.uart_obj.read(nread).decode()
return data
def quecgnss_read(self):
if quecgnss.get_state() == 0:
quecgnss.gnssEnable(1)
while self.break_flag == 0:
self.gps_timer.start(500, 1, self.gps_timer_callback)
quecgnss.read(4096)
self.gps_timer.start(200, 1, self.gps_timer_callback)
data = quecgnss.read(4096)
self.gps_timer.stop()
self.break_flag = 0
data = None
while not data:
data = quecgnss.read(4096)
return data[1].decode()
def read(self):
@ -168,7 +148,6 @@ class GPS(Singleton):
def read_quecIot(self):
data = []
gps_data = self.read()
log.debug('read_quecIot gps_data: %s' % gps_data)
r = self.read_location_GxRMC(gps_data)
if r:
data.append(r)

5
code/main.py
View File

@ -15,15 +15,20 @@ def main():
log.info('PROJECT_NAME: %s' % PROJECT_NAME)
log.info('PROJECT_VERSION: %s' % PROJECT_VERSION)
current_settings = settings.get()
tracker = Tracker()
# Start Device Check
tracker.device_check()
# Start OTA Check
if current_settings['sys']['cloud'] == default_values_sys._cloud.quecIot and \
current_settings['app']['sw_ota'] is True:
tracker.remote.check_ota()
# Start PowerManage
# Init Low Energy Work Mode
tracker.power_manage.low_energy_init()
# Start RTC
tracker.power_manage.start_rtc()

4
code/mpower.py
View File

@ -34,7 +34,6 @@ class PowerManage(Singleton):
self.low_energy_method = None
self.set_period()
self.get_low_energy_method()
self.low_energy_init()
self.rtc = RTC()
self.rtc.register_callback(self.rtc_callback)
@ -46,7 +45,6 @@ class PowerManage(Singleton):
self.period = seconds
def start_rtc(self):
log.debug('start PowerManage start_rtc')
current_settings = settings.get()
if current_settings['app']['work_mode'] == default_values_app._work_mode.intelligent:
if self.tracker.locator.gps:
@ -61,11 +59,9 @@ class PowerManage(Singleton):
atime = utime.localtime(utime.mktime(utime.localtime()) + self.period)
alarm_time = [atime[0], atime[1], atime[2], atime[6], atime[3], atime[4], atime[5], 0]
self.rtc.set_alarm(alarm_time)
log.debug('rtc set_alarm')
self.rtc.enable_alarm(1)
def rtc_callback(self, args):
log.debug('start rtc_callback')
self.rtc.enable_alarm(0)
if self.low_energy_method == 'PM':
self.low_energy_queue.put('wakelock_unlock')

1
code/quecthing.py
View File

@ -113,7 +113,6 @@ class QuecThing(object):
break
self.rm_empty_data(data)
log.info('post_data res: %s' % res)
return res
def eventCB(self, data):

1
code/remote.py
View File

@ -128,7 +128,6 @@ def downlink_process(argv):
('query', ['phone_num',...])
'''
data = self.downlink_queue.get()
log.debug('downlink_queue data:', data)
DownLinkOptionObj = DownLinkOption(tracker=self.tracker)
option_attr = data[0]

1
code/settings.py
View File

@ -103,7 +103,6 @@ class default_values_app(object):
all = 0x7
class _work_mode(object):
none = 0x0
cycle = 0x1
intelligent = 0x2

51
code/test_tracker.py
View File

@ -104,14 +104,16 @@ def test_tracker():
log.info('[.] sleep 3')
utime.sleep(3)
# log.info('[.] test tracker.device_data_report()')
# device_data_report_res = tracker.device_data_report()
# log.info('[.] device_data_report_res:', device_data_report_res)
log.info('[.] tracker.device_data_report()')
device_data_report_res = tracker.device_data_report()
log.info('[.] device_data_report_res:', device_data_report_res)
# log.info('[.] sleep 3')
# utime.sleep(3)
log.info('[.] sleep 3')
utime.sleep(3)
log.info('[.] test tracker.power_manage.start_rtc()')
log.info('[.] tracker.power_manage.low_energy_init()')
tracker.power_manage.low_energy_init()
log.info('[.] tracker.power_manage.start_rtc()')
tracker.power_manage.start_rtc()
log.info('[.] end tracker.power_manage.start_rtc()')
@ -207,12 +209,9 @@ def test_pm():
def test_rtc():
rtc_queue = Queue(maxsize=8)
def rtc_cb(df):
global rtc_queue
print('rtc call back test. [%s]' % df)
rtc_queue.put('rtc')
rtc = RTC()
log.debug('rtc.datatime: %s' % str(rtc.datetime()))
@ -224,11 +223,35 @@ def test_rtc():
rtc.set_alarm(alarm_time)
log.debug('rtc.enable_alarm')
rtc.enable_alarm(1)
rtc_data = rtc_queue.get()
log.debug('rtc_data: %s' % rtc_data)
# log.debug('Power.powerDown')
# Power.powerDown()
gps_uart_queue = Queue(maxsize=64)
def test_gps_uart_cb(args):
global gps_uart_queue
log.debug('[test_gps_uart_cb] args: %s' % str(args))
if args:
if gps_uart_queue.size() >= 64:
gps_uart_queue.get()
gps_uart_queue.put(args)
def test_gps_uart():
global gps_uart_queue
gps_cfg = settings.default_values_sys._gps_cfg
uart_obj = UART(
gps_cfg['UARTn'], gps_cfg['buadrate'], gps_cfg['databits'],
gps_cfg['parity'], gps_cfg['stopbits'], gps_cfg['flowctl']
)
uart_obj.set_callback(test_gps_uart_cb)
while True:
log.debug('[test_gps_uart] gps_uart_queue get')
gps_uart_data = gps_uart_queue.get()
log.debug('[test_gps_uart] gps_uart_data: %s' % str(gps_uart_data))
gps_info = uart_obj.read(gps_uart_data[2]).decode()
log.debug('[test_gps_uart] gps_info size: %s' % len(gps_info))
def main():
@ -242,6 +265,8 @@ def main():
test_tracker()
# test_pm()
# test_rtc()
# test_gps_uart()
if __name__ == '__main__':
main()

67
code/timer.py
View File

@ -1,7 +1,5 @@
import utime
import osTimer
import usr.settings as settings
from usr.common import Singleton
from usr.logging import getLogger
@ -13,71 +11,6 @@ except ImportError:
log = getLogger(__name__)
class TrackerTimer(Singleton):
def __init__(self, tracker):
self.tracker = tracker
self.tracker_timer = osTimer()
self.tracker_timer.start(1000, 1, self.timer_callback)
self.loc_count = 0
self.battery_count = 0
self.gnss_count = 0
self.quec_ota = 0
def timer_callback(self, args):
current_settings = settings.settings.get()
self.loc_count += 1
self.battery_count += 1
self.gnss_count += 1
self.quec_ota += 1
if (current_settings['app']['work_mode'] != settings.default_values_app._work_mode.none) \
and current_settings['app']['work_cycle_period'] \
and self.loc_count >= current_settings['app']['work_cycle_period']:
self.loc_count = 0
self.loc_timer()
if self.battery_count >= 60:
self.battery_count = 0
self.battery_timer()
if current_settings['app']['loc_method'] & settings.default_values_app._loc_method.gps and \
current_settings['sys']['gps_mode'] & settings.default_values_sys._gps_mode.internal:
self.gnss_count = 0
self.gnss_timer()
def loc_timer(self):
current_settings = settings.settings.get()
if current_settings['app']['work_mode'] == settings.default_values_app._work_mode.intelligent:
if self.tracker.locator.gps:
if not self.tracker.locator.gps.read_location_GxVTG_speed():
return
elif float(self.tracker.locator.gps.read_location_GxVTG_speed()) <= 0:
return
else:
return
self.tracker.low_energy_queue.put(True)
def battery_timer(self):
current_settings = settings.settings.get()
energy = self.tracker.battery.energy()
is_charge = USB().getStatus() if USB is not None else 1
if is_charge == 0:
self.tracker.energy_led_show(energy)
if current_settings['app']['sw_low_power_alert']:
if energy <= current_settings['app']['low_power_alert_threshold']:
alert_data = self.tracker.get_alert_data(30002, {'local_time': utime.mktime(utime.localtime())})
self.tracker.device_data_report(event_data=alert_data)
if energy <= current_settings['app']['low_power_shutdown_threshold']:
self.tracker.device_data_report(power_switch=False, msg='power_down')
elif is_charge == 1:
self.tracker.energy_led_show(energy)
def gnss_timer(self):
self.tracker.locator.gps.quecgnss_read()
class LEDTimer(Singleton):
def __init__(self, tracker):
self.period = 500

9
code/tracker.py
View File

@ -17,8 +17,8 @@ from usr.common import numiter
from usr.common import Singleton
from usr.mpower import PowerManage
from usr.logging import getLogger
from usr.location import Location, GPS
# from usr.timer import TrackerTimer
from usr.location import GPS
from usr.location import Location
from usr.timer import LEDTimer
try:
@ -45,7 +45,6 @@ class Tracker(Singleton):
self.remote = Remote(self)
self.power_manage = PowerManage(self)
# self.tracker_timer = TrackerTimer(self)
self.led_timer = LEDTimer(self)
self.num_iter = numiter()
@ -74,11 +73,9 @@ class Tracker(Singleton):
return alert_data
def get_device_data(self, power_switch=True):
log.debug('start get_device_data')
device_data = {}
loc_info = self.locator.read()
log.debug('loc_info: %s' % str(loc_info))
if loc_info:
device_data.update(loc_info[1])
@ -169,7 +166,6 @@ class Tracker(Singleton):
self.power_manage.start_rtc()
def device_data_report(self, power_switch=True, event_data={}, msg=''):
log.debug('start device_data_report')
device_data = self.get_device_data(power_switch)
if event_data:
device_data.update(event_data)
@ -177,7 +173,6 @@ class Tracker(Singleton):
num = self.get_num()
topic = num + '/' + msg if msg else num
sys_bus.subscribe(topic, self.data_report_cb)
log.debug("topic: %s, device_data: %s" % (topic, device_data))
self.remote.post_data(topic, device_data)
def device_check(self):

22
quec_cloud_module.json
View File

@ -2,7 +2,7 @@
"profile":{
"tslVersion":"1.1.0",
"productKey":"p11275",
"version":"20220316114050639"
"version":"20220324132257527"
},
"properties":[
{
@ -75,34 +75,34 @@
{
"specs":{
"unit":"",
"min":"0",
"max":"15",
"min":"1",
"max":"2",
"step":"1"
},
"code":"work_mode",
"dataType":"INT",
"name":"定位工作模式",
"name":"工作模式",
"subType":"RW",
"id":25,
"sort":4,
"type":"PROPERTY",
"desc":"0: 无\n1: 循环\n2: 报警时\n4: 电话呼入\n8: 开启录音时\n15: 全部"
"desc":"1: 周期性模式\n2: 智能模式"
},
{
"specs":{
"unit":"",
"min":"",
"max":"",
"step":""
"unit":"s",
"min":"5",
"max":"2147483647",
"step":"1"
},
"code":"work_cycle_period",
"dataType":"INT",
"name":"工作循环周期",
"name":"工作模式循环周期",
"subType":"RW",
"id":26,
"sort":5,
"type":"PROPERTY",
"desc":"单位s"
"desc":""
},
{
"specs":{