import socket
import threading

HOST = '127.0.0.1'  # 本机地址，如果局域网访问可改为 '0.0.0.0'
PORT = 5000         # 端口号，可自定义

def handle_client(conn, addr):
    print(f"[INFO] 连接来自 {addr}")
    try:
        while True:
            data = conn.recv(1024)  # 接收前端发送的数据
            if not data:
                print(f"[INFO] {addr} 已断开连接")
                break

            message = data.decode('utf-8')
            print(f"[RECV] {addr}: {message}")

            # 简单处理：把消息改成大写作为示例
            response = f"Server received: {message.upper()}"
            conn.sendall(response.encode('utf-8'))
    except Exception as e:
        print(f"[ERROR] {e}")
    finally:
        conn.close()


def start_server():
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.bind((HOST, PORT))
    server.listen()
    print(f"[INFO] 服务器启动，监听 {HOST}:{PORT}")

    while True:
        conn, addr = server.accept()
        client_thread = threading.Thread(target=handle_client, args=(conn, addr))
        client_thread.start()


if __name__ == "__main__":
    start_server()

# pip install websockets
import asyncio
import websockets

async def handler(websocket):
    async for message in websocket:
        print(f"[RECV] {message}")
        await websocket.send(f"Server received: {message.upper()}")

async def main():
    async with websockets.serve(handler, "127.0.0.1", 5000):
        print("[INFO] WebSocket服务器启动 127.0.0.1:5000")
        await asyncio.Future()  # 永久运行

if __name__ == "__main__":
    asyncio.run(main())

# 文件名: server.py
# 安装依赖: pip install websockets
import asyncio
import websockets

connected_clients = set()

async def handler(websocket):
    # 新客户端加入
    connected_clients.add(websocket)
    addr = websocket.remote_address
    print(f"[INFO] 客户端连接: {addr}")
    try:
        async for message in websocket:
            print(f"[RECV] {addr}: {message}")
            
            # 处理消息（示例：命令解析，可自定义逻辑）
            if message.lower() == "hello":
                response = "Hi! This is server."
            else:
                response = f"Server received: {message.upper()}"
            
            # 给当前客户端回传
            await websocket.send(response)

            # 可选：广播给所有其他客户端
            # broadcast = f"[广播] {addr}: {message}"
            # await asyncio.gather(*[c.send(broadcast) for c in connected_clients if c != websocket])

    except websockets.ConnectionClosed:
        print(f"[INFO] 客户端断开: {addr}")
    finally:
        connected_clients.remove(websocket)

async def main():
    server = await websockets.serve(handler, "0.0.0.0", 5000)
    print("[INFO] WebSocket服务器启动 0.0.0.0:5000")
    await server.wait_closed()

if __name__ == "__main__":
    asyncio.run(main())

# server.py
# pip install websockets
import asyncio
import websockets
import json
from datetime import datetime

clients = set()


# =========================
# 工具函数
# =========================
def log(msg):
    print(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}")


async def send_to_all(message: str):
    if clients:
        await asyncio.gather(*[c.send(message) for c in clients])


async def send_to_one(target_id, message: str):
    for c in clients:
        if id(c) == target_id:
            await c.send(message)
            return True
    return False


# =========================
# 命令处理核心
# =========================
async def handle_command(websocket, message: str):
    """
    支持命令格式：
    1. 普通文本
    2. JSON命令
    """

    addr = websocket.remote_address

    # -------- JSON命令模式 --------
    try:
        data = json.loads(message)
        if isinstance(data, dict):
            cmd = data.get("cmd")

            # 广播
            if cmd == "broadcast":
                msg = data.get("msg", "")
                await send_to_all(f"[BROADCAST] {msg}")
                return

            # 点对点
            if cmd == "send":
                target = data.get("target")
                msg = data.get("msg", "")
                ok = await send_to_one(target, f"[PRIVATE] {msg}")
                await websocket.send("OK" if ok else "TARGET NOT FOUND")
                return

            # 获取客户端列表
            if cmd == "list":
                await websocket.send(
                    json.dumps([id(c) for c in clients])
                )
                return

    except json.JSONDecodeError:
        pass  # 不是 JSON 就按普通命令处理

    # -------- 文本命令模式 --------
    msg = message.strip()

    if msg == "ping":
        await websocket.send("pong")
    elif msg == "time":
        await websocket.send(datetime.now().isoformat())
    elif msg.startswith("upper "):
        await websocket.send(msg[6:].upper())
    elif msg == "clients":
        await websocket.send(str(len(clients)))
    else:
        await websocket.send(f"[ECHO] {msg}")


# =========================
# WebSocket连接处理
# =========================
async def handler(websocket):
    clients.add(websocket)
    cid = id(websocket)
    log(f"Client connected: {cid}")

    try:
        async for message in websocket:
            log(f"RECV({cid}): {message}")
            await handle_command(websocket, message)

    except websockets.ConnectionClosed:
        pass
    finally:
        clients.remove(websocket)
        log(f"Client disconnected: {cid}")


