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: 1. api doc; 2. disable alert.py; 3. change location info upload; 4. tracker function;

Browse Source
This commit is contained in:
JackSun-qc 2022-03-16 19:43:12 +08:00
parent 6b935ecf57
commit 3b315528bc
8 changed files with 300 additions and 277 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

81
code/alert.py
View File

@ -1,6 +1,4 @@
import _thread
from queue import Queue
from usr import settings
from usr.logging import getLogger
from usr.common import Singleton
@ -8,76 +6,25 @@ from usr.common import Singleton
log = getLogger(__name__)
ALERTCODE = {
20000: 'fault_alert',
30002: 'low_power_alert',
30003: 'over_speed_alert',
30004: 'sim_out_alert',
30005: 'disassemble_alert',
40000: 'drive_behavior_alert',
50001: 'sos_alert',
}
FAULT_CODE = {
20001: 'net_error',
20002: 'gps_error',
20003: 'temp_sensor_error',
20004: 'light_sensor_error',
20005: 'move_sensor_error',
20006: 'mike_error',
}
DRIVE_BEHAVIOR_CODE = {
40001: 'quick_start',
40002: 'quick_stop',
40003: 'quick_turn_left',
40004: 'quick_turn_right',
}
class AlertMonitorError(Exception):
def __init__(self, value):
self.value = value
def __str__(self):
return repr(self.value)
def alert_process(argv):
'''
alert_signals_queue data format
(30001, {'local_time': 1646731286})
(40000, {'drive_behavior_code': 40001, 'local_time': 1646731286})
'''
self = argv
while True:
data = self.alert_signals_queue.get()
if data:
log.info('alert_signals_queue data: ', data)
if ALERTCODE.get(data[0]):
current_settings = settings.settings.get()
alert_status = current_settings.get('app', {}).get('sw_' + ALERTCODE.get(data[0]))
if alert_status:
if self.alert_read_cb:
self.alert_read_cb(ALERTCODE.get(data[0]), data[1])
else:
log.warn('Alert callback is not defined.')
else:
log.warn('%s status is %s' % (ALERTCODE.get(data[0]), alert_status))
else:
log.error('altercode (%s) is not exists. alert info: %s' % data)
class AlertMonitor(Singleton):
'''
Recv alert signals and process them
'''
def __init__(self, alert_read_cb=None):
self.alert_read_cb = alert_read_cb
self.alert_signals_queue = Queue(maxsize=64)
_thread.start_new_thread(alert_process, (self,))
def post_alert(self, alert_code, alert_info):
self.alert_signals_queue.put((alert_code, alert_info))
if settings.ALERTCODE.get(alert_code):
current_settings = settings.settings.get()
alert_status = current_settings.get('app', {}).get('sw_' + settings.ALERTCODE.get(alert_code))
if alert_status:
if self.alert_read_cb:
return self.alert_read_cb(settings.ALERTCODE.get(alert_code), alert_info)
else:
log.warn('Alert callback is not defined.')
else:
log.warn('%s status is %s' % (settings.ALERTCODE.get(alert_code), alert_status))
else:
log.error('altercode (%s) is not exists. alert info: %s' % (alert_code, alert_info))
return False

90
code/location.py
View File

