16：全球监控摄像头接入：RTSP协议与边缘计算实时流处理

import socket
import re

class RTSPClient:
    def __init__(self, server_ip, server_port=554):
        self.server_ip = server_ip
        self.server_port = server_port
        self.session_id = None
        self.rtp_port = 5000
        self.rtcp_port = 5001
    
    def connect(self):
        """连接到RTSP服务器"""
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.connect((self.server_ip, self.server_port))
    
    def send_request(self, method, url, headers=None):
        """发送RTSP请求"""
        if headers is None:
            headers = {}
        
        # 构建请求行
        request = f"{method} {url} RTSP/1.0\r\n"
        
        # 添加默认 headers
        headers.setdefault("CSeq", "1")
        if self.session_id:
            headers["Session"] = self.session_id
        
        # 添加 headers
        for key, value in headers.items():
            request += f"{key}: {value}\r\n"
        
        # 结束请求
        request += "\r\n"
        
        # 发送请求
        self.sock.sendall(request.encode())
        
        # 接收响应
        response = self.sock.recv(4096)
        return response.decode()
    
    def describe(self, url):
        """获取媒体流描述"""
        response = self.send_request("DESCRIBE", url, {
            "Accept": "application/sdp"
        })
        return response
    
    def setup(self, url):
        """建立媒体传输通道"""
        response = self.send_request("SETUP", url, {
            "Transport": f"RTP/AVP;unicast;client_port={self.rtp_port}-{self.rtcp_port}"
        })
        # 提取会话ID
        match = re.search(r"Session: (\w+)", response)
        if match:
            self.session_id = match.group(1)
        return response
    
    def play(self, url):
        """开始播放媒体流"""
        response = self.send_request("PLAY", url)
        return response
    
    def pause(self, url):
        """暂停播放"""
        response = self.send_request("PAUSE", url)
        return response
    
    def teardown(self, url):
        """关闭媒体会话"""
        response = self.send_request("TEARDOWN", url)
        return response
    
    def close(self):
        """关闭连接"""
        if hasattr(self, 'sock'):
            self.sock.close()

import cv2
import numpy as np
import threading
import time

class EdgeDevice:
    def __init__(self, camera_url, device_id):
        self.camera_url = camera_url
        self.device_id = device_id
        self.cap = None
        self.running = False
        self.thread = None
        self.frame = None
    
    def start(self):
        """启动边缘设备"""
        self.running = True
        self.thread = threading.Thread(target=self._process)
        self.thread.daemon = True
        self.thread.start()
    
    def _process(self):
        """处理视频流"""
        self.cap = cv2.VideoCapture(self.camera_url)
        if not self.cap.isOpened():
            print(f"Failed to open camera: {self.camera_url}")
            self.running = False
            return
        
        while self.running:
            ret, frame = self.cap.read()
            if not ret:
                print(f"Failed to read frame from camera: {self.camera_url}")
                break
            
            # 处理帧
            self.frame = frame
            
            # 进行人脸识别
            faces = self._detect_faces(frame)
            
            # 进行行为分析
            behavior = self._analyze_behavior(frame)
            
            # 发送结果到云端
            self._send_results(faces, behavior)
            
            # 控制帧率
            time.sleep(0.03)  # 约30fps
    
    def _detect_faces(self, frame):
        """检测人脸"""
        # 加载预训练的人脸检测器
        face_cascade = cv2.CascadeClassifier(cv2.data.haarcascades + 'haarcascade_frontalface_default.xml')
        
        # 转换为灰度图像
        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        
        # 检测人脸
        faces = face_cascade.detectMultiScale(gray, scaleFactor=1.1, minNeighbors=5, minSize=(30, 30))
        
        # 提取人脸区域
        face_regions = []
        for (x, y, w, h) in faces:
            face_regions.append(frame[y:y+h, x:x+w])
        
        return face_regions
    
    def _analyze_behavior(self, frame):
        """分析行为"""
        # 简单的行为分析示例
        # 实际应用中可以使用更复杂的算法
        return "normal"
    
    def _send_results(self, faces, behavior):
        """发送结果到云端"""
        # 实际应用中可以使用MQTT、WebSocket等协议
        print(f"Device {self.device_id}: Detected {len(faces)} faces, behavior: {behavior}")
    
    def stop(self):
        """停止边缘设备"""
        self.running = False
        if self.thread:
            self.thread.join()
        if self.cap:
            self.cap.release()

