from machine import ExtInt, Pin,Timer,RTC,UART
import EventMesh
import audio
+from machine import I2C
from Gsensor import SC7A20
from Hsensor import GH3018
import quecgnss
def voice_callback(args):
print('voicecall iinfo:{}'.format(args))
if args[0] == 10:
+ get_UI_high_priority = EventMesh.publish("get_UI_high_priority")#如果在拨出去或者通话中屏蔽来电
+ if get_UI_high_priority == 1:
+ EventMesh.publish("set_interrup_incall",1)
+ return
print('voicecall incoming call, PhoneNO.: ', args[6])
size_len = len(str(args[6]))
tmp = []
EventMesh.publish("load_screen", {"screen": "incoming_call","meta":str(tmp)})
elif args[0] == 11:
+ EventMesh.publish("set_UI_high_priority",0)
+ sos_flag = EventMesh.publish("get_sos_flag")
+ if sos_flag == 1:
+ sos_flag = 0
+ EventMesh.publish("set_sos_flag",0)
+
print('voicecall connected, PhoneNO.: ', args[6])
size_len = len(str(args[6]))
tmp = []
EventMesh.publish("load_screen", {"screen": "on_call","meta":str(tmp)})
elif args[0] == 12:
+ set_interrup_call = EventMesh.publish("get_interrup_incall")#打断这次不需要来电的ui事件 以免通话中退出当前界面进不去了
+ if set_interrup_call == 1:
+ EventMesh.publish("set_interrup_incall",0)
+ return
+
+ EventMesh.publish("set_interrup_incall",0)
+ EventMesh.publish("set_UI_high_priority",0)
print('voicecall disconnect')
- EventMesh.publish("load_screen", {"screen": "main_menu"})
+ EventMesh.publish("load_screen", {"screen": "main_menu"})
+ sos_flag = EventMesh.publish("get_sos_flag")
+ if sos_flag == 0:
+ EventMesh.publish("set_sos_num",0)
+ else:
+ num = EventMesh.publish("get_sos_num")
+ number = EventMesh.publish("get_dial_sos_number")
+ if len(number) > 0:
+ if num == None:
+ num = 0
+ num+=1
+ num_len = 0
+
+ if len(number) == 1:
+ num_len = len(number)
+ elif len(number) == 2:
+ num_len = len(number)+1
+ elif len(number) == 3:
+ num_len = len(number)+2
+ # print("number:{} number_len :{} num:{} num_len:{}".format(number,len(number),num,num_len))
+ if num > num_len :
+ EventMesh.publish("set_sos_flag",0)
+ EventMesh.publish("set_sos_num",0)
+ else:
+ # print("number:{} number_len :{}".format(number,len(number)))
+ EventMesh.publish("set_sos_num",num)
+ if len(number) == 1:
+ EventMesh.publish("set_callnumer_cb",number[0])
+ print("phone 1")
+ elif len(number) == 2:
+ if num == 1 or num == 3:
+ print("phone 2")
+ EventMesh.publish("set_callnumer_cb",number[1])
+ elif num == 2:
+ print("phone 3")
+ EventMesh.publish("set_callnumer_cb",number[0])
+ elif len(number) == 3:
+ if num == 1 or num == 4:
+ EventMesh.publish("set_callnumer_cb",number[1])
+ print("phone 4")
+ elif num == 2 or num == 5:
+ print("phone 5")
+ EventMesh.publish("set_callnumer_cb",number[2])
+ elif num == 3:
+ print("phone 6")
+ EventMesh.publish("set_callnumer_cb",number[0])
+ print('voicecall disconnect end')
+
elif args[0] == 13:
print('voicecall is waiting, PhoneNO.: ', args[6])
elif args[0] == 14:
+
print('voicecall dialing, PhoneNO.: ', args[6])
EventMesh.publish("load_screen", {"screen": "dialing"})
print('GNSS init ok.')
else:
print('GNSS init failed.')
+
+ self.i2c0 = I2C(I2C.I2C0, I2C.STANDARD_MODE)
+ self.stop_temp( None, None)
def post_processor_after_instantiation(self):
print("post_processor_after_instantiation:1")
EventMesh.subscribe(constants.EVENT.GET_BLOOD_OXYGEN, self.get_blood_oxygen)
EventMesh.subscribe(constants.EVENT.GET_HEART, self.get_heart)
+ EventMesh.subscribe(constants.EVENT.START_TEMP, self.start_temp)
+ EventMesh.subscribe(constants.EVENT.STOP_TEMP, self.stop_temp)
+ EventMesh.subscribe(constants.EVENT.GET_TEMP, self.get_temp)
+
print("post_processor_after_instantiation:2")
self.timer_push_schedule.start(10000, 1, self.publish_event)
# self.battery_push_schedule.start(3000, 1, self.__check_battery)
self.timer_push = 0
date_tmp = utime.localtime()
time_tmp = utime.mktime(date_tmp)
+ print("date_tmp......01....{} time_tmp:{}\r\n".format(date_tmp,time_tmp))
+
if math.fabs(time_tmp - self.KA_time) > 7*60: #7分钟一次心跳
self.KA_time = time_tmp
EventMesh.publish(constants.PROTOCOL.KA_SEND)
self.upload_time = EventMesh.publish("get_upload_data")
-
+ #print("self.upload_time:{} math.fabs(time_tmp - self.UD_time):{}".format(self.upload_time,math.fabs(time_tmp - self.UD_time)))
if self.upload_time != None:
if math.fabs(time_tmp - self.UD_time) > self.upload_time: #7分钟一次定位
self.UD_time = time_tmp
self.bat_time = 0
if battery[1] == 20:
EventMesh.publish("load_screen", {"screen": "LOW_POWER"})
- print("battery:{}".format(battery))
+
#消息已满
msg_nums = sms.getMsgNums() #获取短信条数
self.sms_full_flag+=10
if msg_nums >= 50:
EventMesh.publish("load_screen", {"screen": "message_full"})
- #print("hr......01....{}\r\n".format(hr_time))
+
if EventMesh.publish(constants.EVENT.GET_CONNECT) == 1 or battery[1] != 1:
if hr_time != None:
# print("hr......01....{}\r\n".format(hr_time))
if hr_time > 0:
- # print("get_hr_state:{} get_spo2_state:{} self.old_time:{} hr_time:{}".format(self.get_hr_state,self.get_spo2_state,self.old_time,hr_time))
+ #print("get_hr_state:{} get_spo2_state:{} self.old_time:{} hr_time:{}".format(self.get_hr_state,self.get_spo2_state,self.old_time,hr_time))
if self.get_hr_state != 1 and self.get_spo2_state != 1:
if self.get_hr_state == 0 and self.get_spo2_state == 0:
if self.HR_state == 0 or self.HR_state == 3:
EventMesh.publish("set_old_spo2_data",spo2_value)
EventMesh.publish(constants.PROTOCOL.OXYGEN_SEND,spo2_value)
self.HR_state = 3
-
else:
- self.SENSOR_time = time_tmp
self.HR_state = 0
- self.old_time = 0
+ self.old_time = hr_time
+
if self.HR_state == 3 or self.HR_state == 0:#只有不在测量的时候才测单次的
if self.get_spo2_state == 3 or self.get_hr_state == 3:#单次测量
self.HR_state = 0
else:
self.HR_Onestate = 0
self.HR_OneTime = 0
-
else:
- # print("hr......01....{} {} self.HR_Onestate {} {}".format(self.get_hr_state,self.get_spo2_state,self.HR_OneTime,self.HR_Onestate))
+ #print("hr......01....{} {} self.HR_Onestate {} {}".format(self.get_hr_state,self.get_spo2_state,self.HR_OneTime,self.HR_Onestate))
self.SENSOR_time = time_tmp#进入最后一次的时间戳
- self.old_time = 0
if self.get_hr_state != 1 and self.get_spo2_state != 1:#没有设置间隔时间单次测量
if self.get_spo2_state == 3 or self.get_hr_state == 3:#单次测量
def get_heart(self, topic, msg):
data = self.hsensor.get_all_data()
return data[0]
+
+ def start_temp(self, topic, msg):
+ WHO_AM_I = bytearray([0x01, 0])
+ I2C_SLAVE_ADDR = 0x49
+ r_data = bytearray(2)
+ data = bytearray([0x61A0, 0]) # 唤醒
+ return self.i2c0.write(I2C_SLAVE_ADDR, WHO_AM_I, 1, data, 2)
+ def stop_temp(self, topic, msg):
+ WHO_AM_I = bytearray([0x01, 0])
+ I2C_SLAVE_ADDR = 0x49
+ r_data = bytearray(2)
+ data = bytearray([0x60A0, 0]) #休眠
+ return self.i2c0.write(I2C_SLAVE_ADDR, WHO_AM_I, 1, data, 2)
+
+ def get_temp(self, topic, msg):
+ WHO_AM_I = bytearray([0x00, 0])
+ I2C_SLAVE_ADDR = 0x49
+ r_data = bytearray(2)
+ self.i2c0.read(I2C_SLAVE_ADDR, WHO_AM_I, 1, r_data, 2, 0)
+ #print("temp_data:{}".format(r_data))
+ temp_s = ((r_data[0]<<8) | (r_data[1]))>>4
+ #print("temp_s:{}".format(temp_s))
+ temp_data = temp_s * 0.0625
+ #print("temp_data:{}".format(temp_data))
+ return temp_data
_thread.start_new_thread(self.reconnect, ())
break
data = data + r
- if data[-1] != 0x5d and len(data)<10*1024: #']':
+ if data[-1] != 0x5d and len(data)<20*1024: #']':
continue
self.connect_tmp = 1 #有数据，就认为是连接状态
print("TCP:recv: data:{}".format(data))
data = ""
self.location_timer_count = self.location_timer_count +1
self.location_check_data_timer_count = self.location_check_data_timer_count+1
- if self.location_check_data_timer_count > 5 :#5秒查一次
+
+ match_gga_data = []
+ match_gsv_data = []
+
+ if self.location_check_data_timer_count > 2 :#5秒查一次
self.location_check_data_timer_count = 0
#print(self.location_timer_count)
#检查wifi数据
data = quecgnss.read(4096)
#print(data[1].decode())
if data[0] > 0 : #有数据
+ match_gga_data = self.match.GxGGA(data[1]).split(",")
+ match_gsv_data = self.match.GxGSV(data[1]).split(",")
if self.gps.__check_gps_valid(data[1].decode()): #数据齐全
match_gmc_data = self.match.GxRMC(data[1]).split(",")
match_vtg_data = self.match.GxVTG(data[1]).split(",")
if self.location_type == 1: #测试模式
EventMesh.publish("PUSH_LOCATION_DATA", [data,self.wifi_data,self.cell_data])
elif self.location_type == 0: #正常模式
- if self.gps_data != [] or self.location_timer_count > 60: #or( self.wifi_data != [] and self.cell_data != []):
+ if self.gps_data != [] or self.location_timer_count > 40: #or( self.wifi_data != [] and self.cell_data != []):
#print("publish:READ_LOCATION")
EventMesh.publish(constants.EVENT.READ_LOCATION, [self.gps_data,self.wifi_data,self.cell_data])
EventMesh.publish(constants.EVENT.CLOSE_LOCATION)#关闭定位
- # if self.location_timer_count > 60 : #1分钟定位超时，超时发送数据，停止定时器及关闭定位
- # EventMesh.publish(constants.EVENT.CLOSE_LOCATION)#关闭定位
- # EventMesh.publish(constants.EVENT.READ_LOCATION, [self.gps_data,self.wifi_data,self.cell_data])#发送数据
-
+
+ if len(match_gsv_data) >= 3:
+ #print("match_gsv_data[3]:{}".format(match_gsv_data[3]))
+ if (int(match_gsv_data[3]) < 3 and self.wifi_data != [] and self.cell_data != [] and self.location_timer_count > 10): #超过10秒且卫星个数少于3
+ #print("publish:READ_LOCATION")
+ EventMesh.publish(constants.EVENT.READ_LOCATION, [self.gps_data,self.wifi_data,self.cell_data])
+ EventMesh.publish(constants.EVENT.CLOSE_LOCATION)#关闭定位
self.volume_size = 8
self.read_data = {}
check = ql_fs.path_exists("/ext/static/storage_file/config.json")
- print("check:{}".format(check))
+
if check == True:
with open("/ext/static/storage_file/config.json","r") as fd:
self.read_data = ujson.load(fd)
self.volume_size = 8
else:
self.volume_size = 8
+ # print("self.volume_size:{}".format(self.volume_size))
def post_processor_after_instantiation(self):
EventMesh.subscribe("set_volum_size", self.set_volum_size)
EventMesh.subscribe("get_volum_size", self.get_volum_size)
def get_volum_size(self, event, msg):
return self.volume_size
+class getSosFlagManager(Abstract):#设置或获取音量大小数据
+ def __init__(self):
+ self.sos_flag = 0
+ self.sos_num = 0
+ self.call_timer = osTimer()
+ self.sos_numer1 = []
+ self.callnumer =""
+ self.read_data = {}
+ check = ql_fs.path_exists("/ext/static/storage_file/config.json")
+
+ if check == True:
+ with open("/ext/static/storage_file/config.json","r") as fd:
+ self.read_data = ujson.load(fd)
+ if self.read_data != {}:
+ b1 = str("SOS_NUMBER1") in self.read_data.keys()
+ if b1 == True:
+ self.sos_numer1 = [self.read_data["SOS_NUMBER1"]]
+ b2 = str("SOS_NUMBER2") in self.read_data.keys()
+ if b2 == True:
+ self.sos_numer1 = [self.read_data["SOS_NUMBER1"], self.read_data["SOS_NUMBER2"]]
+
+ b3 = str("SOS_NUMBER3") in self.read_data.keys()
+ if b3 == True:
+ self.sos_numer1 = [self.read_data["SOS_NUMBER1"], self.read_data["SOS_NUMBER2"],self.read_data["SOS_NUMBER3"]]
+
+ # print("self.sos_numer1:{} self.read_data:{}".format(self.sos_numer1,self.read_data))
+
+ def post_processor_after_instantiation(self):
+ EventMesh.subscribe("set_sos_flag", self.set_sos_flag)#设置sos拨打标识
+ EventMesh.subscribe("get_sos_flag", self.get_sos_flag)
+
+ EventMesh.subscribe("set_sos_num", self.set_sos_num)#设置拨打循环次数
+ EventMesh.subscribe("get_sos_num", self.get_sos_num)
+
+ EventMesh.subscribe("set_dial_sos_number", self.set_dial_sos_number)
+
+ EventMesh.subscribe("set_callnumer_cb", self.set_callnumer_cb)#延时回调
+
+ EventMesh.subscribe("get_dial_sos_number", self.get_dial_sos_number)#设置家庭号码三个
+
+ def set_sos_flag(self, event, msg):
+ self.sos_flag = msg
+ if self.sos_flag == 1:
+ number = EventMesh.publish("get_dial_sos_number")
+ if len(number) > 0 and number != None:
+ EventMesh.publish("save_phonename",number[0])
+ EventMesh.publish("save_phonenum",number[0])
+ voiceCall.callStart(str(number[0]))
+
+
+ def set_callnumer_cb(self, event, msg):
+ self.callnumer = msg
+ self.call_timer.start(5000, 0, self.call_timer_cb)#延迟1S 继续拨打
+
+ def call_timer_cb(self, *args, **kwargs):
+ # print("call_timer_cb:{}".format(self.callnumer))
+ EventMesh.publish("save_phonename",str(self.callnumer))
+ EventMesh.publish("save_phonenum",str(self.callnumer))
+ voiceCall.callStart(str(self.callnumer))
+
+ def get_sos_flag(self, event, msg):
+ return self.sos_flag
+
+ def set_sos_num(self, event, msg):
+ self.sos_num = msg
+
+ def get_sos_num(self, event, msg):
+ return self.sos_num
+
+ def set_dial_sos_number(self, event, msg):
+ self.sos_numer1 = msg
+ # print("self.sos_numer1:{}".format(self.sos_numer1))
+
+ def get_dial_sos_number(self, event, msg):
+ return self.sos_numer1
+
+
+class getBootOffManager(Abstract):#开关机时间
+ def __init__(self):
+ self.read_data = {}
+ self.off_msg = []
+ check = ql_fs.path_exists("/ext/static/storage_file/config.json")
+
+ if check == True:
+ with open("/ext/static/storage_file/config.json","r") as fd:
+ self.read_data = ujson.load(fd)
+ if self.read_data != {}:
+ b1 = str("off_on") in self.read_data.keys()
+ if b1 == True:
+ self.off_on = self.read_data["off_on"]
+ b2 = str("off_start") in self.read_data.keys()
+ if b2 == True:
+ self.off_start= self.read_data["off_start"]
+
+ b3 = str("off_end") in self.read_data.keys()
+ if b3 == True:
+ self.off_end = self.read_data["off_end"]
+
+ # print("self.sos_numer1:{} self.read_data:{}".format(self.sos_numer1,self.read_data))
+
+ def post_processor_after_instantiation(self):
+ EventMesh.subscribe("set_boot_off_time", self.set_boot_off_time)#设置关机时间
+ EventMesh.subscribe("get_boot_off_time", self.get_boot_off_time)
+
+ # def _rtc_callback(self,args):
+ # print("---!!!!!!alarm_ring!!!!!!!---,idx:",args)
+
+ # def set_boot_off_time(self, event, msg):
+
+ # self.off_msg = msg
+ # if msg[0] == 0:
+ # return
+
+ # rtc_now_time = self.rtc.datetime()
+
+ # # data = RTC().datetime()
+ # # hour = data[4]
+ # # min = data[5]
+
+
+
+ # current_minute = rtc_now_time[4]*60 + rtc_now_time[5] #当前的分钟数不包含年月日
+ # off_msg1 = msg[1]
+ # off_msg2 = msg[2]
+ # msg_tmp1 = off_msg1.split(':')
+ # msg_tmp2 = off_msg2.split(':')
+ # start_min = msg_tmp1[0]*60+msg_tmp1[1]
+ # end_min = msg_tmp2[0]*60+msg_tmp2[1]
+ # #9:00 6:00
+ # if current_minute > start_min:#如果当前的分钟数 大于开始的关机时间分钟数 则认为是第二天6:00需要开机 23:00需要关机
+ # next_alarm_time = utime.localtime(next_alarm_mktime)
+ # pass
+
+
+ # print("04当前的rtc_total_minutes：{}".format(current_minute))
+
+ # if current_minute in minute_list:
+ # print("当前时间和闹钟列表时间重复，删除重复的时间")
+ # minute_list.remove(current_minute)
+
+ # minute_list.append(current_minute)
+
+ # # 对分钟数列表进行排序，并获取当前时间分钟数所在的位置
+ # minute_list = sorted(minute_list)
+ # current_index = minute_list.index(current_minute)
+
+ # print("05将当前时间插入min_list后的min_list：{}".format(minute_list))
+ # print("06当前时间所在位置：{}".format(current_index))
+
+ # next_index = current_index + 1
+ # if next_index >= len(minute_list):
+ # next_minute = minute_list[0]
+ # print("07下一个闹钟的分钟数：", next_minute)
+ # else:
+ # next_minute = minute_list[next_index]
+ # print("07下一个闹钟的分钟数：", next_minute)
+
+ # if next_minute >= current_minute:
+ # interval_minute = next_minute - current_minute
+ # print("08距离下一次响铃的的分钟数：", interval_minute)
+ # else:
+ # interval_minute = end_minute - current_minute + next_minute
+ # print("08距离下一次响铃的的分钟数：", interval_minute)
+
+ # #反向获取时间戳
+ # time = utime.localtime()
+ # print("09获取utime的时间戳：", time)
+ # rtc_mktime = utime.mktime(time)
+ # print("10反向获取utime的mktime：", rtc_mktime)
+
+ # #加上下一次响铃需要的时间
+ # next_alarm_mktime = rtc_mktime + interval_minute*60
+ # print("11下一次响铃的mktime：", next_alarm_mktime)
+
+ # #格式化时间戳，得到下一次闹钟的时间
+ # next_alarm_time = utime.localtime(next_alarm_mktime)
+ # print("12下一次响铃的准确时间：", next_alarm_time)
+ # #将时间转换为rtc的格式
+ # data_l = [2021, 7, 9, 5, 12, 30, 0, 0]
+ # #给rtc时间赋值
+ # data_l[0] = next_alarm_time[0] #year
+ # data_l[1] = next_alarm_time[1] #month
+ # data_l[2] = next_alarm_time[2] #day
+ # data_l[4] = next_alarm_time[3] #hour
+ # data_l[5] = next_alarm_time[4] #minute
+ # #转换为元组
+ # data_e=tuple(data_l)
+ # print("13将该时间转换为RTC元组：", data_e)
+
+ # #设置rtc闹钟时间
+ # self.rtc.register_callback(self._rtc_callback)
+
+ # FLAG = self.rtc.set_alarm(data_e)
+ # if FLAG == 0:
+ # print("成功设置rtc的到期时间：", data_e)
+ # else:
+ # print("设置rtc的到期时间失败：", data_e)
+
+ # self.rtc.enable_alarm(1)
+
+ # def get_boot_off_time(self, event, msg):
+ # return self.off_msg
+
\ No newline at end of file
projects / mirrors / ASR1603_G5.git / commitdiff