这个实现的代码比较简单，我们先看看这个硬件，就是这个样子，很精美哦~因为不和国内市场强份额的话，就会做的很不错，算了。不说了
就是这个样子了，小屏幕很好嗷。
还是彩色的
https://docs.m5stack.com/en/quick_start/m5stickc/m5stickc_quick_start_with_uiflow
可以看看官网的图，好好看
一些设置的界面
这个是产品的一些的资料可以下载
下面的链接里面有arduino的开发API
以及mpy的API等
https://shop.m5stack.com/pages/downloadhttps://docs.m5stack.com/en/mpy/advanced/wifihttps://docs.m5stack.com/en/arduino/arduino_developmenthttps://console.cloud.tencent.com/iotexplorer
其实很过分，国产厂家非要搞个这emmmmm
https://github.com/meganetaaan/m5stack-avatar
https://github.com/botofancalin/M5Stack-ESP32-Oscilloscope
特别的，还可以使用云服务，腾讯云
创新的一点是可以用积木编程
然后是结构后的数据，其实是。只不过它没有格式化吧。。。
我看了看感觉的json，然后我写了个处理的脚本
然后就得到一些格式化的数据
你看这个就是可视化的推拽的生成代码
设计出来就是这样的
整体是这样的
大致就很是这样了，看实现吧。然后支持这个Gopro的HTTP协议，就是标准的mpy库是不可以的。因为不能正确的响应上下文，需要做更改：
import usocketclass Response:def __init__(self, f):self.raw = fself.encoding = "utf-8"self._cached = Nonedef close(self):if self.raw:self.raw.close()self.raw = Noneself._cached = Nonedef content(self):if self._cached is None:try:self._cached = self.raw.read()finally:self.raw.close()self.raw = Nonereturn self._cacheddef text(self):return str(self.content, self.encoding)def json(self):import ujsonreturn ujson.loads(self.content)# 解析 JSON str并返回一个对象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 = 80elif proto == "https:":import usslport = 443else: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 allocfor 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 Noneimport ujsondata = ujson.dumps(json)# 返回一个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"\n" 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()raiseresp = Response(s)resp.status_code = statusresp.reason = reasonreturn respdef 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)
看下结构
主要是更改这个响应部分的代码，我们做这个努力就好了。具体的实现的时候记得装饰器的使用。@property装饰器就是负责把一个方法变成属性调用。
def __init__(self, f):self.raw = fself.encoding = "utf-8"self._cached = Nonedef close(self):if self.raw:self.raw.close()self.raw = Noneself._cached = None@propertydef content(self):if self._cached is None:try:self._cached = self.raw.read()finally:self.raw.close()self.raw = Nonereturn self._cached@propertydef text(self):return str(self.content, self.encoding)def json(self):import ujsonreturn ujson.loads(self.content)
这个是类的初始化，完成了连接和关闭，以及获取回复信息
然后还是想解析json，所以写成这样
其实对于一个ujson来讲，实现的不多，刚好就够用而已
从源地址收到的信息是不透明的，需要自己构建的解析
使用这个函数来创建连接该服务的所有参数
后面是套接字类型
ai = usocket.getaddrinfo(host, port, 0, usocket.SOCK_STREAM)ai = ai[0]
这段相当于的debug方便的处理代码
如果http，端口80
如果是https，就启动SSL
如果不是HTTP的连接协议的话，直接就raise走起。不支持这个协议了，后面把这个url的最前面的str打印一下
这个代码用来分离端口
然后
就是上面的get函数 得到一个连接源的所有必要信息
借助这些就建立和相机的连接
接着就是主要的逻辑了，但是你要注意，现在虽然是连接上了，但是读数据还是从getaddress来处理
http://10.5.5.9/gp/gpControl/execute?p1=gpStream&a1=proto_v2&c1=restart这里写个个URL给你开开眼
https://www.reddit.com/r/gopro/comments/2md8hm/how_to_livestream_from_a_gopro_hero4/cr1b193连接好以后，就是https的连接了
https://docs.micropython.org/en/latest/reference/glossary.html#term-CPython
这里需要使用这个模块
就一个函数
接着就是读写了
其实读写的时候还有一个完全的发送函数
这个是逐个的发送头
先判断json是不是空
然后用一个assert继续判断
接着开始我们捕获这个json文件，import ujson
后面就是发送一个，看我的链接
https://www.cnblogs.com/zh718594493/p/12115579.html
要求发送json
发送数据，反正都是样板代码
Content-Length,在http的协议中Content-Length头部告诉浏览器报文中实体数据的大小。
先从port读data，然后下面去处理
直接分割两次
把空格处理一下
不说了，直接看代码，后面是一个内置的错误处理
这些函数很简单，直接封装了
下面的代码是可以直接写入到ESP32里面的，我就不放了，有需要的私聊我。
import usocketclass Response:def __init__(self, f):self.raw = fself.encoding = "utf-8"self._cached = Nonedef close(self):if self.raw:self.raw.close()self.raw = Noneself._cached = Nonedef content(self):if self._cached is None:try:self._cached = self.raw.read()finally:self.raw.close()self.raw = Nonereturn self._cacheddef text(self):return str(self.content, self.encoding)def json(self):import ujsonreturn ujson.loads(self.content)# 解析 JSON str并返回一个对象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 = 80elif proto == "https:":import usslport = 443else: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)# 迭代键以避免元组分配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 Noneimport ujsondata = ujson.dumps(json)# 返回一个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()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"\n" or l == b"\r\n":breakif 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()raiseresp = Response(s)resp.status_code = statusresp.reason = reasonreturn respdef 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)
改良过后的库完整的写到这里