from pyspark.streaming import StreamingContext
from pyspark import SparkContext

class StreamProcessor:
    def __init__(self, master="local[2]", appName="VideoStreamProcessor"):
        self.sc = SparkContext(master, appName)
        self.ssc = StreamingContext(self.sc, 1)  # 1秒批处理
    
    def start(self, stream_source):
        """开始流处理"""
        # 从流源获取数据
        lines = self.ssc.socketTextStream(*stream_source)
        
        # 处理数据
        processed = lines.map(self._process_data)
        
        # 输出结果
        processed.pprint()
        
        # 启动流处理
        self.ssc.start()
        self.ssc.awaitTermination()
    
    def _process_data(self, data):
        """处理单条数据"""
        # 实际应用中可以进行更复杂的处理
        return f"Processed: {data}"
    
    def stop(self):
        """停止流处理"""
        self.ssc.stop()

import tensorflow as tf
from tensorflow.keras.applications import MobileNetV2
from tensorflow.keras.preprocessing import image
from tensorflow.keras.applications.mobilenet_v2 import preprocess_input, decode_predictions

class VideoAnalyzer:
    def __init__(self):
        # 加载预训练模型
        self.model = MobileNetV2(weights='imagenet')
    
    def analyze_frame(self, frame):
        """分析单帧图像"""
        # 调整图像大小
        img = cv2.resize(frame, (224, 224))
        
        # 预处理图像
        x = image.img_to_array(img)
        x = np.expand_dims(x, axis=0)
        x = preprocess_input(x)
        
        # 预测
        preds = self.model.predict(x)
        
        # 解码预测结果
        results = decode_predictions(preds, top=3)[0]
        
        return results
    
    def analyze_video(self, video_path):
        """分析视频"""
        cap = cv2.VideoCapture(video_path)
        results = []
        
        while cap.isOpened():
            ret, frame = cap.read()
            if not ret:
                break
            
            # 分析帧
            frame_results = self.analyze_frame(frame)
            results.append(frame_results)
        
        cap.release()
        return results

class CameraConnector:
    def __init__(self):
        self.protocols = {
            "rtsp": self._connect_rtsp,
            "http": self._connect_http,
            "onvif": self._connect_onvif
        }
    
    def connect(self, url):
        """连接到摄像头"""
        # 提取协议
        protocol = url.split("://")[0]
        
        # 调用相应的连接方法
        if protocol in self.protocols:
            return self.protocols[protocol](url)
        else:
            raise ValueError(f"Unsupported protocol: {protocol}")
    
    def _connect_rtsp(self, url):
        """连接RTSP摄像头"""
        cap = cv2.VideoCapture(url)
        if cap.isOpened():
            return cap
        else:
            raise Exception(f"Failed to connect to RTSP camera: {url}")
    
    def _connect_http(self, url):
        """连接HTTP摄像头"""
        cap = cv2.VideoCapture(url)
        if cap.isOpened():
            return cap
        else:
            raise Exception(f"Failed to connect to HTTP camera: {url}")
    
    def _connect_onvif(self, url):
        """连接ONVIF摄像头"""
        # 实际应用中需要使用ONVIF库
        raise NotImplementedError("ONVIF support not implemented yet")

class LoadBalancer:
    def __init__(self, edge_devices):
        self.edge_devices = edge_devices
        self.current_index = 0
    
    def get_device(self):
        """获取负载最低的边缘设备"""
        # 简单的轮询策略
        device = self.edge_devices[self.current_index]
        self.current_index = (self.current_index + 1) % len(self.edge_devices)
        return device
    
    def add_device(self, device):
        """添加边缘设备"""
        self.edge_devices.append(device)
    
    def remove_device(self, device):
        """移除边缘设备"""
        if device in self.edge_devices:
            self.edge_devices.remove(device)