使用FOMO物體檢測(cè)算法實(shí)現(xiàn)無(wú)人機(jī)野生動(dòng)物計(jì)數(shù)系統(tǒng)的設(shè)計(jì)

人工智能在牲畜和野生動(dòng)物監(jiān)測(cè)中的作用預(yù)計(jì)將顯著增長(zhǎng)。這個(gè)項(xiàng)目就是一個(gè)例子，展示了人工智能如何使用嵌入式機(jī)器學(xué)習(xí)以快速有效的方式跟蹤和計(jì)數(shù)對(duì)象（動(dòng)物或農(nóng)作物）。該跟蹤系統(tǒng)使用無(wú)人機(jī)飛越場(chǎng)地（向下掃描表面）的計(jì)算機(jī)視覺(jué)，攝像頭朝下。ML 模型將能夠檢測(cè)和區(qū)分動(dòng)物或作物的類(lèi)型，并可以實(shí)時(shí)計(jì)算每種對(duì)象（動(dòng)物/作物）的累積數(shù)量。這使野生動(dòng)物救援隊(duì)能夠監(jiān)測(cè)動(dòng)物/作物的數(shù)量，也可以用于企業(yè)計(jì)算畜牧和農(nóng)業(yè)市場(chǎng)的潛在收入。

本項(xiàng)目使用 Edge Impulse 的 FOMO（Faster Objects， More Objects）物體檢測(cè)算法。野生動(dòng)物/牲畜/資產(chǎn)跟蹤環(huán)境可以通過(guò)選擇灰度圖像塊和具有 2 個(gè)輸出類(lèi)（例如烏龜和鴨子）的 FOMO 對(duì)象檢測(cè)來(lái)模擬和執(zhí)行。該項(xiàng)目利用 FOMO 快速高效的算法對(duì)對(duì)象進(jìn)行計(jì)數(shù)，同時(shí)使用受限微控制器或單板基于 Linux 的計(jì)算機(jī)（如 Raspberry Pi）。

Edge Impulse 模型也在我們的 Python 代碼中實(shí)現(xiàn)，以便它可以累積計(jì)算對(duì)象。該算法將當(dāng)前幀的坐標(biāo)與之前的幀進(jìn)行比較；查看相機(jī)上是否有新對(duì)象，或者該對(duì)象之前是否已計(jì)數(shù)。在我們的測(cè)試中，有時(shí)計(jì)算的對(duì)象數(shù)量仍然不準(zhǔn)確，因?yàn)樵撃Ｐ腿蕴幱诟拍铗?yàn)證階段。不過(guò)我們相信這個(gè)概念可以進(jìn)一步發(fā)展到現(xiàn)實(shí)世界的應(yīng)用中。

該項(xiàng)目共包括 5 個(gè)步驟：

準(zhǔn)備

數(shù)據(jù)采集??和標(biāo)記

使用 FOMO 對(duì)象檢測(cè)訓(xùn)練和構(gòu)建模型

在 Raspberry Pi 上部署和測(cè)試對(duì)象檢測(cè)

構(gòu)建 Python 應(yīng)用程序以檢測(cè)和計(jì)數(shù)（累積）

第 1 步：準(zhǔn)備

使用更新的 Raspberry Pi OS（Buster 或 Bullseye）準(zhǔn)備您的 Raspberry Pi。然后打開(kāi)您的終端應(yīng)用程序并ssh到您的 Pi。

從上方拍攝不同位置的物體（例如鴨子和烏龜），背景不同的照明條件，以確保模型可以在不同的條件下工作（防止過(guò)度擬合）。在這個(gè)項(xiàng)目中，我使用智能手機(jī)攝像頭捕捉圖像以進(jìn)行數(shù)據(jù)收集，以方便使用。

注意：盡量保持圖片中物體大小相似，物體大小的顯著差異會(huì)混淆 FOMO 算法。

項(xiàng)目使用 Edge Impulse 作為機(jī)器學(xué)習(xí)平臺(tái)，所以我們需要登錄（首先創(chuàng)建一個(gè)帳戶），然后轉(zhuǎn)到Edge Impulse并創(chuàng)建新項(xiàng)目。

第 2 步：數(shù)據(jù)采集和標(biāo)記

選擇圖像項(xiàng)目選項(xiàng)，然后分類(lèi)多個(gè)對(duì)象。

在 Dashboard 》 Project Info 中，選擇 Bounding Boxes 進(jìn)行標(biāo)記方法，選擇 Raspberry Pi 4 進(jìn)行延遲計(jì)算。

