在这篇文章中，我们将使用 Raspberry Pi 和 AI 功能创建一个复杂的家庭安全系统。我们的系统将识别家庭成员、检测陌生人并识别包裹递送，同时发送实时网络通知。
硬件要求
- Raspberry Pi 5（这是我默认使用的那个）
- Hailo 8L Raspberry Pi AI 套件（在此处提供）
- NexiGo 网络摄像头（功能物超所值，请在此处查看）
- 互联网连接（幸运的是，Raspberry Pi 5 带有集成的 Wi-Fi 连接并具有以太网端口）
软件设置
首先，让我们使用必要的软件设置我们的 Raspberry Pi。我将跳过作系统安装，因为 CanaKit Raspberry Pi 5 套件附带了预装了系统的 SD 卡：
sudo apt update && sudo apt full-upgrade
sudo apt install hailo-all
git clone https://github.com/hailo-ai/hailo-rpi5-examples.git
cd hailo-rpi5-examples
source setup_env.sh
./compile_postprocess.sh
pip3 install opencv-python-headless numpy supervision pushbullet.py face_recognition
代码
现在，让我们创建 smart_security_system.py 文件：
import cv2
import numpy as np
import supervision as sv
import face_recognition
import time
from hailo_rpi_common import GStreamerApp, app_callback_class
from pushbullet import Pushbullet
from hailo_models import YoloV5PostProcessing
# Initialize Pushbullet for notifications
pb = Pushbullet("YOUR_API_KEY")
# Load known faces
known_face_encodings = []
known_face_names = []
defload_known_faces(directory):
for filename in os.listdir(directory):
if filename.endswith(".jpg") or filename.endswith(".png"):
image = face_recognition.load_image_file(os.path.join(directory, filename))
encoding = face_recognition.face_encodings(image)[0]
known_face_encodings.append(encoding)
known_face_names.append(os.path.splitext(filename)[0])
load_known_faces("known_faces")
# Initialize YOLOv5 object detection
yolo_postprocess = YoloV5PostProcessing()
@app_callback_class
classSmartSecurityCallback:
def__init__(self):
self.last_notification_time = 0
self.face_locations = []
self.face_names = []
self.process_this_frame = True
defapp_callback(self, buffer, caps):
frame = self.get_numpy_from_buffer(buffer, caps)
# Resize frame for faster face recognition processing
small_frame = cv2.resize(frame, (0, 0), fx=0.25, fy=0.25)
rgb_small_frame = small_frame[:, :, ::-1]
if self.process_this_frame:
# Find all faces in the current frame
self.face_locations = face_recognition.face_locations(rgb_small_frame)
face_encodings = face_recognition.face_encodings(rgb_small_frame, self.face_locations)
self.face_names = []
for face_encoding in face_encodings:
matches = face_recognition.compare_faces(known_face_encodings, face_encoding)
name = "Unknown"
face_distances = face_recognition.face_distance(known_face_encodings, face_encoding)
best_match_index = np.argmin(face_distances)
if matches[best_match_index]:
name = known_face_names[best_match_index]
self.face_names.append(name)
self.process_this_frame = not self.process_this_frame
# Detect objects (people and packages)
detections = yolo_postprocess.postprocess(frame)
for detection in detections:
if detection.class_id == 0: # Person
if"Unknown"in self.face_names:
self.send_notification("Stranger detected")
else:
self.send_notification(f"Family member detected: {', '.join(set(self.face_names))}")
elif detection.class_id == 39: # Package (assuming class ID 39 for package in COCO dataset)
self.send_notification("Package delivery detected")
defsend_notification(self, message):
current_time = time.time()
if current_time - self.last_notification_time > 60: # Limit to one notification per minute
push = pb.push_note("Smart Security Alert", message)
self.last_notification_time = current_time
defget_numpy_from_buffer(self, buffer, caps):
# Convert GStreamer buffer to numpy array
# Implementation depends on the specific GStreamer setup
pass
defmain():
app = GStreamerApp("Smart Security System", SmartSecurityCallback())
app.run()
if __name__ == "__main__":
main()
很多人问我如何将 GStreamer 的 cap 缓冲区转换为 NumPy 数组，因此在这里我向大家分享我的解决方案，特别是在 cap 是视频的情况下：
import numpy as np
import gi
gi.require_version('Gst', '1.0')
from gi.repository import Gst
defget_numpy_from_buffer(self, buffer, caps):
"""
Convert GStreamer buffer to numpy array
:param buffer: Gst.Buffer
:param caps: Gst.Caps
:return: numpy.ndarray
"""
# Get the Gst.Structure from the caps
structure = caps.get_structure(0)
# Get the width and height of the video frame
width = structure.get_value("width")
height = structure.get_value("height")
# Get the pixel format (assuming it's in caps)
format_info = structure.get_value("format")
# Map the buffer to memory
success, map_info = buffer.map(Gst.MapFlags.READ)
ifnot success:
raise ValueError("Could not map buffer")
try:
# Get the data from the mapped buffer
data = map_info.data
# Determine the data type and shape based on the pixel format
if format_info == "RGB":
dtype = np.uint8
shape = (height, width, 3)
elif format_info == "RGBA":
dtype = np.uint8
shape = (height, width, 4)
elif format_info == "GRAY8":
dtype = np.uint8
shape = (height, width)
elif format_info == "GRAY16_LE":
dtype = np.uint16
shape = (height, width)
else:
raise ValueError(f"Unsupported format: {format_info}")
# Create numpy array from the buffer data
array = np.ndarray(shape=shape, dtype=dtype, buffer=data)
# Make a copy of the array to ensure it's not tied to the original buffer
return np.array(array)
finally:
# Unmap the buffer
buffer.unmap(map_info)
此实现执行以下作：
- 它从 GStreamer Cap 中提取宽度、高度和像素格式。
- 它将缓冲区映射到内存以访问其数据。
- 根据像素格式，它确定适当的 numpy 数据类型和形状。
- 它从缓冲区数据创建一个 numpy 数组。
- 最后，它返回数组的副本，以确保它不绑定到原始缓冲区。
请注意，此实现采用某些像素格式（RGB、RGBA、GRAY8 GRAY16_LE）。您可能需要添加更多格式处理，具体取决于您的特定用例。此外，请确保您已安装必要的 GStreamer 和 numpy 依赖项：
pip install numpy PyGObject
您可能还需要在系统上安装 GStreamer 开发库。在 Ubuntu 或 Debian 上，您可以通过以下方式执行此作：
sudo apt-get install libgstreamer1.0-dev libgstreamer-plugins-base1.0-dev
如何运作
- 人脸识别：系统使用
face_recognition库来识别已知人脸。您需要使用家庭成员的图像填充known_faces目录。 - 对象检测：我们使用 Hailo 8L 加速的 YOLOv5 来检测视频流中的人员和包裹。
- 通知： 当系统检测到陌生人、家庭成员或包裹递送时，系统会通过 Pushbullet 发送 Web 通知。
- 性能：通过利用 Hailo 8L AI 加速器，我们实现了对视频源的实时处理，确保了快速响应时间。
设置系统
- 将
“YOUR_API_KEY”替换为您的实际 Pushbullet API 密钥。 - 将家庭成员的照片添加到
known_faces目录，并使用人员的姓名（例如，john.jpg）命名每个文件。 - 如果需要，根据您在 Hailo 8L 上使用的特定模型，调整
YoloV5PostProcessing参数。
结论
这个智能家居安全系统展示了将 Raspberry Pi 与 AI 功能相结合的强大功能。Hailo 8L Raspberry Pi AI 套件提供了实时运行复杂 AI 模型所需的处理能力，而 NexiGo 网络摄像头则确保了高质量的视频输入。
通过构建此系统，您不仅可以增强家庭安全性，还可以获得 AI 和计算机视觉方面的宝贵经验。扩展的可能性是无穷无尽的——您可以添加入侵者警报、宠物检测或与智能家居设备集成等功能。
请记住，出色的 DIY 项目的关键是选择正确的组件。Hailo 8L 套件和 NexiGo 相机在性能和价值之间实现了出色的平衡，使其成为该项目的理想选择。
