ESP32 使用MicroPython 控制Gopro Hero Black4

云深无际

发布于 2021-07-23 12:22:08

7650

文章被收录于专栏：云深之无迹云深之无迹

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

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

还是彩色的

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 = 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，所以写成这样

其实对于一个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)

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

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2021-07-11，如有侵权请联系 cloudcommunity@tencent.com 删除

json