然后在數(shù)據(jù)采集中，單擊上傳數(shù)據(jù)選項(xiàng)卡，選擇您的文件，選擇自動(dòng)拆分，然后單擊開(kāi)始上傳。

現(xiàn)在，是時(shí)候貼標(biāo)簽了。單擊標(biāo)簽隊(duì)列選項(xiàng)卡，然后開(kāi)始在對(duì)象周?chē)蟿?dòng)一個(gè)框并標(biāo)記它（鴨或?yàn)觚敚┎⒈４?。重?fù)直到標(biāo)記所有圖像。確保訓(xùn)練和測(cè)試數(shù)據(jù)之間的比率是理想的，大約為 80/20。

第 3 步：使用 FOMO 對(duì)象檢測(cè)訓(xùn)練和構(gòu)建模型

準(zhǔn)備好數(shù)據(jù)集后，轉(zhuǎn)到 Create Impulse 并將 96 x 96 設(shè)置為圖像寬度 - 高度（這有助于使模型的內(nèi)存大小保持較?。Ｈ缓筮x擇擬合最短軸，并選擇圖像和對(duì)象檢測(cè)作為學(xué)習(xí)塊。

轉(zhuǎn)到圖像參數(shù)部分，選擇顏色深度作為灰度，然后按保存參數(shù)。

最后，單擊 Generate features 按鈕，您應(yīng)該會(huì)得到如下圖所示的結(jié)果。

然后，導(dǎo)航到目標(biāo)檢測(cè)部分，并保持神經(jīng)網(wǎng)絡(luò)的訓(xùn)練設(shè)置不變——在我們的例子中是非常平衡的預(yù)訓(xùn)練模型，然后我們選擇 FOMO （MobileNet V2 0.35）。通過(guò)按開(kāi)始訓(xùn)練來(lái)訓(xùn)練模型，您可以看到進(jìn)度。如果一切正常，您應(yīng)該會(huì)看到如下內(nèi)容：

之后我們可以測(cè)試模型，進(jìn)入模型測(cè)試部分并單擊全部分類(lèi)。如果準(zhǔn)確率結(jié)果超過(guò) 80%，那么我們可以進(jìn)行下一步——部署。注意：如果準(zhǔn)確率結(jié)果不如預(yù)期，請(qǐng)重新開(kāi)始使用質(zhì)量數(shù)據(jù)、標(biāo)簽，或者只是通過(guò)訓(xùn)練周期和學(xué)習(xí)率設(shè)置更改重新訓(xùn)練模型。

第 4 步：部署訓(xùn)練好的模型并在 Raspberry Pi 上進(jìn)行測(cè)試

現(xiàn)在，我們可以切換到 Raspberry Pi。確保您的 Pi 已安裝所有依賴項(xiàng)和 Edge Impulse for Linux CLI （如步驟 1 所示）并連接您的 Pi 攝像頭（或 USB 網(wǎng)絡(luò)攝像頭）。然后，通過(guò)終端ssh你的 Pi 并輸入：

$ edge-impulse-linux-runner

（如果您有多個(gè)項(xiàng)目，請(qǐng)?zhí)砑? - clean ）在此過(guò)程中，您將被要求登錄您的 Edge Impulse 帳戶。

這將自動(dòng)下載您的模型并將其編譯到您的 Pi，然后開(kāi)始分類(lèi)。結(jié)果將顯示在終端窗口中。

您還可以在瀏覽器上啟動(dòng)視頻流：http:// 你的樹(shù)莓派 IP 地址：4912

Turtle 和 Duck 已通過(guò) x、y 坐標(biāo)實(shí)時(shí)成功識(shí)別（每次推理時(shí)間非常短）。

在這一步之前，我們已經(jīng)取出數(shù)據(jù)并在 Edge Impulse 平臺(tái)上訓(xùn)練了一個(gè)對(duì)象檢測(cè)模型，并在我們的 Raspberry Pi 板上本地運(yùn)行該模型。因此，可以得出結(jié)論，它已成功部署。

第 5 步：構(gòu)建 Python 程序進(jìn)行檢測(cè)和計(jì)數(shù)

為了使該項(xiàng)目對(duì)特定用例更有意義，我們希望它計(jì)算從移動(dòng)相機(jī)（通過(guò)無(wú)人機(jī)）拍攝的每種類(lèi)型對(duì)象的累積計(jì)數(shù)。我們采用 Edge Impulse 的示例對(duì)象檢測(cè)程序，并通過(guò)解決加權(quán)二分匹配問(wèn)題將其轉(zhuǎn)變?yōu)閷?duì)象跟蹤程序，以便可以跨不同幀跟蹤同一對(duì)象。有關(guān)更多詳細(xì)信息，您可以在下面的代碼中查看。

