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

update: settings.py&main.py&location.py

Browse Source
This commit is contained in:
JackSun-qc 2022-03-04 13:21:48 +08:00
parent 2e9177cd0f
commit 0998516e16
11 changed files with 150 additions and 94 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
code/dev_info.py
View File

@ -1,7 +1,14 @@
version = '1.0.0'
# quecIot = {
# 'PK': 'p1122c',
# 'PS': 'UG9wMGxwS2t1UE5x'
# }
quecIot = {
'PK': 'p1122c',
'PS': 'UG9wMGxwS2t1UE5x'
'PK': 'p11275',
'PS': 'Q0ZQQndaN3pCUFd6',
'DK': 'trackdev0304',
'DS': ''
}

71
code/location.py
View File

@ -1,32 +1,35 @@
import usr.settings as settings
from machine import UART
import cellLocator
import wifilocator
import ure
import _thread
import cellLocator
from wifilocator import wifilocator
import usr.settings as settings
from queue import Queue
from machine import UART
gps_data_retrieve_queue = None
'''
GPS data retrieve callback from UART
When data comes, send a message to queue of data length
'''
def gps_data_retrieve_cb(para_list):
'''
GPS data retrieve callback from UART
When data comes, send a message to queue of data length
'''
global gps_data_retrieve_queue
toRead = para_list[2]
if toRead:
gps_data_retrieve_queue.put(toRead)
'''
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.
'''
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
@ -35,6 +38,7 @@ def gps_data_retrieve_thread(argv):
if toRead:
self.gps_data = self.uart_read(toRead).decode()
class GPS(UART):
def __init__(self, gps_cfg):
global gps_data_retrieve_queue
@ -70,19 +74,28 @@ class GPS(UART):
else:
return ""
class CellLocator(cellLocator):
class CellLocator(object):
def __init__(self, cellLocator_cfg):
self.cellLocator_cfg = cellLocator_cfg
def read(self):
return super(CellLocator, self).getLocation(self.cellLocator_cfg['serverAddr'], self.cellLocator_cfg['port'], self.cellLocator_cfg['token'], self.cellLocator_cfg['timeout'], self.cellLocator_cfg['profileIdx'])
return cellLocator.getLocation(
self.cellLocator_cfg['serverAddr'],
self.cellLocator_cfg['port'],
self.cellLocator_cfg['token'],
self.cellLocator_cfg['timeout'],
self.cellLocator_cfg['profileIdx']
)
class WiFiLocator(wifilocator):
def __init__(self, wifiLocator_cfg):
super(WiFiLocator, self).__init__(wifiLocator_cfg['token'])
super(wifilocator, self).__init__(wifiLocator_cfg['token'])
def read(self):
return super(WiFiLocator, self).getwifilocator()
return super(wifilocator, self).getwifilocator()
def loc_worker(argv):
self = argv
@ -93,33 +106,35 @@ def loc_worker(argv):
if data and self.read_cb:
self.read_cb(data)
class Location(GPS, CellLocator, WiFiLocator):
gps_enabled = False
cellLoc_enabled = False
wifiLoc_enabled = False
def __init__(self, read_cb, **kw):
current_settings = settings.get()
current_settings, _ = settings.get()
# current_settings = settings.current_settings
self.read_cb = read_cb
if current_settings['app']['loc_method'] & settings.default_values_app._loc_method.gps:
if 'gps_cfg' in kw:
super(Location, self).__init__(kw['gps_cfg'])
super(GPS, self).__init__(kw['gps_cfg'])
self.gps_enabled = True
else:
raise ValueError('Invalid gps init parameters.')
if current_settings['app']['loc_method'] & settings.default_values_app._loc_method.cell:
if 'cellLocator_cfg' in kw:
super(GPS, self).__init__(kw['cellLocator_cfg'])
super(CellLocator, self).__init__(kw['cellLocator_cfg'])
self.cellLoc_enabled = True
else:
raise ValueError('Invalid cell-locator init parameters.')
if current_settings['app']['loc_method'] & settings.default_values_app._loc_method.wifi:
if 'wifiLocator_cfg' in kw:
super(CellLocator, self).__init__(kw['wifiLocator_cfg'])
super(WiFiLocator, self).__init__(kw['wifiLocator_cfg'])
self.wifiLoc_enabled = True
else:
raise ValueError('Invalid wifi-locator init parameters.')
@ -130,11 +145,11 @@ class Location(GPS, CellLocator, WiFiLocator):
def read(self):
if self.gps_enabled:
data = []
r = super(Location, self).read_location_GxRMC()
r = super(GPS, self).read_location_GxRMC()
if r:
data.append(r)
r = super(Location, self).read_location_GxGGA()
r = super(GPS, self).read_location_GxGGA()
if r:
data.append(r)
@ -142,12 +157,12 @@ class Location(GPS, CellLocator, WiFiLocator):
return (settings.default_values_app._loc_method.gps, data)
if self.cellLoc_enabled:
data = super(GPS, self).read()
data = super(CellLocator, self).read()
if data:
return (settings.default_values_app._loc_method.cell, data)
if self.wifiLoc_enabled:
data = super(CellLocator, self).read()
data = super(WiFiLocator, self).read()
if data:
return (settings.default_values_app._loc_method.wifi, data)

