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/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-avatarhttps://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)
改良过后的库完整的写到这里
赞 (0)