Python基于WebSocket實現簡易屏幕共享工具
前面寫了兩個簡單的屏幕共享工具,不過那只是為了驗證通過截屏的方式是否可行,因為通常手動截屏的頻率很低,而對于視頻來說它的幀率要求就很高了,至少要一秒30幀率左右。所以,經過實際的截屏工具驗證,我了解了幾個Python截屏庫的特點和限制。例如,多數截屏庫都不支持很高的截屏速度,并且截屏是典型的 CPU 密集任務(我嘗試使用多線程截屏,發(fā)現速度更慢了,之后有時間我也會把這一點整理成文章發(fā)出來)。
所以,我的初始的想法其實是基于 WebSocket 來實現的?,F在,就讓我們對先前的代碼進行重構,采用 WebSocket 來傳輸圖片數據。不過我這里沒有使用到它的雙向傳輸的特性,只是將原來 HTTP 傳輸的圖片換成通過 WebSocket 來傳輸了。不過這里后續(xù)還有很多東西可以開發(fā),如果有時間的話,也可以基于這個做一些有趣的東西。
演示
我這個筆記本的性能可能不太行,我只要打開視頻幀率就降低了很多,哈哈。
截取播放B站視頻

截取攝像頭畫面
這樣甚至可以遠程共享畫面了,如果兩個人都布置一個,就可以各自看到對話了(不過沒有聲音,且效率低下,可能也就只能在局域網使用),不過這樣它是相當于從服務器的地方獲取的數據,而我們平時使用的視頻通話工具都是從客戶端獲取的數據。
而且頁面越多,幀率越低,這里可能要優(yōu)化一下或者它就是這么累贅,只能個人使用。=