因?yàn)槲覀兪褂?Python，所以我們需要安裝 Python 3 Edge Impulse SDK 并從之前的 Edge Impulse 示例中克隆存儲(chǔ)庫(kù)。

您還需要下載經(jīng)過(guò)訓(xùn)練的模型文件，以便我們正在運(yùn)行的程序可以訪問(wèn)它。輸入這個(gè)來(lái)下載它：

$ edge-impulse-linux-runner --download modelfile.eim

確保您/我們的程序 《count_moving_ducks》 放置在正確的目錄中，例如：

$ cd linux-sdk-python/examples/image

然后，使用以下命令運(yùn)行程序：

$ python3 count_moving_ducks.py ~/modelfile.eim

最后，我們成功實(shí)現(xiàn)了 Edge Impulse FOMO 對(duì)象檢測(cè)模型，并在樹(shù)莓派本地運(yùn)行累積計(jì)數(shù)程序。以我們獲得的速度和精度水平，我們有信心這個(gè)項(xiàng)目也可以用于微控制器，如 Arduino 的 Nicla Vision 或 ESP32 CAM，因此更容易安裝到無(wú)人機(jī)上。

count_moving_ducks.py：

'''
Author: Jallson Suryo & Nicholas Patrick
Date: 2022-07-25
License: CC0
Source: Edge Impulse python SDK example file (classify.py) -- modified
Description: Program to count livestock or wildlife from a drone (moving camera) using
Edge Impulse FOMO trained model.
'''
#!/usr/bin/env python

import device_patches # Device specific patches for Jetson Nano (needs to be before importing cv2)

from math import inf, sqrt
from queue import Queue
import cv2
import os
import sys, getopt
import signal
import time
from edge_impulse_linux.image import ImageImpulseRunner

runner = None
# if you don't want to see a camera preview, set this to False
show_camera = True
if (sys.platform == 'linux' and not os.environ.get('DISPLAY')):
show_camera = False

def now():
return round(time.time() * 1000)

def get_webcams():
port_ids = []
for port in range(5):
print("Looking for a camera in port %s:" %port)
camera = cv2.VideoCapture(port)
if camera.isOpened():
ret = camera.read()[0]
if ret:
backendName =camera.getBackendName()
w = camera.get(3)
h = camera.get(4)
print("Camera %s (%s x %s) found in port %s " %(backendName,h,w, port))
port_ids.append(port)
camera.release()
return port_ids

def sigint_handler(sig, frame):
print('Interrupted')
if (runner):
runner.stop()
sys.exit(0)

signal.signal(signal.SIGINT, sigint_handler)

def help():
print('python classify.py ')

def main(argv):
try:
opts, args = getopt.getopt(argv, "h", ["--help"])
except getopt.GetoptError:
help()
sys.exit(2)

for opt, arg in opts:
if opt in ('-h', '--help'):
help()
sys.exit()

if len(args) == 0:
help()
sys.exit(2)

model = args[0]

dir_path = os.path.dirname(os.path.realpath(__file__))
modelfile = os.path.join(dir_path, model)

print('MODEL: ' + modelfile)

with ImageImpulseRunner(modelfile) as runner:
try:
model_info = runner.init()
print('Loaded runner for "' + model_info['project']['owner'] + ' / ' + model_info['project']['name'] + '"')
labels = model_info['model_parameters']['labels']
if len(args)>= 2:
videoCaptureDeviceId = int(args[1])
else:
port_ids = get_webcams()
if len(port_ids) == 0:
raise Exception('Cannot find any webcams')
if len(args)<= 1 and len(port_ids)> 1:
raise Exception("Multiple cameras found. Add the camera port ID as a second argument to use to this script")
videoCaptureDeviceId = int(port_ids[0])

camera = cv2.VideoCapture(videoCaptureDeviceId)
ret = camera.read()[0]
if ret:
backendName = camera.getBackendName()
w = camera.get(3)
h = camera.get(4)
print("Camera %s (%s x %s) in port %s selected." %(backendName,h,w, videoCaptureDeviceId))
camera.release()
else:
raise Exception("Couldn't initialize selected camera.")

HEIGHT = 96
WIDTH = 96

next_frame_start_time = 0
prev_frame_objects = []
cumulative_counts = {'duck' : 0, 'turtle' : 0}

# iterate through frames
for res, img in runner.classifier(videoCaptureDeviceId):
# print('classification runner response', res)

