Skip to content

Commit 7ae1a87

Browse files
Mr-FJCjayceon.fu
Mr-FJC
and
jayceon.fu
authored
code (cellular): 提交蜂窝无线网卡应用文档中的示例代码 (#1)
固件版本: N/A 是否需要文案翻译: 是 Co-authored-by: jayceon.fu <jayceon.fu@quectel.com>
1 parent 100db34 commit 7ae1a87

6 files changed

Lines changed: 553 additions & 0 deletions

‎network-comm/nic/cellular/example_network_exception_handle.py‎

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,66 @@
1+
import net
2+
import dataCall
3+
import _thread
4+
import osTimer
5+
import utime
6+
import sys_bus
7+
8+
9+
CFUN_SWITCH_TOPIC = 'cfun_switch_timer'
10+
NETWORK_EVENT_TOPIC = 'network_event'
11+
12+
network_monitor = None
13+
14+
class NetworkConnectState:
15+
DISCONNECT = 0
16+
CONNECT = 1
17+
18+
19+
class NetworkExceptionService:
20+
def __init__(self):
21+
self.timer_period = 1000 * 60 * 2 # 2 * 60s
22+
self.__cfun_timer = osTimer()
23+
24+
sys_bus.subscribe(CFUN_SWITCH_TOPIC, self.__cfun_switch_timer_handle)
25+
sys_bus.subscribe(NETWORK_EVENT_TOPIC, self.__network_event_handle)
26+
27+
def enable(self):
28+
dataCall.setCallback(self.__network_event_callback)
29+
30+
def __cfun_switch_timer_callback(self, args):
31+
print('cfun switch timer.')
32+
sys_bus.publish(CFUN_SWITCH_TOPIC, 0)
33+
34+
def __cfun_switch_timer_handle(self, topic, msg):
35+
print('recv event form cfun_switch_timer.')
36+
net.setModemFun(0)
37+
utime.sleep(5)
38+
net.setModemFun(1)
39+
40+
def __network_event_callback(self, args):
41+
print('The state of the network connection has changed.')
42+
sys_bus.publish(NETWORK_EVENT_TOPIC, args)
43+
44+
def __network_event_handle(self, topic, msg):
45+
print('recv event form network_event_callback.')
46+
profile_id = msg[0]
47+
conn_state = msg[1]
48+
if conn_state == NetworkConnectState.DISCONNECT:
49+
print('The network connection has been disconnected.')
50+
print('start cfun_switch_timer.')
51+
self.__cfun_timer.start(self.timer_period, 1, self.__cfun_switch_timer_callback)
52+
elif conn_state == NetworkConnectState.CONNECT:
53+
print('The network connection has been connected.')
54+
self.__cfun_timer.stop()
55+
else:
56+
print('unknown state value:{}.'.format(conn_state))
57+
58+
59+
def main():
60+
global network_monitor
61+
network_monitor = NetworkExceptionService()
62+
network_monitor.enable()
63+
64+
65+
if __name__ == '__main__':
66+
main()

‎network-comm/nic/cellular/example_socket_activate_default_nic_no_apn.py‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
import checkNet
2+
import usocket
3+
4+
5+
def main():
6+
stage, state = checkNet.waitNetworkReady(20)
7+
if stage == 3 and state == 1:
8+
print('Network connected successfully.')
9+
# 创建一个socket对象
10+
sock = usocket.socket(usocket.AF_INET, usocket.SOCK_STREAM)
11+
# 解析域名
12+
try:
13+
sockaddr = usocket.getaddrinfo('python.quectel.com', 80)[0][-1]
14+
except Exception:
15+
print('Domain name resolution failed.')
16+
sock.close()
17+
return
18+
# 建立连接
19+
sock.connect(sockaddr)
20+
# 向服务端发送消息
21+
ret = sock.send('GET /News HTTP/1.1\r\nHost: python.quectel.com\r\nAccept-Encoding: deflate\r\nConnection: keep-alive\r\n\r\n')
22+
print('send {} bytes'.format(ret))
23+
24+
# 接收服务端消息
25+
data = sock.recv(256)
26+
print('recv {} bytes:'.format(len(data)))
27+
print(data.decode())
28+
29+
# 关闭连接
30+
sock.close()
31+
else:
32+
print('Network connected failed, stage={}, state={}'.format(stage, state))
33+
34+
35+
if __name__ == '__main__':
36+
main()

‎network-comm/nic/cellular/example_socket_activate_default_nic_set_apn.py‎

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,68 @@
1+
import checkNet
2+
import usocket
3+
import dataCall
4+
from misc import Power
5+
6+
# 用户需要配置的APN信息，根据实际情况修改
7+
usrCfg = {'apn': '3gnet', 'username': '', 'password': ''}
8+
9+
10+
def checkAPN():
11+
# 获取第一路网卡的APN信息，确认当前使用的是否是用户指定的APN
12+
pdpCtx = dataCall.getPDPContext(1)
13+
if pdpCtx != -1:
14+
if pdpCtx[1] != usrCfg['apn']:
15+
# 如果不是用户需要的APN，使用如下方式配置
16+
ret = dataCall.setPDPContext(1, 0, usrCfg['apn'], usrCfg['username'], usrCfg['password'], 0)
17+
if ret == 0:
18+
print('APN configuration successful. Ready to restart to make APN take effect.')
19+
print('Please re-execute this program after restarting.')
20+
# 重启后按照配置的信息进行拨号
21+
Power.powerRestart()
22+
else:
23+
print('APN configuration failed.')
24+
return False
25+
else:
26+
print('The APN is correct and no configuration is required')
27+
return True
28+
else:
29+
print('Failed to get PDP Context.')
30+
return False
31+
32+
33+
def main():
34+
checkpass = checkAPN()
35+
if not checkpass:
36+
return
37+
38+
stage, state = checkNet.waitNetworkReady(20)
39+
if stage == 3 and state == 1:
40+
print('Network connected successfully.')
41+
# 创建一个socket对象
42+
sock = usocket.socket(usocket.AF_INET, usocket.SOCK_STREAM)
43+
# 解析域名
44+
try:
45+
sockaddr = usocket.getaddrinfo('python.quectel.com', 80)[0][-1]
46+
except Exception:
47+
print('Domain name resolution failed.')
48+
sock.close()
49+
return
50+
# 建立连接
51+
sock.connect(sockaddr)
52+
# 向服务端发送消息
53+
ret = sock.send('GET /News HTTP/1.1\r\nHost: python.quectel.com\r\nAccept-Encoding: deflate\r\nConnection: keep-alive\r\n\r\n')
54+
print('send {} bytes'.format(ret))
55+
56+
# 接收服务端消息
57+
data = sock.recv(256)
58+
print('recv {} bytes:'.format(len(data)))
59+
print(data.decode())
60+
61+
# 关闭连接
62+
sock.close()
63+
else:
64+
print('Network connected failed, stage={}, state={}'.format(stage, state))
65+
66+
67+
if __name__ == '__main__':
68+
main()

‎network-comm/nic/cellular/example_socket_activate_multiple_nic_set_apn.py‎

Lines changed: 132 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,132 @@
1+
import checkNet
2+
import usocket
3+
import dataCall
4+
from misc import Power
5+
6+
7+
# 用户需要配置的APN信息，根据实际情况修改
8+
usrCfg1 = {'profileID': 1, 'apn': '3gnet', 'username': '', 'password': ''}
9+
usrCfg2 = {'profileID': 2, 'apn': '3gwap', 'username': '', 'password': ''}
10+
11+
12+
def checkAPN(usrCfg, reboot=False):
13+
if type(usrCfg) != dict:
14+
print("Error: Input is not a dictionary.")
15+
return False
16+
17+
print('Check the APN configuration of the {} network card.'.format(usrCfg['profileID']))
18+
# 获取网卡的APN信息，确认当前使用的是否是用户指定的APN
19+
pdpCtx = dataCall.getPDPContext(usrCfg['profileID'])
20+
if pdpCtx != -1:
21+
if pdpCtx[1] != usrCfg['apn']:
22+
# 如果不是用户需要的APN，使用如下方式配置
23+
ret = dataCall.setPDPContext(usrCfg['profileID'], 0, usrCfg['apn'], usrCfg['username'], usrCfg['password'], 0)
24+
if ret == 0:
25+
print('APN configuration successful.')
26+
# 重启后按照配置的信息进行拨号
27+
if reboot:
28+
print('Ready to restart to make APN take effect.')
29+
print('Please re-execute this program after restarting.')
30+
Power.powerRestart()
31+
else:
32+
return True
33+
else:
34+
print('APN configuration failed.')
35+
return False
36+
else:
37+
print('The APN is correct and no configuration is required')
38+
return True
39+
else:
40+
print('Failed to get PDP Context.')
41+
return False
42+
43+
44+
def main():
45+
# 使能第一路网卡开机自动激活功能
46+
dataCall.setAutoActivate(1, 1)
47+
# 使能第一路网卡自动重连功能
48+
dataCall.setAutoConnect(1, 1)
49+
# 使能第二路网卡开机自动激活功能
50+
dataCall.setAutoActivate(2, 1)
51+
# 使能第二路网卡自动重连功能
52+
dataCall.setAutoConnect(2, 1)
53+
54+
# 检查第一路网卡的APN配置，暂时不重启
55+
checkpass = checkAPN(usrCfg1, reboot=False)
56+
if not checkpass:
57+
return
58+
# 检查第二路网卡的APN配置，配置后重启
59+
checkpass = checkAPN(usrCfg2, reboot=True)
60+
if not checkpass:
61+
return
62+
63+
stage, state = checkNet.waitNetworkReady(20)
64+
if stage == 3 and state == 1:
65+
print('Network connected successfully.')
66+
# 分别获取第一路和第二路网卡的IP地址信息
67+
ret1 = dataCall.getInfo(usrCfg1['profileID'], 0)
68+
ret2 = dataCall.getInfo(usrCfg2['profileID'], 0)
69+
print('NIC{}：{}'.format(usrCfg1['profileID'], ret1))
70+
print('NIC{}：{}'.format(usrCfg2['profileID'], ret2))
71+
ip_nic1 = None
72+
ip_nic2 = None
73+
if ret1 == -1 or ret1[2][2] == '0.0.0.0':
74+
print("Error: Failed to get the IP of the NIC{}.".format(usrCfg1['profileID']))
75+
return
76+
else:
77+
ip_nic1 = ret1[2][2]
78+
if ret2 == -1 or ret2[2][2] == '0.0.0.0':
79+
print("Error: Failed to get the IP of the NIC{}.".format(usrCfg2['profileID']))
80+
return
81+
else:
82+
ip_nic2 = ret2[2][2]
83+
print('NIC{} IP：{}'.format(usrCfg1['profileID'], ip_nic1))
84+
print('NIC{} ip：{}'.format(usrCfg2['profileID'], ip_nic2))
85+
86+
print('---------------sock1 test-----------------')
87+
# 创建socket对象
88+
sock1 = usocket.socket(usocket.AF_INET, usocket.SOCK_STREAM)
89+
# 解析域名
90+
try:
91+
sockaddr = usocket.getaddrinfo('python.quectel.com', 80)[0][-1]
92+
except Exception:
93+
print('Domain name resolution failed.')
94+
sock1.close()
95+
return
96+
# 建立连接
97+
sock1.connect(sockaddr)
98+
# 向服务端发送消息
99+
ret = sock1.send('GET /News HTTP/1.1\r\nHost: python.quectel.com\r\nAccept-Encoding: deflate\r\nConnection: keep-alive\r\n\r\n')
100+
print('send {} bytes'.format(ret))
101+
# 接收服务端消息
102+
data = sock1.recv(256)
103+
print('recv {} bytes:'.format(len(data)))
104+
print(data.decode())
105+
# 关闭连接
106+
sock1.close()
107+
print('---------------sock2 test-----------------')
108+
sock2 = usocket.socket(usocket.AF_INET, usocket.SOCK_STREAM, usocket.TCP_CUSTOMIZE_PORT)
109+
sock2.bind((ip_nic2, 0))
110+
sock2.settimeout(10)
111+
# 服务器IP和端口，下面的IP和端口仅作示例参考
112+
server_addr = ('220.180.239.212', 8305)
113+
# 建立连接
114+
sock2.connect(server_addr)
115+
# 向服务器发送消息
116+
ret = sock2.send('test data.')
117+
print('send {} bytes'.format(ret))
118+
# 接收服务端消息
119+
try:
120+
data = sock2.recv(256)
121+
print('recv {} bytes:'.format(len(data)))
122+
print(data.decode())
123+
except Exception:
124+
print('No reply from server.')
125+
# 关闭连接
126+
sock2.close()
127+
else:
128+
print('Network connected failed, stage={}, state={}'.format(stage, state))
129+
130+
131+
if __name__ == '__main__':
132+
main()

0 commit comments

Comments
 (0)