描述

介紹

在美國，全國約有 200 萬公里的污水管道，為 2.4 億美國公民提供服務(wù)。據(jù)報道，美國每年至少有 23,000–75,000 起下水道管道故障，這會向環(huán)境排放多達 3–100 億加侖的未經(jīng)處理的污水。這會導(dǎo)致巨大的經(jīng)濟損失、水污染，并威脅到公共健康。

這些下水道管道需要每年維護，以保持正常運行并避免下水道溢出。在大多數(shù)情況下，下水道檢查由專家檢查員在現(xiàn)場進行，他們通常使用帶有攝像頭的遙控機器人手動檢查下水道的內(nèi)部結(jié)構(gòu)。這個過程非常耗時，而且由于工作的困難和令人厭煩的性質(zhì)，時間常常會導(dǎo)致對下水道管道的檢查有缺陷。

小下水道

TinySewer 是一個獨立的攝像頭模塊，可使用 tinyML 識別下水道缺陷。該模塊旨在安裝在現(xiàn)有的機器人下水道檢查平臺上，為平臺提供機器視覺功能，以在檢查過程中識別下水道故障。

安裝在機器人汽車上的 TinySewer

?

該模塊允許自主下水道檢查并減少檢查員的工作量。檢查員可以簡單地緩慢駕駛汽車并觀察屏幕上的 TinySewer 應(yīng)用程序是否有任何檢測，或者他們可以停下來手動檢查。此外，TinySewer 將準確判斷存在哪種類型的故障，因此無需專業(yè)的下水道檢查員、普通檢查員甚至入門級檢查員就足夠了。

為了進一步支持自主檢測，TinySewer 客戶端應(yīng)用程序在視頻時間線上記錄所有帶有檢測標簽的鏡頭，以便檢查員查看鏡頭并輕松選擇故障發(fā)生的時間點。這允許檢查員在進行下水道檢查的同時處理其他任務(wù)。

可擴展性

在可擴展性方面，TinySewer 非常便宜，因為它每臺僅需 150 美元左右，并且可以輕松集成到現(xiàn)有的機器人系統(tǒng)或下水道檢查工具中。此外，TinySewer 強大的故障檢測系統(tǒng)允許創(chuàng)建一個更大的系統(tǒng)，其中自主下水道檢查機器人定期進行下水道檢查并將帶有缺陷報告的鏡頭發(fā)送到單個主機計算機可以對各種下水道故障報告進行分類并指定人員修復(fù)有缺陷的下水道.

以 TinySewer 作為機器視覺系統(tǒng)的自主下水道檢測系統(tǒng)

?

低功耗能力

Arduino Portenta

?

TinySewer 使用 Arduino Portenta H7 作為其主要計算單元。Arduino Portenta H7 采用雙核低功耗 Cortex M7 處理器，有助于降低功耗。

此外，TinySewer 允許其操作員在檢查待機或完成時關(guān)閉攝像頭。這將總共節(jié)省大約 40mA。最后，操作員可以選擇關(guān)閉 Tiny Sewer，這將使設(shè)備進入深度睡眠模式，直到有外部中斷將其喚醒。

原理圖

小下水道示意圖

?

TinySewer 模塊包含一個帶有外圍設(shè)備的 Arduino Portenta 微控制器，包括一個通過高密度連接器連接的 Vision Shied，用于相機饋送。還有兩個直接連接到 Arduino Portenta PH15 引腳并由 PWM 控制的白色 LED。一切都由 5V、2.4 安培便攜式電池供電。

套管

TinySewer外殼使用 PLA 燈絲制成，可以使用本文提供的文件從 3D 打印機輕松打印。外殼包括頂部、底部和蓋子。

?

該裝置如下組裝，Arduino Portenta H7 先進入，然后將兩個 LED 放在頂部 2 個中間孔上，并將它們連接到 Arduino Portenta H7 的電線上。然后將頂部放在底部，然后在 4 個角上放 4 個 m3 螺釘。最后，將蓋子蓋在裸露的插銷部分上，以防止水和灰塵進入設(shè)備內(nèi)部。