說明
注意,這里的測試環(huán)境是 Windows,因為有些庫依賴于 Windows 提供的特性,所以需要在 Windows 上運行它。在程序啟動時,它會開啟一個截屏的線程,不斷去獲取屏幕的截屏,如果有用戶訪問,它就會通過 WebSocket 連接,把獲取的屏幕截圖數據傳送給前端,前端的邏輯就是將它繪制在 canvas 上,并添加幀率顯示。
注意:這部分前端的代碼是 AI 生成,對于驗證小功能來說,AI 真是太完美了。而且,其實這整個部分都可以讓 AI 來做,但是具體是哪些的組合還是自己選擇的,畢竟每個人的偏好和技術棧不同。
代碼
所有的代碼都在這里了,大概60行代碼,我把前端壓縮成一行代碼了。如果要運行代碼先要安裝 flask, flask_sock, pillow, dxcam 庫。
import time
from flask import Flask
from flask_sock import Sock
from io import BytesIO
from PIL import Image
import dxcam
# 創(chuàng)建應用
app = Flask(__name__)
sockets = Sock(app)
# 整個應用只創(chuàng)建一個即可
camera = dxcam.create(device_idx=0, output_idx=1) # output_idx 0 是第一塊屏幕,1 是第二塊屏幕
camera.start(target_fps=60, video_mode=True)
JPEG_QUALITY = 80 # 默認的jpeg圖片質量,越高需要的計算量越大,同時越清晰
# 直接前后端寫一起了,這個只是一個演示的demo
INDEX_HTML = """
<!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <meta name="viewport" content="width=device-width, initial-scale=1.0"> <title>Screen Sharing</title> <style> body { display: flex; justify-content: center; align-items: center; height: 100vh; margin: 0; background-color: #f0f0f0; } canvas { border: 1px solid black; } </style> </head> <body> <canvas id="screenCanvas" width="960" height="540"></canvas> <script> const canvas = document.getElementById('screenCanvas'); const ctx = canvas.getContext('2d'); const ws = new WebSocket('ws://'+window.location.host+'/remote_desktop'); ws.onopen = () => { console.log("WebSocket connected"); }; ws.onerror = (event) => { console.error("WebSocket error:", event); }; let lastFrameTime = performance.now(); let frameCount = 0; let fps = 0; ws.onmessage = (event) => { const image = new Image(); image.src = URL.createObjectURL(event.data); image.onload = () => { ctx.drawImage(image, 0, 0, canvas.width, canvas.height); frameCount++; const now = performance.now(); const elapsed = now - lastFrameTime; if (elapsed >= 1000) { fps = frameCount / (elapsed / 1000); frameCount = 0; lastFrameTime = now; } displayFPS(); }; }; function displayFPS() { ctx.font = '30px Arial'; ctx.fillStyle = 'red'; ctx.fillText(`FPS: ${Math.round(fps)}`, 50, 50); } </script> </body> </html>
"""
@app.route('/', methods=['GET'])
def index():
"""簡單的前端"""
return INDEX_HTML
@sockets.route('/remote_desktop')
def get_desktop(ws):
"""
獲取一幀圖片,并發(fā)送給前端
"""
buffer = BytesIO()
fps = 0 # 計算后端生成的幀率
frame_count = 0
last_frame_time = time.perf_counter()
while True:
reset_buffer(buffer) # 每次重置buffer,方便復用
img_data = get_frame(buffer)
frame_count += 1
elapsed = time.perf_counter() - last_frame_time
if elapsed >= 1:
fps = frame_count // elapsed
last_frame_time = time.perf_counter()
frame_count = 0
print("backend real frame fps: ", fps)
ws.send(img_data)
def get_frame(buffer):
"""
獲取一幀,然后處理成jpeg并返回二進制數據
"""
image_array = camera.get_latest_frame()
Image.fromarray(image_array).save(buffer, format="JPEG", quality=JPEG_QUALITY)
return buffer.getvalue()
def reset_buffer(buffer):
# 重置buffer,方便復用
buffer.truncate(0)
buffer.seek(0)
if __name__ == '__main__':
print("remote desktop server starting...")
app.run("0.0.0.0", 9000)
方法補充
除了上文的方法,小編為大家整理了其他兩個實現屏幕共享的方法,希望對大家有所幫助
方法一:50 行代碼實現簡易屏幕共享工具
思路
它的思路很簡單,程序不斷的截屏然后發(fā)送給瀏覽器去顯示,效果也還湊合吧。這里最重要的就是 multipart/x-mixed-replace; boundary=frame 這個首部的使用了,好幾年前還在大學時,我買了一個樹莓派+攝像頭,使用一個開源的軟件叫 motion,搭建了一個局域網的視頻 監(jiān)控。這個 motion 的網頁端就是使用該首部實現的,因為我當時很好奇就去了解了它。
代碼
如果要運行代碼需要安裝對應的依賴:
pip install flask, mss, PIL
import time
from io import BytesIO
from flask import Flask, Response
import mss
from PIL import Image, ImageDraw, ImageFont
app = Flask(__name__)
@app.route('/monitor', methods=['GET'])
def monitor():
"""獲取屏幕數據并返回"""
def gen_img():
with mss.mss() as sct:
# 獲取屏幕的全部區(qū)域
screen = sct.monitors[1]
buffer = BytesIO()
fps = 0 # 幀率
frame_count = 0 # 計算幀率,這是服務端生成的速率,到了客戶端還會更低
start = time.time()
while True:
# 抓取屏幕并返回
screenshot = sct.grab(screen)
# 將截圖數據轉換為 JPEG 格式的二進制數據
img = Image.frombytes("RGB", screenshot.size, screenshot.bgra, "raw", "BGRX")
if fps > 0:
img = add_fps(img, fps)
img.save(buffer, format="JPEG")
jpeg_data = buffer.getvalue()
yield (b'--frame\r\n' +
b'Content-Type: image/jpeg\r\n\r\n' + jpeg_data + b'\r\n')
# 重置buffer,方便復用
buffer.seek(0)
buffer.truncate(0)
frame_count += 1
elapsed = time.time() - start
if elapsed >= 1:
fps = frame_count // elapsed
frame_count = 0
start = time.time()
return Response(gen_img(), mimetype="multipart/x-mixed-replace; boundary=frame")
def add_fps(img, fps):
"""在圖片上添加fps"""
draw = ImageDraw.Draw(img)
draw.text((100, 100), f"fps: {fps}", (255, 0, 0), ImageFont.truetype("arial.ttf", 100))
return img
if __name__ == '__main__':
print("monitoring...")
app.run("127.0.0.1", 9000)
方法二:簡易共享屏幕工具改進版
改進
1.降低分辨率
方法一那個測試的幀率低,最大的原因是電腦的分辨率有點高了,它是2.5K屏幕,所以這次換成了 1920*1080 的屏幕來測試,效果就好很多了,基本都在 50 幀左右。
2.更換截屏庫
更換了一個速度更快的截屏庫,這個截屏庫的速度很快,比方法一那個快多了,但是反應到實際的幀率上就低了,所以還是后續(xù)其他的操作太過于耗時,但是暫時也沒有想到啥優(yōu)化的好方法。
3.降低圖片的質量
JPEG 格式的圖片是可以選擇壓縮質量的,適當的調低質量可以提高處理的速度。
代碼
from flask import Flask, Response
from io import BytesIO
from PIL import Image, ImageDraw, ImageFont
import time
import dxcam
app = Flask(__name__)
# 整個應用只創(chuàng)建一個即可
camera = dxcam.create(device_idx=0, output_idx=1) # output_idx 0 是第一塊屏幕,1 是第二塊屏幕
camera.start(target_fps=60, video_mode=True)
@app.route('/monitor', methods=['GET'])
def monitor():
"""獲取屏幕數據并返回"""
def gen_img():
# 獲取屏幕的全部區(qū)域
fps = 0 # 幀率
frame_count = 0 # 計算幀率,這是服務端生成的速率,到了客戶端還會更低
start_time = time.perf_counter()
buffer = BytesIO()
while True:
reset_buffer(buffer) # 每次重置buffer,方便復用
t1 = time.perf_counter()
yield get_frame(buffer, fps)
print("get frame fps: ", 1.0 / (time.perf_counter() - t1))
frame_count += 1
elapsed_time = time.perf_counter() - start_time
if elapsed_time >= 1:
fps = frame_count // elapsed_time
frame_count = 0
start_time = time.perf_counter()
print("real frame fps: ", fps)
return Response(gen_img(), mimetype="multipart/x-mixed-replace; boundary=frame")
def get_frame(buffer, fps):
"""
獲取一幀,然后處理成jpeg并返回二進制數據
"""
image_array = camera.get_latest_frame()
img = Image.fromarray(image_array)
if fps > 0:
img = add_fps(img, fps)
img.save(buffer, format="JPEG", quality=20)
return (b'--frame\r\n' +
b'Content-Type: image/jpeg\r\n\r\n' + buffer.getvalue() + b'\r\n')
def reset_buffer(buffer):
# 重置buffer,方便復用
buffer.truncate(0)
buffer.seek(0)
def add_fps(img, fps):
"""在圖片上添加fps"""
draw = ImageDraw.Draw(img)
draw.text((100, 100), f"fps: {fps}", (255, 0, 0), ImageFont.truetype("arial.ttf", 100))
return img
if __name__ == '__main__':
print("monitoring...")
app.run("127.0.0.1", 9000)
到此這篇關于Python基于WebSocket實現簡易屏幕共享工具的文章就介紹到這了,更多相關Python WebSocket屏幕共享內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家!
相關文章
Python selenium環(huán)境搭建實現過程解析
這篇文章主要介紹了Python selenium環(huán)境搭建實現過程解析,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2020-09-09
python腳本框架webpy入門安裝及應用創(chuàng)建
這篇文章主要為大家介紹了python腳本框架web.py的入門安裝及應用創(chuàng)建過程,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步2021-11-11
解決導入django_filters不成功問題No module named ''django_filter''
這篇文章主要介紹了解決導入django_filters不成功問題No module named 'django_filter',具有很好的參考價值,希望對大家有所幫助。一起跟隨小編過來看看吧2020-07-07
python中cv2.imdecode()與cv2.imencode()的使用小結
本文介紹了cv2.imencode()和cv2.imdecode()函數的使用,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧2025-01-01
Python使用Selenium+BeautifulSoup爬取淘寶搜索頁
這篇文章主要為大家詳細介紹了Python使用Selenium+BeautifulSoup爬取淘寶搜索頁,具有一定的參考價值,感興趣的小伙伴們可以參考一下2018-02-02

