已经很久沒有更新博客,主要是现在工作太忙,一天压得比较紧,回家之后就就没有太多精力去写博客了。
最近更新了一个Python工具,主要是用来视频监控。基本想法是通过延时摄影的方式,每隔一段时间拍摄一张照片,然后组合成视频,这样就可以看到一段时间内的变化。
想做这个工具是因为我的电脑不太方便关或者锁屏,所以一般是打开的状态,但是我又想知道在我不在的时候,我的电脑有没有被别人使用,所以就想到了这个工具。目前处于可以使用的状态,后续我想加一些检测功能,比如检测到非自己的人脸就提高检测频率之类的,或者考虑视频压缩,可以减少一些重复、静态帧。
安装
相关的包已经上传到了pypi,可以通过pip安装。依赖的包有opencv-python,numpy。
1
|
pip3 install work-monitor
|
或者通过源码直接使用。
1
2
3
|
git clone git@github.com:caibingcheng/work-monitor.git
cd work-monitor
python3 -m monitor
|
使用
首先是启动监控服务。
然后可以通过客户端来设置或查看服务端的状态,比如查看服务端的配置。
1
|
work-monitor get_config
|
或者设置服务端的配置。
1
|
work-monitor set_config <keys...> <value>
|
help命令可以查看更多的命令。
1
2
3
4
5
6
7
8
9
|
$ work-monitor help
Usage: python3 -m monitor <command> [arguments]
Commands:
help: Print help
server: Start server [video_path], default video_path is empty
stop: Client command, stop server
restart: Client command, restart server
get_config: Client command, get config
set_config: Client command, set config
|
代码讲解
源码地址: https://github.com/caibingcheng/work-monitor
几个模块分类如下:
- app.py: 入口,调用command模块
- command.py: 命令行模块,在其中配置该项目支持的命令
- config.py: 配置模块,用于配置服务端的配置
- server.py: 服务端模块,用于启动服务端,客户端和服务端的通信也在其中实现
- log.py: 日志模块,用于记录日志
- capture.py: 拍照模块,用于拍照和保存图片
- video.py: 视频模块,用于将图片组合成视频
- policy.py: 策略模块,用于配置拍照和视频的策略,比如拍照的间隔时间,视频的长度等
入口
主要关注server的入口。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
@add_command("server", "Start server [video_path], default video_path is empty")
def server(*args):
log_info("Starting")
policy = config["policy"]
log_info(f"Using policy {policy}")
# str to function
policy = globals()[policy]
from monitor.server import start_server, should_stop
start_server()
video_path_for_debug = "" if len(args) == 0 else args[0]
while not should_stop():
try:
policy(video_path_for_debug)
except Exception as e:
log_error(e)
# backtrace
import traceback
traceback.print_exc()
stop()
log_info("Stopped")
|
先找到policy,然后启动命令监听的server,然后执行对应的policy。如果发生异常时,会给server发送stop命令,然后退出。(这里有问题,如果是server异常,则命令不一定能发送到server,所以需要改进。)
配置
原始配置如下:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
raw_config = {
"camera_id": 0,
"video_dir": "$HOME/Videos",
"frames_dir": "$HOME/Pictures/work-monitor",
"config_dir": "$HOME/.work-monitor",
"log_dir": "$HOME/.work-monitor/log",
"fps": 60,
"quality": 75,
"frames_save": False,
"policy": "easy_policy",
"easy_policy": {"frames_interval": 10, "frames_per_video": 1000},
"server": {
"port": 22311,
},
}
|
各项配置的含义如下:
- camera_id: 摄像头id,如果有多个摄像头,可以通过这个配置来选择摄像头
- video_dir: 视频保存的目录
- frames_dir: 帧图片保存的目录
- config_dir: 配置文件保存的目录
- log_dir: 日志保存的目录
- fps: 视频的帧率
- quality: 帧图片的质量,取值范围[0, 100],0表示最差,100表示最好
- frames_save: 是否保存帧图片,一般设置为False,因为帧图片会占用很大的空间
- policy: 策略,目前只有easy_policy,后续可以添加更多的策略
- easy_policy: easy_policy的配置,包括帧图片的间隔时间和视频的长度
- server: 服务端的配置,目前只有端口号
提示
对于frames_save配置目前还有一些问题,如果设置值为True,并且policy是通过帧图片数判断是否要保存视频的话,则在第一次触发阈值之后会频繁触发且保存相同的帧。
配置是通过一个全局变量config来保存的,初始化流程如下:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
try:
config = initialize_config(config)
verify_config(config)
except Exception as e:
print(e)
choise = input("Reset config? (y/n)")
if choise == "y":
print("Resetting config")
config = raw_config
config = initialize_config(config, force=True)
else:
print("Using raw config")
config = raw_config
config = preprocess_config(config)
verify_config(config)
|
initialize_config会对原始的config作预处理,替换其中的环境变量,比如$HOME;判断值范围是否正确,比如quality的值范围是[0, 100],如果不在范围内,则会抛出异常;创建目录,比如video_dir,frames_dir等;如果是第一次启动,则会将配置写入到文件。预处理之后会调用verify_config来验证配置是否正确,主要是检查config和raw_config的key是否一致,如果不一致,则会抛出异常。
当有异常的时候会尝试用raw_config来初始化config,如果用户允许的话,会覆盖掉原来的配置文件。
在server中可以动态更新参数,对应的修改会更新到配置文件中。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
@add_server_command("set_config")
def set_config_server(*args):
if len(args) < 2:
raise Exception("Not enough arguments")
current = config.copy()
current_header = current
keys = args[:-1]
value = args[-1]
for key in keys[:-1]:
if key not in current or not isinstance(current[key], dict):
raise Exception(f"Key {key} not found")
current = current[key]
current[keys[-1]] = value
current_header = initialize_config(current_header, force=True)
config_str = json.dumps(current_header, indent=4).encode("utf-8")
log_info(f"Sending config {config_str}")
update_config(config, current_header)
return config_str
|
比如设置video_dir。
1
|
work-monitor set_config video_dir /home/bing/Videos
|
通信
server在主线程中启动,在子线程中监听客户端的连接。
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
|
def start_server():
# create a socket object
serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# get local machine name
host = socket.gethostname()
port = config["server"]["port"]
# force to release the port
serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
# set keep alive
serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
serversocket.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 1)
serversocket.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 1)
serversocket.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 5)
# bind to the port
serversocket.bind((host, port))
# queue up to 5 requests
serversocket.listen(5)
def server_loop():
log_info("Server started")
while True:
# establish a connection
clientsocket, addr = serversocket.accept()
log_info(f"Got a connection from {addr}")
msg = clientsocket.recv(1024).decode("utf-8").split()
log_info(f"Received {msg}")
argument = msg[1:]
msg = msg[0]
try:
if msg not in server_command:
log_error(f"Unknown command {msg}")
clientsocket.send("failed".encode("utf-8"))
else:
clientsocket.send(server_command[msg](*argument))
except Exception as e:
# show backtrace
log_error("Server failed", e)
log_error("Config", config)
clientsocket.send("failed".encode("utf-8"))
finally:
clientsocket.close()
threading.Thread(target=server_loop).start()
|
客户端如下,是一般定式写法。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
def send_msg_to_server(msg):
# create a socket object
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# get local machine name
host = socket.gethostname()
port = config["server"]["port"]
try:
# connection to hostname on the port.
client.connect((host, port))
except Exception as e:
log_error("Client failed to connect to server", e)
exit(1)
# Receive no more than 1024 bytes
client.send(msg.encode("utf-8"))
response = client.recv(1024).decode("utf-8")
client.close()
return response
|
视频
拍照时,会将图片保存到frames_dir中,然后通过video模块来组合成视频。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
def generate_video():
# only one video can be generated at a time
global video_in_progress
if video_in_progress:
return
log_info("Generating video")
frames, frames_date_range = load_frames(config["frames_dir"])
log_info(f"Total frames: {len(frames)}, date range: {frames_date_range}")
if not frames:
return
video_in_progress = True
import threading
threading.Thread(
target=generate_video_from_frames,
args=(frames, frames_date_range, config["video_dir"], config["fps"]),
).start()
|
视频的生成是在子线程中进行的,考虑到一些边界情况,目前只允许一个视频生成线程在运行。
首先是检查frames_dir中是否有图片,如果没有则直接返回。然后将video_in_progress设置为True,表示有视频正在生成,然后在子线程中调用generate_video_from_frames来生成视频。
在检查frames_dir中是否有图片时,只是拿到图片的地址,此处不会读图,主要是考虑到frames_dir中的图片可能会很多,如果一次性读入内存,可能会导致内存不足,所以在生成视频的时候,会一张张的读入图片,然后组合成视频。
在图片保存的时候,会在图片中插入时间信息,这样在生成的视频中可以直观的看到时间信息。不过这样也会导致图片之间存在差异,不易于判断图片的相似度。(可以对相关区域做mask屏蔽。)
小结
总算更新了一篇博客,这个工具还有很多可以改进的地方,但是比如之前写的fstats、fkfish之类的工具在第一版之后就已经很久没有更新了,目前看起来也够用,总之随心所欲吧。也有几个其他的工具还在推进中,还没有成品,后续慢慢更新。
另外,感觉我这段时间有些闭塞,在制作这个工具的过程中,知道了以下更新:
- python3.10支持pyproject.toml了,并且setup.py是不推荐的了
- pypi用github action + twine发布会报错了,可以通过Trusted Publisher Management来管理发布者
不过,这段时间也不是废了,只是接触的领域不同。目前对DMA、Misra规则、QNX系统等等有一些接触和了解,也是扩展了自己的知识面,并且目前所在的激光雷达领域对我来说,就像是我在相机领域的延续,所以这也是缘分了。后续的年终总结中会想介绍介绍我的想法。