【HZHY-AI300G智能盒试用连载体验】驻车辅助系统

车载应用的项目大都需要大量的传感器和执行器使用。除了使用常见的GPIO外，很多传感器还会用到I2C, SPI等串口通讯方式或是模拟量输入输出。这类通信一般在PCB内进行使用，但到了车载应用这种动辄单位就是“米”的场景，就显得不在可靠。

因此，一般的解决方案都是先通过一个MCU获取传感器信息或控制执行器，再由这个MCU集中与上位机进行通讯，也就是我们的HZHY-AI300G智能盒。

这次的评测中，由于有无线倒车可视雷达的测试，而且作为后添加的辅助设备，无线通讯可以极大的方便安装，因此我们都使用无线方式完成HZHY-AI300G智能盒和MCU之间的通信。具体会使用MQTT方案完成传感器与执行器的数据交互，视频则使用http流媒体方式实现。

由于是车载独立环境，我们不能使用互联网上现成的MQTT测试服务器。因此第一步我们要先在上位机中搭建MQTT服务器，也就是Broker。这里我选择的是Mosquitto。

安装方式非常简单，使用apt就可以完成全部安装：

sudo apt update

sudo apt install mosquitto

在正式使用前，还需要简单配置一下，才可以正常使用。我们先看一下默认的配置文件：

cat /etc/mosquitto/mosquitto.conf

运行后应该可以看到以下输出：

通过阅读这个文件，我们知道官方希望我们修改配置的方法并不是直接修改该文件，而是将我们自己的配置文件放到/etc/mosquitto/conf.d/这个文件夹中。那么渠道这个文件夹，创建一个以.conf结尾的配置文件，写入以下简单配置：

allow_anonymous true

listener 1883 0.0.0.0

max_connections -1

这三条命令的含义分别是允许匿名访问；开启1883端口监听所有来源的访问；以及取消最大连接数限制。

更改保存后，不要忘记重启一下mosquitto，这样它才可以加载到我们最新的配置：

sudo service mosquitto restart

接下来，我们可以用python来测试一下。由于测试过程会有一个发送一个接收，因此需要两个窗口同时工作。这里我们可以开启两个终端，当然也可以使用screen这样的工具实现。安装的方式也非常简单，以下命令就可以。

sudo apt install screen

在编写python代码前，我们先创建一个虚拟环境，并补齐需要安装的安装包：

sudo apt install python3-dev

sudo apt install python3-venv

python3 -m venv venv --system-site-packages

source venv/bin/activate

pip install wheel

接着安装MQTT客户端

pip install paho-mqtt

准备好后，我们就可以编写测试代码。首先是发送端：

import time

from paho.mqtt import client as mqtt_client



broker = '127.0.0.1'

port = 1883

topic = "/test"

client_id = "sender"

# username = 'user'

# password = 'password'



def connect_mqtt():

    def on_connect(client, userdata, flags, rc, properties):

        if rc == 0:

            print("Connected to MQTT Broker!")

        else:

            print("Failed to connect, return code %d\\n", rc)

    # Set Connecting Client ID

    client = mqtt_client.Client(client_id=client_id, callback_api_version=mqtt_client.CallbackAPIVersion.VERSION2)



    # client.username_pw_set(username, password)

    client.on_connect = on_connect

    client.connect(broker, port)

    return client





def publish(client):

    msg_count = 0

    while True:

        time.sleep(1)

        msg = f"messages: {msg_count}"

        result = client.publish(topic, msg)

        # result: [0, 1]

        status = result[0]

        if status == 0:

            print(f"Send `{msg}` to topic `{topic}`")

        else:

            print(f"Failed to send message to topic {topic}")

        msg_count += 1





def run():

    client = connect_mqtt()

    client.loop_start()

    publish(client)

    client.loop_stop()





if __name__ == '__main__':

    run()

运行后，我们如果可以看到如下输出，说明已经成功连接到MQTT服务器并发送数据：

接着是接收端的代码：

from paho.mqtt import client as mqtt_client





broker = '127.0.0.1'

port = 1883

topic = "/test"

client_id = "receiver"

# username = 'user'

# password = 'password'





def connect_mqtt():

    def on_connect(client, userdata, flags, rc, properties):

        if rc == 0:

            print("Connected to MQTT Broker!")

        else:

            print("Failed to connect, return code %d\\n", rc)

    # Set Connecting Client ID

    client = mqtt_client.Client(client_id=client_id, callback_api_version=mqtt_client.CallbackAPIVersion.VERSION2)



    # client.username_pw_set(username, password)

    client.on_connect = on_connect

    client.connect(broker, port)

    return client