@ -1,6 +1,6 @@
import ure
import utime
# import utime
import _thread
import cellLocator
import usr.settings as settings
@ -139,9 +139,6 @@ class CellLocator(object):
self.cellLocator_cfg = cellLocator_cfg
def read(self):
return ['LBS']
def get_location(self):
return cellLocator.getLocation(
self.cellLocator_cfg['serverAddr'],
self.cellLocator_cfg['port'],
@ -150,34 +147,37 @@ class CellLocator(object):
self.cellLocator_cfg['profileIdx']
)
def read_quecIot(self):
return ['LBS']
class WiFiLocator(object):
def __init__(self, wifiLocator_cfg):
self.wifilocator_obj = wifilocator(wifiLocator_cfg['token'])
def read(self):
return []
def get_location(self):
return self.wifilocator_obj.getwifilocator()
def read_quecIot(self):
return []
def loc_worker(argv):
self = argv
while True:
trigger = self.trigger_queue.get()
if trigger:
data = None
retry = 0
while retry < 3:
data = self.read()
if data:
break
else:
retry += 1
utime.sleep(1)
if data and self.loc_read_cb:
self.loc_read_cb(data)
# def loc_worker(argv):
# self = argv
# while True:
# trigger = self.trigger_queue.get()
# if trigger:
# data = None
# retry = 0
# while retry < 3:
# data = self.read()
# if data:
# break
# else:
# retry += 1
# utime.sleep(1)
# if data and self.loc_read_cb:
# self.loc_read_cb(data)
class Location(Singleton):
@ -185,10 +185,10 @@ class Location(Singleton):
cellLoc = None
wifiLoc = None
def __init__(self, loc_read_cb):
self.loc_read_cb = loc_read_cb
self.trigger_queue = Queue(maxsize=64)
_thread.start_new_thread(loc_worker, (self,))
def __init__(self):
# self.loc_read_cb = loc_read_cb
# self.trigger_queue = Queue(maxsize=64)
# _thread.start_new_thread(loc_worker, (self,))
self._locater_init()
def _locater_init(self):
@ -224,35 +224,45 @@ class Location(Singleton):
def read(self):
self._locater_init()
current_settings = settings.settings.get()
if self.gps:
data = []
r = self.gps.read_location_GxRMC()
if r:
data.append(r)
if current_settings['sys']['cloud'] == settings.default_values_sys._cloud.quecIot:
r = self.gps.read_location_GxRMC()
if r:
data.append(r)
r = self.gps.read_location_GxGGA()
if r:
data.append(r)
r = self.gps.read_location_GxGGA()
if r:
data.append(r)
r = self.gps.read_location_GxVTG()
if r:
data.append(r)
r = self.gps.read_location_GxVTG()
if r:
data.append(r)
if len(data):
return (settings.default_values_app._loc_method.gps, data)
if self.cellLoc:
data = self.cellLoc.read()
if current_settings['sys']['cloud'] == settings.default_values_sys._cloud.quecIot:
data = self.cellLoc.read_quecIot()
else:
data = self.cellLoc.read()
if data:
return (settings.default_values_app._loc_method.cell, data)
if self.wifiLoc:
data = self.wifiLoc.read()
if current_settings['sys']['cloud'] == settings.default_values_sys._cloud.quecIot:
data = self.wifiLoc.read_quecIot()
else:
data = self.wifiLoc.read()
if data:
return (settings.default_values_app._loc_method.wifi, data)
return ()
def trigger(self):
self.trigger_queue.put(True)
# def trigger(self):
# self.trigger_queue.put(True)

39
code/main.py
View File

@ -1,6 +1,4 @@
import osTimer
import usr.settings as settings
from usr.tracker import Tracker
from usr.logging import getLogger
@ -10,40 +8,11 @@ PROJECT_NAME = 'QuecPython_Tracker'
PROJECT_VERSION = '2.0.0'
tracker = None
current_settings = settings.settings.get()
def loc_read_cb(data):
if data:
loc_method = data[0]
loc_data = data[1]
log.info("loc_method:", loc_method)
log.info("loc_data:", loc_data)
if loc_method == settings.default_values_app._loc_method.gps:
data_type = tracker.remote.DATA_LOCA_GPS
else:
data_type = tracker.remote.DATA_LOCA_NON_GPS
tracker.remote.post_data(data_type, loc_data)
def alert_read_cb(data):
if data:
data_type = tracker.remote.DATA_NON_LOCA
alert_data = {data[0]: data[1]}
tracker.remote.post_data(data_type, alert_data)
tracker = Tracker(loc_read_cb, alert_read_cb)
def loc_timer_cb(argv):
tracker.locator.trigger()
def main():
tracker = Tracker()
log.info(tracker.locator.read())
if __name__ == '__main__':
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)
main()

6
code/remote.py
View File

@ -363,7 +363,11 @@ class Remote(Singleton):
return True
def set_block_io(self, val):
self.block_io = val
try:
self.block_io = val
return True
except:
return False
def check_ota(self):
current_settings = settings.settings.get()

