《2025 DigiKey AI应用创意挑战赛》基于FRDM-IMX93与EtherCAT工业AIoT智能控制系统
本帖最后由 eefocus_3863048 于 2026-2-3 10:41 编辑# 《2025 DigiKey AI应用创意挑战赛》基于FRDM-IMX93与EtherCAT工业AIoT智能控制系统
## 一，项目名称
基于FRDM-IMX93与EtherCAT工业AIoT智能控制系统
## 二，项目概述
本项目设计了一套面向工业AIoT场景的智能工业边缘控制系统。该系统采用云-边-端协同架构，利用FRDM-IMX93作为EtherCAT主站负责视频监控，AI推理和危险行为报警，HPM6E作为EtherCAT从站负责实时数据采集和控制执行。系统面向工业AIoT场景，实现实时状态监测和AI安全规范实时监控的一体化解决方案。
功能特性
1. EtherCAT 工业通信
* 主站/从站通信支持
* 实时工业控制
* 报警信号输出
* 支持内核模块动态加载
2. ZLMediaKit 媒体服务器
* **RTSP/RTMP/HTTP/WebRTC** 多种协议支持
* **HLS/DASH** 自适应码流
* **GB28181** 国标协议对接
* **Web 管理界面** (Element UI)
* **RESTful API** + Swagger 文档
* **FFmpeg** 硬件加速编解码
* **x264** H.264 编码支持
3. 跌倒检测系统
* **YOLOv5 TensorFlow Lite** 边缘推理
* **NPU 加速**
* **实时视频流** RTMP 推送
* **EtherCAT 报警联动**
* **云端告警** HTTP 上报
* **本地存储** 事件图片保存
* **多模型支持** (INT8/VELA 优化)
## 三，系统框架
**整体系统架构图**!(https://www.eefocus.com/forum/data/attachment/forum/202602/02/165124w9b8cdfktvx090gv.png)
**线程与队列架构图**
!(https://www.eefocus.com/forum/data/attachment/forum/202602/02/021522oam6hoed1s1w7w7k.png)
**数据流与处理管道**
!(https://www.eefocus.com/forum/data/attachment/forum/202602/02/021540ix7ihikh97twg97a.png)
**ZLMediaKit 推流架构**
!(https://www.eefocus.com/forum/data/attachment/forum/202602/02/021654yww9qd28228ztnny.png)
**系统时序图**
!(https://www.eefocus.com/forum/data/attachment/forum/202602/02/022124kltp1t26140466zr.png)
## 四，硬件部分
**硬件部分主要有一下部分组成：**
* **EtherCat从站-HPM6E**!(https://www.eefocus.com/forum/data/attachment/forum/202602/02/021739i9fbcsr24yljvahs.png)
* **EtherCat主站-FRDM-IMX93**!(https://www.eefocus.com/forum/data/attachment/forum/202602/02/021751k83fhzn1z89ff3f3.png)
**硬件连接图：**
!(https://www.eefocus.com/forum/data/attachment/forum/202602/02/022330sdgk949264drrrk1.png)
## 五，软件部分
#### FRDM-IMX93核心代码
```
# camera_simple_imx93_with_cloud_and_streaming.py
import argparse
import time
import os
import numpy as np
import cv2
import signal
import sys
import subprocess
import threading
import queue
import atexit
import json
import urllib.request
import urllib.error
import tflite_runtime.interpreter as tflite
from datetime import datetime
def load_labels(filename):
"""Load label file, ignore empty lines"""
labels = []
with open(filename, 'r') as f:
for line in f:
line = line.strip()
if line:# Only add non-empty lines
labels.append(line)
return labels
class SimpleCameraInference:
"""Simple camera inference for IMX93 with cloud server integration"""
def __init__(self, model_file, label_file, conf_thres=0.25,
delegate_path=None, camera_device="/dev/video0",
rtmp_url=None, ethercat_enable=False,
cloud_server_url=None, device_id="imx93_camera_001",
location="Industrial Area A"):
self.conf_thres = conf_thres
self.camera_device = camera_device
self.rtmp_url = rtmp_url
self.ethercat_enable = ethercat_enable
self.cloud_server_url = cloud_server_url
self.device_id = device_id
self.location = location
self.running = True
# Alarm status tracking
self.alarm_status = 0# 0: No alarm, 1: Alarm
self.last_alarm_status = 0
self.alarm_count = 0
self.alarm_threshold = 3# Need 3 consecutive detections
# Cloud alert tracking
self.last_cloud_alert_time = 0
self.cloud_alert_cooldown = 30# Minimum 30 seconds between cloud alerts
self.cloud_alert_count = 0
# Load labels
self.labels = load_labels(label_file)
print(f"Loaded {len(self.labels)} labels: {self.labels}")
# Load model
print(f"Loading model: {model_file}")
interpreter_options = {}
if delegate_path and os.path.exists(delegate_path):
print(f"Using Ethos-U delegate: {delegate_path}")
try:
delegate =
interpreter_options['experimental_delegates'] = delegate
except Exception as e:
print(f"Delegate error: {e}")
self.interpreter = tflite.Interpreter(
model_path=model_file,
**interpreter_options
)
self.interpreter.allocate_tensors()
# Get model info
self.input_details = self.interpreter.get_input_details()
self.output_details = self.output_details = self.interpreter.get_output_details()
self.input_shape = self.input_details['shape']
self.input_height = self.input_shape
self.input_width = self.input_shape
print(f"Model input: {self.input_width}x{self.input_height}")
# Cloud server status
if self.cloud_server_url:
print(f"Cloud server enabled: {self.cloud_server_url}")
self.test_cloud_connection()
# FFmpeg streaming
self.ffmpeg_process = None
if self.rtmp_url:
self.init_ffmpeg_stream()
# EtherCAT communication
if self.ethercat_enable:
self.init_ethercat()
# Initialize camera with direct V4L2
self.init_camera()
# Statistics
self.frame_count = 0
self.total_inference_time = 0
self.last_stats_time = time.time()
# Create save directory
self.save_dir = "saved_frames"
os.makedirs(self.save_dir, exist_ok=True)
# Heartbeat thread for cloud server
if self.cloud_server_url:
self.heartbeat_interval = 60# Send heartbeat every 60 seconds
self.heartbeat_thread = threading.Thread(target=self.heartbeat_worker, daemon=True)
self.heartbeat_thread.start()
print(f"Heartbeat thread started (interval: {self.heartbeat_interval}s)")
# Streaming queue
self.stream_queue = queue.Queue(maxsize=10)
self.stream_thread = None
if self.rtmp_url:
self.start_stream_thread()
# Register cleanup
atexit.register(self.cleanup)
def test_cloud_connection(self):
"""Test connection to cloud server"""
try:
url = f"{self.cloud_server_url}/api/health"
req = urllib.request.Request(url, method='GET')
response = urllib.request.urlopen(req, timeout=5)
data = json.loads(response.read().decode('utf-8'))
print(f"✓ Cloud server connection successful: {data.get('status')}")
return True
except Exception as e:
print(f"✗ Cloud server connection failed: {e}")
return False
def send_to_cloud(self, alert_data):
"""Send alert data to cloud server"""
if not self.cloud_server_url:
return False
current_time = time.time()
if current_time - self.last_cloud_alert_time < self.cloud_alert_cooldown:
print(f"Cloud alert cooldown: {int(self.cloud_alert_cooldown - (current_time - self.last_cloud_alert_time))}s remaining")
return False
try:
url = f"{self.cloud_server_url}/api/alerts"
# Ensure required fields
alert_data['device_id'] = self.device_id
alert_data['location'] = self.location
alert_data['timestamp'] = datetime.now().isoformat()
# Convert to JSON
json_data = json.dumps(alert_data).encode('utf-8')
# Create request
req = urllib.request.Request(url, data=json_data, method='POST')
req.add_header('Content-Type', 'application/json')
# Send request
response = urllib.request.urlopen(req, timeout=10)
result = json.loads(response.read().decode('utf-8'))
self.last_cloud_alert_time = current_time
self.cloud_alert_count += 1
print(f"✓ Alert sent to cloud: {result.get('message')}")
return True
except urllib.error.URLError as e:
print(f"✗ Network error sending to cloud: {e}")
except Exception as e:
print(f"✗ Error sending to cloud: {e}")
return False
def heartbeat_worker(self):
"""Send periodic heartbeat to cloud server"""
while self.running:
try:
time.sleep(self.heartbeat_interval)
if self.cloud_server_url:
heartbeat_data = {
'device_id': self.device_id,
'event_type': 'heartbeat',
'timestamp': datetime.now().isoformat(),
'location': self.location,
'message': 'Device heartbeat',
'status': 'active',
'fps': 0,# Will be updated
'alarm_status': self.alarm_status,
'cloud_alerts_sent': self.cloud_alert_count
}
# Get FPS if available
if hasattr(self, 'last_fps'):
heartbeat_data['fps'] = self.last_fps
self.send_to_cloud(heartbeat_data)
except Exception as e:
print(f"Heartbeat error: {e}")
time.sleep(30)# Wait longer on error
def init_ffmpeg_stream(self):
"""Initialize FFmpeg streaming process"""
print(f"\nInitializing FFmpeg stream to: {self.rtmp_url}")
# Stream settings
width = 640
height = 480
fps = 15
# FFmpeg command
ffmpeg_cmd = [
'ffmpeg',
'-y',# Overwrite output files
'-f', 'rawvideo',
'-vcodec', 'rawvideo',
'-pix_fmt', 'bgr24',
'-s', f'{width}x{height}',
'-r', str(fps),
'-i', '-',# Read from stdin
'-c:v', 'libx264',
'-preset', 'ultrafast',
'-tune', 'zerolatency',
'-pix_fmt', 'yuv420p',
'-f', 'flv',
self.rtmp_url
]
try:
self.ffmpeg_process = subprocess.Popen(
ffmpeg_cmd,
stdin=subprocess.PIPE,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
)
print("FFmpeg stream initialized successfully")
except Exception as e:
print(f"Failed to initialize FFmpeg stream: {e}")
self.ffmpeg_process = None
def init_ethercat(self):
"""Initialize EtherCAT communication"""
print("\nInitializing EtherCAT communication...")
# Check ethercat command
try:
result = subprocess.run(['which', 'ethercat'],
capture_output=True, text=True)
if result.returncode != 0:
print("Warning: ethercat command not found")
self.ethercat_enable = False
return
# Test EtherCAT
test_cmd = ['ethercat', 'slaves']
result = subprocess.run(test_cmd, capture_output=True, text=True)
if result.returncode == 0:
print("EtherCAT initialized successfully")
print(f"Slaves: {result.stdout}")
# Reset alarm state
self.set_ethercat_alarm(0)
else:
print(f"EtherCAT test failed: {result.stderr}")
self.ethercat_enable = False
except Exception as e:
print(f"EtherCAT initialization error: {e}")
self.ethercat_enable = False
def set_ethercat_alarm(self, alarm_value):
"""Set EtherCAT alarm value (0=no alarm, 1=alarm)"""
if not self.ethercat_enable:
return False
try:
if alarm_value == 0:
cmd = ['ethercat', 'download', '-p', '0', '--type', 'uint16', '0x7010', '0', '0x00']
else:
cmd = ['ethercat', 'download', '-p', '0', '--type', 'uint16', '0x7010', '0', '0x01']
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode == 0:
status = "ALARM" if alarm_value == 1 else "NORMAL"
print(f"EtherCAT output set: {status}")
return True
else:
print(f"EtherCAT command failed: {result.stderr}")
return False
except Exception as e:
print(f"EtherCAT communication error: {e}")
return False
def start_stream_thread(self):
"""Start streaming thread"""
if not self.ffmpeg_process:
return
def stream_worker():
while self.running:
try:
# Get frame from queue with timeout
frame = self.stream_queue.get(timeout=1)
if frame is not None and self.ffmpeg_process:
try:
# Write to FFmpeg
self.ffmpeg_process.stdin.write(frame.tobytes())
except Exception as e:
print(f"Stream write error: {e}")
break
except queue.Empty:
continue
except Exception as e:
print(f"Stream thread error: {e}")
break
self.stream_thread = threading.Thread(target=stream_worker, daemon=True)
self.stream_thread.start()
print("Streaming thread started")
def push_to_stream(self, frame):
"""Push frame to streaming queue"""
if self.ffmpeg_process and self.stream_queue.qsize() < 5:
try:
# Resize for streaming
stream_frame = cv2.resize(frame, (640, 480))
self.stream_queue.put_nowait(stream_frame)
except queue.Full:
pass# Skip frame if queue is full
except Exception as e:
print(f"Push to stream error: {e}")
def init_camera(self):
"""Initialize camera using direct V4L2 device path"""
print(f"\nInitializing camera: {self.camera_device}")
# Try different approaches
try:
# Method 1: Direct device path with V4L2
print("Trying direct V4L2 device path...")
self.cap = cv2.VideoCapture(self.camera_device, cv2.CAP_V4L2)
if not self.cap.isOpened():
# Method 2: Try as index if path is numeric
print("Trying as camera index...")
try:
idx = int(self.camera_device.split('/dev/video')[-1])
self.cap = cv2.VideoCapture(idx, cv2.CAP_V4L2)
except:
pass
if not self.cap.isOpened():
# Method 3: Try default
print("Trying default camera...")
self.cap = cv2.VideoCapture(0, cv2.CAP_V4L2)
if self.cap.isOpened():
print("Camera opened successfully!")
# Set basic properties
self.cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640)
self.cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 480)
self.cap.set(cv2.CAP_PROP_FPS, 15)
# Get actual properties
self.camera_width = int(self.cap.get(cv2.CAP_PROP_FRAME_WIDTH))
self.camera_height = int(self.cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
fps = self.cap.get(cv2.CAP_PROP_FPS)
print(f"Camera resolution: {self.camera_width}x{self.camera_height}")
print(f"Camera FPS: {fps:.1f}")
# Calculate scaling
self.scale_x = self.camera_width / self.input_width
self.scale_y = self.camera_height / self.input_height
# Test read
ret, frame = self.cap.read()
if ret:
print(f"Test frame read: {frame.shape}")
else:
print("Warning: Could not read test frame")
return
except Exception as e:
print(f"Camera initialization error: {e}")
print("ERROR: Could not initialize camera")
print("Switching to test mode...")
self.cap = None
self.test_mode = True
def get_test_frame(self):
"""Generate test frame for demonstration"""
frame = np.zeros((480, 640, 3), dtype=np.uint8)
# Add timestamp
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
cv2.putText(frame, f"TEST MODE | {timestamp}",
(10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 2)
# Add some test objects
t = time.time()
# Person
px = 320 + int(100 * np.sin(t))
py = 240 + int(80 * np.cos(t * 0.7))
cv2.rectangle(frame, (px-50, py-100), (px+50, py+100), (0, 255, 0), 2)
cv2.putText(frame, "person: 0.85", (px-50, py-110),
cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 2)
# Fall (occasionally)
if int(t) % 5 == 0:
fx = 160 + int(50 * np.cos(t * 0.5))
fy = 360 + int(30 * np.sin(t * 0.3))
cv2.rectangle(frame, (fx-40, fy-60), (fx+40, fy+60), (0, 0, 255), 2)
cv2.putText(frame, "fall: 0.72", (fx-40, fy-70),
cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 255), 2)
return frame
def get_frame(self):
"""Get frame from camera or generate test frame"""
if self.cap and self.cap.isOpened():
ret, frame = self.cap.read()
if ret and frame is not None and frame.size > 0:
return frame
# If camera fails, generate test frame
return self.get_test_frame()
def preprocess(self, frame):
"""Preprocess frame for model"""
# Resize to model input
frame_resized = cv2.resize(frame, (self.input_width, self.input_height))
# Convert BGR to RGB
frame_rgb = cv2.cvtColor(frame_resized, cv2.COLOR_BGR2RGB)
# For uint8 model
input_data = frame_rgb.astype(np.uint8)
input_data = np.expand_dims(input_data, axis=0)
return frame, input_data
def inference(self, input_data):
"""Run inference"""
self.interpreter.set_tensor(self.input_details['index'], input_data)
start_time = time.perf_counter()
self.interpreter.invoke()
inference_time = (time.perf_counter() - start_time) * 1000
output = self.interpreter.get_tensor(self.output_details['index'])
return output, inference_time
def postprocess(self, output, frame_shape):
"""Post-process inference results - FIXED VERSION"""
detections = []
# Get predictions
predictions = output# Shape:
# Dequantize if needed
if self.output_details.get('quantization', (0, 0)) != 0:
scale, zero_point = self.output_details['quantization']
# Dequantize predictions
predictions = (predictions.astype(np.float32) - zero_point) * scale
# Calculate combined confidence for all predictions
# Format:
obj_conf = predictions[:, 4]
class_prob = predictions[:, 6]
combined_conf = obj_conf * class_prob
# Filter by confidence threshold
valid_mask = combined_conf > self.conf_thres
valid_predictions = predictions
# Limit the number of predictions to process for performance
max_predictions = min(100, len(valid_predictions))
for i in range(max_predictions):
pred = valid_predictions
# Extract values by index (correct way)
x_center_norm = pred
y_center_norm = pred
width_norm = pred
height_norm = pred
obj_conf_val = pred
class_id = int(pred)# class_id is already an integer index
class_prob_val = pred
# Combined confidence
conf = obj_conf_val * class_prob_val
# Convert to pixel coordinates
if hasattr(self, 'scale_x'):
x_center_px = x_center_norm * self.input_width * self.scale_x
y_center_px = y_center_norm * self.input_height * self.scale_y
width_px = width_norm * self.input_width * self.scale_x
height_px = height_norm * self.input_height * self.scale_y
else:
# Test mode scaling
scale_factor = frame_shape / self.input_width
x_center_px = x_center_norm * self.input_width * scale_factor
y_center_px = y_center_norm * self.input_height * scale_factor
width_px = width_norm * self.input_width * scale_factor
height_px = height_norm * self.input_height * scale_factor
# Calculate bounding box
x1 = int(x_center_px - width_px / 2)
y1 = int(y_center_px - height_px / 2)
x2 = int(x_center_px + width_px / 2)
y2 = int(y_center_px + height_px / 2)
# Clamp to image boundaries
x1 = max(0, x1)
y1 = max(0, y1)
x2 = min(frame_shape, x2)
y2 = min(frame_shape, y2)
if x2 <= x1 or y2 <= y1:
continue
if class_id < len(self.labels):
class_name = self.labels
else:
class_name = f"class_{class_id}"
detections.append({
'bbox': ,
'confidence': float(conf),
'class_id': class_id,
'class_name': class_name
})
# Apply simple Non-Maximum Suppression to remove overlapping boxes
if len(detections) > 1:
detections = self.simple_nms(detections)
return detections
def simple_nms(self, detections, iou_threshold=0.45):
"""Simple Non-Maximum Suppression"""
if len(detections) == 0:
return []
# Sort by confidence (highest first)
sorted_detections = sorted(detections, key=lambda x: x['confidence'], reverse=True)
keep = []
while sorted_detections:
# Take the detection with highest confidence
current = sorted_detections.pop(0)
keep.append(current)
if not sorted_detections:
break
# Calculate IoU with remaining detections
remaining = []
current_bbox = np.array(current['bbox'])
current_area = (current_bbox - current_bbox) * (current_bbox - current_bbox)
for det in sorted_detections:
det_bbox = np.array(det['bbox'])
det_area = (det_bbox - det_bbox) * (det_bbox - det_bbox)
# Calculate intersection
x1 = max(current_bbox, det_bbox)
y1 = max(current_bbox, det_bbox)
x2 = min(current_bbox, det_bbox)
y2 = min(current_bbox, det_bbox)
if x2 <= x1 or y2 <= y1:
# No overlap
remaining.append(det)
continue
intersection = (x2 - x1) * (y2 - y1)
union = current_area + det_area - intersection
iou = intersection / union if union > 0 else 0
# Keep if IoU is below threshold
if iou < iou_threshold:
remaining.append(det)
sorted_detections = remaining
return keep
def draw_results(self, frame, detections, stats):
"""Draw results on frame"""
# Draw detections
for det in detections:
x1, y1, x2, y2 = det['bbox']
# Color coding: green for person, red for fall
if det['class_name'] == 'person':
color = (0, 255, 0)# Green
elif det['class_name'] == 'fall':
color = (0, 0, 255)# Red
else:
color = (255, 255, 0)# Yellow for other classes
# Draw bounding box
cv2.rectangle(frame, (x1, y1), (x2, y2), color, 2)
# Draw label with confidence
label = f"{det['class_name']}: {det['confidence']:.2f}"
# Calculate text size for background
font = cv2.FONT_HERSHEY_SIMPLEX
font_scale = 0.5
thickness = 2
(text_width, text_height), baseline = cv2.getTextSize(label, font, font_scale, thickness)
# Draw text background
cv2.rectangle(frame, (x1, y1 - text_height - 5),
(x1 + text_width, y1), color, -1)# -1 means filled
# Draw text
cv2.putText(frame, label, (x1, y1 - 5),
font, font_scale, (255, 255, 255), thickness)
# Draw statistics
y_offset = 30
stats_text = [
f"FPS: {stats['FPS']}",
f"Inference: {stats['Inference']}",
f"Persons: {stats['Persons']}",
f"Falls: {stats['Falls']}",
f"Frame: {stats['Frame']}",
f"Alarm: {'ON' if self.alarm_status == 1 else 'OFF'}",
f"Cloud: {'ON' if self.cloud_server_url else 'OFF'}",
f"Stream: {'ON' if self.rtmp_url else 'OFF'}",
f"EtherCAT: {'ON' if self.ethercat_enable else 'OFF'}"
]
for i, text in enumerate(stats_text):
cv2.putText(frame, text, (10, y_offset + i * 25),
cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2)
# Draw alarm indicator
if self.alarm_status == 1:
cv2.putText(frame, "ALARM! FALL DETECTED!",
(frame.shape//2 - 150, 50),
cv2.FONT_HERSHEY_SIMPLEX, 1.0, (0, 0, 255), 3)
# Draw red border
cv2.rectangle(frame, (5, 5), (frame.shape-5, frame.shape-5), (0, 0, 255), 10)
return frame
def handle_alarm(self, fall_detected, detections, frame):
"""Handle fall alarm logic with cloud notification"""
# Update alarm counter
if fall_detected:
self.alarm_count = min(self.alarm_count + 1, self.alarm_threshold)
else:
self.alarm_count = max(self.alarm_count - 1, 0)
# Determine new alarm status
new_alarm_status = 1 if self.alarm_count >= self.alarm_threshold else 0
# Check if alarm status changed
alarm_changed = new_alarm_status != self.alarm_status
self.alarm_status = new_alarm_status
# If alarm just triggered, send to cloud
if alarm_changed and self.alarm_status == 1:
print(f"⚠️ALARM TRIGGERED! Sending to cloud...")
# Calculate confidence for fall detection
fall_confidences = for d in detections if d['class_name'] == 'fall']
max_confidence = max(fall_confidences) if fall_confidences else 0.0
# Prepare cloud alert data
cloud_alert = {
'device_id': self.device_id,
'event_type': 'fall_detected',
'timestamp': datetime.now().isoformat(),
'fall_count': len( == 'fall']),
'person_count': len( == 'person']),
'confidence': max_confidence,
'location': self.location,
'message': f'Fall detected at {self.location}',
'alarm_status': self.alarm_status,
'camera_resolution': f'{frame.shape}x{frame.shape}'
}
# Send to cloud server
success = self.send_to_cloud(cloud_alert)
if success:
# Save frame with fall
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"fall_{timestamp}.jpg"
save_path = os.path.join(self.save_dir, filename)
cv2.imwrite(save_path, frame)
print(f"⚠️Fall frame saved: {filename}")
# Update EtherCAT output if changed
if alarm_changed and self.ethercat_enable:
success = self.set_ethercat_alarm(self.alarm_status)
if success:
print(f"EtherCAT alarm set to: {self.alarm_status}")
# If alarm just cleared, send recovery notification
if alarm_changed and self.alarm_status == 0 and self.last_alarm_status == 1:
print(f"✓ Alarm cleared")
if self.cloud_server_url:
recovery_alert = {
'device_id': self.device_id,
'event_type': 'alarm_cleared',
'timestamp': datetime.now().isoformat(),
'message': f'Fall alarm cleared at {self.location}',
'location': self.location
}
self.send_to_cloud(recovery_alert)
# Update last alarm status
self.last_alarm_status = self.alarm_status
def run(self):
"""Main loop"""
print("\n" + "="*60)
print("Industrial Fall Detection - IMX93 with Cloud Integration")
print(f"Device ID: {self.device_id}")
print(f"Location: {self.location}")
print(f"Cloud Server: {self.cloud_server_url if self.cloud_server_url else 'Disabled'}")
print(f"RTMP Stream: {self.rtmp_url if self.rtmp_url else 'Disabled'}")
print(f"EtherCAT Alarm: {'Enabled' if self.ethercat_enable else 'Disabled'}")
print("="*60)
print("Press 'q' or ESC to exit")
print("="*60)
stats_interval = 2.0
last_stats_print = time.time()
try:
while self.running:
# Get frame
frame = self.get_frame()
# Preprocess
original_frame, input_data = self.preprocess(frame)
# Inference
output, inference_time = self.inference(input_data)
# Post-process
detections = self.postprocess(output, frame.shape)
# Count falls and persons
fall_count = sum(1 for d in detections if d['class_name'] == 'fall')
fall_detected = fall_count > 0
# Handle alarm logic with cloud notification
self.handle_alarm(fall_detected, detections, original_frame)
# Update statistics
self.frame_count += 1
self.total_inference_time += inference_time
# Calculate FPS
if inference_time > 0:
fps = 1000 / inference_time
self.last_fps = fps
else:
fps = 0
self.last_fps = 0
# Count detections
person_count = sum(1 for d in detections if d['class_name'] == 'person')
# Create stats
stats = {
'FPS': f'{fps:.1f}',
'Inference': f'{inference_time:.1f}ms',
'Persons': person_count,
'Falls': fall_count,
'Frame': self.frame_count
}
# Draw results
frame_with_results = self.draw_results(original_frame.copy(), detections, stats)
# Push to streaming queue
if self.rtmp_url:
self.push_to_stream(frame_with_results)
# Display (if not in no-display mode)
if hasattr(self, 'show_display') and self.show_display:
cv2.imshow('Fall Detection', frame_with_results)
key = cv2.waitKey(1) & 0xFF
if key == ord('q') or key == 27:
break
# Print statistics periodically
current_time = time.time()
if current_time - last_stats_print >= stats_interval:
avg_inference = self.total_inference_time / self.frame_count if self.frame_count > 0 else 0
cloud_info = f"Cloud: {self.cloud_alert_count}" if self.cloud_server_url else ""
print(f"[{datetime.now().strftime('%H:%M:%S')}] "
f"FPS: {fps:.1f}, "
f"Inference: {inference_time:.1f}ms, "
f"Persons: {person_count}, Falls: {fall_count}, "
f"Alarm: {'ON' if self.alarm_status == 1 else 'OFF'} "
f"{cloud_info}")
# Reset counters for next interval
self.frame_count = 0
self.total_inference_time = 0
last_stats_print = current_time
# Small delay
time.sleep(0.01)
except KeyboardInterrupt:
print("\nExiting...")
finally:
# Cleanup
self.cleanup()
# Print final stats
print("\n" + "="*60)
print("Session ended")
if self.cloud_server_url:
print(f"Cloud alerts sent: {self.cloud_alert_count}")
print("="*60)
def cleanup(self):
"""Cleanup resources"""
print("\nCleaning up resources...")
# Stop camera
if self.cap:
self.cap.release()
# Stop FFmpeg stream
if self.ffmpeg_process:
try:
self.ffmpeg_process.stdin.close()
self.ffmpeg_process.wait(timeout=2)
print("FFmpeg stream closed")
except:
self.ffmpeg_process.terminate()
# Reset EtherCAT alarm
if self.ethercat_enable and self.alarm_status == 1:
print("Resetting EtherCAT alarm to OFF")
self.set_ethercat_alarm(0)
# Send final heartbeat
if self.cloud_server_url:
final_heartbeat = {
'device_id': self.device_id,
'event_type': 'shutdown',
'timestamp': datetime.now().isoformat(),
'message': f'Device {self.device_id} shutting down',
'location': self.location,
'cloud_alerts_sent': self.cloud_alert_count
}
try:
self.send_to_cloud(final_heartbeat)
except:
pass
# Close OpenCV windows
cv2.destroyAllWindows()
print("Cleanup completed")
def main():
parser = argparse.ArgumentParser(description='Industrial Fall Detection with Cloud Integration')
parser.add_argument('-m', '--model', required=True, help='TFLite model file')
parser.add_argument('-l', '--labels', default='labels.txt', help='Label file')
parser.add_argument('-d', '--delegate', default='/usr/lib/libethosu_delegate.so',
help='Ethos-U delegate path')
parser.add_argument('-c', '--camera', default='/dev/video0',
help='Camera device (e.g., /dev/video0)')
parser.add_argument('--rtmp', help='RTMP stream URL (e.g., rtmp://server/live/stream)')
parser.add_argument('--cloud', help='Cloud server URL (e.g., http://115.190.220.148:8000)')
parser.add_argument('--device-id', default='imx93_camera_001',
help='Device identifier for cloud server')
parser.add_argument('--location', default='Industrial Area A',
help='Location description for alerts')
parser.add_argument('--ethercat', action='store_true',
help='Enable EtherCAT alarm output')
parser.add_argument('--conf', type=float, default=0.25,
help='Confidence threshold')
parser.add_argument('--no-display', action='store_true',
help='Do not display window')
args = parser.parse_args()
# Check files
if not os.path.exists(args.model):
print(f"ERROR: Model file not found: {args.model}")
sys.exit(1)
if not os.path.exists(args.labels):
print("Creating default labels file...")
with open(args.labels, 'w') as f:
f.write("person\n")
f.write("fall\n")
# Check delegate
if args.delegate and not os.path.exists(args.delegate):
print(f"WARNING: Delegate not found: {args.delegate}")
args.delegate = None
# Check FFmpeg
if args.rtmp:
try:
subprocess.run(['which', 'ffmpeg'], capture_output=True, check=True)
except:
print("WARNING: ffmpeg not found, streaming disabled")
args.rtmp = None
try:
detector = SimpleCameraInference(
model_file=args.model,
label_file=args.labels,
conf_thres=args.conf,
delegate_path=args.delegate,
camera_device=args.camera,
rtmp_url=args.rtmp,
ethercat_enable=args.ethercat,
cloud_server_url=args.cloud,
device_id=args.device_id,
location=args.location
)
# Add display flag
detector.show_display = not args.no_display
detector.run()
except Exception as e:
print(f"ERROR: {e}")
import traceback
traceback.print_exc()
if __name__ == '__main__':
main()
```
运行
```
python3 camera_simple_imx93_with_cloud_and_streaming.py \
--model vela_models/best_int8_quantized_vela.tflite \
--labels labels.txt \
--cloud http://115.190.220.148:8000 \
--device-id "imx93_camera_01" \
--location "Assembly Line Station 3" \
--rtmprtmp://127.0.0.1/live/test --ethercat
```
#### 云服务器端代码
服务器
```
#!/usr/bin/env python3
"""
Industrial Fall Detection Cloud Server - English Version
No Chinese characters to avoid encoding issues
"""
import json
import os
import time
import logging
from datetime import datetime
from http.server import HTTPServer, BaseHTTPRequestHandler
# Setup logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class AlertServer(BaseHTTPRequestHandler):
"""Simple HTTP server for fall detection alerts"""
def do_GET(self):
"""Handle GET requests"""
if self.path == '/':
# Return HTML dashboard
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
html = self.generate_dashboard()
self.wfile.write(html.encode('utf-8'))
elif self.path == '/api/health':
# Health check endpoint
self.send_response(200)
self.send_header('Content-type', 'application/json')
self.end_headers()
response = {
'status': 'ok',
'time': datetime.now().isoformat(),
'service': 'fall-detection-server'
}
self.wfile.write(json.dumps(response).encode('utf-8'))
elif self.path == '/api/alerts':
# Get alert list
self.send_response(200)
self.send_header('Content-type', 'application/json')
self.end_headers()
alerts = self.get_alerts()
self.wfile.write(json.dumps(alerts, ensure_ascii=False).encode('utf-8'))
else:
self.send_response(404)
self.end_headers()
def do_POST(self):
"""Handle POST requests"""
if self.path == '/api/alerts':
try:
content_length = int(self.headers['Content-Length'])
post_data = self.rfile.read(content_length)
# Parse JSON data
alert_data = json.loads(post_data.decode('utf-8'))
# Save alert
self.save_alert(alert_data)
# Log to console
logger.info(f"Alert received: {alert_data.get('device_id', 'unknown')}")
# Return success
self.send_response(200)
self.send_header('Content-type', 'application/json')
self.end_headers()
response = {
'status': 'success',
'message': 'Alert received',
'timestamp': datetime.now().isoformat()
}
self.wfile.write(json.dumps(response).encode('utf-8'))
except Exception as e:
logger.error(f"Error processing alert: {e}")
self.send_response(500)
self.end_headers()
else:
self.send_response(404)
self.end_headers()
def save_alert(self, alert_data):
"""Save alert to file"""
try:
# Create data directory
os.makedirs('data', exist_ok=True)
# Generate filename
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
device_id = alert_data.get('device_id', 'unknown').replace('/', '_')
filename = f"data/alert_{device_id}_{timestamp}.json"
# Add server timestamp
alert_data['server_time'] = datetime.now().isoformat()
alert_data['server_timestamp'] = int(time.time())
# Save to file
with open(filename, 'w', encoding='utf-8') as f:
json.dump(alert_data, f, indent=2, ensure_ascii=False)
# Also append to log file
log_file = 'data/alerts.log'
with open(log_file, 'a', encoding='utf-8') as f:
f.write(f"{alert_data['server_time']} | {alert_data.get('device_id')} | {alert_data.get('event_type')}\n")
return True
except Exception as e:
logger.error(f"Error saving alert: {e}")
return False
def get_alerts(self):
"""Get recent alerts"""
alerts = []
try:
if os.path.exists('data'):
for filename in os.listdir('data'):
if filename.startswith('alert_') and filename.endswith('.json'):
filepath = os.path.join('data', filename)
try:
with open(filepath, 'r', encoding='utf-8') as f:
alert = json.load(f)
alerts.append(alert)
except:
continue
# Sort by timestamp (newest first)
alerts.sort(key=lambda x: x.get('server_timestamp', 0), reverse=True)
return alerts[:50]# Return last 50 alerts
except Exception as e:
logger.error(f"Error getting alerts: {e}")
return []
def generate_dashboard(self):
"""Generate HTML dashboard"""
alerts = self.get_alerts()
html = f"""<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Fall Detection Monitor</title>
<style>
* {{ margin: 0; padding: 0; box-sizing: border-box; }}
body {{ font-family: Arial, sans-serif; background: #f5f5f5; color: #333; }}
.container {{ max-width: 1200px; margin: 0 auto; padding: 20px; }}
.header {{ background: #2c3e50; color: white; padding: 2rem; border-radius: 10px; margin-bottom: 2rem; }}
.stats {{ display: grid; grid-template-columns: repeat(auto-fit, minmax(200px, 1fr)); gap: 1rem; margin-bottom: 2rem; }}
.stat-card {{ background: white; padding: 1.5rem; border-radius: 10px; box-shadow: 0 2px 10px rgba(0,0,0,0.1); text-align: center; }}
.stat-card h3 {{ font-size: 2rem; color: #3498db; margin-bottom: 0.5rem; }}
.alerts {{ background: white; border-radius: 10px; padding: 1.5rem; box-shadow: 0 2px 10px rgba(0,0,0,0.1); }}
.alert-item {{ border-bottom: 1px solid #eee; padding: 1rem 0; }}
.alert-item:last-child {{ border-bottom: none; }}
.alert-time {{ color: #666; font-size: 0.9rem; }}
.alert-device {{ color: #333; font-weight: bold; }}
.alert-type.fall {{ color: #e74c3c; }}
.alert-type.heartbeat {{ color: #2ecc71; }}
.alert-details {{ margin-top: 0.5rem; color: #555; }}
.last-update {{ text-align: center; margin-top: 1rem; color: #999; }}
</style>
<script>
// Auto refresh every 10 seconds
setTimeout(function() {{ location.reload(); }}, 10000);
// Send test alert
function testAlert() {{
fetch('/api/alerts', {{
method: 'POST',
headers: {{'Content-Type': 'application/json'}},
body: JSON.stringify({{
device_id: 'test_device_001',
timestamp: new Date().toISOString(),
event_type: 'fall_detected',
fall_count: 1,
person_count: 1,
confidence: 0.85,
location: 'Test Area',
message: 'Test alert for demonstration'
}})
}})
.then(response => response.json())
.then(data => {{
alert('Test alert sent: ' + data.message);
setTimeout(() => location.reload(), 500);
}});
}}
</script>
</head>
<body>
<div class="container">
<div class="header">
<h1>Fall Detection Monitoring System</h1>
<p>Real-time monitoring of fall detection devices</p>
</div>
<div class="stats">
<div class="stat-card">
<h3>{len(alerts)}</h3>
<p>Total Alerts</p>
</div>
<div class="stat-card">
<h3>{sum(1 for a in alerts if a.get('event_type') == 'fall_detected')}</h3>
<p>Fall Events</p>
</div>
<div class="stat-card">
<h3>{len(set(a.get('device_id', '') for a in alerts))}</h3>
<p>Active Devices</p>
</div>
<div class="stat-card">
<h3>Online</h3>
<p>System Status</p>
</div>
</div>
<div style="text-align: center; margin-bottom: 20px;">
<button onclick="testAlert()" style="padding: 10px 20px; background: #3498db; color: white; border: none; border-radius: 5px; cursor: pointer; margin-right: 10px;">
Send Test Alert
</button>
<button onclick="location.reload()" style="padding: 10px 20px; background: #27ae60; color: white; border: none; border-radius: 5px; cursor: pointer;">
Refresh Data
</button>
</div>
<div class="alerts">
<h2 style="margin-bottom: 1rem;">Recent Alert Records</h2>
{"".join() if alerts else '<p style="color: #999; text-align: center;">No alerts yet</p>'}
</div>
<div class="last-update">
<p>Last update: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} | System running normally</p>
</div>
</div>
</body>
</html>"""
return html
def generate_alert_item(self, alert):
"""Generate HTML for a single alert item"""
event_type = alert.get('event_type', 'unknown')
timestamp = alert.get('timestamp', alert.get('server_time', 'Unknown time'))
# Format time
if 'T' in timestamp:
time_display = timestamp.replace('T', ' ').split('.')
else:
time_display = timestamp
# Alert styling
if event_type == 'fall_detected':
alert_class = "fall"
alert_icon = "⚠️"
alert_color = "#e74c3c"
else:
alert_class = "heartbeat"
alert_icon = "💚"
alert_color = "#2ecc71"
return f"""
<div class="alert-item" style="border-left: 4px solid {alert_color}; padding-left: 1rem;">
<div style="display: flex; justify-content: space-between; align-items: center;">
<div>
<span class="alert-time">{time_display}</span>
<span class="alert-device">{alert.get('device_id', 'Unknown device')}</span>
</div>
<span class="alert-type {alert_class}">{alert_icon} {event_type}</span>
</div>
<div class="alert-details">
<strong>Location:</strong> {alert.get('location', 'Unknown')} |
<strong>Falls:</strong> {alert.get('fall_count', 0)} |
<strong>Persons:</strong> {alert.get('person_count', 0)} |
<strong>Confidence:</strong> {alert.get('confidence', 0):.2f}
{f'<br><strong>Message:</strong> {alert.get("message", "")}' if alert.get("message") else ""}
</div>
</div>
"""
def log_message(self, format, *args):
"""Override logging to use our logger"""
logger.info("%s - %s", self.address_string(), format % args)
def run_server(port=8000):
"""Run the HTTP server"""
server_address = ('0.0.0.0', port)
httpd = HTTPServer(server_address, AlertServer)
print("=" * 60)
print("Fall Detection Cloud Server - English Version")
print(f"Start time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print(f"Local: http://127.0.0.1:{port}")
print(f"Network: http://192.168.27.59:{port}")
print(f"Public: http://115.190.220.148:{port}")
print("=" * 60)
print("Endpoints:")
print(f"GET/ - Web dashboard")
print(f"GET/api/health- Health check")
print(f"GET/api/alerts- Get alerts")
print(f"POST /api/alerts- Send alert")
print("=" * 60)
print("Press Ctrl+C to stop")
print("=" * 60)
try:
httpd.serve_forever()
except KeyboardInterrupt:
print("\nServer stopped")
except Exception as e:
logger.error(f"Server error: {e}")
if __name__ == '__main__':
run_server(8000)
```
启动脚本
```
#!/bin/bash
echo "Starting Fall Detection Server..."
echo "=================================="
# Go to script directory
cd "$(dirname "$0")"
# Check Python
if ! command -v python3 &> /dev/null; then
echo "Error: Python3 not found"
echo "Please install: sudo apt install python3"
exit 1
fi
# Kill any existing process on port 8000
echo "Checking port 8000..."
if command -v lsof &> /dev/null; then
lsof -ti:8000 | xargs kill -9 2>/dev/null || true
elif command -v fuser &> /dev/null; then
fuser -k 8000/tcp 2>/dev/null || true
fi
# Create data directory
mkdir -p data
# Get IP address
IP=$(hostname -I | awk '{print $1}' 2>/dev/null)
if [ -z "$IP" ]; then
IP="127.0.0.1"
fi
echo ""
echo "Server starting on:"
echo "Local: http://localhost:8000"
echo "Network:http://$IP:8000"
echo "Public: http://115.190.220.148:8000"
echo ""
echo "Press Ctrl+C to stop"
echo ""
# Start server
python3 cloud_server_en.py
```
监控脚本
```
#!/bin/bash
echo "Server Status Check"
echo "==================="
echo ""
# Check process
echo "1. Check Server Process:"
ps aux | grep -E "python.*cloud_server" | grep -v grep
if [ $? -eq 0 ]; then
echo "✓ Server is running"
else
echo "✗ Server is not running"
fi
echo ""
# Check port
echo "2. Check Port 8000:"
if command -v ss &> /dev/null; then
ss -tlnp | grep :8000
elif command -v netstat &> /dev/null; then
netstat -tlnp | grep :8000
fi
if [ $? -eq 0 ]; then
echo "✓ Port 8000 is listening"
else
echo "✗ Port 8000 is not listening"
fi
echo ""
# Test API
echo "3. Test API Connection:"
curl -s http://localhost:8000/api/health 2>/dev/null | grep -q status
if [ $? -eq 0 ]; then
echo "✓ API is working"
curl -s http://localhost:8000/api/health | python3 -m json.tool 2>/dev/null
else
echo "✗ API is not responding"
fi
echo ""
# Check data directory
echo "4. Check Data Directory:"
if [ -d "data" ]; then
echo "✓ Data directory exists"
echo "Alert files: $(ls -la data/alert_*.json 2>/dev/null | wc -l)"
echo "Log file: $(wc -l < data/alerts.log 2>/dev/null || echo 0) lines"
else
echo "✗ Data directory does not exist"
fi
echo ""
# URLs
echo "5. Access URLs:"
echo "Local: http://localhost:8000"
echo "Public: http://115.190.220.148:8000"
echo ""
```
部署步骤
```
#Set permissions
chmod +x start_simple.sh
chmod +x check_server.sh
chmod +x cloud_server_en.py
#Start the server
./start_simple.sh
```
## 六，视频演示
推理结果：
!(https://www.eefocus.com/forum/data/attachment/forum/202602/02/043737ob1ywuvuz68o6vgp.jpg)
ZLmediakit流媒体服务器：
!(https://www.eefocus.com/forum/data/attachment/forum/202602/02/043808gr2zhpefeoeuttyr.png)
云平台：
!(https://www.eefocus.com/forum/data/attachment/forum/202602/02/043937fqkmkrqpycsbqqq7.png)
实时推理演示：
!(https://www.eefocus.com/forum/data/attachment/forum/202602/03/104116qhbdvdhenhtdattb.png)
<iframe src="//player.bilibili.com/player.html?isOutside=true&aid=115997168110106&bvid=BV1H768B9Eao&cid=35760508163&p=1" scrolling="no" border="0" frameborder="no" framespacing="0" allowfullscreen="true"></iframe>
源码：https://gitee.com/cool-ditch-repair/frdm-imx93-ether-cat.git
过程贴：(https://www.nxpic.org.cn/module/forum/thread-810725-1-1.html)
(https://www.nxpic.org.cn/module/forum/thread-810726-1-1.html)
(https://www.nxpic.org.cn/module/forum/thread-810727-1-1.html)
页:
[1]