-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtest.py
136 lines (107 loc) · 4.22 KB
/
test.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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
import cv2
import socket
import numpy as np
import struct
import time # Import time for recording start time
from feature import calculate_compression_profile
import sys
import streamers.mjpeg as mjpeg
def cap_compression_profile(matrix):
transformed_matrix = matrix * 100 * 15
transformed_matrix = np.clip(transformed_matrix, 1, 50)
transformed_matrix = transformed_matrix.astype(int)
return transformed_matrix
def send_tile(client_socket, tile, quality):
encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), int(quality)]
_, tile_encoded = cv2.imencode('.jpg', tile, encode_param)
tile_data = tile_encoded.tobytes()
header = struct.pack('!I', len(tile_data))
client_socket.sendall(header)
client_socket.sendall(tile_data)
def send_image(client_socket, image, qualities):
rows = len(qualities)
cols = len(qualities[0])
h, w, _ = image.shape
tile_height, tile_width = h // rows, w // cols
for i in range(rows):
for j in range(cols):
tile = image[i * tile_height:(i + 1) * tile_height, j * tile_width:(j + 1) * tile_width]
send_tile(client_socket, tile, qualities[i][j])
def test_read_frame():
target_fps = 8.35
cap = cv2.VideoCapture('videos/climbing.mp4')
# Calculate the time to wait between frames
frame_time = 1.0 / target_fps
frames_read = 0
test_start_time = time.time()
while True:
start_time = time.time()
ret, frame = cap.read()
frames_read += 1
if not ret:
print("Failed to capture frame")
print(f'Actual frame rate: {frames_read / (time.time() - test_start_time)}')
break
# Calculate elapsed time and sleep if necessary
elapsed_time = time.time() - start_time
time_to_wait = frame_time - elapsed_time
if time_to_wait > 0:
time.sleep(time_to_wait)
def test_read_frame_and_compression():
target_fps = 8.35
cap = cv2.VideoCapture('videos/climbing.mp4')
# Calculate the time to wait between frames
frame_time = 1.0 / target_fps
frames_read = 0
test_start_time = time.time()
while True:
start_time = time.time()
ret, frame = cap.read()
frames_read += 1
if not ret:
print("Failed to capture frame")
print(f'Actual frame rate: {frames_read / (time.time() - test_start_time)}')
break
encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), 50]
_, frame_encoded = cv2.imencode('.jpg', frame, encode_param)
# Calculate elapsed time and sleep if necessary
elapsed_time = time.time() - start_time
time_to_wait = frame_time - elapsed_time
if time_to_wait > 0:
time.sleep(time_to_wait)
def test_read_frame_and_tile_compression():
target_fps = 8.35
cap = cv2.VideoCapture('videos/climbing.mp4')
# Calculate the time to wait between frames
frame_time = 1.0 / target_fps
frames_read = 0
test_start_time = time.time()
while True:
start_time = time.time()
ret, frame = cap.read()
frames_read += 1
if not ret:
print("Failed to capture frame")
print(f'Actual frame rate: {frames_read / (time.time() - test_start_time)}')
break
qualities = cap_compression_profile(calculate_compression_profile(frame, 2, 4))
print("Compression Profile:\n", qualities)
rows = len(qualities)
cols = len(qualities[0])
h, w, _ = frame.shape
tile_height, tile_width = h // rows, w // cols
for i in range(rows):
for j in range(cols):
tile = frame[i * tile_height:(i + 1) * tile_height, j * tile_width:(j + 1) * tile_width]
encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), int(qualities[i][j])]
_, tile_encoded = cv2.imencode('.jpg', tile, encode_param)
# Calculate elapsed time and sleep if necessary
elapsed_time = time.time() - start_time
time_to_wait = frame_time - elapsed_time
if time_to_wait > 0:
time.sleep(time_to_wait)
if __name__ == '__main__':
main()
# test_read_frame()
# test_read_frame_and_tile_compression()
# test_read_frame_and_compression()