68
code/settings.py
View File

@ -12,6 +12,32 @@ PROJECT_NAME = 'QuecPython_Tracker'
PROJECT_VERSION = '2.0.0'
ALERTCODE = {
20000: 'fault_alert',
30002: 'low_power_alert',
30003: 'over_speed_alert',
30004: 'sim_out_alert',
30005: 'disassemble_alert',
40000: 'drive_behavior_alert',
50001: 'sos_alert',
}
FAULT_CODE = {
20001: 'net_error',
20002: 'gps_error',
20003: 'temp_sensor_error',
20004: 'light_sensor_error',
20005: 'move_sensor_error',
20006: 'mike_error',
}
DRIVE_BEHAVIOR_CODE = {
40001: 'quick_start',
40002: 'quick_stop',
40003: 'quick_turn_left',
40004: 'quick_turn_right',
}
tracker_settings_file = '/usr/tracker_settings.json'
_settings_lock = _thread.allocate_lock()
@ -240,20 +266,24 @@ class Settings(Singleton):
@settings_lock('Settings.init')
def init(self):
default_values_sys.locator_init_params = default_values_sys._get_locator_init_params(default_values_app.loc_method)
default_values_sys.cloud_init_params = default_values_sys._get_cloud_init_params(default_values_sys.cloud)
try:
default_values_sys.locator_init_params = default_values_sys._get_locator_init_params(default_values_app.loc_method)
default_values_sys.cloud_init_params = default_values_sys._get_cloud_init_params(default_values_sys.cloud)
default_settings_app = {k: v for k, v in default_values_app.__dict__.items() if not k.startswith('_')}
default_settings_sys = {k: v for k, v in default_values_sys.__dict__.items() if not k.startswith('_')}
default_settings = {'app': default_settings_app, 'sys': default_settings_sys}
default_settings_app = {k: v for k, v in default_values_app.__dict__.items() if not k.startswith('_')}
default_settings_sys = {k: v for k, v in default_values_sys.__dict__.items() if not k.startswith('_')}
default_settings = {'app': default_settings_app, 'sys': default_settings_sys}
if not ql_fs.path_exists(tracker_settings_file):
with open(tracker_settings_file, 'w') as f:
ujson.dump(default_settings, f)
self.current_settings = dict(default_settings)
else:
with open(tracker_settings_file, 'r') as f:
self.current_settings = ujson.load(f)
if not ql_fs.path_exists(tracker_settings_file):
with open(tracker_settings_file, 'w') as f:
ujson.dump(default_settings, f)
self.current_settings = dict(default_settings)
else:
with open(tracker_settings_file, 'r') as f:
self.current_settings = ujson.load(f)
return True
except:
return False
@settings_lock('Settings.get')
def get(self):
@ -331,12 +361,20 @@ class Settings(Singleton):
@settings_lock('Settings.save')
def save(self):
with open(tracker_settings_file, 'w') as f:
ujson.dump(self.current_settings, f)
try:
with open(tracker_settings_file, 'w') as f:
ujson.dump(self.current_settings, f)
return True
except:
return False
@settings_lock('Settings.reset')
def reset(self):
uos.remove(tracker_settings_file)
try:
uos.remove(tracker_settings_file)
return True
except:
return False
settings = Settings()

4
code/timer.py
View File

@ -57,7 +57,7 @@ class TrackerTimer(Singleton):
self.quecthing_ota_timer()
def loc_timer(self):
self.tracker.locator.trigger()
self.tracker.loc_report()
def battery_timer(self):
current_settings = settings.settings.get()
@ -67,7 +67,7 @@ class TrackerTimer(Singleton):
self.tracker.energy_led_show(energy)
if current_settings['app']['sw_low_power_alert']:
if energy <= current_settings['app']['low_power_alert_threshold']:
self.tracker.alert.post_alert(30002, {'local_time': utime.mktime(utime.localtime())})
self.tracker.alert_report(30002, {'local_time': utime.mktime(utime.localtime())})
self.tracker.machine_info_report()
if energy <= current_settings['app']['low_power_shutdown_threshold']:
self.tracker.machine_info_report(power_switch=False, block_io=True)

