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

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

Edge Impulse 模型也在我們的 Python 代碼中實現(xiàn)，以便它可以累積計算對象。該算法將當前幀的坐標與之前的幀進行比較；查看相機上是否有新對象，或者該對象之前是否已計數(shù)。在我們的測試中，有時計算的對象數(shù)量仍然不準確，因為該模型仍處于概念驗證階段。不過我們相信這個概念可以進一步發(fā)展到現(xiàn)實世界的應(yīng)用中。

該項目共包括 5 個步驟：

準備

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

使用 FOMO 對象檢測訓練和構(gòu)建模型

在 Raspberry Pi 上部署和測試對象檢測

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

第 1 步：準備

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

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

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

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

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

選擇圖像項目選項，然后分類多個對象。

在 Dashboard 》 Project Info 中，選擇 Bounding Boxes 進行標記方法，選擇 Raspberry Pi 4 進行延遲計算。

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

現(xiàn)在，是時候貼標簽了。單擊標簽隊列選項卡，然后開始在對象周圍拖動一個框并標記它（鴨或烏龜）并保存。重復直到標記所有圖像。確保訓練和測試數(shù)據(jù)之間的比率是理想的，大約為 80/20。

第 3 步：使用 FOMO 對象檢測訓練和構(gòu)建模型

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

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

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

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

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

第 4 步：部署訓練好的模型并在 Raspberry Pi 上進行測試

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

$ edge-impulse-linux-runner

（如果您有多個項目，請?zhí)砑? - clean ）在此過程中，您將被要求登錄您的 Edge Impulse 帳戶。

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

您還可以在瀏覽器上啟動視頻流：http:// 你的樹莓派 IP 地址：4912

Turtle 和 Duck 已通過 x、y 坐標實時成功識別（每次推理時間非常短）。

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

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

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

因為我們使用 Python，所以我們需要安裝 Python 3 Edge Impulse SDK 并從之前的 Edge Impulse 示例中克隆存儲庫。

您還需要下載經(jīng)過訓練的模型文件，以便我們正在運行的程序可以訪問它。輸入這個來下載它：

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

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

$ cd linux-sdk-python/examples/image

然后，使用以下命令運行程序：

$ python3 count_moving_ducks.py ~/modelfile.eim

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

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:])