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

update: settings; device error handling; device loction check; modules.

Browse Source
This commit is contained in:
JackSun 2022-05-25 11:46:41 +08:00
parent 85ae15f9df
commit 849a4b3f89
4 changed files with 53 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
code/modules

@ -1 +1 @@
Subproject commit 02f26f83e73cdf090729a32454c600c01225342c
Subproject commit 853e2ccafe0a6a715bfd86f4134759ecdbf00c16

6
code/settings.py
View File

@ -124,16 +124,16 @@ class Settings(Singleton):
elif opt in ("sw_ota", "sw_ota_auto_upgrade", "sw_voice_listen", "sw_voice_record",
"sw_fault_alert", "sw_low_power_alert", "sw_over_speed_alert",
"sw_sim_abnormal_alert", "sw_disassemble_alert", "sw_drive_behavior_alert"):
if not isinstance(val, bool):
if not isinstance(val, bool) and val not in (0, 1):
return False
self.current_settings["user_cfg"][opt] = val
self.current_settings["user_cfg"][opt] = bool(val)
return True
elif opt == "ota_status":
if not isinstance(val, dict):
return False
self.current_settings["user_cfg"][opt] = val
return True
elif opt in ("user_ota_action", "drive_behavior_code"):
elif opt in ("user_ota_action", "drive_behavior_code", "loc_gps_read_timeout", "work_mode_timeline"):
if not isinstance(val, int):
return False
self.current_settings["sys"][opt] = val

41
code/tracker_collector.py
View File

@ -54,6 +54,9 @@ class Collector(Singleton):
self.__gps_match = GPSMatch()
self.__gps_parse = GPSParse()
self.__net_status = False
self.__loc_status = False
def __format_loc_method(self, data):
"""Decimal to Binary for loc method
The first is gps, second is cell, third is wifi from binary right.
@ -349,6 +352,8 @@ class Collector(Singleton):
net_status = self.__devicecheck.net()
location_status = self.__devicecheck.location()
self.__net_status = True if net_status == (3, 1) else False
self.__loc_status = location_status
temp_status = self.__devicecheck.temp()
light_status = self.__devicecheck.light()
triaxial_status = self.__devicecheck.triaxial()
@ -422,6 +427,7 @@ class Collector(Singleton):
device_data.update({"loc_method": self.__format_loc_method(current_settings["user_cfg"]["loc_method"])})
# Get cloud location data
if self.__loc_status:
loc_info = self.__read_location()
cloud_loc = self.__read_cloud_location(loc_info)
device_data.update(cloud_loc)
@ -580,7 +586,9 @@ class Collector(Singleton):
def event_query(self, *args, **kwargs):
"""Hanle quering object model downlink message from cloud."""
return self.device_data_report()
power_switch = kwargs.get("power_switch", 1)
power_switch = bool(power_switch)
return self.device_data_report(power_switch=power_switch)
def event_ota_plain(self, *args, **kwargs):
"""Hanle OTA plain from cloud."""
@ -649,11 +657,12 @@ class Collector(Singleton):
log.debug("RRPC data: %s" % data)
self.__controller.remote_rrpc_response(message_id, data)
def power_switch(self, onoff=None):
def power_switch(self, onoff=1):
"""Control device power"""
if not self.__controller:
raise TypeError("self.__controller is not registered.")
onoff = bool(onoff)
self.event_query(power_switch=onoff)
if onoff is False:
self.__controller.power_down()
@ -749,30 +758,36 @@ class Collector(Singleton):
def low_engery_option(self, low_energy_method):
"""Business option after low energy waking up."""
log.debug("start low_engery_option")
if not self.__controller:
raise TypeError("self.__controller is not registered.")
self.report_history()
self.__controller.remote_device_report()
self.__controller.remote_ota_check()
report_flag = True
current_settings = settings.get()
if current_settings["user_cfg"]["work_mode"] == UserConfig._work_mode.intelligent:
speed_info = self.__check_speed()
if speed_info.get("current_speed") > 0:
self.device_data_report()
else:
self.device_data_report()
# TODO: Check speed by sensor
if self.__loc_status:
loc_info = self.__read_location()
gps_data = loc_info.get(_loc_method.gps)
speed_info = self.__check_speed(gps_data)
if speed_info.get("current_speed") <= 0:
report_flag = False
if report_flag is True:
self.device_status_check()
if self.__net_status:
self.__controller.remote_device_report()
self.__controller.remote_ota_check()
# Check battery low enery power down.
self.__check_battery_low_energy_power_down()
self.__controller.low_energy_start()
if low_energy_method == "PSM":
# TODO: PSM option.
pass
elif low_energy_method == "POWERDOWN":
if low_energy_method == "POWERDOWN":
self.__controller.power_down()
log.debug("end low_engery_option")
def thing_services(self, data):
log.debug("thing_services data: %s" % str(data))

15
code/tracker_devicecheck.py
View File

@ -57,6 +57,11 @@ class DeviceCheck(object):
sleep_time = 1
while retry < 5:
if retry > 0:
retry += 1
utime.sleep(sleep_time)
sleep_time *= 2
if current_settings["user_cfg"].get("loc_method"):
loc_method = current_settings["user_cfg"].get("loc_method")
elif current_settings["sys"]["base_cfg"]["LocConfig"]:
@ -64,13 +69,13 @@ class DeviceCheck(object):
else:
loc_method = 7
gps_data = self.__locator.read(loc_method)
loc_info = self.__locator.read(loc_method)
for k, v in loc_info.items():
gps_data = v
if gps_data:
break
if gps_data:
break
else:
retry += 1
utime.sleep(sleep_time)
sleep_time *= 2
if gps_data:
return True