def subscribe(client: mqtt_client):

    def on_message(client, userdata, msg):

        print(f"Received `{msg.payload.decode()}` from `{msg.topic}` topic")



    client.subscribe(topic)

    client.on_message = on_message





def run():

    client = connect_mqtt()

    subscribe(client)

    client.loop_forever()





if __name__ == '__main__':

    run()

运行后，保持send,py持续运行的状态下，如果看到以下输出，说明成功接收到send.py发过来的信息：

都成功的话，说明我们的MQTT服务器已经可以正常工作，同时，我们也知道了上位机与服务器交互的方式。

要实现车窗智能防结冰，方法是对车内的温度实时监控，当车内外温差过大时启动风扇让空气流通，降低温差。

那么实现项目的本质就是，如何通过MCU检测到温度变化，并实现与HZHY-AI300G智能盒的双向通讯，不但可以把温度上报给HZHY-AI300G智能盒，同时当HZHY-AI300G智能盒下发通风指令时，MCU也可以正确执行。

我搭了一个简单的电路来进行测试。其中使用BMP280作为温度传感器进行数据测量，而用一个小LED作为执行器，用来代表同风扇。MCU使用的是ESP32-S3开发板，开发环境方便起见使用的是Circuitpython。

Circuitpython开发环境准备这一步我就直接跳过，因为和我们评测的HZHY-AI300G智能盒无关。

我们通过两个MQTT Topic来进行通信。第一个Topic是test/topic，这个topic用来从mcu上报传感器数据到HZHY-AI300G智能盒；另一个topic是test/cmd，用来让HZHY-AI300G智能盒下发指令给MCU。

MCU的主要功能为，测量温度并每秒钟上报；如果接收到HZHY-AI300G智能盒下发的信息则按指令开关LED。具体代码如下，注意要把MQTT broker的地址改为HZHY-AI300G智能盒的IP：

import time

import wifi

import socketpool

import ssl

import adafruit_minimqtt.adafruit_minimqtt as MQTT

import json





# Define callback methods which are called when events occur

def connect(client, userdata, flags, rc):

    # This function will be called when the client is connected

    # successfully to the broker.

    print("Connected to MQTT Broker!")

    print("Flags: {0}\\n RC: {1}".format(flags, rc))





def disconnect(client, userdata, rc):

    # This method is called when the client disconnects

    # from the broker.

    print("Disconnected from MQTT Broker!")





def subscribe(client, userdata, topic, granted_qos):

    # This method is called when the client subscribes to a new feed.

    print("Subscribed to {0} with QOS level {1}".format(topic, granted_qos))





def unsubscribe(client, userdata, topic, pid):

    # This method is called when the client unsubscribes from a feed.

    print("Unsubscribed from {0} with PID {1}".format(topic, pid))





def publish(client, userdata, topic, pid):

    # This method is called when the client publishes data to a feed.

    print("Published to {0} with PID {1}".format(topic, pid))





def message(client, topic, message):

    # This method is called when a topic the client is subscribed to

    # has a new message.

    print(f"New message on topic {topic}: {message}")





pool = socketpool.SocketPool(wifi.radio)

ssl_context = ssl.create_default_context()



# Set up a MiniMQTT Client

# NOTE: We'll need to connect insecurely for ethernet configurations.

mqtt_client = MQTT.MQTT(

    broker="192.168.x.x",

    port=1883,

    username="",

    password="",

    is_ssl=False,

    socket_pool=pool,

    ssl_context=ssl_context,

)



# Connect callback handlers to client

mqtt_client.on_connect = connect

mqtt_client.on_disconnect = disconnect

mqtt_client.on_subscribe = subscribe

mqtt_client.on_unsubscribe = unsubscribe

mqtt_client.on_publish = publish

mqtt_client.on_message = message





def func():

    pass





# MQTT Topic

# Use this topic if you'd like to connect to a standard MQTT broker

mqtt_topic = "test/topic"



print("Attempting to connect to %s" % mqtt_client.broker)

try:

    mqtt_client.disconnect()

except:

    pass

mqtt_client.connect()



# print("Subscribing to %s" % mqtt_topic)

