事件驱动的工业视觉系统

# camera_sdk.py
import cv2
import time
from collections import deque

class CameraFrame:
    def __init__(self, frame, ts):
        self.frame = frame
        self.timestamp = ts


class CameraSDK:
    """
    工业级CameraSDK模拟：
    - 支持buffer
    - 支持时间戳
    - 支持回放基础数据
    """

    def __init__(self, src=0, buffer_size=300):
        self.cap = cv2.VideoCapture(src)
        self.buffer = deque(maxlen=buffer_size)

    def read(self):
        ret, frame = self.cap.read()
        if not ret:
            return None

        ts = time.time()
        f = CameraFrame(frame, ts)
        self.buffer.append(f)
        return f

    def get_prebuffer(self, seconds=5):
        now = time.time()
        return [f for f in self.buffer if now - f.timestamp <= seconds]

# algorithm_sdk.py
import cv2
import numpy as np

class Event:
    def __init__(self, event_type, score, frame):
        self.type = event_type
        self.score = score
        self.frame = frame


class AlgorithmSDK:
    """
    工业视觉算法层：
    - anomaly detection
    - motion detection (demo)
    """

    def __init__(self, threshold=25):
        self.threshold = threshold
        self.last_frame = None

    def analyze(self, camera_frame):
        frame = camera_frame.frame

        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)

        if self.last_frame is None:
            self.last_frame = gray
            return None

        diff = cv2.absdiff(gray, self.last_frame)
        score = np.mean(diff)

        self.last_frame = gray

        if score > self.threshold:
            return Event("motion_anomaly", score, frame)

        return None

# event_recorder.py
from collections import deque
import time

class EventRecorder:
    """
    事件驱动回溯系统：
    - pre-buffer 30s
    - post-buffer 30s
    """

    def __init__(self, pre_sec=30, post_sec=30):
        self.pre_sec = pre_sec
        self.post_sec = post_sec
        self.buffer = deque()
        self.active_events = []

    def push_frame(self, frame):
        self.buffer.append(frame)

        now = time.time()
        # 清理旧数据
        while self.buffer and now - self.buffer[0].timestamp > self.pre_sec:
            self.buffer.popleft()

    def trigger_event(self, event):
        print(f"[EVENT] {event.type}, score={event.score}")

        event_record = {
            "event": event,
            "prebuffer": list(self.buffer),
            "postbuffer": []
        }

        self.active_events.append(event_record)
        return event_record

    def append_post_frame(self, event_record, frame):
        event_record["postbuffer"].append(frame)

# webrtc_track.py
from aiortc import VideoStreamTrack
from av import VideoFrame
import numpy as np

class AITrackedVideo(VideoStreamTrack):
    def __init__(self, camera, algo, recorder):
        super().__init__()
        self.camera = camera
        self.algo = algo
        self.recorder = recorder

    async def recv(self):
        pts, time_base = await self.next_timestamp()

        cam_frame = self.camera.read()
        if cam_frame is None:
            img = np.zeros((480, 640, 3), dtype=np.uint8)
        else:
            img = cam_frame.frame

            # push into recorder
            self.recorder.push_frame(cam_frame)

            # run AI
            event = self.algo.analyze(cam_frame)

            if event:
                ev = self.recorder.trigger_event(event)
                print("Event captured with prebuffer:", len(ev["prebuffer"]))

        vf = VideoFrame.from_ndarray(img, format="bgr24")
        vf.pts = pts
        vf.time_base = time_base
        return vf

# server.py
import asyncio
import json

from aiohttp import web
from aiortc import RTCPeerConnection, RTCSessionDescription

from camera_sdk import CameraSDK
from algorithm_sdk import AlgorithmSDK
from event_recorder import EventRecorder
from webrtc_track import AITrackedVideo

pcs = set()

camera = CameraSDK(0)
algo = AlgorithmSDK()
recorder = EventRecorder()


async def offer(request):
    params = await request.json()

    pc = RTCPeerConnection()
    pcs.add(pc)

    pc.addTrack(AITrackedVideo(camera, algo, recorder))

    offer = RTCSessionDescription(sdp=params["sdp"], type=params["type"])
    await pc.setRemoteDescription(offer)

    answer = await pc.createAnswer()
    await pc.setLocalDescription(answer)

    return web.json_response({
        "sdp": pc.localDescription.sdp,
        "type": pc.localDescription.type
    })


# WebSocket: control + event stream
async def ws_handler(request):
    ws = web.WebSocketResponse()
    await ws.prepare(request)

    async for msg in ws:
        if msg.type == web.WSMsgType.TEXT:
            data = json.loads(msg.data)

            if data.get("cmd") == "get_status":
                await ws.send_str(json.dumps({
                    "buffer_size": len(camera.buffer),
                }))

    return ws


app = web.Application()
app.add_routes([
    web.post("/offer", offer),
    web.get("/ws", ws_handler)
])

web.run_app(app, port=8080)