# =========================
# 后台控制台线程
# =========================
async def console_loop():
    """
    服务器控制台命令（真正后台控制）
    """
    while True:
        cmd = await asyncio.to_thread(input, "SERVER> ")

        if cmd == "help":
            print("""
命令列表:
  clients                 查看连接数
  broadcast <msg>         广播消息
  list                    输出客户端ID列表
  exit                    关闭服务器
            """)

        elif cmd == "clients":
            print(f"在线客户端: {len(clients)}")

        elif cmd.startswith("broadcast "):
            msg = cmd[len("broadcast "):]
            await send_to_all(f"[SERVER BROADCAST] {msg}")

        elif cmd == "list":
            print([id(c) for c in clients])

        elif cmd == "exit":
            print("Shutting down...")
            for c in clients:
                await c.close()
            asyncio.get_event_loop().stop()
            break

        else:
            print("未知命令，输入 help")


# =========================
# 启动
# =========================
async def main():
    server = await websockets.serve(handler, "0.0.0.0", 5000)
    log("WebSocket Server started on ws://0.0.0.0:5000")

    await asyncio.gather(
        server.wait_closed(),
        console_loop()
    )


if __name__ == "__main__":
    asyncio.run(main())

import threading
import queue
import time
import random
from dataclasses import dataclass
from typing import Optional


# =========================================================
# 1️ 事件定义层（Event Model）
# =========================================================

@dataclass
class FrameEvent:
    camera_id: str
    frame_id: int
    data: int  # 模拟图像数据（可替换 numpy / cv2 / GPU buffer）


@dataclass
class InferenceEvent:
    camera_id: str
    frame_id: int
    score: float
    label: str


# =========================================================
# 2️ 全局事件队列（Event Bus）
# =========================================================

class EventBus:
    def __init__(self, maxsize=100):
        self.frame_queue = queue.Queue(maxsize=maxsize)
        self.result_queue = queue.Queue(maxsize=maxsize)


bus = EventBus()


# =========================================================
# 3️ Camera采集线程（Producer）
# =========================================================

class CameraThread(threading.Thread):
    def __init__(self, camera_id: str, fps: float = 10):
        super().__init__(daemon=True)
        self.camera_id = camera_id
        self.fps = fps
        self.running = True
        self.frame_id = 0

    def run(self):
        while self.running:
            frame = FrameEvent(
                camera_id=self.camera_id,
                frame_id=self.frame_id,
                data=random.randint(0, 255)
            )

            try:
                bus.frame_queue.put(frame, timeout=1)
                print(f"[CAMERA] {self.camera_id} -> frame {self.frame_id}")
                self.frame_id += 1
            except queue.Full:
                print("[CAMERA] frame queue full (drop frame)")

            time.sleep(1 / self.fps)

    def stop(self):
        self.running = False


# =========================================================
# 4️ AI推理线程（Consumer + Producer）
# =========================================================

class AIThread(threading.Thread):
    def __init__(self, worker_id: int = 0):
        super().__init__(daemon=True)
        self.worker_id = worker_id
        self.running = True

    def run(self):
        while self.running:
            frame: FrameEvent = bus.frame_queue.get()

            # ===== 模拟AI推理 =====
            time.sleep(0.05)

            score = random.random()
            label = "NG" if score > 0.7 else "OK"

            result = InferenceEvent(
                camera_id=frame.camera_id,
                frame_id=frame.frame_id,
                score=score,
                label=label
            )

            bus.result_queue.put(result)

            print(
                f"[AI-{self.worker_id}] "
                f"{frame.camera_id}:{frame.frame_id} -> {label} ({score:.2f})"
            )

    def stop(self):
        self.running = False


# =========================================================
# 5️ 预警 / 决策线程（Consumer）
# =========================================================

class AlarmThread(threading.Thread):
    def __init__(self):
        super().__init__(daemon=True)
        self.running = True

    def run(self):
        while self.running:
            result: InferenceEvent = bus.result_queue.get()

            if result.label == "NG":
                self.alarm(result)
            else:
                print(f"[OK] {result.camera_id}:{result.frame_id}")

    def alarm(self, result: InferenceEvent):
        print(
            "\n ALARM TRIGGERED\n"
            f"Camera : {result.camera_id}\n"
            f"Frame  : {result.frame_id}\n"
            f"Score  : {result.score:.2f}\n"
        )

    def stop(self):
        self.running = False


# =========================================================
# 6️ 系统启动器（Runtime）
# =========================================================

class VisionRuntime:
    def __init__(self):
        self.camera = CameraThread("cam_01", fps=5)
        self.ai_workers = [AIThread(i) for i in range(2)]
        self.alarm = AlarmThread()

    def start(self):
        print("[SYSTEM] Starting Vision Runtime...")

        self.camera.start()

        for w in self.ai_workers:
            w.start()

        self.alarm.start()

        print("[SYSTEM] Running...")

    def stop(self):
        print("[SYSTEM] Stopping...")

        self.camera.stop()
        for w in self.ai_workers:
            w.stop()
        self.alarm.stop()


# =========================================================
# 7️ main入口
# =========================================================

if __name__ == "__main__":
    runtime = VisionRuntime()
    runtime.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        runtime.stop()

CameraThread(cam_02)
CameraThread(cam_03)

AIThread(0)
AIThread(1)
AIThread(2)