# mqtt_client.subscribe(mqtt_topic)



# print("Publishing to %s" % mqtt_topic)

# mqtt_client.publish(mqtt_topic, "Hello Broker!")



# print("Unsubscribing from %s" % mqtt_topic)

# mqtt_client.unsubscribe(mqtt_topic)



# print("Disconnecting from %s" % mqtt_client.broker)

# mqtt_client.disconnect()



import board

import busio



i2c = busio.I2C(scl=board.GPIO7, sda=board.GPIO6)

assert i2c.try_lock()

print(i2c.scan())

i2c.unlock()





# 温度测试

if 1:

    import adafruit_bmp280



    bmp280 = adafruit_bmp280.Adafruit_BMP280_I2C(i2c, address=0x76)

    bmp280.sea_level_pressure = 1013.25



    import digitalio



    led = digitalio.DigitalInOut(board.GPIO15)

    led.direction = digitalio.Direction.OUTPUT

    led.value = True



    mqtt_client.subscribe("test/cmd")



    def func(client, topic, message):

        led.value = int(message)

        print(f"New message on topic {topic}: {message}")



    mqtt_client.on_message = func



    while True:

        # Poll the message queue

        mqtt_client.loop(timeout=1)



        msg = {"Temperature": bmp280.temperature}

        # Send a new message

        mqtt_client.publish(mqtt_topic, json.dumps(msg))

        print(msg)

        time.sleep(1)

接线方式：

SCL: 7

SDA: 6

LED: 15

HZHY-AI300G智能盒的代码我们可以基于上一篇MQTT测试代码修改。两边信息传递使用的json文本，这是一种非常有效的指令及数据传递方式。代码如下，如果检测到上报的温度大于31度，则会要求开启LED，否则则熄灭LED。

from paho.mqtt import client as mqtt_client

import json



broker = '127.0.0.1'

port = 1883

topic = "test/topic"

client_id = "receiver"

# username = 'user'

# password = 'password'





def connect_mqtt():

    def on_connect(client, userdata, flags, rc, properties):

        if rc == 0:

            print("Connected to MQTT Broker!")

        else:

            print("Failed to connect, return code %d\\n", rc)

    # Set Connecting Client ID

    client = mqtt_client.Client(client_id=client_id, callback_api_version=mqtt_client.CallbackAPIVersion.VERSION2)



    # client.username_pw_set(username, password)

    client.on_connect = on_connect

    client.connect(broker, port)

    return client





def subscribe(client: mqtt_client):

    def on_message(client, userdata, msg):

        print(f"Received `{msg.payload.decode()}` from `{msg.topic}` topic")

        my_dict = json.loads(msg.payload.decode())

        if 'Temperature' in my_dict:

            temp = int(my_dict["Temperature"])

            if temp > 31:

                client.publish("test/cmd", "0")

                print("Too Hot!!!")

            else:

                client.publish("test/cmd", "1")





    client.subscribe(topic)

    client.on_message = on_message





def run():

    client = connect_mqtt()

    subscribe(client)

    client.loop_forever()





if __name__ == '__main__':

    run()

同时运行两边的代码，可以看到智能盒打印出了收到的温度信息，当温度高于31度时，打印对应文本，并点亮MCU上的LED。

倒车雷达两种方案的验证，同样也是使用MCU获取两个传感器得到的信号，并把数据一起上报给HZHY-AI300G智能盒。

MCU的代码可以基于之前的代码上做增量修改，因为MQTT通讯部份都是一样的。增加如下部分就可以实现读取两个距离传感器的数据，并上报给HZHY-AI300G智能盒：

# 测距测试

if 0:

    import adafruit_vl53l0x



    vl53 = adafruit_vl53l0x.VL53L0X(i2c)

    import adafruit_us100



    # Connect TX to TX, RX to RX

    uart = busio.UART(board.GPIO4, board.GPIO5, baudrate=9600)

    sonar = adafruit_us100.US100(uart)



    while True:

        # Poll the message queue

        mqtt_client.loop(timeout=1)



        msg = {"vl53l0x": vl53.range, "sonar": int(sonar.distance * 10)}

        # Send a new message

        mqtt_client.publish(mqtt_topic, json.dumps(msg))

        print(msg)

        time.sleep(1)

HZHY-AI300G智能盒部分的代码可以不用做修改。两边同时运行，我们可以看到HZHY-AI300G智能盒有如下输出：