?

?

?

?

1 / 3

?

集會

?

該模型

TinySewer 使用深度學(xué)習(xí)神經(jīng)網(wǎng)絡(luò)架構(gòu)來分類和識別各種下水道故障類型。目前，TinySewer 可以以至少 85% 的置信度檢測四種最常見的下水道故障類型（裂縫、根部侵入、阻塞、位移）

TinySewer 可以檢測到的下水道故障類型。(a) 阻塞，(b) 根部侵入，(c) 位移，(d) 裂縫

?

該模型是使用Edge Impulse機器學(xué)習(xí)平臺創(chuàng)建的。首先，我從ScienceData獲取圖像。數(shù)據(jù)集附帶一個 CSV，其中包含圖像名稱及其故障類型。我只是創(chuàng)建了一個簡單的 python 腳本來讀取這個 CSV 并將圖像分類到其各自的故障文件夾中。然后將這些圖像上傳到 Edge Impluse 進行訓(xùn)練。總的來說，我們的模型有 5 個不同的類別：正常、裂縫-斷裂-塌陷、障礙物、根部和位移。

接下來，我將創(chuàng)建一個沖動頁面來設(shè)置工作流程。選擇 96x96 作為圖像寬 x 高，處理塊作為圖像，遷移學(xué)習(xí)圖像，然后單擊“生成參數(shù)”。

創(chuàng)造沖動

?

接下來，轉(zhuǎn)到圖像選項卡以生成特征參數(shù)。請記住為顏色深度選擇灰度，因為 Arduino Portenta Vision Shield 是單色相機。

生成特征

?

最后，選擇遷移學(xué)習(xí)選項卡來訓(xùn)練您的模型。對于 TinySewer，我使用 MobileNetV2，學(xué)習(xí)率為 0.35，最后一層有 40 個神經(jīng)元。

遷移學(xué)習(xí)

?

該模型使用 50 個 epoch 進行訓(xùn)練，數(shù)據(jù)增強開啟。該模型整體準確率在94%左右。

準確性

?

最后，為 Arduino Portenta 生成模型文件和標簽文件。我轉(zhuǎn)到部署選項卡并選擇 OpenMV 并單擊構(gòu)建。這將生成一個 zip 文件，其中包括label.txt （標簽文件）、train.tflite （模型文件）和 ei_image_classification.py（python 分類腳本）。將label.txt和train.tflite復(fù)制并粘貼到 Arduino Portenta 內(nèi)部存儲中。該腳本需要修改以提供 WLAN、視頻和數(shù)據(jù)傳輸功能。這些修改將在固件部分討論

部署

?

固件

固件是使用 MicroPython 制作的，它只是 python3 的一個實現(xiàn)，帶有一個標準 Python 庫的子集，經(jīng)過優(yōu)化可以在微控制器上運行。

首先是設(shè)置wifi，這可以通過使用WLAN方法簡單地完成。然后創(chuàng)建一個套接字端口，以便客戶端可以與同一網(wǎng)絡(luò)上的 TinySewer 通信。

# Create server socket
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)

# Bind and listen
print(PORT)
s.bind([HOST, PORT])
s.listen(5)

# Set server socket to blocking
s.setblocking(True)

# Create server socket
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)

# Bind and listen
print(PORT)
s.bind([HOST, PORT])
s.listen(5)

接下來，程序初始化camera對象、MQTT對象，并設(shè)置變量load為model和label

# Init Camera
sensor.reset()
sensor.set_framesize(sensor.QVGA)
sensor.set_pixformat(sensor.GRAYSCALE)


# Load in Model and labels
net = "trained.tflite"
labels = [line.rstrip('\n') for line in open("labels.txt")]

#Setup MQTT
payload = MQTTClient("openmv", "test.mosquitto.org", port=1883)
payload.connect()

接下來我們定義流式傳輸函數(shù)，該流式傳輸函數(shù)使用 MJPEG 協(xié)議將來自 TinySewer 攝像頭的視頻流傳輸回客戶端應(yīng)用程序，

