unsigned char Gsensor_init_flag = 0;
float x,y,z;
bool SC7A20_INT_STATUS = true;
+unsigned int SC7A20_STATUS_COUNT = 0;
unsigned int SL_SC7A20_PEDO_KCAL_WRIST_SLEEP_SWAY_L_ALGO(void)
{
#if SL_Sensor_Algo_Release_Enable==0x00
/*******read int status******/
SC7A20_INT_STATUS=SL_INT_STATUS_READ();
+ if(SC7A20_INT_STATUS==1)
+ SC7A20_STATUS_COUNT++;
if((SC7A20_INT_STATUS==1)||(SL_STEP_TEMP!= SL_STEP))
{
if(SL_INT_TIME_CNT>=SL_INT_TIME_TH) SC7A20_ODR_SET(1);
return mp_const_none;
}
-//获取运动状态
+//获取运动状态次数
STATIC mp_obj_t mp_SC7A20_get_status(mp_obj_t self_in)
{
- if(SC7A20_INT_STATUS)
- return mp_obj_new_int(1);
- else
- return mp_obj_new_int(0);
+
+ return mp_obj_new_int(SC7A20_STATUS_COUNT);
+}
+//清除运动状态次数
+STATIC mp_obj_t mp_SC7A20_clean_status(mp_obj_t self_in)
+{
+ SC7A20_STATUS_COUNT = 0;
+ return mp_obj_new_int(1);
}
//获取步数
STATIC mp_obj_t mp_SC7A20_get_step(mp_obj_t self_in)
STATIC MP_DEFINE_CONST_FUN_OBJ_1(mp_activate_SC7A20_obj, mp_activate_SC7A20);//
//STATIC MP_DEFINE_CONST_FUN_OBJ_1(mp_SC7A20_poll_obj, mp_SC7A20_poll);//
STATIC MP_DEFINE_CONST_FUN_OBJ_1(mp_SC7A20_get_status_obj, mp_SC7A20_get_status);//
+STATIC MP_DEFINE_CONST_FUN_OBJ_1(mp_SC7A20_clean_status_obj, mp_SC7A20_clean_status);//
+
STATIC MP_DEFINE_CONST_FUN_OBJ_1(mp_SC7A20_get_step_obj, mp_SC7A20_get_step);//
STATIC MP_DEFINE_CONST_FUN_OBJ_1(mp_SC7A20_reset_step_obj, mp_SC7A20_reset_step);//
STATIC MP_DEFINE_CONST_FUN_OBJ_1(mp_SC7A20_get_x_y_z_obj, mp_SC7A20_get_x_y_z);//
{ MP_ROM_QSTR(MP_QSTR_activate), MP_ROM_PTR(&mp_activate_SC7A20_obj) },
//{ MP_ROM_QSTR(MP_QSTR_poll), MP_ROM_PTR(&mp_SC7A20_poll_obj) },
{ MP_ROM_QSTR(MP_QSTR_get_status), MP_ROM_PTR(&mp_SC7A20_get_status_obj) },
+ { MP_ROM_QSTR(MP_QSTR_clean_status), MP_ROM_PTR(&mp_SC7A20_clean_status_obj) },
+
{ MP_ROM_QSTR(MP_QSTR_get_step), MP_ROM_PTR(&mp_SC7A20_get_step_obj) },
{ MP_ROM_QSTR(MP_QSTR_reset_step), MP_ROM_PTR(&mp_SC7A20_reset_step_obj) },
{ MP_ROM_QSTR(MP_QSTR_get_x_y_z), MP_ROM_PTR(&mp_SC7A20_get_x_y_z_obj) },
def add_filter(flt):
event_store.add_filter(flt)
+import utime
+Txt_Log = 0
+def txt_log(data):
+ if Txt_Log != 0 :
+ with open("/ext/log.txt","a",encoding="utf-8") as fd:
+ d_time = utime.localtime()
+ data_tmp = "{}-{} {}:{}:{} :".format(d_time[1],d_time[2],d_time[3],d_time[4],d_time[5])
+ fd.write(data_tmp+data+"\n")
def callback_wrapper(callback):#按键回调
def wrapped_callback(*args, **kwargs):
import audio
import utime
import ALIPAY
+import usys
from misc import Power
factorymode_cont_y = 0
factorymode_screen = lv.obj()
factorymode_hsensor_test_label.add_style(style_font_youyuan_20, lv.PART.MAIN | lv.STATE.DEFAULT)
factorymode_hsensor_test.add_style(style_btn, lv.PART.MAIN | lv.STATE.DEFAULT)
+factorymode_cont_y+=60
+factorymode_log = lv.btn(factorymode_cont)
+factorymode_log.set_pos(0, factorymode_cont_y)
+factorymode_log.set_size(240, 60)
+factorymode_log_label = lv.label(factorymode_log)
+factorymode_log_label.set_text("日志")
+factorymode_log.set_style_pad_all(0, lv.STATE.DEFAULT)
+factorymode_log_label.align(lv.ALIGN.LEFT_MID, 70, 0)
+factorymode_log_label.set_style_text_color(lv.color_make(0xff, 0xff, 0xff), lv.STATE.DEFAULT)
+factorymode_log_label.add_style(style_font_youyuan_20, lv.PART.MAIN | lv.STATE.DEFAULT)
+factorymode_log.add_style(style_btn, lv.PART.MAIN | lv.STATE.DEFAULT)
+
factorymode_cont_y+=200
factorymode_alipay_init = lv.btn(factorymode_cont)
factorymode_alipay_init.set_pos(0, factorymode_cont_y)
factorymode_alipay_init.add_event_cb(lambda e: EventMesh.callback_wrapper(self.load_screen(e, {"screen": "FactoryModeAlipayInit"})), lv.EVENT.CLICKED, None)
factorymode_bat_test.add_event_cb(lambda e: EventMesh.callback_wrapper(self.load_screen(e, {"screen": "FactoryModeBatTest"})), lv.EVENT.CLICKED, None)
factorymode_hsensor_test.add_event_cb(lambda e: EventMesh.callback_wrapper(self.load_screen(e, {"screen": "FactoryModeHsensorTest"})), lv.EVENT.CLICKED, None)
-
+ factorymode_log.add_event_cb(lambda e: EventMesh.callback_wrapper(self.load_screen(e, {"screen": "FactoryModeLog"})), lv.EVENT.CLICKED, None)
def done_return(self):
EventMesh.publish("load_screen", {"screen": "main_menu"})
def btn8_press(self):
EventMesh.publish("load_screen", {"screen": "main_menu"})
+#显示txt日志
+FactoryModeLog_screen = lv.obj()
+FactoryModeLog_screen.center()
+FactoryModeLog_screen.set_size(240, 320)
+FactoryModeLog_screen.add_style(style_screen, lv.PART.MAIN | lv.STATE.DEFAULT)
+
+
+FactoryModeLog_label_obj = lv.obj(FactoryModeLog_screen)
+FactoryModeLog_label_obj.center()
+FactoryModeLog_label_obj.set_size(240, 290)
+FactoryModeLog_label_obj.add_style(style_screen, lv.PART.MAIN | lv.STATE.DEFAULT)
+FactoryModeLog_label_obj.set_pos(0, 0)
+
+FactoryModeLog_label = lv.label(FactoryModeLog_label_obj)
+FactoryModeLog_label.set_text("")
+FactoryModeLog_label.set_size(240, lv.SIZE.CONTENT)
+FactoryModeLog_label.set_pos(0, 0)
+FactoryModeLog_label.set_long_mode(lv.label.LONG.WRAP)
+FactoryModeLog_label.set_style_text_color(lv.color_make(0xff, 0xff, 0xff), lv.STATE.DEFAULT)
+FactoryModeLog_label.add_style(style_font_youyuan_20, lv.PART.MAIN | lv.STATE.DEFAULT)
+
+FactoryModeLogClean_btn = lv.btn(FactoryModeLog_screen)
+FactoryModeLogClean_btn.set_pos(5, 290)
+FactoryModeLogClean_btn.set_size(110, 30)
+FactoryModeLogClean_btn.add_style(style_btn_1, lv.PART.MAIN | lv.STATE.DEFAULT)
+FactoryModeLogClean_btn.set_style_pad_all(0, lv.STATE.DEFAULT)
+FactoryModeLogClean_label = lv.label(FactoryModeLogClean_btn)
+FactoryModeLogClean_label.set_text("清空日志")
+FactoryModeLogClean_label.align(lv.ALIGN.CENTER, 0, 0)
+FactoryModeLogClean_label.add_style(style_font_W_B_28, lv.PART.MAIN | lv.STATE.DEFAULT)
+FactoryModeLogClean_label.set_style_text_align(lv.TEXT_ALIGN.CENTER, 0)
+FactoryModeLogClean_label.set_style_text_color(lv.color_make(0x16, 0x2c, 0xd4), lv.STATE.DEFAULT)
+#FactoryModeLogClean_label.set_style_bg_color(lv.color_make(0x00, 0xff, 0x2a), lv.PART.MAIN | lv.STATE.DEFAULT)
+
+FactoryModeLogNext_btn = lv.btn(FactoryModeLog_screen)
+FactoryModeLogNext_btn.set_pos(120, 290)
+FactoryModeLogNext_btn.set_size(110, 30)
+FactoryModeLogNext_btn.add_style(style_btn_1, lv.PART.MAIN | lv.STATE.DEFAULT)
+FactoryModeLogNext_btn.set_style_pad_all(0, lv.STATE.DEFAULT)
+FactoryModeLogNext_label = lv.label(FactoryModeLogNext_btn)
+FactoryModeLogNext_label.set_text("下一页")
+FactoryModeLogNext_label.align(lv.ALIGN.CENTER, 0, 0)
+FactoryModeLogNext_label.add_style(style_font_W_B_28, lv.PART.MAIN | lv.STATE.DEFAULT)
+FactoryModeLogNext_label.set_style_text_align(lv.TEXT_ALIGN.CENTER, 0)
+FactoryModeLogNext_label.set_style_text_color(lv.color_make(0x16, 0x2c, 0xd4), lv.STATE.DEFAULT)
+#FactoryModeLogNext_label.set_style_bg_color(lv.color_make(0x00, 0xff, 0x2a), lv.PART.MAIN | lv.STATE.DEFAULT)
+screen_log_len = 250
+class FactoryModeLogScreen(Screen):
+ NAME = "FactoryModeLog"
+ def __init__(self):
+ super().__init__()
+ self.meta = FactoryModeLog_screen
+ self.Log_timer_count = 0
+ self.Log_timer = osTimer()
+ self.log_data = ""
+ self.log_count = 0
+ FactoryModeLogClean_btn.add_event_cb(lambda e: EventMesh.callback_wrapper(self.LogClean_cb(e)), lv.EVENT.CLICKED, None)
+ FactoryModeLogNext_btn.add_event_cb(lambda e: EventMesh.callback_wrapper(self.LogNext_cb(e)), lv.EVENT.CLICKED, None)
+ def initialization(self):
+ print("FactoryModeLogScreen")
+ self.Log_timer.start(1000,1,self.Log_timer_cb)
+ with open("/ext/log.txt","r",encoding="utf-8") as fd:
+ self.log_data = fd.read()
+ print(self.log_data)
+ if len(self.log_data) < screen_log_len:
+ FactoryModeLog_label.set_text("第1页\n"+self.log_data)
+ else:
+ FactoryModeLog_label.set_text("第1页 {}/{}\n".format(self.log_count+1,0,len(self.log_data))+self.log_data[:screen_log_len])
+ self.log_count = 0
+ def post_processor_after_initialization(self):
+ pass
+
+ def LogClean_cb(self,e):
+ self.log_data = ""
+ with open("/ext/log.txt","w",encoding="utf-8") as fd:
+ fd.write("")
+
+ def LogNext_cb(self,e):
+ self.log_count += 1
+ if self.log_count*screen_log_len > len(self.log_data):
+ if self.log_count != 1:
+ FactoryModeLog_label.set_text("末尾页\n".format(self.log_count+1)+self.log_data[(self.log_count-1)*screen_log_len:])
+ self.log_count = 0
+ else :
+ FactoryModeLog_label.set_text("第{}页 {}/{}\n".format(self.log_count+1,(self.log_count)*screen_log_len,len(self.log_data))+self.log_data[(self.log_count-1)*screen_log_len:(self.log_count)*screen_log_len])
+
+ def Log_timer_cb(self,e):
+ EventMesh.publish("system_sleep_rest")#防止黑屏
+
+
+ def Close(self):
+ self.Log_timer.stop()
+ self.log_data = ""
+ pass
+
+ def done_return(self):
+ EventMesh.publish("load_screen", {"screen": "factorymode"})
+
+ def btn1_release(self):
+ EventMesh.publish("load_screen", {"screen": "factorymode"})
+
+ def btn8_press(self):
+ EventMesh.publish("load_screen", {"screen": "main_menu"})
+
#hsensor测试
FactoryModeHsensorTest_screen = lv.obj()
FactoryModeHsensorTest_screen.center()
FactoryModeHsensorLeakLight_label.set_style_text_color(lv.color_make(0x16, 0x2c, 0xd4), lv.STATE.DEFAULT)
#FactoryModeHsensorLeakLight_label.set_style_bg_color(lv.color_make(0x00, 0xff, 0x2a), lv.PART.MAIN | lv.STATE.DEFAULT)
-
-
class FactoryModeHsensorTestScreen(Screen):
NAME = "FactoryModeHsensorTest"
def __init__(self):
EventMesh.publish("system_sleep_rest")#防止黑屏
def GET_LOCATION_DATA(self, event, msg):
- print("GPS:{}".format(msg[0]))
- print("WIFI:{}".format(msg[1]))
- print("CALL:{}".format(msg[2]))
+ # print("GPS:{}".format(msg[0]))
+ # print("WIFI:{}".format(msg[1]))
+ # print("CALL:{}".format(msg[2]))
EventMesh.publish("system_sleep_rest")#防止黑屏
_DATA = ""
time_tmp = utime.mktime(utime.localtime())
for i in range(len(gsv_tmp)) :
if i % 4 == 0:
GSV_dict[gsv_tmp[i]] = gsv_tmp[i+3]
- print("GSV_dict:{}:[{}]".format(gsv_tmp[i],gsv_tmp[i+3]))
+ #print("GSV_dict:{}:[{}]".format(gsv_tmp[i],gsv_tmp[i+3]))
if len(gsv_tmp)-i < 4:
break
count = 0
- for key in GSV_dict :
- #print("GSV_dict[key]:len:{}".format(len(GSV_dict[key])))
- if len(GSV_dict[key]) >= 1 :
- GNSS_DATA+=" {:<3}:[{:<3}] \n".format(key,GSV_dict[key])#:<3 宽度为3左对齐
- count+=1
- if count > 14:
- break
- print("GxGSV:{}".format(GNSS_DATA))
+ try:
+ for j in range(200):
+ for key in GSV_dict :
+
+ if(int(key) != j):
+ continue
+ #print("GSV_dict[key]:len:{}".format(len(GSV_dict[key])))
+ if len(GSV_dict[key]) >= 1 :
+ GNSS_DATA+=" {:<3}:[{:<3}] ".format(key,GSV_dict[key])#:<3 宽度为3左对齐
+ count+=1
+ if count % 2 == 0:
+ GNSS_DATA+="\n"
+
+ if count > 30:
+ break
+ except Exception as e:
+ usys.print_exception(e)
+ #print("GxGSV:{}".format(GNSS_DATA))
_DATA += GNSS_DATA
#FactoryModeLocation_GNSS_label.set_text(GNSS_DATA)
if msg[1] != [] : #WIFI
WIFI_len = len(msg[1][1])
- _DATA += "WIFI:------\n"
+ _DATA += "\nWIFI:------\n"
if WIFI_len > 3:
WIFI_len = 3
for i in range(WIFI_len) :
from ui_main_menu import MainMenuScreen
# from ui_WatchFaceScreen import WatchFaceScreen
- from appFactoryMode import FactoryModeLocationScreen,FactoryModeScreen ,FactoryModeAlipayInitScreen,FactoryModeBatTestScreen,FactoryModeHsensorTestScreen
+ from appFactoryMode import FactoryModeLocationScreen,FactoryModeScreen ,FactoryModeAlipayInitScreen,FactoryModeBatTestScreen,FactoryModeHsensorTestScreen,FactoryModeLogScreen
from mgr import setSmsIndexManager,ConfigStoreManager,testSmsSetIPManager,playPowerAudioManager,sosInfoCallManager,getBootOffManager,getSosFlagManager,CountdownManager, LocationManager, getVolumSizeManager,gethistoryDataManager,getUpLoadDataManager, getplayAudiodStateManager,\
AddrBookManager,MsgdetailManager,PhoneCallingManager,getOldTimeManager,getStorageDataManager,getBrightnessManager,getHrTimeManager,getTempTimeManager,getLowBatManager
ui.add_screen(FactoryModeAlipayInitScreen())
ui.add_screen(FactoryModeBatTestScreen())
ui.add_screen(FactoryModeHsensorTestScreen())
+ ui.add_screen(FactoryModeLogScreen())
ui.add_screen(PopupAlarmScreen())
except Exception as e:
usys.print_exception(e)
if self.upload_time != None :
if math.fabs(time_tmp - self.UD_time) > self.upload_time: #30分钟一次定位
self.UD_time = time_tmp
- step_tmp = self.gsensor.get_step()
- if EventMesh.publish(constants.EVENT.GET_CONNECT) == 1 and step_tmp != self.GPS_step_t :
+ status_tmp = self.gsensor.get_status()#获取运动次数
+ EventMesh.txt_log("status_count:{}".format(status_tmp))
+ if EventMesh.publish(constants.EVENT.GET_CONNECT) == 1 and status_tmp > 3 :
EventMesh.publish(constants.EVENT.START_LOCATION,constants.PROTOCOL.UD_SEND) #开始UD定位
- self.GPS_step_t = step_tmp#记录步数
+ self.gsensor.clean_status()#清除运动次数
EventMesh.publish(constants.EVENT.PUSH_STEP, self.get_step_count())
"""电池平均值"""
self.wifi_data = []
self.gps_data = []
self.cell_data = []
- gnss_r_data = ""
+ gnss_r_data = []
print("location_thread:1")
EventMesh.publish(constants.EVENT.GPSOPEN)
print("location_thread:2")
EventMesh.publish(constants.EVENT.WIFISCANOPEN)
print("location_thread:3")
utime.sleep_ms(1000)
- quecgnss.read(4096)#去掉第一次错误数据
+ self.location_timer_count
+ #quecgnss.read(4096)#去掉第一次错误数据
match_gga_data = []
- match_gsv_data = []
-
- #检查GNSS数据
- if (self.gps_data == [] or self.location_type == 1) and quecgnss.get_state() == 2:
- self.gps.__gps_nmea_data_clean()
- gnss_r_data = quecgnss.read(4096)
- print("location gnss_r_data:{}".format(gnss_r_data[1].decode()))
- if gnss_r_data[0] > 0 : #有数据
- match_gga_data = self.match.GxGGA(gnss_r_data[1]).split(",")
- match_gsv_data = self.match.GxGSV(gnss_r_data[1]).split(",")
- if self.gps.__check_gps_valid(gnss_r_data[1].decode()): #数据齐全
- match_gmc_data = self.match.GxRMC(gnss_r_data[1]).split(",")
- match_vtg_data = self.match.GxVTG(gnss_r_data[1]).split(",")
- match_gga_data = self.match.GxGGA(gnss_r_data[1]).split(",")
- gps_data_tmp = self.gps.read_coordinates(gnss_r_data[1])
- latitude_tmp = str(gps_data_tmp[1]).split(".")
- if len(latitude_tmp) >= 2 and len(latitude_tmp[1])>=5:
- latitude = "%s.%s"%(latitude_tmp[0],latitude_tmp[1][:6])
- latitude = "%s,%s"%(latitude,match_gmc_data[4] )
- longitude_tmp = str(gps_data_tmp[0]).split(".")
- if len(longitude_tmp) >= 2 and len(longitude_tmp[1])>=5:
- longitude = "%s.%s"%(longitude_tmp[0],longitude_tmp[1][:6])
- longitude = "%s,%s"%(longitude,match_gmc_data[6] ) #match_gmc_data[6] + str(gnss_r_data[1])
- high = str(gps_data_tmp[2])
- self.gps_data = {
- "latitude": latitude,
- "longitude": longitude,
- "high": high,
- "direction": match_vtg_data[1],
- "speed": match_vtg_data[5],
- "number": match_gga_data[7],
- }
- if self.location_check_data_timer_count > 5 :#5秒查一次基站数据和wifi数据
+ #match_gsv_data = []
+ gnss_r_data = []
+ if self.location_check_data_timer_count > 3 :#5秒查一次基站数据和wifi数据
self.location_check_data_timer_count = 0
#print(self.location_timer_count)
- #检查wifi数据
- if self.wifi_data == [] or (self.wifi_data != [] and self.wifi_data[0] == 0):#and self.wifi_data[0] <= 0:
- EventMesh.publish(constants.EVENT.WIFISCANOPEN)
+ GSV_dict= {}
+ GSV_sig_ok_count = 0
+ #检查GNSS数据
+ if (self.gps_data == [] or self.location_type == 1) and quecgnss.get_state() == 2:
+ self.gps.__gps_nmea_data_clean()
+ #gnss_r_data = [0,1]
+ gnss_r_data = quecgnss.read(4096)
+ #print("location gnss_r_data:{}".format(gnss_r_data[1].decode()))
+ print("gnss_r_data len:{}".format(gnss_r_data[0]))
+ if gnss_r_data[0] > 0 : #有数据
+ gsv_data = gnss_r_data[1].split("GSV")
+ if len(gsv_data) >= 2 :
+ gsv_data = gsv_data[1:]
+ for i in range(len(gsv_data)) :
+ gsv_data[i] = gsv_data[i].split("\r\n")[0]
+ gsv_tmp = gsv_data[i].split(",")
+ if len(gsv_tmp) < 7 :
+ break
+ gsv_tmp = gsv_tmp[4:]
+ #print("gsv_tmp:{}".format(gsv_tmp))
+ for i in range(len(gsv_tmp)) :
+ if i % 4 == 0:
+ GSV_dict[gsv_tmp[i]] = gsv_tmp[i+3]
+ try:
+ #print("gsv_tmp[i+3]:{}".format(gsv_tmp[i+3]))
+ if gsv_tmp[i+3] != "" :
+ if int(gsv_tmp[i+3])>=24:
+ GSV_sig_ok_count+=1
+ except Exception as e:
+ usys.print_exception(e)
+ #print("GSV_dict:{}:[{}]".format(gsv_tmp[i],gsv_tmp[i+3]))
+ if len(gsv_tmp)-i < 4:
+ break
+ #match_gga_data = self.match.GxGGA(gnss_r_data[1]).split(",")
+ #match_gsv_data = self.match.GxGSV(gnss_r_data[1]).split(",")
+ if self.gps.__check_gps_valid(gnss_r_data[1].decode()): #数据齐全
+ match_gmc_data = self.match.GxRMC(gnss_r_data[1]).split(",")
+ match_vtg_data = self.match.GxVTG(gnss_r_data[1]).split(",")
+ match_gga_data = self.match.GxGGA(gnss_r_data[1]).split(",")
+ gps_data_tmp = self.gps.read_coordinates(gnss_r_data[1])
+ latitude_tmp = str(gps_data_tmp[1]).split(".")
+ if len(latitude_tmp) >= 2 and len(latitude_tmp[1])>=5:
+ latitude = "%s.%s"%(latitude_tmp[0],latitude_tmp[1][:6])
+ latitude = "%s,%s"%(latitude,match_gmc_data[4] )
+ longitude_tmp = str(gps_data_tmp[0]).split(".")
+ if len(longitude_tmp) >= 2 and len(longitude_tmp[1])>=5:
+ longitude = "%s.%s"%(longitude_tmp[0],longitude_tmp[1][:6])
+ longitude = "%s,%s"%(longitude,match_gmc_data[6] ) #match_gmc_data[6] + str(gnss_r_data[1])
+ high = str(gps_data_tmp[2])
+ self.gps_data = {
+ "latitude": latitude,
+ "longitude": longitude,
+ "high": high,
+ "direction": match_vtg_data[1],
+ "speed": match_vtg_data[5],
+ "number": match_gga_data[7],
+ }
+ #检查wifi数据
+ if self.wifi_data == [] or (self.wifi_data != [] and self.wifi_data[0] == 0):#and self.wifi_data[0] <= 0:
+ EventMesh.publish(constants.EVENT.WIFISCANOPEN)
- #检查基站数据
- if self.cell_data == []:
- data_tmp = net.getCellInfo()
- for cell in data_tmp:
- if cell and cell[0][0] == 0 and len(cell[0]) > 1:
- self.cell_data = cell
- break
- if self.location_timer_count >= 60*4 : #4分钟关闭
+ #检查基站数据
+ if self.cell_data == []:
+ data_tmp = net.getCellInfo()
+ for cell in data_tmp:
+ if cell and cell[0][0] == 0 and len(cell[0]) > 1:
+ self.cell_data = cell
+ break
+ if self.location_timer_count >= 60*7 : #4分钟关闭
EventMesh.publish(constants.EVENT.CLOSE_LOCATION)#关闭定位
#break
+ #EventMesh.txt_log("GSV_sig_ok_count :{}".format(GSV_sig_ok_count))
if self.location_type == 1: #测试模式
- EventMesh.publish("PUSH_LOCATION_DATA", [gnss_r_data,self.wifi_data,self.cell_data])
+ if len(gnss_r_data) >= 1:
+ if gnss_r_data[0] > 0:
+ EventMesh.publish("PUSH_LOCATION_DATA", [gnss_r_data,self.wifi_data,self.cell_data])
elif self.location_type == 0: #正常模式
- if self.gps_data != [] or self.location_timer_count > 70: #or( self.wifi_data != [] and self.cell_data != []):
+ if self.gps_data != [] or self.location_timer_count > 180: #or( self.wifi_data != [] and self.cell_data != []):
self.Location_send([self.gps_data,self.wifi_data,self.cell_data])
EventMesh.publish(constants.EVENT.CLOSE_LOCATION)#关闭定位
+ EventMesh.txt_log("CLOSE_LOCATION:location_timer_count:{}".format(self.location_timer_count))
+ #break
+
+ if len(GSV_dict) < 2 and GSV_sig_ok_count < 1 and self.wifi_data != [] and self.cell_data != [] and self.location_timer_count > 20: #超过20秒且卫星个数少于3
+ self.Location_send([self.gps_data,self.wifi_data,self.cell_data])
+ EventMesh.publish(constants.EVENT.CLOSE_LOCATION)#关闭定位
+ EventMesh.txt_log("len(GSV_dict) < 2")
+ if GSV_sig_ok_count < 3 and self.wifi_data != [] and self.cell_data != [] and self.location_timer_count > 40: #超过40秒且信号大于30的不足3个
+ self.Location_send([self.gps_data,self.wifi_data,self.cell_data])
+ EventMesh.publish(constants.EVENT.CLOSE_LOCATION)#关闭定位
+ EventMesh.txt_log("GSV_sig_ok_count < 3")
#break
- if len(match_gsv_data) >= 3:
- #print("match_gsv_data[3]:{}".format(match_gsv_data[3]))
- #if (self.wifi_data != [] and self.cell_data != [] and self.location_timer_count > 10): #超过10秒且卫星个数少于3
- if (int(match_gsv_data[3]) < 3 and self.wifi_data != [] and self.cell_data != [] and self.location_timer_count > 20): #超过10秒且卫星个数少于3
- self.Location_send([self.gps_data,self.wifi_data,self.cell_data])
-
- EventMesh.publish(constants.EVENT.CLOSE_LOCATION)#关闭定位
- #break
utime.sleep_ms(1000)
#print("location_timer_count:{}".format(self.location_timer_count))
else :
def server_init(server_type = None):
- #服务器选择1，防走丢，2大锤,3三基同创
+ #服务器选择1，防走丢，2大锤,3三基同创,4童信
if server_type == None:
- server_type = 1
+ server_type = 4
#print("server_init:server_type:{}".format(server_type))
d_server_addr = "api.xcloudtech.com"
d_server_port = 9000
d_server_port = 2346
d_md5_key = ""
d_PID = "3G"
+ elif server_type == 4:
+ d_server_addr = "tcp.tx.kidxin.com"
+ d_server_port = 20500
return [d_server_addr,d_server_port,d_md5_key,d_PID]
from queue import Queue
def unicode_to_utf8(src):
res = uos.statvfs("/ext")
res = list(res)
print('总ext空间:{} MB，剩余空间:{} MB'.format((((res[0] * res[2]) / 1024 / 1024)),((res[0] * res[3]) / 1024 / 1024)))
-from lcd import tp_cst816
-from btn_device_ec800m import BtnDevice
+# from lcd import tp_cst816
+# from btn_device_ec800m import BtnDevice
import lvgl as lv
import utime
import osTimer
import pm
import ql_fs
-from machine import Pin
-from lcd import lcd,lcd_ctrl,tp_cst816,lcd_ctrl
-from misc import PWM
-import uos
-from machine import I2C
-from Hsensor import GH3018
-from Gsensor import SC7A20
-btn_device = BtnDevice(tp_cst816)
-btn_device.start()
-i2c0 = I2C(I2C.I2C0, I2C.FAST_MODE)
-WHO_AM_I = bytearray([0x01, 0])
-I2C_SLAVE_ADDR = 0x49
-r_data = bytearray(2)
-data = bytearray([0x60A0, 0]) #休眠
-i2c0.write(I2C_SLAVE_ADDR, WHO_AM_I, 1, data, 2)
-def gsensor_cb(self, para):
- print("gsensor_cb:{}".format(para))
-gsensor = SC7A20()
-print("DeviceInfoManager:1.0")
-gsensor.activate()
-print("DeviceInfoManager:1.1")
-gsensor.init()
-print("DeviceInfoManager:2")
-gsensor.set_callback(gsensor_cb)
-hsensor = GH3018()
-hsensor.activate()
-hsensor.init()
-hsensor.open_blood_oxygen()
-utime.sleep(5)
-PWM_backlight = PWM(PWM.PWM1, PWM.ABOVE_MS, 0, 1)
-PWM_backlight.open()
-hsensor.close()
-tp_cst816.suspend()
-lcd.lcd_display_off()
-#lcd_ctrl.write(0)
-lv.autoSleep(1)
-pm.autosleep(1)
-print("test end2")
\ No newline at end of file
+# from machine import Pin
+# from lcd import lcd,lcd_ctrl,tp_cst816,lcd_ctrl
+# from misc import PWM
+# import uos
+# from machine import I2C
+# from Hsensor import GH3018
+# from Gsensor import SC7A20
+# btn_device = BtnDevice(tp_cst816)
+# btn_device.start()
+# i2c0 = I2C(I2C.I2C0, I2C.FAST_MODE)
+# WHO_AM_I = bytearray([0x01, 0])
+# I2C_SLAVE_ADDR = 0x49
+# r_data = bytearray(2)
+# data = bytearray([0x60A0, 0]) #休眠
+# i2c0.write(I2C_SLAVE_ADDR, WHO_AM_I, 1, data, 2)
+# def gsensor_cb(self, para):
+# print("gsensor_cb:{}".format(para))
+# gsensor = SC7A20()
+# print("DeviceInfoManager:1.0")
+# gsensor.activate()
+# print("DeviceInfoManager:1.1")
+# gsensor.init()
+# print("DeviceInfoManager:2")
+# gsensor.set_callback(gsensor_cb)
+# hsensor = GH3018()
+# hsensor.activate()
+# hsensor.init()
+# hsensor.open_blood_oxygen()
+# utime.sleep(5)
+# PWM_backlight = PWM(PWM.PWM1, PWM.ABOVE_MS, 0, 1)
+# PWM_backlight.open()
+# hsensor.close()
+# tp_cst816.suspend()
+# lcd.lcd_display_off()
+# #lcd_ctrl.write(0)
+# lv.autoSleep(1)
+# pm.autosleep(1)
+# print("test end2")
+
+
+# import request
+# import ujson
+
+
+# def manual_url_encode(s: str) -> str:
+# """Encodes a string for use in a URL query string."""
+# safe_chars = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-.*"
+# encoded_str = ""
+
+# for char in s:
+# if char in safe_chars:
+# encoded_str += char
+# else:
+# encoded_str += ''.join(['%' + hex(c)[2:].upper() for c in char.encode('utf-8')])
+# return encoded_str
+
+# def generate_url(base_uri: str, params: dict) -> str:
+# """Generates a full URL given a base URI and a dictionary of query parameters."""
+# # Encode each key-value pair in the params dictionary
+# encoded_params = "&".join([key + "=" + manual_url_encode(str(value)) for key, value in params.items()])
+# return base_uri + "?" + encoded_params
+
+# # Test the function with the given example
+# base_uri = "http://ability.raisound.com:8096/bwtts"
+# params = {
+# "text": "这是一个测试用例,一二三四五,上山打老虎",
+# "voice_name": "xiaomu",
+# "sample_rate": 16000,
+# "audio_format": "mp3"
+# }
+# generated_url = generate_url(base_uri, params)
+# print("url:{}".format(generated_url))
+# response = request.get(generated_url)
+# print("status_code:{}".format(response.status_code))
+# play_flag = 2 # = 1流播放 =2保存文件播放
+# import audio
+# from machine import Pin
+# record_test = audio.Record()
+# record_test.gain(4,12)
+# aud = audio.Audio(0)
+# aud.set_pa(Pin.GPIO29)
+# aud.setVolume(11)
+# tmp = bytearray()
+# size = 10*1024
+# ret = 0
+# if play_flag == 1:
+# for i in response.content:
+# #print("type:{}".format(type(i)))
+# tmp+= bytearray(i)
+# print("data len:{}".format(len(tmp)))
+# #print("data:{}".format(tmp))
+# ret = aud.playStream(3, tmp)
+# utime.sleep_ms(20)
+# else:
+# len_tmp = 0
+# with open('/ext/test.mp3', 'w+')as fp:
+# for i in response.content:
+# fp.write(i)
+# len_tmp+=len(i)
+# print("data len:{}".format(len_tmp))
+# #ret = aud.play(1, 1, '/ext/test.mp3')
+# with open("/ext/test.mp3", "rb")as f:
+# while 1:
+# b = f.read(size) # read
+# if not b:
+# break
+# ret =aud.playStream(3, b)
+# utime.sleep_ms(20)
+# print("ret:{}".format(ret))
+
+
+
+
+import quecgnss
+import utime
+
+import quecgnss
+import utime
+import ure
+
+gnss_init_flag = 0
+ret = -1
+fix_num = 0
+
+def gnss_nmea_debug(gnss_data):
+ global fix_num
+ global gnss_init_flag
+ try:
+ rmc_info = ure.search("GNRMC(.+?)M", gnss_data)
+ if rmc_info.group(0).split(",")[2] == 'A':
+ if fix_num > 1:
+ print("输出有效定位！\r\n")
+ else:
+ fix_num = fix_num + 1
+
+ except:
+ pass
+
+
+def gnss_nmea_get_sat(gnss_data):
+ try:
+ nema_str_list = gnss_data.split('\r\n')
+ for i in range(len(nema_str_list)):
+ if(ure.search("GPGSV(.+?)", nema_str_list[i]) != None):
+ # print(nema_str_list[i].split(",")[3])
+ try:
+ print('GPS SAT NUM:{},SAT1 {} SNR {},SAT2 {} SNR {},SAT3 {} SNR {},SAT4 {} SNR {}\r\n'.format(nema_str_list[i].split(",")[3], \
+ nema_str_list[i].split(",")[4], nema_str_list[i].split(",")[7], nema_str_list[i].split(",")[8], \
+ nema_str_list[i].split(",")[11], nema_str_list[i].split(",")[12], nema_str_list[i].split(",")[15], \
+ nema_str_list[i].split(",")[16], nema_str_list[i].split(",")[19]))
+
+ except:
+ pass
+
+ elif(ure.search("GBGSV(.+?)", nema_str_list[i]) != None):
+ # print(nema_str_list[i].split(",")[3])
+ try:
+ print('BDS SAT NUM:{},SAT1 {} SNR {},SAT2 {} SNR {},SAT3 {} SNR {},SAT4 {} SNR {}\r\n'.format(nema_str_list[i].split(",")[3], \
+ nema_str_list[i].split(",")[4], nema_str_list[i].split(",")[7], nema_str_list[i].split(",")[8], \
+ nema_str_list[i].split(",")[11], nema_str_list[i].split(",")[12], nema_str_list[i].split(",")[15], \
+ nema_str_list[i].split(",")[16], nema_str_list[i].split(",")[19]))
+ except:
+ pass
+
+ else:
+ pass
+ except Exception as e:
+ print(str(e))
+
+
+def main():
+ global fix_num
+ global gnss_init_flag
+ global ret
+ if(gnss_init_flag == 0):
+ ret = quecgnss.init()
+ if ret == 0:
+ print('GNSS init ok.\r\n')
+ gnss_init_flag = 1
+ else:
+ print('GNSS init failed.\r\n')
+ return -1
+ else:
+ print('GNSS has been init.\r\n')
+ quecgnss.gnssEnable(1)
+
+ for i in range(60):
+ print("==========GNSS READTIME {}==========\r\n".format(i+1))
+ data = quecgnss.read(4096)
+ gnss_data = data[1].decode()
+ print(gnss_data)
+ gnss_nmea_get_sat(gnss_data)
+ utime.sleep(1)
+
+ quecgnss.gnssEnable(0)
+ fix_num = 0
+ utime.sleep(5)
+ quecgnss.gnssEnable(1)
+ print("==========GNSS RESTART==========\r\n")
+
+ for i in range(60):
+ print("==========GNSS READTIME {}==========\r\n".format(i+1))
+ data = quecgnss.read(4096)
+ gnss_data = data[1].decode()
+ print(gnss_data)
+ gnss_nmea_get_sat(gnss_data)
+ utime.sleep(1)
+
+ quecgnss.gnssEnable(0)
+ fix_num = 0
+
+
+
+if __name__ == '__main__':
+ main()
\ No newline at end of file
self.meta = rrestor_confirm_n_screen
self.timer1.start(1000,1,self.timer_cb_onescond)
- self.timer2.start(3000,0,self.timer_cb_threescond)
+
ret = 0
ret = ql_fs.path_exists("/ext/static/storage_file")#电话本
if ret == True:
EventMesh.publish("alipay_unbind")
if EventMesh.publish(constants.EVENT.GET_CONNECT) == 1:#已连接服务器发指令恢复出厂
EventMesh.publish(constants.PROTOCOL.DLT_SEND)
-
+ self.timer2.start(4000,0,self.timer_cb_threescond)
def Close(self):
projects / mirrors / ASR1603_G5.git / commitdiff