"""
def __init__(self):
- self.file_name = "/usr/conf_store.json"
+ self.file_name = "//conf_store.json"
self.lock = Lock()
self.map = dict()
def set_sos_info_call(self, event, msg):
hangup_mode = EventMesh.publish("get_hangup_mode")
- print('hangup_mode ',hangup_mode)
if hangup_mode == 1:
EventMesh.publish("set_sos_flag",0)
EventMesh.publish("set_sos_num",0)
self.log = LogAdapter(self.__class__.__name__)
self.rtc = RTC()
self.aud = audio.Audio(0)
- print("DeviceInfoManager:1")
self.gsensor = SC7A20()
print("DeviceInfoManager:1.0")
self.gsensor.activate()
self.gsensor.init()
print("DeviceInfoManager:2")
self.gsensor.set_callback(self.gsensor_cb)
- # try :
- # self.hsensor = HX3918()
- # except Exception as e:
- # print("Hsensor none HX3918:{}".format(e))
- # try :
- # self.hsensor = GH3018()
- # except Exception as e:
- # print("Hsensor none GH3018:{}".format(e))
- # #
- # self.hsensor.activate()
- # self.hsensor.init()
- # self.hsensor.close()
+
+
self.old_temp_time = 0
# #print("hsensor.open_heart:%d\r\n"%(self.hsensor.open_heart()))
print("DeviceInfoManager:3")
EventMesh.subscribe(constants.EVENT.STOP_HEART, self.stop_heart)
EventMesh.subscribe(constants.EVENT.GET_BLOOD_OXYGEN, self.get_blood_oxygen)
EventMesh.subscribe(constants.EVENT.GET_HEART, self.get_heart)
+ EventMesh.subscribe(constants.EVENT.GET_WEAR, self.get_wear)
+ EventMesh.subscribe("get_Hsensor", self.get_Hsensor)
EventMesh.subscribe("Hsensor_set_sleep", self.Hsensor_set_sleep)
EventMesh.subscribe(constants.EVENT.START_TEMP, self.start_temp)
def gsensor_cb(self, para):
print("gsensor_cb:{}".format(para))
+ flip_sw = EventMesh.publish("get_storage_config","flip_sw")
+
if para == 0: #跌倒报警
pass
#EventMesh.publish("load_screen", {"screen": "FIND"})
elif para == 1: #抬腕
- pass
- # EventMesh.publish(constants.EVENT.SYSTEM_UP)
+
+ if flip_sw == 1:
+ EventMesh.publish(constants.EVENT.SYSTEM_UP)
+
elif para == 2: #放手
pass
- # EventMesh.publish(constants.EVENT.SYSTEM_SLEEP)
+ # if flip_sw == 1:
+ # EventMesh.publish(constants.EVENT.SYSTEM_SLEEP)
def publish_event_cb(self, *args, **kwargs):
EventMesh.publish("publish_message_event_DeviceInfo",{"func":self.publish_event,"data":0})
bat_total_value = bat_total_value//curr_count#一分钟更新一次电量 10个平均值
EventMesh.publish("set_curr_battery_level",bat_total_value)
- print("self.vbat_buf:{}".format(self.vbat_buf))
+ # print("self.vbat_buf:{} curr_count :{} bat_total_value:{} temp:{}".format(self.vbat_buf,curr_count,bat_total_value,temp))
bat_tmp = EventMesh.publish("screen_get_battery")
if min <= 1:
if self.clear_step == 0:
step = EventMesh.publish("get_old_step_value")
+ EventMesh.publish("set_old_time_data",time_tmp)
if step > 0:
EventMesh.publish("set_old_step_value",0)
self.gsensor.reset_step()
else:
self.sms_full_flag = 0
+ WEAR = EventMesh.publish(constants.EVENT.GET_WEAR)#佩戴数据
get_temp_time = EventMesh.publish("get_temp_time")
get_temp_state = EventMesh.publish("get_temp_state")
hr_time = {}
hr_time = EventMesh.publish("get_time_hr_spo2")#获取时间是否大于0
- if EventMesh.publish(constants.EVENT.GET_CONNECT) == 1 and battery[0] != "1":
+ if get_temp_state != None : #突然脱腕
+ if get_temp_state > 1 and WEAR != 1:
+ EventMesh.publish(constants.EVENT.STOP_TEMP)
+ EventMesh.publish("set_temp_state",0)
+ if (self.get_hr_state > 1 or self.get_spo2_state > 1) and WEAR != 1:
+ if self.get_hr_state > 1 :
+ EventMesh.publish("set_hr_state",0)#2代表后台进入测试，定时测心率 1ui界面测试洗心率
+ EventMesh.publish(constants.EVENT.STOP_HEART)
+ elif self.get_spo2_state > 1:
+ EventMesh.publish("set_spo2_state",0)#2代表后台进入测试
+ EventMesh.publish(constants.EVENT.STOP_BLOOD_OXYGEN)
+ if 0:
+
# print("get_temp_time:{} self.temp_one_state:{}".format(get_temp_time,self.temp_one_state))
if get_temp_time != None:#定时测量体温
if get_temp_time > 0:
self.temp_state = 0
self.temp_one_state = 0
temp = EventMesh.publish(constants.EVENT.GET_TEMP)
+ temp += 0.5
if temp < 25 or temp > 42:
temp = 0
tmp_f = str(temp).split('.')[0]
elif self.temp_one_state > 6:
self.temp_one_state = 0
temp = EventMesh.publish(constants.EVENT.GET_TEMP)
+ temp += 0.5
if temp < 25 or temp > 42:
temp = 0
tmp_f = str(temp).split('.')[0]
elif self.temp_one_state > 6:
self.temp_one_state = 0
temp = EventMesh.publish(constants.EVENT.GET_TEMP)
+ temp += 0.5
if temp < 25 or temp > 42:
temp = 0
tmp_f = str(temp).split('.')[0]
if get_temp_state == 2 or get_temp_state == 3:
self.temp_one_state = 0
self.temp_state = 0
- temp = EventMesh.publish(constants.EVENT.GET_TEMP)
+ temp = EventMesh.publish(constants.EVENT.GET_TEMP)
+ temp += 0.5
if temp < 25 or temp > 42:
temp = 0
d = int(
time_msg / 86400) # The int call removes the decimals. Conveniently, it always rounds down. int(2.9) returns 2 instead of 3, for example.
time_msg -= (
- d * 86400) # This updates the value of x to show that the already counted seconds won't be double counted or anything.
+ d * 86400) # This updates the value of x to show that the already counted seconds won't be double counted or anything.
h = int(time_msg / 3600)
time_msg -= (h * 3600)
m = int(time_msg / 60)
return self.connect
def start_blood_oxygen(self, topic, msg):
- print("start_blood_oxygen!!!")
pass
- # self.hsensor.open_blood_oxygen()
def start_heart(self, topic, msg):
- print("start_heart!!!")
pass
- # self.hsensor.open_heart()
def stop_blood_oxygen(self, topic, msg):
- print("stop_blood_oxygen!!!")
pass
- # self.hsensor.close()
def stop_heart(self, topic, msg):
- print("stop_heart!!!")
pass
- # self.hsensor.close()
def get_blood_oxygen(self, topic, msg):
- # data = self.hsensor.get_all_data()
+
return 0
def get_heart(self, topic, msg):
- # data = self.hsensor.get_all_data()
+
+ return 0
+
+ def get_wear(self, topic, msg):
+ return 0
+
+ def get_Hsensor(self, topic, msg):
return 0
def Hsensor_set_sleep(self, topic, msg):
def start_temp(self, topic, msg):
pass
- # 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):
pass
- # 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):
- pass
- # 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
+
+ return 0
step = 0
step = EventMesh.publish("get_old_step_value")
if step != 0:
- read_data = EventMesh.publish("get_storage_data")
- b = str("old_step_new") in read_data.keys()
- if b == True:
- old_step = read_data["old_step_new"]
- else:
- old_step = 0
+
+
+ tmp_data = EventMesh.publish("get_storage_config","old_step_new")
+ if tmp_data != None:
+ old_step = tmp_data
+
if old_step != step:
- read_data["old_step"] = step
- read_data["old_step_new"] = step
- EventMesh.publish("set_storage_data", read_data)
- with open("/usr/static/storage_file/config.json","w",encoding="utf-8") as fd:
- ujson.dump(read_data,fd)
+ EventMesh.publish("set_storage_config",["old_step",step])
+ EventMesh.publish("set_storage_config",["old_step_new",step])
+
old_step = step
# step = str(EventMesh.publish(constants.EVENT.GET_STEP)[0])
#翻转次数
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 > 40: #or( self.wifi_data != [] and self.cell_data != []):
+ if self.gps_data != [] or self.location_timer_count > 60: #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)#关闭定位
#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 > 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)#关闭定位
self.read_histroy_temp_data = []
self.read_histroy_spo2_data = []
- read_data = EventMesh.publish("get_storage_data")
- if read_data != None:
- b = str("old_hr") in read_data.keys()
- if b == True:
- self.oldHr = read_data["old_hr"]
+ oldHr = EventMesh.publish("get_storage_config","old_hr")
+ if oldHr != None:
+ self.oldHr = oldHr
+ old_spo2 = EventMesh.publish("get_storage_config","old_spo2")
+ if old_spo2 != None:
+ self.oldspo2 = old_spo2
+ old_step = EventMesh.publish("get_storage_config","old_step")
+ if old_step != None:
+ self.oldstep = old_step
+
- old_spo2 = str("old_spo2") in read_data.keys()
- if old_spo2 == True:
- self.oldspo2 = read_data["old_spo2"]
-
- old_step = str("old_step") in read_data.keys()
- if old_step == True:
- self.oldstep = read_data["old_step"]
-
- # check_json= ql_fs.path_exists("/ext/static/storage_file/config.json")
- # if check_json == True :
- # with open("/ext/static/storage_file/config.json","r") as fd:#音频文件编号
- # self.read_data = ujson.load(fd)
- # if self.read_data != {}:
- # b = str("old_hr") in self.read_data.keys()
- # if b == True:
- # self.oldHr = self.read_data["old_hr"]
- # else:
- # self.oldHr = 0
- # old_spo2 = str("old_spo2") in self.read_data.keys()
- # if old_spo2 == True:
- # self.oldspo2 = self.read_data["old_spo2"]
- # else:
- # self.oldspo2 = 0
- # old_step = str("old_step") in self.read_data.keys()
- # if old_step == True:
- # self.oldstep = self.read_data["old_step"]
- # else:
- # self.oldstep = 0
- # else:
- # self.oldHr = 0
- # self.oldspo2 = 0
- # self.oldstep = 0
check_hr= ql_fs.path_exists("/usr/static/storage_file/history_hr.json")
if check_hr == True :
with open("/usr/static/storage_file/history_hr.json","r") as fd:#音频文件编号
EventMesh.subscribe("get_history_spo2_value", self.get_history_spo2_value)
EventMesh.subscribe("set_history_spo2_value", self.set_history_spo2_value)
- def get_history_spo2_value(self, event, msg): #近7次体温历史数据统计显示
+ def get_history_spo2_value(self, event, msg): #近7次血氧历史数据统计显示
return self.read_histroy_spo2_data
def set_history_spo2_value(self, event, msg):
else:
self.read_histroy_temp_data.append({"timeSec": time_tmp, "temp": self.temp_value})
- with open("/usr/static/storage_file/history_spo2.json","w",encoding="utf-8") as fd:
+ with open("/usr/static/storage_file/history_temp.json","w",encoding="utf-8") as fd:
ujson.dump(self.read_histroy_temp_data,fd)
#print(self.read_histroy_temp_data)
def save_hr_value(self, event, msg):
if msg >= 50:
-
- read_data_hr = EventMesh.publish("get_storage_data")
- read_data_hr["old_hr"] = msg
+ EventMesh.publish("set_storage_config",["old_hr",msg])
self.oldHr = msg
- EventMesh.publish("set_storage_data", read_data_hr)
+
date_tmp = utime.localtime()
time_tmp = utime.mktime(date_tmp)
else:
self.read_histroy_hr_data.append({"timeSec": time_tmp, "hr": self.oldHr})
- with open("/usr/static/storage_file/config.json","w",encoding="utf-8") as fd:
- ujson.dump(read_data_hr,fd)
with open("/usr/static/storage_file/history_hr.json","w",encoding="utf-8") as fd:
ujson.dump(self.read_histroy_hr_data,fd)
def save_spo2_value(self, event, msg):
if msg >= 85:
- read_data_spo2 = EventMesh.publish("get_storage_data")
- read_data_spo2["old_spo2"] = msg
+ EventMesh.publish("set_storage_config",["old_spo2",msg])
self.oldspo2 = msg
- EventMesh.publish("set_storage_data", read_data_spo2)
+
date_tmp = utime.localtime()
time_tmp = utime.mktime(date_tmp)
else:
self.read_histroy_spo2_data.append({"timeSec": time_tmp, "spo2": self.oldspo2})
- with open("/usr/static/storage_file/config.json","w",encoding="utf-8") as fd:
- ujson.dump(read_data_spo2,fd)
-
with open("/usr/static/storage_file/history_spo2.json","w",encoding="utf-8") as fd:
ujson.dump(self.read_histroy_spo2_data,fd)
#print(self.read_histroy_spo2_data)
# 获取天气数据
def __init__(self):
self.time_sec = 0
- read_data = EventMesh.publish("get_storage_data")
- if read_data != None:
- b = str("old_time") in read_data.keys()
- if b == True:
- self.time_sec = read_data["old_time"]
-
- # check_json= ql_fs.path_exists("/ext/static/storage_file/config.json")
- # if check_json == True :
- # with open("/ext/static/storage_file/config.json","r") as fd:#音频文件编号
- # self.read_data = ujson.load(fd)
- # if self.read_data != {}:
- # b = str("old_time") in self.read_data.keys()
- # if b == True:
- # self.time_sec = self.read_data["old_time"]
-
+ tmp_data = EventMesh.publish("get_storage_config","old_time")
+ if tmp_data !=None :
+ self.time_sec = tmp_data
+
def post_processor_after_instantiation(self):
EventMesh.subscribe("get_old_time_data", self.get_old_time_data)
EventMesh.subscribe("set_old_time_data", self.set_old_time_data)
def set_old_time_data(self, event, msg):
self.time_sec = msg
- read_data_time = EventMesh.publish("get_storage_data")
- read_data_time["old_time"] = msg
- EventMesh.publish("set_storage_data", read_data_time)
- with open("/usr/static/storage_file/config.json","w",encoding="utf-8") as fd:
- ujson.dump(read_data_time,fd)
+ EventMesh.publish("set_storage_config",["old_time",msg])
+
+
self.set_hr = 0 #0关闭 1手动 2 自动测量按照时间间隔测试心率
self.set_spo2 = 0
self.open_btn = 0
- self.get_temp = 0
- read_data = EventMesh.publish("get_storage_data")
+ read_data = EventMesh.publish("get_storage_config","measure_time")
+ if read_data != None:
+ self.set_time = read_data
+ read_data = EventMesh.publish("get_storage_config","open_hr_btn")
+
if read_data != None:
- b = str("measure_time") in read_data.keys()
- if b == True:
- self.set_time = read_data["measure_time"]
- btn = str("open_hr_btn") in read_data.keys()
- if btn == True:
- self.open_btn = read_data["measure_time"]
-
- # 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 != {}:
- # b = str("measure_time") in self.read_data.keys()
- # if b == True:
- # self.set_time = self.read_data["measure_time"]
- # btn = str("open_hr_btn") in self.read_data.keys()
- # if btn == True:
- # self.open_btn = self.read_data["measure_time"]
+ self.open_btn = read_data
+
+
def post_processor_after_instantiation(self):
EventMesh.subscribe("set_time_hr_spo2", self.set_time_hr_spo2)
def set_time_hr_spo2(self, event, msg):
self.open_btn = msg[0]
self.set_time = msg[1]
+ EventMesh.publish("set_storage_config",["measure_time",self.set_time])
+ EventMesh.publish("set_storage_config",["open_hr_btn",self.open_btn])
def get_time_hr_spo2(self, event, msg):
-
return [self.open_btn,self.set_time]
def set_hr_state(self, event, msg):
def __init__(self):
self.set_time = 0
self.get_temp = 0
- read_data = EventMesh.publish("get_storage_data")
+ read_data = EventMesh.publish("get_storage_config","temp_time")
+
if read_data != None:
- b = str("temp_time") in read_data.keys()
- if b == True:
- self.set_time = read_data["temp_time"]
- # 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 != {}:
- # b = str("temp_time") in self.read_data.keys()
- # if b == True:
- # self.set_time = self.read_data["temp_time"]
+ self.set_time = read_data
+
def post_processor_after_instantiation(self):
EventMesh.subscribe("set_temp_time", self.set_temp_time)
def set_temp_time(self, event, msg):
self.set_time = msg
+ EventMesh.publish("set_storage_config",["set_temp_time",self.set_time])
def get_temp_time(self, event, msg):
-
return self.set_time
def set_temp_state(self, event, msg):
# self.read_data = {}
self.time = 5
self.PWM_backlight = None
- # 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)
+
def post_processor_after_instantiation(self):
EventMesh.subscribe("get_brightness_level", self.get_brightness_level)
def set_brightness_level(self, event, msg):
self.level = msg
-
+ EventMesh.publish("set_storage_config", ["brightness_level",self.level])
+
def get_brightness_level(self, event, msg):
- read_data = EventMesh.publish("get_storage_data")
- if read_data != {}:
- b = str("brightness_level") in read_data.keys()
- if b == True:
- self.level = read_data["brightness_level"]
+ tmp = EventMesh.publish("get_storage_config","brightness_level")
+ if tmp != None:
+ self.level = tmp
return self.level
def set_brightness_time(self, event, msg):
self.time = msg
+ EventMesh.publish("set_storage_config", ["brightness_time",self.time])
def get_brightness_time(self, event, msg):
- # print("get_brightness_time self.read_data:{} self.time:{}".format(self.read_data,self.time))
- read_data = EventMesh.publish("get_storage_data")
- if read_data != {}:
- b = str("brightness_time") in read_data.keys()
- if b == True:
- self.time = read_data["brightness_time"]
+ tmp = EventMesh.publish("get_storage_config","brightness_time")
+ if tmp != None:
+ self.time = tmp
return self.time
def add(self, event, msg):
if self.level < 10:
self.level = self.level + 2
- if self.level >= 10:
- self.level = 10
- print("add self.level:{}".format(self.level))
self.PWM_backlight = PWM(PWM.PWM1, PWM.ABOVE_10US, self.level*100, 1000)
self.PWM_backlight.open()
-
- read_data = EventMesh.publish("get_storage_data")
- read_data["brightness_level"] = self.level
+ EventMesh.publish("set_storage_config", ["brightness_level",self.level])
+ level = EventMesh.publish("get_storage_config","brightness_level")
- EventMesh.publish("set_storage_data", read_data)
- with open("/usr/static/storage_file/config.json","w",encoding="utf-8") as fd:
- ujson.dump(read_data,fd)
def reduce(self, event, msg):
if self.level > 0:
self.level = self.level - 2
- print("reduce self.level:{}".format(self.level))
if self.level <= 0:
self.PWM_backlight = PWM(PWM.PWM1, PWM.ABOVE_10US, 1*100, 1000)
else:
self.PWM_backlight = PWM(PWM.PWM1, PWM.ABOVE_10US,self.level*100, 1000)
self.PWM_backlight.open()
- read_data = EventMesh.publish("get_storage_data")
- read_data["brightness_level"] = self.level
- EventMesh.publish("set_storage_data", read_data)
- with open("/usr/static/storage_file/config.json","w",encoding="utf-8") as fd:
- ujson.dump(read_data,fd)
+ EventMesh.publish("set_storage_config", ["brightness_level",self.level])
+
class getStorageDataManager(Abstract):#获取内存数据
def __init__(self):
- self.read_data = {}
+ self.config_data = {}
check = ql_fs.path_exists("/usr/static/storage_file/config.json")
if check == True:
with open("/usr/static/storage_file/config.json","r") as fd:
- self.read_data = ujson.load(fd)
+ self.config_data = ujson.load(fd)
+ else :
+ self.config_data = {'old_time': 0, 'flip_sw': 0}
+ pass
+ print("self.config_data:{}".format(self.config_data))
def post_processor_after_instantiation(self):
EventMesh.subscribe("set_storage_data", self.set_storage_data)
+ EventMesh.subscribe("set_storage_config", self.set_storage_config)
EventMesh.subscribe("get_storage_data", self.get_storage_data)
+ EventMesh.subscribe("get_storage_config", self.get_storage_config)
def set_storage_data(self, event, msg):
+ # print("set_storage_data:{}".format(msg))
if msg != None:
- self.read_data = msg
-
- def get_storage_data(self, event, msg):
- return self.read_data
+ self.config_data = msg
+ with open("/usr/static/storage_file/config.json",'w+',encoding="utf-8") as fd:
+ ujson.dump(self.config_data,fd)
+ def set_storage_config(self,event,msg):
+ # print("set_storage_key_data1:{}".format(msg))
+ if len(msg) >= 2:
+ self.config_data[msg[0]]=msg[1]
+ # print("set_storage_key_data2:{}".format(self.config_data[msg[0]]))
+ EventMesh.publish("publish_message_event_ui",{"func":self.set_storage_save,"data":0})
+ def set_storage_save(self, data):
+ print("set_storage_key_data2:{}".format(self.config_data))
+ with open("/usr/static/storage_file/config.json",'w+',encoding="utf-8") as fd:
+ ujson.dump(self.config_data,fd)
+ def get_storage_data(self, event, msg):
+ # print("get_storage_data:{}".format(self.config_data))
+ return self.config_data
+ def get_storage_config(self,event,msg):
+ # print("get_storage_key_data1:{}".format(msg))
+ b = msg in self.config_data.keys()
+ if b == True:
+ print("get_storage_key_data2:{}".format(self.config_data[msg]))
+ return self.config_data[msg]
+ else :
+ return None
+ # try:
+ # return self.config_data[msg]
+ # except:
+ # print("get_storage_key_data:none key:{}",msg)
+ # return None
#定位频率
class getUpLoadDataManager(Abstract):#
def __init__(self):
self.upload_time = 1800
- read_data = EventMesh.publish("get_storage_data")
+ read_data = EventMesh.publish("get_storage_config","upload_time")
if read_data != None:
- b = str("upload_time") in read_data.keys()
- if b == True:
- self.upload_time = read_data["upload_time"]
- # 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 != {}:
- # b = str("upload_time") in self.read_data.keys()
- # if b == True:
- # self.upload_time = self.read_data["upload_time"]
+ self.upload_time = read_data
+
def post_processor_after_instantiation(self):
EventMesh.subscribe("set_upload_data", self.set_upload_data)
class getVolumSizeManager(Abstract):#设置或获取音量大小数据
def __init__(self):
self.volume_size = 8
- read_data = EventMesh.publish("get_storage_data")
+ read_data = EventMesh.publish("get_storage_config","volum")
if read_data != None:
- b = str("volum") in read_data.keys()
- if b == True:
- self.volume_size = read_data["volum"]
- # 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 != {}:
- # b = str("volum") in self.read_data.keys()
- # if b == True:
- # self.volume_size = self.read_data["volum"]
- # else:
- # self.volume_size = 8
- # else:
- # self.volume_size = 8
- # print("self.volume_size:{}".format(self.volume_size))
+ self.volume_size = read_data
+
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 set_volum_size(self, event, msg):
self.volume_size = msg
+ EventMesh.publish("set_storage_config", ["volum",self.volume_size])
def get_volum_size(self, event, msg):
return self.volume_size
self.call_timer = osTimer()
self.sos_numer1 = []
self.callnumer =""
- read_data = EventMesh.publish("get_storage_data")
- if read_data != None:
- b1 = str("SOS_NUMBER1") in read_data.keys()
- if b1 == True:
- self.sos_numer1 = [read_data["SOS_NUMBER1"]]
- b2 = str("SOS_NUMBER2") in read_data.keys()
- if b2 == True:
- self.sos_numer1 = [read_data["SOS_NUMBER1"], read_data["SOS_NUMBER2"]]
-
- b3 = str("SOS_NUMBER3") in read_data.keys()
- if b3 == True:
- self.sos_numer1 = [read_data["SOS_NUMBER1"], read_data["SOS_NUMBER2"],read_data["SOS_NUMBER3"]]
- # 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))
+ SOS_NUMBER1 = EventMesh.publish("get_storage_config","SOS_NUMBER1")
+ if SOS_NUMBER1 != None:
+ self.sos_numer1 = SOS_NUMBER1
+ SOS_NUMBER2 = EventMesh.publish("get_storage_config","SOS_NUMBER2")
+ if SOS_NUMBER1 != None:
+ self.sos_numer1 = [SOS_NUMBER1, SOS_NUMBER2]
+
+ SOS_NUMBER3 = EventMesh.publish("get_storage_config","SOS_NUMBER3")
+ if SOS_NUMBER3 != None:
+ self.sos_numer1 = [SOS_NUMBER1, SOS_NUMBER2,SOS_NUMBER3]
+
def post_processor_after_instantiation(self):
EventMesh.subscribe("set_sos_flag", self.set_sos_flag)#设置sos拨打标识
def __init__(self):
self.low_btn = 0
- read_data = EventMesh.publish("get_storage_data")
+ read_data = EventMesh.publish("get_storage_config","low_on")
if read_data != None:
- b1 = str("low_on") in read_data.keys()
- if b1 == True:
- self.low_btn = read_data["low_on"]
- # 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("low_on") in self.read_data.keys()
- # if b1 == True:
- # self.low_btn = self.read_data["low_on"]
+ self.low_btn = read_data
def post_processor_after_instantiation(self):
EventMesh.subscribe("set_low_bat_btn", self.set_low_bat_btn)#
def set_low_bat_btn(self, event, msg):
self.low_btn = msg
+ EventMesh.publish("set_storage_config",["low_on",self.low_btn])
def get_low_bat_btn(self, event, msg):
return self.low_btn
-class getCountDownManager(Abstract):#倒计时
- def __init__(self):
- self.read_data = {}
- self.low_btn = 0
-
- def post_processor_after_instantiation(self):
- EventMesh.subscribe("set_low_bat_btn", self.set_low_bat_btn)#
- EventMesh.subscribe("get_low_bat_btn", self.get_low_bat_btn)
-
- def set_low_bat_btn(self, event, msg):
- self.low_btn = msg
-
- def get_low_bat_btn(self, event, msg):
- return self.low_btn
class playPowerAudioManager(Abstract):#播放开关机音频文件
def __init__(self):
self.off_msg = {}
self.off_flag = 0
self.rtc = RTC()
- read_data = EventMesh.publish("get_storage_data")
+
+ read_data = EventMesh.publish("get_storage_config","off_on")
if read_data != None:
- b1 = str("off_on") in read_data.keys()
- if b1 == True:
- self.off_on = read_data["off_on"]
- b2 = str("off_start") in read_data.keys()
- if b2 == True:
- self.off_start= read_data["off_start"]
-
- b3 = str("off_end") in read_data.keys()
- if b3 == True:
- self.off_end = read_data["off_end"]
- # 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))
+ self.off_on = read_data
+ read_data = EventMesh.publish("get_storage_config","off_start")
+ if read_data != None:
+ self.off_start= read_data
+ read_data = EventMesh.publish("get_storage_config","off_end")
+ if read_data != None:
+ self.off_end = 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 set_boot_off_time(self, event, msg):
- read_data = EventMesh.publish("get_storage_data")
- read_data["off_on"] = msg[0]
- read_data["off_start"] = msg[1]
- read_data["off_end"] = msg[2]
- with open("/usr/static/storage_file/config.json", 'w+',encoding='utf-8') as fd:
- ujson.dump(read_data,fd)
+ EventMesh.publish("get_storage_config",["off_on",msg[0]])
+ EventMesh.publish("get_storage_config",["off_start",msg[1]])
+ EventMesh.publish("get_storage_config",["off_end",msg[2]])
self.off_msg = msg
- # print("read_data:{} self.off_msg:{}".format(read_data,self.off_msg))
+
def get_boot_off_time(self, event, msg):
flag = 0
def set_sms_index(self, event, msg):
self.index = msg
- # print("read_data:{} self.off_msg:{}".format(read_data,self.off_msg))
def get_sms_index(self, event, msg):
class testSmsSetIPManager(Abstract):#
def __init__(self):
self.timer = osTimer()
- self.newip = ""#默认防走丢协议
- self.newcom = 0
-
- read_data = EventMesh.publish("get_storage_data")
- if read_data != None:
- b1 = str("set_new_ip_addr") in read_data.keys()
- if b1 == True:
- self.newip = read_data["set_new_ip_addr"]
- else:
- self.newip = "api.xcloudtech.com"
-
- b2 = str("set_new_com") in read_data.keys()
- if b2 == True:
- self.newcom = read_data["set_new_com"]
- else:
- self.newcom = 9000
-
- b3 = str("set_new_md5") in read_data.keys()
- if b3 == True:
- self.newmd5 = read_data["set_new_md5"]
- else:
- self.newmd5 = "bZld0ALsZTt15tKEZhOlLPgbsO75BLTP"
-
- # 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)
- # print("config.json:{}".format(self.read_data))
- # b1 = str("set_new_ip_addr") in self.read_data.keys()
- # if b1 == True:
- # self.newip = self.read_data["set_new_ip_addr"]
- # else:
- # self.newip = "api.xcloudtech.com"
-
- # b2 = str("set_new_com") in self.read_data.keys()
- # if b2 == True:
- # self.newcom = self.read_data["set_new_com"]
- # else:
- # self.newcom = 9000
-
- # b3 = str("set_new_md5") in self.read_data.keys()
- # if b3 == True:
- # self.newmd5 = self.read_data["set_new_md5"]
- # else:
- # self.newmd5 = "bZld0ALsZTt15tKEZhOlLPgbsO75BLTP"
-
- # self.read_data["set_new_ip_addr"] = self.newip
- # self.read_data["set_new_com"] = self.newcom
- # # self.read_data["set_new_md5"] = self.newmd5
+ self.newip = "api.xcloudtech.com"#默认防走丢协议
+ self.newcom = 9000
+ self.newmd5 = "bZld0ALsZTt15tKEZhOlLPgbsO75BLTP"
+ tmp_data = EventMesh.publish("get_storage_config","set_new_ip_addr")
+ if tmp_data != None:
+ self.newip = tmp_data
+ tmp_data = EventMesh.publish("get_storage_config","set_new_com")
+ if tmp_data != None:
+ self.newcom = tmp_data
+ tmp_data = EventMesh.publish("get_storage_config","set_new_md5")
+ if tmp_data != None:
+ self.newmd5 = tmp_data
- # with open("/ext/static/storage_file/config.json","w",encoding="utf-8") as fd:
- # ujson.dump(self.read_data,fd)
- # else:
- # self.newip = "api.xcloudtech.com"#默认防走丢协议
- # self.newcom = 9000
- # # self.newmd5 = "bZld0ALsZTt15tKEZhOlLPgbsO75BLTP"
- # print("config.json:{}".format(self.read_data))
- # server_addr = "tcptest.tx.kidxin.com"
- # #SETIP:api.xcloudtech.com:9000
- # server_port = 11500
-
def post_processor_after_instantiation(self):
EventMesh.subscribe("set_new_ip", self.set_new_ip)#设置关机时间
self.newcom = int(msg[1])
# self.newmd5 = str(msg[2])
#短信格式：SETIP:api.xcloudtech.com:9000:bZld0ALsZTt15tKEZhOlLPgbsO75BLTP 以冒号分隔 SETIP:ip地址：ip端口号:ip密钥key
- read_data = EventMesh.publish("get_storage_data")
- read_data["set_new_ip_addr"] = self.newip #通过短信设置ip地址
- read_data["set_new_com"]= self.newcom #通过短信设置ip端口号
+
# self.read_data["set_new_md5"] = self.newmd5 #通过短信设置ip的key密钥
-
- with open("/usr/static/storage_file/config.json","w",encoding="utf-8") as fd:
- ujson.dump(read_data,fd)
- with open("/usr/static/storage_file/config.json","r") as fd:
- read_data = ujson.load(fd)
- print("config.json:{}".format(read_data))
- # print("msg:{} self.read_data:{}".format(msg,self.read_data))
+
+ EventMesh.publish("set_storage_config",["set_new_ip_addr",self.newip])
+ EventMesh.publish("set_storage_config",["set_new_com",self.newcom])
+
number = str(msg[2])
sms.sendPduMsg(number, 'set ok', 'UCS2')
- self.timer.start(5*1000,0,self.fun)
+ self.timer.start(10*1000,0,self.fun)
def fun(self,*argc):
Power.powerRestart()
projects / mirrors / ASR1603_G5.git / commitdiff