树莓派60/100 - Pico连接物联网NB-IoT，给http网页发送POST请求
前面试验通过了SIM7020C的http get 方法，将它封装为一个独立的类，今天试验通过了HTTP POST方法，sim7020.py的源代码。
import machineimport utimeimport ubinasciidef ends_with(s, suffix): l = len(suffix) if len(s) >= l: return s[-l::] == suffix else: return Falsedef get_client_id(resp): id = -1 if '\r\nOK\r\n' in resp: lines = resp.split('\r\n') lines = list(filter(None, lines)) #移除空串 for line in lines: if line.startswith('+CHTTPCREATE:'): print('found: ', line) paras = line.replace('+CHTTPCREATE:', '').split(',') id = int(paras[0]) break return idclass SIM7020: HTTP_GET = 0 HTTP_POST = 1 def __init__(self, uart, power_pin=14): self.uart = uart self.power_enable = machine.Pin(power_pin, machine.Pin.OUT) def read_uart(self, timeout = 2000, fast_return = False): ticks_start = utime.ticks_ms() resp = b'' while utime.ticks_diff(utime.ticks_ms(), ticks_start) < timeout: if self.uart.any(): line = self.uart.readline() resp += line #print("■■■■", utime.ticks_diff(utime.ticks_ms(), ticks_start), line) if ends_with(resp, b'\r\nERROR\r\n'): break if fast_return and ends_with(resp, b'\r\nOK\r\n'): break ticks_start = utime.ticks_ms() return resp def exec_cmd(self, cmd, timeout=2000): #print(">>>AT命令>>>", cmd) self.uart.write((cmd + '\r\n').encode()) resp = self.read_uart(timeout).decode() #print("<<<<响应<<<<", resp) return resp def exec_cmd_fast(self, cmd): self.uart.write((cmd + '\r\n').encode()) resp = "" try: resp = self.read_uart(1000, True).decode() except: resp = "ERROR: UnicodeError" return resp def reboot(self, force = False): if force: self.power_enable.off() utime.sleep(0.1) self.power_enable.on() utime.sleep(2) for _ in range(5): resp = self.exec_cmd_fast('AT') if '\r\nOK\r\n' in resp: print("发现SIM7020C模块，重启成功") break else: print('给SIM7020C发送AT指令，没有得到OK响应，重试……') utime.sleep(1) def get_signal_strength(self): signal_strength = 0 resp = self.exec_cmd_fast("AT+CSQ") if '\r\nOK\r\n' in resp: lines = resp.split('\r\n') lines = list(filter(None, lines)) #移除空串 for line in lines: if line.startswith('+CSQ:'): paras = line.replace('+CSQ:', '').split(',') signal_strength = int(paras[0]) break return signal_strength ### 自检，测试几条AT指令的返回结果 def self_test(self): self.exec_cmd_fast("ATE1") resp = self.exec_cmd_fast("AT+CGMM;+CPIN?;+CGREG?;+COPS?") return resp def get_IP(self): # +CGCONTRDP: 1,5,"cmnbiot","100.1.2.3.255.255.255.0" resp = self.exec_cmd_fast("AT+CGCONTRDP") paras = resp.split(',') if len(paras) >= 4: ip_and_mask = paras[3] ip = ip_and_mask.split('.') if len(ip) >= 4: return '.'.join(ip[0:4]) return '0.0.0.0' def is_ready(self): ip = self.get_IP() if ip != "0.0.0.0": print("IP地址：", ip) return True return False def http_get(self, url): index_slash = url.find('/', 8) # 跳过http://或https:// 找下一个斜杠的位置 host = url[:index_slash] path = url[index_slash:] # CHTTPCREATE会返回一个ID，拿着这个ID，进行访问网页的手续操作 resp = self.exec_cmd_fast('AT+CHTTPCREATE="' + host + '"') id = get_client_id(resp) if id < 0: print(resp) return "" print("■■■■client id: ", id) self.exec_cmd_fast('AT+CHTTPCON=' + str(id)) resp = self.exec_cmd('AT+CHTTPSEND=' + str(id) + ',' + str(self.HTTP_GET) + ',"' + path + '"') print("HTTP SEND的反馈：", resp) self.exec_cmd_fast('AT+CHTTPDISCON=' + str(id)) self.exec_cmd_fast('AT+CHTTPDESTROY=' + str(id)) return resp # +CHTTPSEND: (0-4),(0-3),"path","http header","http content type","http content" def http_post(self, url, content, content_type = 'application/x-www-form-urlencoded'): index_slash = url.find('/', 8) # 跳过http://或https:// 找下一个斜杠的位置 host = url[:index_slash] print('host:' , host) path = url[index_slash:] print('path:' , path) # CHTTPCREATE会返回一个ID，拿着这个ID，进行访问网页的手续操作 resp = self.exec_cmd_fast('AT+CHTTPCREATE="' + host + '"') id = get_client_id(resp) if id < 0: print(resp) return "" print("■■■■client id: ", id) self.exec_cmd_fast('AT+CHTTPCON=' + str(id)) http_header = b'Accept: */*\r\nConnection: Keep-Alive\r\nUser-Agent: SIMCOM_MODULE\r\n' hex_http_header = ubinascii.hexlify(http_header).decode() hex_content = ubinascii.hexlify(content).decode() cmd = 'AT+CHTTPSEND=' + str(id) + ',' + str(self.HTTP_POST) + ',"' + path + '",' \ + hex_http_header + ',"'+ content_type + '",' + hex_content print('CMD: ', cmd) resp = self.exec_cmd(cmd, 5000) print("HTTP SEND的反馈：", resp) self.exec_cmd_fast('AT+CHTTPDISCON=' + str(id)) self.exec_cmd_fast('AT+CHTTPDESTROY=' + str(id)) return resp简单说明一下几个函数：
1）self_test()，AT命令是可以一次发送多条的，中间用分号分隔即可
2）is_readey() 判断模块是否准备就绪，这里以它是否获得本地IP地址为准
3）http_get()，client_id不是硬编码为0，而是判断CHTTPCREATE的返回结果，这个client_id只能是0到4之间的整数，代码还没有加上try…except…，出现异常时，可能造成client_id不能关闭和释放，这个时候只能重启SIM7020C模块
4）http_post()，调用网页的POST方法，关键的一条AT命令是AT+CHTTPSEND: (0-4),(0-3),“path”,“http header”,“http content type”,“http content”，第一个参数是client_id，第二个参数是1，表示POST方法。header和content需要用十六进制字符进行编码
主程序是：
import sim7020import utimeimport ubinasciiuart = machine.UART(0, 115200, tx=machine.Pin(0), rx=machine.Pin(1), txbuf=1024, rxbuf=1024)print(uart)sim = sim7020.SIM7020(uart)sim.reboot(force=True)#sim.reboot()def get_nmic(resp): lines = resp.split('\r\n') nmic = list(filter(None, lines))[-1] if nmic.startswith('+CHTTPNMIC:'): nmic = nmic.replace('+CHTTPNMIC:', '') return nmic.split(',')[-1].strip() else: return ''def hex2str(hex_str): str_bin = ubinascii.unhexlify(hex_str) return str_bin.decode()while not sim.is_ready(): utime.sleep(1)content = b'Time=2021-12-09 12:34:56&Ip=12.34.56.78&Type=%E7%83%9F%E7%81%AB%E4%BA%8B%E4%BB%B6&Authorization=Basic token_example_by_shenlongbin='resp = sim.http_post('http://12.3.4.5:6789/shen_long_bin_test_page', content)print(resp)nmic = get_nmic(resp)content = hex2str(nmic)print(content)在调用http_post()时需要注意几点：
1）准备一个支持POST的服务器网页
2）content type要填对，我测试的页面用的是：application/x-www-form-urlencoded
3）content是给网页传参数，开头不要写’?’，直接用key1=value1&key2=value2的格式
4）调用成功后 CHTTPNMIC里得到返回结果，我的测试程序是一个json字符串
http_get()和http_post()当前访问https协议时会报错，原因未知。
推荐阅读：
树莓派Pico开发系列文章