def start_streaming(s):
    print ('Waiting for connections..')
    client, addr = s.accept()
    # set client socket timeout to 5s
    client.settimeout(5.0)
    print ('Connected to ' + addr[0] + ':' + str(addr[1]))
    # Read request from client
    data = client.recv(1024)
    # Should parse client request here
    # Send multipart header
    client.sendall("HTTP/1.1 200 OK\r\n" \
                "Server: OpenMV\r\n" \
                "Content-Type: multipart/x-mixed-replace;boundary=openmv\r\n" \
                "Cache-Control: no-cache\r\n" \
                "Pragma: no-cache\r\n\r\n")
    # FPS clock
    clock = time.clock()
    # Start streaming images

    while (True):
        clock.tick() # Track elapsed milliseconds between snapshots().
        frame = sensor.snapshot()
        cframe = frame.compressed(quality=35)
        predict = prediction(frame)
        #print(predict)
        header = "\r\n--openmv\r\n" \
                 "Content-Type: image/jpeg\r\n"\
                 "Content-Length:"+str(cframe.size())+"\r\n\r\n"
        client.sendall(header)
        client.sendall(cframe)
        #client.sendall(bytes('POST /%s HTTP/1.0\r\nHost: 127.0.0.1:9990\r\n\r\n' % (predict), 'utf8'))
        payload.publish("openmv/test", str(predict))
        payload.check_msg() # poll for messages.
        print(clock.fps())

接下來，我定義了一種預(yù)測方法，它只查看當前幀并使用 tinyML 模型計算每個標簽的置信度。標簽和各自的置信度被連接成一個字符串，然后通過 MQTT 發(fā)送到客戶端應(yīng)用程序

def prediction(img):
    prediction = ""
    #print("predict call")
    for obj in tf.classify(net, img, min_scale=1.0, scale_mul=0.8, x_overlap=0.5, y_overlap=0.5):
        #print("**********\nPredictions at [x=%d,y=%d,w=%d,h=%d]" % obj.rect())
        #frame.draw_rectangle(obj.rect())
        # This combines the labels and confidence values into a list of tuples
        predictions_list = list(zip(labels, obj.output()))
        #prediction = predictions_list[0][1] #defect confidence
        for i in range(len(predictions_list)):
            #print("%s = %f" % (predictions_list[i][0], predictions_list[i][1]))
            label = str(predictions_list[i][0])
            confident = str(predictions_list[i][1])
            prediction += label + ":" + confident + ","
            #name = (predictions_list[i][0])
            #if label != "normal" :
                #prediction = predictions_list[i][1]
    return prediction

然后我有一個 lightcontrol() 方法來控制兩個 LED 的亮度。該方法采用 0 到 100 之間的整數(shù)，0 表示最亮，100 表示沒有光

def lightControl(percent):
    for k, pwm in pwms.items():
        tim = Timer(pwm.tim, freq=1000) # Frequency in Hz
        ch  = tim.channel(pwm.ch, Timer.PWM, pin=Pin(pwm.pin), pulse_width_percent=percent)

最后，還有一個設(shè)置初始光照值的主while循環(huán)，稱為視頻流函數(shù)。

while (True):
    try:
        lightControl(50)
        start_streaming(s)
        print("main call")
    except OSError as e:
        print("socket error: ", e)
        #sys.print_exception(e)

完整的實現(xiàn)在 Github 上名為 sewer.py 的文件中

軟件

該軟件是使用名為 Electron 的框架制作的。Electron 允許使用Node.js等 Web 技術(shù)開發(fā)桌面 GUI 應(yīng)用程序。TinySewer 客戶端分為兩個選項卡。第一個選項卡包含來自 TinySewer 的視頻流、用于錄制視頻的按鈕、用于燈光控制的按鈕以及當前下水道故障及其置信度的顯示框。

視頻流頁面

?

第二個選項卡用于視頻分析。Stream 會自動保存為 .mp4 視頻文件，可以回放以供進一步分析。此外，還有一個視頻時間線，其中包含檢測到下水道故障時的持續(xù)時間亮點

分析頁面

?

小下水道在行動

?