两个传感器都可以测量距离，但是原理不同。一个是基于超声波，使用声纳的方式来测距；另外一个是基于激光，测量方式也和声纳类似，但在激光上被叫做TOF。

两者很难说优略，与其说是替代的关系，不如说是互补的关系。因为激光波长远小于超声波，因此超声波测量范围会比较大，但是精度稍低；而激光精度会高一些，但是范围比较小。因此两种传感器搭配使用的话会获得更好的结果。

这一期我们来看一下执行器的控制。譬如后视镜调整，警报驱动，窗机启停等功能，实际上都是HZHY-AI300G智能盒向MCU下发指令控制执行器的过程。

我们使用一个舵机作为执行器，依然先完成MCU部分的增量代码。代码内容非常简单，把接收到的舵机角度直接传递给舵机即可。

# 舵机测试

if 1:

    import pwmio

    from adafruit_motor import servo



    pwm = pwmio.PWMOut(board.GPIO14, frequency=50)

    s1 = servo.Servo(pwm, min_pulse=500, max_pulse=2500)



    mqtt_client.subscribe("test/cmd")



    def func(client, topic, message):

        print(f"New message on topic {topic}: {message}")

        s1.angle = int(message)

        msg = {"angle": int(message)}

        mqtt_client.publish(mqtt_topic, json.dumps(msg))



    mqtt_client.on_message = func



    while True:

        # Poll the message queue

        mqtt_client.loop(timeout=1)

舵机记得要使用5V供电，接在14号引脚上。

HZHY-AI300G智能盒由于是单纯的下发指令，因此这次代码我们可以基于一开始的发送代码进行修改。出于测试目的，我们让HZHY-AI300G智能盒下发每间隔1S旋转舵机180度的指令：

import time

from paho.mqtt import client as mqtt_client



broker = '127.0.0.1'

port = 1883

topic = "test/topic"

client_id = "sender"

# username = 'user'

# password = 'password'



def connect_mqtt():

    def on_connect(client, userdata, flags, rc, properties):

        if rc == 0:

            print("Connected to MQTT Broker!")

        else:

            print("Failed to connect, return code %d\\n", rc)

    # Set Connecting Client ID

    client = mqtt_client.Client(client_id=client_id, callback_api_version=mqtt_client.CallbackAPIVersion.VERSION2)



    # client.username_pw_set(username, password)

    client.on_connect = on_connect

    client.connect(broker, port)

    return client





def publish(client):

    msg_count = 0

    while True:

        time.sleep(1)

        msg = f"messages: {msg_count}"

        result = client.publish(topic, msg)

        # result: [0, 1]

        status = result[0]

        if status == 0:

            print(f"Send `{msg}` to topic `{topic}`")

        else:

            print(f"Failed to send message to topic {topic}")

        msg_count += 1





def run():

    client = connect_mqtt()

    # client.loop_start()

    # publish(client)

    # client.loop_stop()



    # 舵机测试

    if 1:

        while True:

            client.publish("test/cmd", "0")

            time.sleep(1)

            client.publish("test/cmd", "180")

            time.sleep(1)



if __name__ == '__main__':

    run()

两边代码同时运行，我们就可以观察到舵机按照要求运动起来。观察MCU的控制台输出，可以看到打印出了接收到的数据信息：

而如果我们同步运行HZHY-AI300G智能盒上的接收代码，也可以看到对应的反馈信息：

接下来我们要测试一下无线倒车雷达的实践。正式进行测试前，需要先做一些准备工作。

要想让HZHY-AI300G智能盒显示图像，可以有多种方案，比如说用RS485串口屏，或者HDMI直接输出。目前我手边暂时没有合适的显示设备，因此我们需要换一种方式来把HZHY-AI300G智能盒接收到的视频流显示出来。

之前我对HZHY-AI300G智能盒的所有操作，都是通过ssh完成的。而ssh有一个功能，叫做X11-Forwading。如果平时使用MobaXterm的话，应该会有注意到每次连接都会显示这么一个图像：

想要知道这个功能是什么，需要简单科普一下：

Linux 本身是没有图形化界面的，所谓的图形化界面系统只不过中 Linux 下的应用程序。这一点和 Windows 不一样。Windows 从 Windows 95 开始，图形界面就直接在系统内核中实现了，是操作系统不可或缺的一部分。Linux 的图形化界面，底层都是基于 X 协议。

