树莓派与Win11通信【一对一】（四）

news2025/11/1 1:20:09

树莓派与Win11通信【一对一】（四）

树莓派与Win11通信【一对一】（四）的代码优化版，

最近给代码添加了打开摄像头与否的验证，以及文件的保存，定时拍摄

1.Server端

import socket
import time
# from picamera2 import Picamera2
from datetime import datetime

from PIL import Image
from io import BytesIO
import numpy as np
from detect_by_cv import *
import socket
import cv2
import numpy as np
from detect_by_cv import Yolov5
from PIL import Image
from io import BytesIO


def main():
    global ret, cap
    yolo = Yolov5()
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.bind(('127.0.0.1', 8083))  # 使用服务器的IP地址
    server_socket.listen(1)
    print("服务器启动，等待连接...")

    while True:  # 主循环，用于接受新的连接
        client_socket, addr = server_socket.accept()
        print(f"接受到来自 {addr} 的连接")

        try:
            while True:
                # 每2秒启动一次摄像头进行拍照
                time.sleep(2)  # 等待2秒
                try:
                    cap = cv2.VideoCapture(0)  # 尝试打开摄像头
                    if not cap.isOpened():  # 检查摄像头是否成功打开
                        raise IOError("摄像头打开失败")

                    ret, frame = cap.read()
                except Exception as cap_error:
                    photo = cv2.imread('./Serverimg/fail/fail.jpg')
                    im = Image.fromarray(cv2.cvtColor(photo, cv2.COLOR_BGR2RGB))
                    buffer = BytesIO()
                    im.save(buffer, format='JPEG')
                    buffer = buffer.getvalue()
                    client_socket.sendall(len(buffer).to_bytes(4, 'big'))
                    client_socket.sendall(buffer)

                    # 发送最大最小值数据
                    max_min_str = f"fail,fail,fail,fail"

                    client_socket.sendall(len(max_min_str).to_bytes(4, 'big'))
                    client_socket.sendall(max_min_str.encode())
                finally:
                    cap.release()  # 确保摄像头资源被释放

                if ret:
                    box, scores, classid, _ = yolo.detect(frame)
                    resultlist = []
                    for i in range(len(classid)):
                        resultlist.append([classid[i], box[i], scores[i]])
                    for item in resultlist:
                        category, bbox, confidence = item
                        clslist = ["fire"]
                        # 边界框坐标
                        x, y, x2, y2 = bbox
                        # 绘制边界框和置信度文本
                        cv2.rectangle(frame, (x, y), (x2, y2), (0, 255, 0), 2)
                        label = f'{confidence:.2f},{clslist[category]}'
                        cv2.putText(frame, label, (x, y - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.9, (0, 255, 0), 2)
                    save_directory = "./Serverimg"
                    if not os.path.exists(save_directory):
                        os.makedirs(save_directory)

                    # 生成以当前时间戳为文件名的图片
                    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
                    filename = f"{timestamp}.jpg"
                    filepath = os.path.join(save_directory, filename)
                    cv2.imwrite(filepath, frame)

                    # 检查目录中的图片数量，如果达到50张，删除最早的图片
                    files = [os.path.join(save_directory, f) for f in os.listdir(save_directory) if f.endswith('.jpg')]
                    files.sort(key=os.path.getmtime)

                    if len(files) > 10:
                        os.remove(files[0])  # 删除最早的文件

                    photo = frame

                    im = Image.fromarray(photo)
                    if im.mode != 'RGB':
                        im = im.convert('RGB')
                    buffer = BytesIO()
                    im.save(buffer, format='JPEG')
                    buffer = buffer.getvalue()

                    max_pixel_value = np.max(photo)
                    min_pixel_value = np.min(photo)
                    max_min_str = f"{max_pixel_value},{min_pixel_value},{max_pixel_value},{min_pixel_value}"

                    # 发送图像数据长度和数据
                    client_socket.sendall(len(buffer).to_bytes(4, 'big'))
                    client_socket.sendall(buffer)

                    # 发送最大最小值数据
                    client_socket.sendall(len(max_min_str).to_bytes(4, 'big'))
                    client_socket.sendall(max_min_str.encode())

        except Exception as e:
            print(f"发生异常: {e}")
        finally:
            client_socket.close()
            print("等待新的连接...")

if __name__ == '__main__':
    main()