【原】ESP32 使用MicroPython 控制Gopro Hero Black4

这个实现的代码比较简单，我们先看看这个硬件，就是这个样子，很精美哦~因为不和国内市场强份额的话，就会做的很不错，算了。不说了

就是这个样子了，小屏幕很好嗷。

还是彩色的

https://docs.m5stack.com/en/quick_start/m5stickc/m5stickc_quick_start_with_uiflow

可以看看官网的图，好好看

一些设置的界面

这个是产品的一些的资料可以下载

下面的链接里面有arduino的开发API

以及mpy的API等

https://shop.m5stack.com/pages/download

https://docs.m5stack.com/en/mpy/advanced/wifi

https://docs.m5stack.com/en/arduino/arduino_development

https://console.cloud.tencent.com/iotexplorer

其实很过分，国产厂家非要搞个这emmmmm

https://github.com/meganetaaan/m5stack-avatar

https://github.com/botofancalin/M5Stack-ESP32-Oscilloscope

特别的，还可以使用云服务，腾讯云

创新的一点是可以用积木编程

然后是结构后的数据，其实是。只不过它没有格式化吧。。。

我看了看感觉的json，然后我写了个处理的脚本

然后就得到一些格式化的数据

你看这个就是可视化的推拽的生成代码

设计出来就是这样的

整体是这样的

大致就很是这样了，看实现吧。然后支持这个Gopro的HTTP协议，就是标准的mpy库是不可以的。因为不能正确的响应上下文，需要做更改：

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.read() 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) # 解析 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 = 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) # 返回一个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() 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)


看下结构

主要是更改这个响应部分的代码，我们做这个努力就好了。具体的实现的时候记得装饰器的使用。@property装饰器就是负责把一个方法变成属性调用。

def __init__(self, f): self.raw = f self.encoding = "utf-8" self._cached = Nonedef close(self): if self.raw: self.raw.close() self.raw = None self._cached = None@propertydef content(self): if self._cached is None: try: self._cached = self.raw.read() finally: self.raw.close() self.raw = None return self._cached@propertydef text(self): return str(self.content, self.encoding)def json(self): import ujson return 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 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.read() 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) # 解析 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 = 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) # 迭代键以避免元组分配 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) # 返回一个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": break 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)


改良过后的库完整的写到这里