-
Notifications
You must be signed in to change notification settings - Fork 0
/
camera_standalone.py
64 lines (48 loc) · 1.64 KB
/
camera_standalone.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
import cv2
import websockets
import asyncio
import numpy as np
import threading
import base64
async def send_frames():
while True:
uri = "wss://bernhackt23-backend.web01.dalcloud.net/"
try:
async with websockets.connect(uri) as websocket:
cap = init_camera(0)
if cap is None:
print("Failed to initialize camera. Retrying in 2 seconds...")
await asyncio.sleep(2)
continue
while True:
ret, frame = cap.read()
if not ret:
print("Failed to get frame")
continue
_, buffer = cv2.imencode('.jpg', frame)
base64_frame = base64.b64encode(buffer).decode('utf-8')
await websocket.send(base64_frame)
await asyncio.sleep(0.1)
cap.release()
except Exception as e:
print(f"WebSocket Error: {e}. Reconnecting in 2 seconds...")
await asyncio.sleep(2)
def init_camera(index):
cap = cv2.VideoCapture(index)
if not cap.isOpened():
print(f"Error: Couldn't open camera {index}.")
return None
return cap
def start_websocket_thread():
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(send_frames())
def main():
websocket_thread = threading.Thread(target=start_websocket_thread)
websocket_thread.start()
while True:
if cv2.waitKey(1) & 0xFF == ord('q'):
break
cv2.destroyAllWindows()
if __name__ == "__main__":
main()