检测摄像头的fps

需求

项目中经常遇到不是摄像头就是网线的问题，曾经遇到一个项目算法日志一直报 warning，经过好几个小时的远程排查，发现是摄像头的 fps 不稳定，而且出现 fps 逐渐降低的情况，所以算法跑着跑着就挂了。

于是就需要开发一个测试 fps 的工具，工具倒是不复杂，主要依赖 opencv-python 库读取摄像头视频流。

代码

最基础版本

import argparse
import sys
import time

import cv2

parser = argparse.ArgumentParser()
ip = parser.add_argument("-i", "--ip", type=str)
args = parser.parse_args()

frame_id = 0

cap = cv2.VideoCapture(f"rtsp://username:password@{args.ip}:554/Streaming/Channels/101")
while True:
    try:
        ret, frame = cap.read()
        if not ret:
            break
        else:
            frame_id += 1

            if frame_id < 100:
                remainder = frame_id % 5
                if remainder == 0:
                    print(f"wait:{'-' * (frame_id // 5)}>", end="\r")
            if frame_id == 100:
                time_start = time.time()

            if frame_id > 100:
                time_end = time.time()
                print(
                    f"fps: {(frame_id - 100) / (time_end - time_start)}",
                    end="\r",
                )
    except KeyboardInterrupt:
        exit(-1)

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

\r 是回车符，光标会回到行首。但是只有一串数字输出，所以经过我的改良有了下面的一版：
改良版本

import argparse
import atexit
import sys
import time
from collections import OrderedDict
from multiprocessing import Event, Process, Queue
from typing import Any, List

import cv2
import keyboard

parser = argparse.ArgumentParser()
ip = parser.add_argument("-i", "--ip", type=str)
args = parser.parse_args()
alive_set = Event()

exit_msg = "\nAll processes have exited."


def test_fps(args: Any, event: Event, fps_q: Queue, name: str):
    frame_id = 0

    cap = cv2.VideoCapture(
        f"rtsp://username:password@{args.ip}:554/Streaming/Channels/101"
    )
    string = ""
    current_time = time.time()
    if cap.isOpened():
        print(f"Enter `q` to exit...\n{'-' * 40}")
    else:
        print("The camera is not opened yet.")
        exit(0)

    while True:
        try:
            ret, _ = cap.read()
            if not ret:
                break
            else:
                frame_id += 1

                if frame_id < 100:
                    remainder = frame_id % 5
                    first_100_time = time.time()

                    if remainder == 0 and current_time < first_100_time:
                        string = (
                            f"wait: {(first_100_time - current_time) * 1000:.4f} ms"
                        )
                if frame_id == 100:
                    time_start = time.time()
                    event.set()
                    string = ""

                if frame_id > 100:
                    time_end = time.time()
                    string = f"{(frame_id - 100) / (time_end - time_start):.8f}"

            fps_q.put([name, string])

        except KeyboardInterrupt:
            exit(0)


def test_time(time_start: float, event: Event, time_q: Queue, name: str):
    while True:
        try:
            if event.is_set():
                try:
                    time_now = time.time()
                    cost = f"{time_now - time_start:.3f}s."
                    time_q.put([name, cost])
                except (KeyboardInterrupt, Exception) as e:
                    exit(0)
        except (KeyboardInterrupt, Exception) as e:
            exit(0)


def print_progress(progress: "OrderedDict"):
    # sys.stdout.write("\033[2J\033[H")  # clear screen
    out_list = []
    for name, data in progress.items():
        if data:
            out_list.append(f"{name}: {data}")

    string = ""
    if len(out_list) == 1:
        string = "\r" + out_list[0].replace("fps:", "").strip()

    elif len(out_list) > 1:
        string = "\r" + " | ".join(out_list)
    else:
        string = "\r" + "waiting..."

    sys.stdout.write(string)
    sys.stdout.flush()


def kill_pid(processes: List[Process]):
    try:
        for p in processes:
            p.terminate()

    except Exception as e:
        sys.stderr.write(f"Detailed error: {e}")
        exit(0)


def main():
    print("Test the fps of camera...")
    status = Queue(maxsize=2)
    progress = OrderedDict()
    processes = []

    try:
        time_start = time.time()
        tasks = [
            Process(target=test_fps, args=(args, alive_set, status, "fps"), name="fps"),
            Process(
                target=test_time,
                args=(time_start, alive_set, status, "time"),
                name="time",
            ),
        ]

        for t in tasks:
            t.start()
            processes.append(t)
            progress[t.name] = None

        while any(i.is_alive() for i in tasks):
            while not status.empty():
                try:
                    name, data = status.get()
                    progress[name] = data
                    print_progress(progress)

                    keyboard.add_hotkey("q", kill_pid, args=(processes,))

                except KeyboardInterrupt:
                    break

    except KeyboardInterrupt:
        atexit.register(kill_pid, *(processes,))
        print(exit_msg)
        exit(0)


if __name__ == "__main__":
    main()
    print(exit_msg)

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
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152

改良版使用了 多进程 和 队列 通信和 Event 以及 keyboard 库，支持动态单行局部字符串一直刷新和 q 键退出代码的功能。

结果

$ python.exe .\test_fps.py -i 192.168.0.67
Test the fps of camera...
Enter `q` to exit...
----------------------------------------
fps: 20.03542377 | time: 43.861s.
All processes have exited.
1
2
3
4
5
6

下面有动图:

最后推荐一个好用的windows下的录屏工具，上面的gif就是这个工具录制的。ScreenToGif. 再次感叹开源的伟大。🤓