#设置通话音量
voiceCall.setVolume(10)
-#clog.info("voice call volume {}".format(voiceCall.getVolume()))
-# lpm_fd = pm.create_wakelock("system_lock", len("system_lock"))
-pm.autosleep(1)
-# pm.wakelock_lock(lpm_fd)
-#clog.info("create_wakelock {}".format(pm.get_wakelock_num()))
+clog.info("voice call volume {}".format(voiceCall.getVolume()))
+lpm_fd = pm.create_wakelock("system_lock", len("system_lock"))
+# pm.autosleep(1)
+pm.wakelock_lock(lpm_fd)
+clog.info("create_wakelock {}".format(pm.get_wakelock_num()))
telecomTimerCount = 0
def get_versionkeyname_by_py(app_type):
return str("{}version".format(app_type))
ql_fs.touch(CONFIG_JSON,data)
def set_version_to_json_config(app_type,keyvalue):
- config_data = ql_fs.read_json(CONFIG_JSON)
- config_data.update( {str(get_versionkeyname_by_py(app_type)):keyvalue})
- ql_fs.touch(CONFIG_JSON, config_data)
- #clog.info("{}={}".format(get_versionkeyname_by_py(app_type),keyvalue))
- #clog.info("{}".format(config_data))
+ try :
+ config_data = ql_fs.read_json(CONFIG_JSON)
+ config_data.update( {str(get_versionkeyname_by_py(app_type)):keyvalue})
+ ql_fs.touch(CONFIG_JSON, config_data)
+ clog.info("{}={}".format(get_versionkeyname_by_py(app_type),keyvalue))
+ clog.info("{}".format(config_data))
+ except Exception as e:
+ clog.error("set_version_to_json_config")
def get_version_from_json_config(app_type):
config_data = ql_fs.read_json(CONFIG_JSON)
keyvalue = config_data.get( get_versionkeyname_by_py(app_type),get_default_versionvalue_by_py(app_type))
- #clog.info("{}={}".format(get_versionkeyname_by_py(app_type),keyvalue))
+ clog.info("{}={}".format(get_versionkeyname_by_py(app_type),keyvalue))
return keyvalue
set_version_to_json_config(CURRENT_APP_TYPE,PROJECT_VERSION)
# level - 引脚电平，int类型，0 - 设置引脚为低电平, 1- 设置引脚为高电平。
# recvDataPin = Pin(Pin.GPIO22, Pin.IN, Pin.PULL_PU, 0)
def ExtIntfun(args):
- # #clog.info('### interrupt {} ###'.format(args)) # args[0]:gpio号 args[1]:上升沿或下降沿
+ # clog.info('### interrupt {} ###'.format(args)) # args[0]:gpio号 args[1]:上升沿或下降沿
if args[1] == 0:
- #clog.info("disable autosleep")
- pm.autosleep(0)
- # pm.wakelock_lock(lpm_fd)
+ clog.info("disable autosleep")
+ # pm.autosleep(0)
+ pm.wakelock_lock(lpm_fd)
elif args[1] == 1:
- #clog.info("enable autosleep")
- # pm.wakelock_unlock(lpm_fd)
- pm.autosleep(1)
- # #clog.info("recvDataPin={}".format(recvDataPin.read()))
+ clog.info("enable autosleep")
+ pm.wakelock_unlock(lpm_fd)
+ # pm.autosleep(1)
+ # clog.info("recvDataPin={}".format(recvDataPin.read()))
extint = ExtInt(ExtInt.GPIO22, ExtInt.IRQ_RISING_FALLING, ExtInt.PULL_DISABLE, ExtIntfun)
extint.enable()
self.__wait_resp = 0
self.__waiting_netstat = 0
slip.destroy()
- #clog.info('slip uart={0}'.format (self.__uart))
+ clog.info('slip uart={0}'.format (self.__uart))
ret = slip.construct(self.__uart, self.__mode, 0)
if ret != 0:
clog.error('slip netif construct fail')
return None
- #clog.info('slip network card create success')
+ clog.info('slip network card create success')
# self.__queue = Queue(1)
self.__sock = self.__socket_init()
self.__threadid = _thread.start_new_thread(self.__Socket_Thread, ())
# cell_data = hex(cmd_tmp)
# decimal_number = int(cell_data, 16)
# cmd = "%02x"%(decimal_number) # 使用两位表示，不足前面补0
- #clog.info(cmd) # 输出: 07
- #clog.info("btmcu_cmd_handler cmd {0} , args {1}".format(cmd,parm))
+ clog.info(cmd) # 输出: 07
+ clog.info("btmcu_cmd_handler cmd {0} , args {1}".format(cmd,parm))
if cmd == "00":
- #clog.info(cmd)
+ clog.info(cmd)
# UDP_CMD_GET_CAT1VER
modem_ver = modem.getDevFwVersion()
self.reply_cmd('00',"pyVer:{}@modemVer:{}".format(PROJECT_VERSION,modem_ver))
ret = voiceCall.callStart(str(parm))
# if ret != 0:
# ret = voiceCall.callStart(str(parm))
- #clog.info("ret={} {} {}".format(ret,cmd,parm))
+ clog.info("ret={} {} {}".format(ret,cmd,parm))
if ret != 0:
self.reply_cmd('0F',"callcmd_fail")
else:
else:
self.reply_cmd('0F',"callcmd_fail")
elif cmd == "02":
- #clog.info(cmd)
+ clog.info(cmd)
# UDP_CMD_HUNGUP
ret = voiceCall.callEnd()
if ret != 0:
else:
self.reply_cmd('0F',"callend_success")
elif cmd == "03":
- #clog.info(cmd)
+ clog.info(cmd)
# UDP_CMD_ANSWER
ret = voiceCall.callAnswer()
if ret != 0:
clog.info(cmd)
# UDP_CMD_SEND_SMS
elif cmd == "06":
- #clog.info(cmd)
+ clog.info(cmd)
# UDP_CMD_GET_TIME
time=utime.localtime()
zone=utime.getTimeZone()
#(2024, 3, 2, 18, 31, 32, 5, 62),8
# 返回当前时间戳，参数为元组
# timestamp = utime.mktime(time)
- # #clog.info(timestamp)
+ # clog.info(timestamp)
self.reply_cmd('06',"{0}@{1}".format(time,zone))
elif cmd == "07":
- #clog.info("cmd ==== 07!")
+ clog.info("cmd ==== 07!")
# UDP_CMD_GET_RECV_NETWORKSTAT
simstat = sim.getStatus()
imei = modem.getDevImei()
# ([voice_state, voice_lac, voice_cid, voice_rat, voice_reject_cause, voice_psc], [data_state, data_lac, data_cid, data_rat, data_reject_cause, data_psc])
operator = net.operatorName()
signal = net.csqQueryPoll()
- #clog.info("imei:{0},sim:{1},oper:{2},signal:{3},net:{4}".format(imei,simstat,operator,signal,netstat))
+ clog.info("imei:{0},sim:{1},oper:{2},signal:{3},net:{4}".format(imei,simstat,operator,signal,netstat))
self.reply_cmd('07',"imei:{0}@sim:{1}@oper:{2}@signal:{3}@net:{4}".format(imei,simstat,operator,signal,netstat))
elif cmd == "09":
- #clog.info("{},{}".format(cmd,parm))
+ clog.info("{},{}".format(cmd,parm))
# UDP_CMD_ATCMD
# net.setModemFun(0)
# net.setModemFun(1)
resp=bytearray(50)
ret=atcmd.sendSync("{}\r\n".format(parm),resp,'',10)
if ret == 0:
- #clog.info("reply:{},ret:{}".format(resp.decode(),ret))
+ clog.info("reply:{},ret:{}".format(resp.decode(),ret))
self.reply_cmd('09',"{}".format(resp.decode()))
else:
- #clog.info("cmd error ret:{}".format(ret))
+ clog.info("cmd error ret:{}".format(ret))
self.reply_cmd('09',"{}".format("at error"))
elif cmd == "10" or cmd == "12" or cmd == "14": #获取固件版本号或者获取BT保存的SIM卡的ICCID BLE那边是否开启电信入库功能
pass
elif cmd == "13":
on = int(parm)
- #clog.info("set_dmp_change_mode on:{}".format(on))
+ clog.info("set_dmp_change_mode on:{}".format(on))
eventMesh.publish("set_dmp_change_mode",on)
elif cmd == "16":
if cmd == "10":
self.software_version = str(parm)
eventMesh.publish("set_dev_version",self.software_version)
- #clog.info("software_version:{},{}".format(self.software_version,parm))
+ clog.info("software_version:{},{}".format(self.software_version,parm))
elif cmd == "12":
pass
elif cmd == "14":#
if cmd == "10":
self.software_version = str(parm)
eventMesh.publish("set_dev_version",self.software_version)
- #clog.info("software_version:{},{}".format(self.software_version,parm))
+ clog.info("software_version:{},{}".format(self.software_version,parm))
elif cmd == "16": #获取BLE mac地址
eventMesh.publish("set_dev_mac",str(parm))
- #clog.info("parm:{}\r\n".format(parm))
+ clog.info("parm:{}\r\n".format(parm))
pass
def get_factory_status(self,event,msg):
def delete_sms_cb(self,*argc):
eventMesh.publish_async("remove_all_sms_cb","",1*1024)
def remove_all_sms_cb(self, event, msg):
- #clog.info("remove_all_sms_cb\r\n")
+ clog.info("remove_all_sms_cb\r\n")
msg_nums = sms.getMsgNums() #短信数量
if msg_nums == -1:
msg_nums = 0
sms_data = sms.getSaveLoc()
- #clog.info("{}\r\n".format(sms_data))
+ clog.info("{}\r\n".format(sms_data))
if sms_data == -1:
return
sms_max = sms_data[0][2] #最大存储短信数量
if sms_data == -1:
return
sms_max = sms_data[0][2] #最大存储短信数量
- #clog.info("recv:{}".format(int(msg)))
+ clog.info("recv:{}".format(int(msg)))
message=self.sms_get_message_info(int(msg))
if message == -1: #说明没有这条短信
return
SMS_lis={}
SMS_lis_count = 0
sms_data = sms.getSaveLoc()
- #clog.info(sms_data)
+ clog.info(sms_data)
if sms_data == -1:
return
sms_max = sms_data[0][2]#最大存储短信数量
- #clog.info(sms_max)
+ clog.info(sms_max)
self.msg_nums = sms.getMsgNums()
if self.msg_nums_old != self.msg_nums:
self.send_data = []
count = len(self.send_data)
for i in range(count):
text = self.send_data[i]
- #clog.info("text:{} {} {}\r\n".format(text[0],text[1],text[2]))
+ clog.info("text:{} {} {}\r\n".format(text[0],text[1],text[2]))
msg_total = "@code:"+str(text[0])+"@totalcodenum:"+str(text[1])+"@currcodeid:"+str(text[2])+"@data:"+str(text[3])
self.reply_cmd('17',msg_total)
return 0
self.__waiting_netstat = 1
stage, state = checkNet.waitNetworkReady(30)
- # pm.wakelock_unlock(lpm_fd)
+ pm.wakelock_unlock(lpm_fd)
if stage == 3 and state == 1:
- #clog.info('Network connection successful.')
+ clog.info('Network connection successful.')
self.set_network_config()
self.__net_connected = 1
eventMesh.publish("set_network_state",self.__net_connected)
self.__waiting_netstat = 0
return 0
else:
- #clog.info('Network connection failed, stage={}, state={}'.format(stage, state))
+ clog.info('Network connection failed, stage={}, state={}'.format(stage, state))
self.reset_network_config()
self.__net_connected = 0
eventMesh.publish("set_network_state",self.__net_connected)
def reply_cmd(self,cmd,_timesync):
try :
- #clog.info('reply_cmd 1')
+ clog.info('reply_cmd 1')
self.clear_remain()
message = str(_timesync)
- #clog.info('reply_cmd 2')
+ clog.info('reply_cmd 2')
self.__Socket_UDP_noreply(cmd, message)
except:
clog.info('reply_cmd')
def reset_network_config(self):
if (self.__net_connected == 0):
- #clog.info("skip reset_network_config")
+ clog.info("skip reset_network_config")
return 0
Info = dataCall.getInfo(1,0)
- #clog.info("Info{}".format(Info))
+ clog.info("Info{}".format(Info))
default_NIC_ip = Info[2][2] # 假设默认网卡ip地址为10.10.1.1，请根据实际情况更改
segment_ip = '172.16.4.0' # 假设ap模式网段范围是[192.168.4.0 ~ 192.168.4.255]，请根据实际情况更改
mask = '255.255.255.0' # 假设子网掩码为255.255.255.0，请根据实际情况更改
# 配置默认网卡，通过默认网卡进行网络转发
cat1service.set_default_NIC(default_NIC_ip)
- #clog.info("default_NIC_ip {0} segment_ip {1}".format(default_NIC_ip,segment_ip))
+ clog.info("default_NIC_ip {0} segment_ip {1}".format(default_NIC_ip,segment_ip))
#添加路由规则，只有该网段IP才进行转发
cat1service.router_add(segment_ip, mask)
if (self.__net_connected == 1) :
- #clog.info("skip set_network_config")
+ clog.info("skip set_network_config")
return 0
Info = dataCall.getInfo(1,0)
- #clog.info("Info{}".format(Info))
+ clog.info("Info{}".format(Info))
default_NIC_ip = Info[2][2] # 假设默认网卡ip地址为10.10.1.1，请根据实际情况更改
segment_ip = '172.16.1.0' # 假设ap模式网段范围是[192.168.4.0 ~ 192.168.4.255]，请根据实际情况更改
mask = '255.255.255.0' # 假设子网掩码为255.255.255.0，请根据实际情况更改
# 配置默认网卡，通过默认网卡进行网络转发
cat1service.set_default_NIC(default_NIC_ip)
- #clog.info("default_NIC_ip {0} segment_ip {1}".format(default_NIC_ip,segment_ip))
+ clog.info("default_NIC_ip {0} segment_ip {1}".format(default_NIC_ip,segment_ip))
#添加路由规则，只有该网段IP才进行转发
cat1service.router_add(segment_ip, mask)
return tlv_pack
# 解码tlv数据包
def __unpack_tlv_format(self, msg):
- #clog.info('__unpack_tlv_format {}\r\n'.format(msg))
+ clog.info('__unpack_tlv_format {}\r\n'.format(msg))
self.msg = str(msg)[2:]#这是去除head FE
tag = self.msg[0:2]
length = self.msg[2:6]
if len(value) == 0 or len(value) > 9999 or len(tag) != 2 or len_str != self.msg[2:6]:
clog.error('__unpack_tlv_format err')
return -1
- #clog.info('__unpack_tlv_format: {} {} {}\r\n'.format(tag,length,value))
+ clog.info('__unpack_tlv_format: {} {} {}\r\n'.format(tag,length,value))
unpack = (tag,length,value)
return unpack
# socket通信(UDP)模块
def Socket_SEND_Thread(self):
while True:
try:
- #clog.info("send====")
+ clog.info("send====")
data=self.send_data_q.get()
self.ack_sem=1
# # if data[2]:
self.__Socket_UDP_write( data[0], data[1])
- #clog.info("Socket_SEND_Thread data:{} {}".format(data[0],data[1]))
+ clog.info("Socket_SEND_Thread data:{} {}".format(data[0],data[1]))
count=20
while count & self.ack_sem :
# else :
self.__Socket_UDP_write( data[0], data[1])
- #clog.info("Socket_SEND_Thread while data:{} {}".format(data[0],data[1]))
+ clog.info("Socket_SEND_Thread while data:{} {}".format(data[0],data[1]))
if data:
clog.info("data ok!!")
# 向服务端发送消息
server_addr = ('172.16.1.5',1000)
- #clog.info("__Socket_UDP send msg {}".format(msg))
+ clog.info("__Socket_UDP send msg {}".format(msg))
self.__err = 0
self.__sock.sendto(msg,server_addr)
- #clog.info("__Socket_UDP send 2")
+ clog.info("__Socket_UDP send 2")
self.__wait_resp = 1
# data = self.__queue.get()
self.__wait_resp = 0
# 向服务端发送消息
server_addr = ('172.16.1.5',1000)
- #clog.info("__Socket_UDP_noreply send msg {}".format(msg))
+ clog.info("__Socket_UDP_noreply send msg {}".format(msg))
self.__err = 0
data_size = self.__sock.sendto(msg,server_addr)
- #clog.info("self.__sock data_size {}".format(data_size))
+ clog.info("self.__sock data_size {}".format(data_size))
self.__wait_resp = 0
return 0
def __Socket_Thread(self):
while True:
try:
- #clog.info("recv====")
+ clog.info("recv====")
data=self.__sock.recv(1024)
- #clog.info("data:{}".format(data))
+ clog.info("data:{}".format(data))
if data:
value = self.__unpack_tlv_format(data.decode())
if value != -1:
- #clog.info('mode: {0}, recv {1} bytes, Data: {2}'.format (value[0],len(data), value[2]))
+ clog.info('mode: {0}, recv {1} bytes, Data: {2}'.format (value[0],len(data), value[2]))
if value[1] == 'FF':
self.btmcu_cmd_handler(value[0],value[2])
else:
# self.__value = value
# self.__queue.put('1')
self.btmcu_cmd_handler(value[0],value[2])
- #clog.info("End else Thread")
+ clog.info("End else Thread")
else:
clog.error("data error!!")
except Exception as recverr:
- #clog.info("End Socket Thread")
+ clog.info("End Socket Thread")
# UDP_CMD_GET_TIME = '6', // 获取Cat1 时间 ，YYYYMMDD HHmmss zone
# UDP_CMD_GET_NETWORKSTAT = '7', //获取Cat1 信号信息，VOLTE信息，信号强度，SIM卡状态
# 下文进行系统配置
-#clog.info("Cat1Service start")
+clog.info("Cat1Service start")
cat1service = Cat1Service()
#voice call callback function
def voice_callback(args):
if args[0] == 10:
- #clog.info('voicecall incoming call, PhoneNO: ', args[6])
if eventMesh.publish("get_volte_switch") == 0: #关闭VOLTE功能
- #clog.info('voicecall auto end:')
+ clog.info('voicecall auto end:')
voiceCall.callEnd()
else:
# gpioWakeBT.write(1)
cat1service.notifyIncomingcall(args[6])
elif args[0] == 11:
if eventMesh.publish("get_volte_switch") == 1: #开启VOLTE功能
- #clog.info('voicecall connected, PhoneNO: ', args[6])
+ clog.info('voicecall connected, PhoneNO: \n'.format(args[6]))
cat1service.notifyConnecteCall(args[6])
elif args[0] == 12:
- #clog.info('voicecall disconnect')
+ clog.info('voicecall disconnect')
if eventMesh.publish("get_volte_switch") == 1: #开启VOLTE功能
- #clog.info('voicecall disconnected, PhoneNO: ', args[6])
+ clog.info('voicecall disconnected, PhoneNO: \n'.format(args[6]))
cat1service.notifyDisconnectCall(args[6])
elif args[0] == 13:
- clog.info('voicecall is waiting, PhoneNO: ', args[6])
+ clog.info('voicecall is waiting, PhoneNO: \n'.format(args[6]))
elif args[0] == 14:
- clog.info('voicecall dialing, PhoneNO: ', args[6])
+ clog.info('voicecall dialing, PhoneNO: \n'.format(args[6]))
elif args[0] == 15:
- clog.info('voicecall alerting, PhoneNO: ', args[6])
+ clog.info('voicecall alerting, PhoneNO: \n'.format(args[6]))
elif args[0] == 16:
- clog.info('voicecall holding, PhoneNO: ', args[6])
+ clog.info('voicecall holding, PhoneNO: \n'.format(args[6]))
else:
- clog.info('voice_callback , args[0]: ', args[0])
+ clog.info('voice_callback , args[0]: \n'.format(args[6]))
#电话消息反馈
voiceCall.setCallback(voice_callback)
def checknet_cb(arg):
- #clog.info("start checktime!!")
+ clog.info("start checktime!!")
# if 1: #CHINA_TELECOM_DM_FUN_ENABLE == 1:#这是手表每30s检查一次
# if eventMesh.publish("get_dmp_change_mode") == 1:
# telecomDmpCb.telecom_dmp_power_on_count()
delaybootupmsg_timer.start(1000,0,bootupmsg_cb)
# def checkpin_cb(arg):
-# #clog.info("recvDataPin={}".format(recvDataPin.read()))
+# clog.info("recvDataPin={}".format(recvDataPin.read()))
# timerCheckRevPin = osTimer()
# timerCheckRevPin.start(1000,1,checkpin_cb)
projects / mirrors / FD230.git / commitdiff