if "classification" in res["result"].keys():
print('Result (%d ms.) ' % (res['timing']['dsp'] + res['timing']['classification']), end='')
for label in labels:
score = res['result']['classification'][label]
print('%s: %.2f\t' % (label, score), end='')
print('', flush=True)

elif "bounding_boxes" in res["result"].keys():
curr_frame_objects = res["result"]["bounding_boxes"]
m, n = len(prev_frame_objects), len(curr_frame_objects)
print('Found %d bounding boxes (%d ms.)' % (n, res['timing']['dsp'] + res['timing']['classification']))
# iterate through identified objects
for bb in curr_frame_objects:
print('\t%s (%.2f): x=%d y=%d w=%d h=%d' % (bb['label'], bb['value'], bb['x'], bb['y'], bb['width'], bb['height']))
img = cv2.rectangle(img, (bb['x'], bb['y']), (bb['x'] + bb['width'], bb['y'] + bb['height']), (255, 0, 0), 1)

# Pairs objects seen in both the previous frame and the current frame.
# To get a good pairing, each potential pair is given a cost. The problem
# then transforms into minimum cost maximum cardinality bipartite matching.

# populate table
def get_c(a0, a1):
# computes cost of pairs. A cost of inf implies no edge.
A, B = sqrt(HEIGHT ** 2 + WIDTH ** 2) / 8, 5
if a0['label'] != a1['label']: return inf
d2 = (a0['x'] - a1['x']) ** 2 + (a0['x'] - a1['x']) ** 2
dn4 = d2 ** -2 if d2 else 10**20
val = a0['value'] * a1['value'] * (((1 + B) * dn4) / (dn4 + A ** -4) - B)
return inf if val <= 0 else 1 - val
match_c = [[get_c(i, j) for j in curr_frame_objects] for i in prev_frame_objects]

# solves the matching problem in O(V^2E) by repeatedly finding augmenting paths
# using shortest path faster algorithm (SPFA).
# A modified Hungarian algorithm could also have been used.
# 0..m-1: prev, left
# m..m+n-1: this, right
# m+n: source
# m+n+1: sink
source, sink, V = m + n, m + n + 1, m + n + 2
matched = [-1] * (m + n + 2)
adjLis = [[] for i in range(m)] + [[(sink, 0)] for _ in range(n)] + [[(i, 0) for i in range(m)], []]
# left right source sink
for i in range(m):
for j in range(n):
if match_c[i][j] != inf:
adjLis[i].append((j + m, match_c[i][j]))

# finds augmenting paths until no more are found.
while True:
# SPFA
distance = [inf] * V
distance[source] = 0
parent = [-1] * V
Q, inQ = Queue(), [False] * V
Q.put(source); inQ[source] = True
while not Q.empty():
u = Q.get(); inQ[u] = False
for v, w in adjLis[u]:
if u < m and matched[u] == v: continue
if u == source and matched[v] != -1: continue
if distance[u] + w < distance[v]:
distance[v] = distance[u] + w
parent[v] = u
if not inQ[v]: Q.put(v); inQ[v] = True
aug = parent[sink]
if aug == -1: break
# augment the shortest path
while aug != source:
v = aug
aug = parent[aug]
u = aug
aug = parent[aug]
adjLis[v] = [(u, -match_c[u][v - m])]
matched[u], matched[v] = v, u

# updating cumulative_counts by the unmatched new objects
for i in range(n):
if matched[m + i] == -1:
cumulative_counts[curr_frame_objects[i]['label']] += 1

# preparing prev_frame_objects for the next frame
next_prev_frame_objects = curr_frame_objects
# considering objects that became invisible (false negative) for a few frames.
for i in range(m):
if matched[i] != -1: continue
prev_frame_objects[i]['value'] *= 0.7
if prev_frame_objects[i]['value'] >= 0.35:
next_prev_frame_objects.append(prev_frame_objects[i])
prev_frame_objects = next_prev_frame_objects

print("current cumulative_counts:\n %d ducks, %d turtles" % (cumulative_counts['duck'], cumulative_counts['turtle']))

if (show_camera):
cv2.imshow('edgeimpulse', cv2.cvtColor(img, cv2.COLOR_RGB2BGR))
if cv2.waitKey(1) == ord('q'):
break

if (next_frame_start_time > now()):
time.sleep((next_frame_start_time - now()) / 1000)
# operates at a maximum of 5fps
next_frame_start_time = now() + 200
finally:
if (runner):
runner.stop()

if __name__ == "__main__":
main(sys.argv[1:])