120
code/tracker.py
View File

@ -8,7 +8,6 @@ from usr.sensor import Sensor
from usr.remote import Remote
from usr.battery import Battery
from usr.common import Singleton
from usr.alert import AlertMonitor
from usr.logging import getLogger
from usr.location import Location, GPS
from usr.timer import TrackerTimer, LEDTimer
@ -33,8 +32,7 @@ class Tracker(Singleton):
self.energy_led = LED()
self.running_led = LED()
self.sensor = Sensor()
self.locator = Location(self.loc_read_cb)
self.alert = AlertMonitor(self.alert_read_cb)
self.locator = Location()
self.battery = Battery()
self.remote = Remote(self)
@ -48,7 +46,17 @@ class Tracker(Singleton):
self.usb = USB()
self.usb.setCallback(self.usb_callback)
def loc_read_cb(self, data):
def loc_report(self):
data = None
retry = 0
while retry < 3:
data = self.locator.read()
if data:
break
else:
retry += 1
utime.sleep(1)
if data:
loc_method = data[0]
loc_data = data[1]
@ -56,28 +64,75 @@ class Tracker(Singleton):
data_type = self.remote.DATA_LOCA_GPS
else:
data_type = self.remote.DATA_LOCA_NON_GPS
self.remote.post_data(data_type, loc_data)
def alert_read_cb(self, *data):
if data:
data_type = self.remote.DATA_NON_LOCA
alert_data = {data[0]: data[1]}
self.remote.post_data(data_type, alert_data)
current_settings = settings.settings.get()
if current_settings['sys']['cloud'] == settings.default_values_sys._cloud.quecIot:
if loc_method == settings.default_values_app._loc_method.cell:
loc_data = ['LBS']
elif loc_method == settings.default_values_app._loc_method.wifi:
loc_data = []
def machine_info_report(self, power_switch=True, block_io=True):
current_settings = settings.settings.get()
self.locator.trigger()
# TODO: Other Machine Info.
machine_info = {
'power_switch': power_switch,
'energy': self.battery.energy(),
'local_time': utime.mktime(utime.localtime()),
'ota_status': current_settings['sys']['ota_status']
}
machine_info.update(current_settings['app'])
if self.remote.block_io != block_io:
self.remote.set_block_io(block_io)
self.remote.post_data(self.remote.DATA_NON_LOCA, machine_info)
return self.remote.post_data(data_type, loc_data)
else:
log.warn('Location data is not ready.')
return False
def alert_report(self, alert_code, alert_info):
if settings.ALERTCODE.get(alert_code):
current_settings = settings.settings.get()
alert_status = current_settings.get('app', {}).get('sw_' + settings.ALERTCODE.get(alert_code))
if alert_status:
data_type = self.remote.DATA_NON_LOCA
alert_data = {settings.ALERTCODE.get(alert_code): alert_info}
return self.remote.post_data(data_type, alert_data)
else:
log.warn('%s status is %s' % (settings.ALERTCODE.get(alert_code), alert_status))
else:
log.error('altercode (%s) is not exists. alert info: %s' % (alert_code, alert_info))
return False
def machine_info_report(self, power_switch=True):
if self.loc_report():
# TODO: Other Machine Info.
current_settings = settings.settings.get()
machine_info = {
'power_switch': power_switch,
'energy': self.battery.energy(),
'local_time': utime.mktime(utime.localtime()),
'ota_status': current_settings['sys']['ota_status']
}
machine_info.update(current_settings['app'])
return self.remote.post_data(self.remote.DATA_NON_LOCA, machine_info)
return False
def machine_check(self):
net_check_res = self.check.net_check()
gps_check_res = self.check.gps_check()
sensor_check_res = self.check.sensor_check()
alert_code = 20000
fault_code = 0
if net_check_res and gps_check_res and sensor_check_res:
self.running_led.period = 2
else:
self.running_led.period = 0.5
if not net_check_res:
fault_code = 20001
alert_info = {'fault_code': fault_code, 'local_time': utime.mktime(utime.localtime())}
if not gps_check_res:
fault_code = 20002
alert_info = {'fault_code': fault_code, 'local_time': utime.mktime(utime.localtime())}
if not sensor_check_res:
# TODO: Need To Check What Sensor Error To Report.
pass
self.alert_report(alert_code, alert_info)
self.machine_info_report()
return fault_code
def energy_led_show(self, energy):
current_settings = settings.settings.get()
@ -89,23 +144,6 @@ class Tracker(Singleton):
elif current_settings['app']['low_power_alert_threshold'] < energy:
self.energy_led.period = 0
def machine_check(self):
net_check_res = self.check.net_check()
gps_check_res = self.check.gps_check()
sensor_check_res = self.check.sensor_check()
if net_check_res and gps_check_res and sensor_check_res:
self.running_led.period = 2
else:
self.running_led.period = 0.5
if not net_check_res:
self.alert.post_alert(20000, {'fault_code': 20001, 'local_time': utime.mktime(utime.localtime())})
if not gps_check_res:
self.alert.post_alert(20000, {'fault_code': 20002, 'local_time': utime.mktime(utime.localtime())})
if not sensor_check_res:
# TODO: Need To Check What Sensor Error To Report.
pass
self.machine_info_report()
def pwk_callback(self, status):
if status == 0:
log.info('PowerKey Release.')

