python----/远程桌面测试.ipynb

In [ ]:

import cv2
import numpy as np
from mss import mss

class FrameDiffProcessor:
    def __init__(self):
        self.prev_frame = None
        self.threshold = 25  # 差异阈值（0-255）
        self.min_contour_area = 100  # 最小变化区域面积（像素）

    def _preprocess(self, frame):
        """预处理：转为灰度图并高斯模糊"""
        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        return cv2.GaussianBlur(gray, (21, 21), 0)

    def _get_diff_rects(self, current_frame):
        """计算差异区域边界框"""
        # 计算绝对差异并二值化
        diff = cv2.absdiff(self.prev_frame, current_frame)
        mean_diff = np.mean(diff)
        dynamic_threshold = max(15, min(mean_diff * 0.7, 50))
        _, thresh = cv2.threshold(diff, dynamic_threshold, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU  )
        
        # 查找轮廓
        contours, _2 = cv2.findContours(
            thresh, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE
        )
        
        # 过滤小面积区域
        rects = []
        for cnt in contours:
            if cv2.contourArea(cnt) > self.min_contour_area:
                x, y, w, h = cv2.boundingRect(cnt)
                rects.append((x, y, x + w, y + h))  # (x1,y1,x2,y2)
        cv2.imshow("Debug: Prev|Current|Diff|Thresh", debug_img)
        return rects

    def process(self, frame):
        """主处理流程"""
        processed = self._preprocess(frame)
        if self.prev_frame is None:
            self.prev_frame = processed
            return None  # 首帧不处理
        
        rects = self._get_diff_rects(processed)
        self.prev_frame = processed  # 更新前一帧
        
        return rects

In [17]:

sct = mss()
m1  = sct.monitors[1]

In [18]:

sct_img = sct.grab(m1)
frame = np.array(sct_img)

In [19]:

sct_img2 = sct.grab(m1)
frame2 = np.array(sct_img2)

In [20]:

diff_processor = FrameDiffProcessor()

In [21]:

diff_processor.process(frame)

In [22]:

rect = diff_processor.process(frame2)

In [24]:

img3= frame2.copy() 
for item in rect:
    x1, y1, x2, y2 = item
    cv2.rectangle(img3, (x1, y1), (x2, y2), (0, 255, 0), 2)
cv2.imshow("Result", img3)
cv2.waitKey(0)
cv2.destroyAllWindows()

In [25]:

cv2.imshow("Result", frame)
cv2.waitKey(0)
cv2.destroyAllWindows()
cv2.imshow("Result", frame2)
cv2.waitKey(0)
cv2.destroyAllWindows()

In [26]:

# async def send_screen(websocket):
#     sct = mss()
#     monitor = sct.monitors[1]
#     diff_processor = FrameDiffProcessor()
    
#     while True:
#         # 捕获原始帧
#         sct_img = sct.grab(monitor)
#         frame = np.array(sct_img)
        
#         # 计算差异区域
#         diff_rects = diff_processor.process(frame)
        
#         if diff_rects:
#             # 提取变化区域并压缩
#             payload = []
#             for (x1, y1, x2, y2) in diff_rects:
#                 roi = frame[y1:y2, x1:x2]  # 截取变化区域
#                 _, buffer = cv2.imencode('.jpg', roi, [cv2.IMWRITE_JPEG_QUALITY, 85])
#                 payload.append({
#                     'x': x1,
#                     'y': y1,
#                     'data': buffer.tobytes()
#                 })
            
#             # 序列化并发送
#             await websocket.send(json.dumps(payload))
#         else:
#             # 无变化时发送心跳包
#             await websocket.send("no_change")
        
#         await asyncio.sleep(0.05)  # 控制帧率