【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 `