169
docs/tracker公版方案用户指导手册.md
View File

@ -14,57 +14,6 @@ from usr.tracker import Tracker
tracker = Tracker()
```
### alert 告警功能
该功能提供`post_alert`方法, 将定义好的报警编码与报警信息上报到云端。
>`tracker.alert.post_alert`
- 例:
```python
import utime
alert_code = 20000
alert_info = {
'fault_code': 20001,
'local_time': utime.mktime(utime.localtime())
}
tracker.alert.post_alert(alert_code, alert_info)
```
- 参数:
|参数|参数类型|参数说明|
|:---|---|---|
|alert_code|int|报警编码|
|alert_info|dict|报警信息, key为标识符, value为具体信息|
- 返回值:
无
- 报警编码与标识符
|报警编码|子码|标识符|说明|
|:---|:---|:---|:---|
|20000|-|`fault_alert`|故障报警|
|-|20001|`net_error`|网络异常|
|-|20002|`gps_error`|GPS异常|
|-|20003|`temp_sensor_error`|温度传感器异常|
|-|20004|`light_sensor_error`|照度传感器异常|
|-|20005|`move_sensor_error`|三轴加速度传感器异常|
|-|20006|`mike_error`|麦克风异常|
|30002|-|`low_power_alert`|低电量报警|
|30003|-|`over_speed_alert`|超速报警|
|30004|-|`sim_out_alert`|拔卡报警|
|30005|-|`disassemble_alert`|拆卸报警|
|40000|-|`drive_behavior_alert`|驾驶行为监测报警|
|-|40001|`quick_start`|急起|
|-|40002|`quick_stop`|急停|
|-|40003|`quick_turn_left`|左急转弯|
|-|40004|`quick_turn_right`|右急转弯|
|50001|-|`sos_alert`|SOS求救报警|
### battery 电池功能
改功能提供`energy`方法查询当前电池电量。
@ -86,9 +35,9 @@ battery_energy = tracker.battery.energy()
### locator 定位功能
该功能提供了`read`和`trigger`两个方法, 定位模式, 定位方式, 定位信息上报模式在`settings`模块中配置, 亦可通过云端远程进行消息控制。
该功能提供了`read`查询当前模块的定位信息, 定位模式, 定位方式, 定位信息上报模式在`settings`模块中配置, 亦可通过云端远程进行消息控制。
#### `read` 查询当前发送云端的定位信息。
#### `read` 查询当前模块的定位信息。
>`tracker.locator.read`
@ -121,22 +70,6 @@ location_info = tracker.locator.read()
- `loc_method` -- 4
- `[]`
#### `trigger` 用于立即向云端报告设备定位信息功能。
>`tracker.locator.trigger`
- 例:
```python
tracker.locator.trigger()
```
- 参数:
无
- 返回值:
无
### remote 信息通信功能
#### `post_data` 向云端进行消息发送功能。
@ -217,7 +150,7 @@ tracker.remote.post_data(tracker.remote.DATA_NON_LOCA, data)
- 返回值:
返回`bool`类型数据, `True`发送成功, `False`发送失败。
返回`bool`类型数据, `True`成功, `False`失败。
#### `set_block_io` 设置消息发送为阻塞或非阻塞方式。
@ -235,6 +168,78 @@ tracker.remote.set_block_io(False)
|:---|---|---|
|val|bool|是否阻塞发送消息, 默认True, True:阻塞;False:非阻塞|
- 返回值:
返回`bool`类型数据, `True`成功, `False`失败。
### alert_report 告警功能
该功能提供`alert_report`方法, 将定义好的报警编码与报警信息上报到云端。
>`tracker.alert_report`
- 例:
```python
import utime
alert_code = 20000
alert_info = {
'fault_code': 20001,
'local_time': utime.mktime(utime.localtime())
}
tracker.alert_report(alert_code, alert_info)
```
- 参数:
|参数|参数类型|参数说明|
|:---|---|---|
|alert_code|int|报警编码|
|alert_info|dict|报警信息, key为标识符, value为具体信息|
- 返回值:
返回`bool`类型数据, `True`成功, `False`失败。
- 报警编码与标识符
|报警编码|子码|标识符|说明|
|:---|:---|:---|:---|
|20000|-|`fault_alert`|故障报警|
|-|20001|`net_error`|网络异常|
|-|20002|`gps_error`|GPS异常|
|-|20003|`temp_sensor_error`|温度传感器异常|
|-|20004|`light_sensor_error`|照度传感器异常|
|-|20005|`move_sensor_error`|三轴加速度传感器异常|
|-|20006|`mike_error`|麦克风异常|
|30002|-|`low_power_alert`|低电量报警|
|30003|-|`over_speed_alert`|超速报警|
|30004|-|`sim_out_alert`|拔卡报警|
|30005|-|`disassemble_alert`|拆卸报警|
|40000|-|`drive_behavior_alert`|驾驶行为监测报警|
|-|40001|`quick_start`|急起|
|-|40002|`quick_stop`|急停|
|-|40003|`quick_turn_left`|左急转弯|
|-|40004|`quick_turn_right`|右急转弯|
|50001|-|`sos_alert`|SOS求救报警|
### `loc_report` 用于立即向云端报告设备定位信息功能。
>`tracker.loc_report`
- 例:
```python
tracker.loc_report()
```
- 参数:
无
- 返回值:
返回`bool`类型数据, `True`成功, `False`失败。
### machine_info_report 机器信息上报功能
该模块实现了机器信息的汇总上报功能, 会将机器的位置信息, 开机状态, 电池电量等相关设置信息全部实时上报云端。
@ -251,7 +256,7 @@ tracker.machine_info_report()
- 返回值:
无
返回`bool`类型数据, `True`成功, `False`失败。
### machine_check 机器自检功能
@ -269,7 +274,17 @@ tracker.machine_check()
- 返回值:
无
返回码，数值类型`int`
|返回码|说明|
|:---|:---|
|0|无异常|
|20001|网络异常|
|20002|GPS异常|
|20003|温度传感器异常|
|20004|照度传感器异常|
|20005|三轴加速度传感器异常|
|20006|麦克风异常|
## settings
@ -303,7 +318,7 @@ settings.init()
- 返回值:
无
返回`bool`类型数据, `True`成功, `False`失败。
### get 获取配置参数
@ -387,7 +402,9 @@ settings.set(opt, val)
- 返回值:
无
返回`bool`类型数据, `True`成功, `False`失败。
- 配置参数标识符列表，见`移远云物模型属性功能定义标识符`表中可写权限数据
### save 持久化保存配置参数
@ -403,7 +420,7 @@ settings.save()
- 返回值:
无
返回`bool`类型数据, `True`成功, `False`失败。
### reset 重置配置参数
@ -419,4 +436,4 @@ settings.reset()
- 返回值:
无
返回`bool`类型数据, `True`成功, `False`失败。