??在这个智能硬件和物联网时代,MicroPython和树莓派PICO正以其独特的优势引领着嵌入式开发的新潮流。MicroPython作为一种精简优化的Python 3语言,为微控制器和嵌入式设备提供了高效开发和简易调试的
??当我们结合WIZnet W5100S/W5500网络模块,MicroPython和树莓派PICO的开发潜力被进一步放大。这两款模块都内置了TCP/IP协议栈,使得在嵌入式设备上实现网络连接变得更加容易。无论是进行数据传输、远程控制,还是构建物联网应用,它们都提供了强大的支持。
??本章我们将以WIZnet W5100S为例,以MicroPython的开发方式,连接至OneNET的HTTP服务器上,并定时通过POST上传DHT11温湿度传感器采集的温湿度信息,再通过GET获取温湿度信息。
??首先进入开发者中心,在多协议接入的HTTP界面下创建好产品以及设备,然后添加apikey。
注意:如果您登录之后并没有多协议接入这个选项,则需先进行实名认证。
在数据流模板界面创建温度和湿度的数据流模板。
WIZnet 主流硬件协议栈以太网芯片参数对比
Model | Embedded Core | Host I/F | TX/RX Buffer | HW Socket | Network Performance |
---|---|---|---|---|---|
W5100S | TCP/IPv4, MAC & PHY | 8bit BUS, SPI | 16KB | 4 | Max 25Mbps |
W6100 | TCP/IPv4/IPv6, MAC & PHY | 8bit BUS, Fast SPI | 32KB | 8 | Max 25Mbps |
W5500 | TCP/IPv4, MAC & PHY | Fast SPI | 32KB | 8 | Max 15Mbps |
相较于软件协议栈,WIZnet的硬件协议栈以太网芯片有以下优点:
软件:
硬件:
??我们直接打开http_client.onenet.py文件。
第一步:可以看到在w5x00_init()函数中,进行了SPI的初始化。以及将spi相关引脚和复位引脚注册到库中,后续则是激活网络,并使用DHCP配置网络地址信息,当DHCP失败时,则配置静态网络地址信息。当未配置成功时,会打印出网络地址相关寄存器的信息,可以帮助我们更好的排查问题。
第二步:打开定时器,例程设定是5秒触发一次。
第三步:触发定时器之后,会进入到request()函数,在这个函数中,首先会进行校准传感器,然后先通过POST的方式把温湿度信息上传到平台上,再通过GET的方式获取刚刚上传的温湿度信息,再通过串口打印出来。
需注意:要将程序中的apiaddress替换成您的HTTP api地址,将apikey替换成您的HTTP APIKey
from usocket import socket
from machine import Pin,SPI,Timer
import dht
import urequests
import network
import time
#Timer-related definitions
message_interval = 5
timer_1s_count = 0
tim = Timer()
#HTTP of OneNET
apiaddress = "http://api.heclouds.com/devices/1168040732"
apikey = "fh=hneBwiD6m4o4X==1u=Q6HXp8="
#DHT11 definitions
pin = Pin(2,Pin.OUT)
sensor = dht.DHT11(pin)
"""
W5x00 chip initialization.
param: None
returns: None
"""
def w5x00_init():
spi=SPI(0,2_000_000, mosi=Pin(19),miso=Pin(16),sck=Pin(18))
nic = network.WIZNET5K(spi,Pin(17),Pin(20)) #spi,cs,reset pin
nic.active(True)
try:
#DHCP
print("\r\nConfiguring DHCP")
nic.ifconfig('dhcp')
except:
#None DHCP
print("\r\nDHCP fails, use static configuration")
nic.ifconfig(('192.168.1.20','255.255.255.0','192.168.1.1','8.8.8.8'))#Set static network address information
#Print network address information
print("IP :",nic.ifconfig()[0])
print("Subnet Mask:",nic.ifconfig()[1])
print("Gateway :",nic.ifconfig()[2])
print("DNS :",nic.ifconfig()[3],"\r\n")
#If there is no network connection, the register address information is printed
while not nic.isconnected():
time.sleep(1)
print(nic.regs())
"""
POST submission data and GET request data.
param: None
returns: None
"""
def request():
sensor.measure()
r= urequests.request('POST',apiaddress+'/datapoints',headers={"api-key":apikey},json={'datastreams':[{'id':'CurrentTemperature','datapoints':[{'value':sensor.temperature()}]},{'id':'CurrentHumidity','datapoints':[{'value':sensor.humidity()}]}]})
if not r:
print('spreadsheet: no response received')
print("\r\npost response:",r.json())
r = None
r = urequests.request('GET',apiaddress+'/datapoints',headers={"api-key":apikey})
if r.status_code == 200:
print("get request succ\r\n",r.text)
else:
print("get request error")
"""
1-second timer callback function.
param1: class timer
returns: None
"""
def tick(timer):
global timer_1s_count
timer_1s_count += 1
if timer_1s_count >= message_interval:
timer_1s_count = 0
request()
def main():
print("WIZnet chip HTTP Client of OneNET example")
w5x00_init()
tim.init(freq=1, callback=tick)
while True:
time.sleep(1)
if __name__ == "__main__":
main()
其中还需要用到urequests.py库,代码如下:
import usocket
class Response:
def __init__(self, f):
self.raw = f
self.encoding = "utf-8"
self._cached = None
def close(self):
if self.raw:
self.raw.close()
self.raw = None
self._cached = None
@property
def content(self):
if self._cached is None:
try:
self._cached = self.raw.recv(4096)
finally:
self.raw.close()
self.raw = None
return self._cached
@property
def text(self):
return str(self.content, self.encoding)
def json(self):
import ujson
return ujson.loads(self.content)
def request(method, url, data=None, json=None, headers={}, stream=None):
try:
proto, dummy, host, path = url.split("/", 3)
except ValueError:
proto, dummy, host = url.split("/", 2)
path = ""
if proto == "http:":
port = 80
elif proto == "https:":
import ussl
port = 443
else:
raise ValueError("Unsupported protocol: " + proto)
if ":" in host:
host, port = host.split(":", 1)
port = int(port)
ai = usocket.getaddrinfo(host, port, 0, usocket.SOCK_STREAM)
ai = ai[0]
s = usocket.socket(ai[0], ai[1], ai[2])
try:
s.connect(ai[-1])
if proto == "https:":
s = ussl.wrap_socket(s, server_hostname=host)
s.write(b"%s /%s HTTP/1.0\r\n" % (method, path))
if not "Host" in headers:
s.write(b"Host: %s\r\n" % host)
# Iterate over keys to avoid tuple alloc
for k in headers:
s.write(k)
s.write(b": ")
s.write(headers[k])
s.write(b"\r\n")
if json is not None:
assert data is None
import ujson
data = ujson.dumps(json)
s.write(b"Content-Type: application/json\r\n")
if data:
s.write(b"Content-Length: %d\r\n" % len(data))
s.write(b"\r\n")
if data:
s.write(data)
l = s.readline()
#print(l)
l = l.split(None, 2)
status = int(l[1])
reason = ""
if len(l) > 2:
reason = l[2].rstrip()
while True:
l = s.readline()
if not l or l == b"\r\n":
break
#print(l)
if l.startswith(b"Transfer-Encoding:"):
if b"chunked" in l:
raise ValueError("Unsupported " + l)
elif l.startswith(b"Location:") and not 200 <= status <= 299:
raise NotImplementedError("Redirects not yet supported")
except OSError:
s.close()
raise
resp = Response(s)
resp.status_code = status
resp.reason = reason
return resp
def head(url, **kw):
return request("HEAD", url, **kw)
def get(url, **kw):
return request("GET", url, **kw)
def post(url, **kw):
return request("POST", url, **kw)
def put(url, **kw):
return request("PUT", url, **kw)
def patch(url, **kw):
return request("PATCH", url, **kw)
def delete(url, **kw):
return request("DELETE", url, **kw)
要测试以太网示例,必须将开发环境配置为使用Raspberry Pi Pico。
第一步:将程序复制到Thonny中,然后选择环境为Raspberry Pi Pico
第二步:将urequests.py这个库文件保存到树莓派Pico开发板中
第二步:运行http_client_onenet.py程序,可以在OneNET平台上看到上报的传感器信息。
注意:因为MicroPython的print函数是启用了stdout缓冲的,所以有时候并不会第一时间打印出内容。
想了解更多,评论留言哦!