5
code/logging.py
View File

@ -1,9 +1,8 @@
import utime
def asyncLog(name, level, *message, timeout=None, await_connection=True):
'''
pass
#
@ -12,6 +11,7 @@ def asyncLog(name, level, *message, timeout=None, await_connection=True):
'''
pass
def log(name, level, *message, local_only=False, return_only=False, timeout=None):
if hasattr(utime, "strftime"):
@ -26,6 +26,7 @@ def log(name, level, *message, local_only=False, return_only=False, timeout=None
if not local_only:
pass
class Logger:
def __init__(self, name):
self.name = name

50
code/main.py
View File

@ -1,44 +1,15 @@
import osTimer
import usr.settings as settings
from usr.tracker import Tracker
from usr.logging import getLogger
import UART
import osTimer
log = getLogger(__name__)
tracker = None
log = getLogger('tracker')
PROFILE_IDX = 0
settings.init()
current_settings = settings.get()
locator_init_params = {}
if current_settings['app']['loc_method'] & settings.default_values_app._loc_method.gps:
locator_init_params['gps_cfg'] = {
'UARTn': UART.UART0,
'buadrate': 115200,
'databits': 8,
'parity': 0,
'stopbits': 1,
'flowctl': 0
}
if current_settings['app']['loc_method'] & settings.default_values_app._loc_method.cell:
locator_init_params['cellLocator_cfg'] = {
'serverAddr': 'www.queclocator.com',
'port': 80,
'token': 'xGP77d2z0i91s67n',
'timeout': 3,
'profileIdx': PROFILE_IDX
}
if current_settings['app']['loc_method'] & settings.default_values_app._loc_method.wifi:
locator_init_params['wifiLocator_cfg'] = {
'token': 'xGP77d2z0i91s67n'
}
def loc_read_cb(data):
if data:
@ -52,11 +23,18 @@ def loc_read_cb(data):
data_type = tracker.remote.DATA_LOCA_NON_GPS
tracker.remote.post_data(data_type, loc_data)
tracker = Tracker(loc_read_cb, **locator_init_params)
tracker = Tracker(loc_read_cb, **settings.locator_init_params)
def loc_timer_cb(argv):
tracker.trigger()
if (current_settings['app']['loc_mode'] & settings.default_values_app._loc_mode.cycle) and current_settings['app']['loc_cycle_period']:
loc_timer = osTimer()
loc_timer.start(current_settings['app']['loc_cycle_period'] * 1000, 1, loc_timer_cb)
if __name__ == '__main__':
settings.init()
# current_settings, locator_init_params = settings.get()
if (settings.current_settings['app']['loc_mode'] & settings.default_values_app._loc_mode.cycle) \
and settings.current_settings['app']['loc_cycle_period']:
loc_timer = osTimer()
loc_timer.start(settings.current_settings['app']['loc_cycle_period'] * 1000, 1, loc_timer_cb)

7
code/quecthing.py
View File

@ -6,14 +6,17 @@ DATA_NON_LOCA = 0x0
DATA_LOCA_NON_GPS = 0x1
DATA_LOCA_GPS = 0x2
log = getLogger('QuecThing')
# log = getLogger('QuecThing')
log = getLogger(__name__)
class QuecThing(object):
def __init__(self, pk, ps, downlink_queue):
def __init__(self, pk, ps, dk, ds, downlink_queue):
self.downlink_queue = downlink_queue
quecIot.init()
quecIot.setEventCB(self.eventCB)
quecIot.setProductinfo(pk, ps)
quecIot.setDkDs(dk, ds)
quecIot.setServer(1, "iot-south.quectel.com:2883")
quecIot.setConnmode(1)

24
code/remote.py
View File

@ -1,5 +1,5 @@
from logging import exception
# from logging import exception
import utime
import ql_fs
import ujson
@ -14,6 +14,7 @@ current_settings = settings.get()
if current_settings['sys']['cloud'] == settings.default_values_sys._cloud.quecIot:
from usr.quecthing import QuecThing
class RemoteError(Exception):
def __init__(self, value):
self.value = value
@ -21,6 +22,7 @@ class RemoteError(Exception):
def __str__(self):
return repr(self.value)
def downlink_process(argv):
self = argv
while True:
@ -38,9 +40,10 @@ def downlink_process(argv):
TODO: processing for settings or control commands from downlink channel
'''
def uplink_process(argv):
self = argv
ret = False
# ret = False
while True:
'''
@ -70,14 +73,14 @@ def uplink_process(argv):
# Try at most 3 times to post data to server.
while not self.cloud.post_data(data_type, data):
ntry += 1
if ntry >= 3: # Data post failed after 3 times, maybe network error?
raise RemoteError('Data post failed.') # Stop posting more data, go to exception handler.
if ntry >= 3: # Data post failed after 3 times, maybe network error?
raise RemoteError('Data post failed.') # Stop posting more data, go to exception handler.
utime.sleep(1)
else:
value.pop(i) # Pop data from data-list after posting sueecss.
need_refresh = True # Data in hist-dictionary changed, need to refresh history file.
value.pop(i) # Pop data from data-list after posting sueecss.
need_refresh = True # Data in hist-dictionary changed, need to refresh history file.
except Exception:
while True: # Put all data in uplink_queue to hist-dictionary.
while True: # Put all data in uplink_queue to hist-dictionary.
if self.uplink_queue.size():
msg = self.uplink_queue.get()
if msg:
@ -123,6 +126,7 @@ def uplink_process(argv):
else:
continue
class Remote(object):
_history = '/usr/tracker_data.hist'
@ -130,7 +134,7 @@ class Remote(object):
self.downlink_queue = Queue(maxsize=64)
self.uplink_queue = Queue(maxsize=64)
if current_settings['sys']['cloud'] == settings.default_values_sys._cloud.quecIot:
self.cloud = QuecThing(dev_info.quecIot['PK'], dev_info.quecIot['PS'], self.downlink_queue)
self.cloud = QuecThing(dev_info.quecIot['PK'], dev_info.quecIot['PS'], dev_info.quecIot['DK'], dev_info.quecIot['DS'], self.downlink_queue)
self.DATA_NON_LOCA = QuecThing.DATA_NON_LOCA
self.DATA_LOCA_NON_GPS = QuecThing.DATA_LOCA_NON_GPS
self.DATA_LOCA_GPS = QuecThing.DATA_LOCA_GPS
@ -190,7 +194,7 @@ class Remote(object):
if key not in res:
res[key] = []
res[key].append(data)
return self.refresh_history(res)
@ -198,7 +202,7 @@ class Remote(object):
def refresh_history(self, hist_dict):
try:
with open(self._history, 'w') as f:
ujson.dump(hist_dict, f, indent = 4)
ujson.dump(hist_dict, f, indent=4)
return True
except Exception:
return False

4
code/selfcheck.py
View File

@ -8,10 +8,12 @@ def netcheck():
# return True if OK
pass
def gps_check():
# return True if OK
pass
def sensor_check():
# return True if OK
pass
pass

2
code/sensor.py
View File

@ -1,4 +1,4 @@
class Sensor():
def __init__(self):
pass
pass

69
code/settings.py
View File

@ -3,9 +3,11 @@ import ql_fs
import ujson
import uos
import ure
from machine import UART
tracker_settings_file = '/usr/tracker_settings.json'
class default_values_app(object):
'''
App default settings
@ -50,6 +52,8 @@ class default_values_app(object):
sw_ota = True
sw_auto_upgrade = True
sw_ota_auto_upgrade = True
sw_voice_listen = False
@ -88,28 +92,66 @@ class default_values_sys(object):
cloud = _cloud.quecIot
default_settings_app = {k:v for k,v in default_values_app.__dict__.items() if not k.startswith('_')}
default_settings_app = {k: v for k, v in default_values_app.__dict__.items() if not k.startswith('_')}
current_settings_app = {}
default_settings_sys = {k:v for k,v in default_values_sys.__dict__.items() if not k.startswith('_')}
default_settings_sys = {k: v for k, v in default_values_sys.__dict__.items() if not k.startswith('_')}
current_settings_sys = {}
default_settings = {'app':default_settings_app, 'sys':default_settings_sys}
default_settings = {'app': default_settings_app, 'sys': default_settings_sys}
current_settings = {}
PROFILE_IDX = 0
locator_init_params = {}
all_locator_init_params = {
'gps_cfg': {
'UARTn': UART.UART0,
'buadrate': 115200,
'databits': 8,
'parity': 0,
'stopbits': 1,
'flowctl': 0
},
'cellLocator_cfg': {
'serverAddr': 'www.queclocator.com',
'port': 80,
'token': 'xGP77d2z0i91s67n',
'timeout': 3,
'profileIdx': PROFILE_IDX
},
'wifiLocator_cfg': {
'token': 'xGP77d2z0i91s67n'
}
}
def init():
global current_settings
global locator_init_params
if not ql_fs.path_exists(tracker_settings_file):
with open(tracker_settings_file, 'w') as f:
ujson.dump(default_settings, f, indent = 4)
ujson.dump(default_settings, f)
current_settings = dict(default_settings)
else:
with open(tracker_settings_file, 'r') as f:
current_settings = ujson.load(f)
if current_settings['app']['loc_method'] & default_values_app._loc_method.gps:
locator_init_params['gps_cfg'] = all_locator_init_params['gps_cfg']
elif current_settings['app']['loc_method'] & default_values_app._loc_method.cell:
locator_init_params['cellLocator_cfg'] = all_locator_init_params['cellLocator_cfg']
elif current_settings['app']['loc_method'] & default_values_app._loc_method.wifi:
locator_init_params['wifiLocator_cfg'] = all_locator_init_params['wifiLocator_cfg']
def get():
global current_settings
return current_settings
global locator_init_params
return current_settings, locator_init_params
def set(opt, val):
if opt in current_settings['app']:
@ -129,7 +171,7 @@ def set(opt, val):
return False
current_settings['app'][opt] = val
return True
elif opt == 'loc_mode':
if not isinstance(val, int):
return False
@ -145,7 +187,6 @@ def set(opt, val):
return False
current_settings['app'][opt] = val
return True
elif opt == 'low_power_alert_threshold' or opt == 'low_power_shutdown_threshold':
if not isinstance(val, int):
return False
@ -153,10 +194,11 @@ def set(opt, val):
return False
current_settings['app'][opt] = val
return True
elif opt == 'sw_ota' or opt == 'sw_ota_auto_upgrade' or opt == 'sw_voice_listen' or opt == 'sw_voice_record' \
or opt == 'sw_fault_alert' or opt == 'sw_low_power_alert' or opt == 'sw_over_speed_alert' or opt == 'sw_sim_out_alert' \
or opt == 'sw_disassemble_alert' or opt == 'sw_drive_behavior_alert':
elif opt in (
'sw_ota', 'sw_auto_upgrade', 'sw_electric_fence', 'sw_phone_call',
'sw_voice_record', 'sw_jtt808', 'sw_fault_alert', 'sw_low_power_alert',
'sw_over_speed_alert', 'sw_sim_out_alert', 'sw_disassemble_alert',
'sw_vibrate_alert', 'sw_drive_behavior_alert'):
if not isinstance(val, bool):
return False
current_settings['app'][opt] = val
@ -168,13 +210,16 @@ def set(opt, val):
else:
return False
def save():
with open(tracker_settings_file, 'w') as f:
ujson.dump(current_settings, f, indent = 4)
ujson.dump(current_settings, f, indent=4)
def reset():
uos.remove(tracker_settings_file)
class Error(Exception):
def __init__(self, value):
self.value = value

1
code/tracker.py
View File

@ -4,6 +4,7 @@ from usr.location import Remote
from usr.sensor import Sensor
from usr.led import LED
class Tracker():
def __init__(self, loc_read_cb, **kw):
self.locator = Location(loc_read_cb, **kw)

BIN
docs/tracker功能定义.xlsx
View File

Binary file not shown.