X 协议由 X server 和 X client 组成：

X server 管理主机上与显示相关的硬件设置（如显卡、硬盘、鼠标等），它负责屏幕画面的绘制与显示，以及将输入设置（如键盘、鼠标）的动作告知 X client。X client (即 X 应用程序) 则主要负责事件的处理（即程序的逻辑）。

举个例子，如果用户点击了鼠标左键，因为鼠标归 X server 管理，于是 X server 就捕捉到了鼠标点击这个动作，然后它将这个动作告诉 X client，因为 X client 负责程序逻辑，于是 X client 就根据程序预先设定的逻辑（例如画一个圆），告诉 X server 说：“请在鼠标点击的位置，画一个圆”。最后，X server 就响应 X client 的请求，在鼠标点击的位置，绘制并显示出一个圆。

那么所谓的X11-Forwarding，就是当X server和X client在不同的机器上时，通过X协议进行远程通讯。在本地显示 (X server)运行在服务器上的 GUI 程序 (X client)。这样的操作可以通过 SSH X11 Forwarding 来实现。X11 中的 X 指的就是 X 协议，11 指的是采用 X 协议的第 11 个版本。

有了X11 Forwarding，通过SSH连接并运行Linux上有GUI的程序，就像是在Windows下运行GUI程序一样方便。很多时候，这样的机制可以方便有图形显示的程序的调试，也就可以通过SSH显示我们的传送到HZHY-AI300G智能盒的视频流。

首先需要在HZHY-AI300G智能盒上安装一下图形化软件：

sudo apt update

sudo apt install xorg

接下来还需要修改下HZHY-AI300G智能盒上的SSHD配置：

sudo nano /etc/ssh/sshd_config

把其中的X11Forwarding改为yes。

重启一下sshd服务：

sudo systemctl restart sshd

最后，在MobaXterm中，我们要在连接配置中勾选X11-Forwarding选项：

再次重新连接SSH，看到如下X11-Forwarding部分已经被搭上了勾，就说明已经启用完毕。

可以测试一下，HZHY-AI300G智能盒本身就带图形界面，预装有firefox浏览器。我们在ssh中直接运行，可以看到HZHY-AI300G智能盒上的firefox浏览器直接就在我们电脑上运行了起来：

firefox

完成上面的步骤后，我们就可以开始准备网络摄像头，来远程推送视频。摄像头我使用的是最常见的ESP32CAM模块：

模块上的代码可以使用arduino的官方CameraWebServer例程：

例程代码里有几处需要修改：

首先我们需要定义我们使用的开发板：

#define CAMERA_MODEL_AI_THINKER

其次是要更改wifi连接信息，也就是以下两行：

const char* ssid = "REPLACE_WITH_YOUR_SSID";

const char* password = "REPLACE_WITH_YOUR_PASSWORD";

上传代码后，如果一切正常，我们可以在串口监视器里看到开发板的ip。

接下来回到HZHY-AI300G智能盒，进入虚拟环境后，先安装所需的库文件：

pip install opencv-python

然后新建一个cam.py的文件，写入下列代码：

import cv2



url = "http://192.168.1.91:81/stream"

cap = cv2.VideoCapture(url)

#size = (640, 480)



while (True):

  ret, frame = cap.read()

  #img = cv2.resize(frame, size)

  cv2.imshow('frame', frame)

  if cv2.waitKey(1) & 0xFF == ord('q'):

    break



cap.release()

cv2.destroyAllWindows()

这段代码的内容是通过opencv从html流媒体上获取视频数据，然后将获取到的图像逐帧显示在窗口上。

运行这段代码，图像通过上期配置好的X11-Forwarding就可以直接显示在我们的电脑上：

最后一期，我将之前所有实验所需的源代码都附上。其中代码部分包括两块：

第一块是MCU部分的代码，所有需要的库文件都已打包在lib文件夹下。

里面每一个独立模块上我都加了if 0:判断，启用哪部分测试模块，就将哪部分的0改为1即可。注意因为每个模块里面均是死循环，因此同时只可启用一个模块；如果启用了多个模块，只会运行第一个启用的。

另外一块是HZHY-AI300G智能盒上的代码。在运行代码前需要先新建一个虚拟环境，然后安装所需的库。库文件列表我已经提取出来，可以直接使用下面的命令安装：

pip install -r requirements.txt

源代码打包文件可以从这里下载：
*附件：source_code.zip