监控老板一举一动 99行python助你无风险摸鱼
场景
办公室内，所有人都在热火朝天的努力工作，而在春末的暖风中，你却心神荡漾，于是你决定暂停一下手头的工作（虽然你今天可能还没有开始），去愉快的网上冲浪一下。可是你不知道，一个鬼祟且恐怖的身影已经站在你背后，像盯着老鼠洞的老贼猫一样，静静地等待这猎物犯错 ... ...
这时候如果有办法可以不张扬地盯着背后就好了，我就找到了这个办法。
背景
当然，很明显老板是不会花钱让我来研究这个的，真实的背景是我们要做一个汽车的哨兵系统，需要实现一个远程摄像头的功能。
我：“这个很简单，汽车中的设备作为客户端采集摄像头数据，然后推流到服务器，服务器再转给手机APP就行，技术上没有问题，就是要搭建一套视频系统”
老板：“很好，就交给你负责了，不过吧，我们不可能花钱搭建一整套系统，上次demo延时的服务器还在，你将就用一下吧”
我： “... ...” （呵呵，那个下行带宽才2M，2M啊大哥，你是不是不知道这点带宽能干啥）
老板： “还有就是，车端那边的开发同事任务比较紧，你尽量做的简单点”
我： “... ...” （呵呵，8个月时间，才把现成的mqtt客户端库调用成功，还他喵的只会单线程应用，能不吃紧吗）
老板： “还有就是，APP的开发外包到期了，前端的人手也不足，尽量做简单点，但是一定要酷炫”
我： “... ...” （呵呵，人家实习生马上就出国上学了，用说的这么委婉吗）
老板： “还有就是，我们经费不够了，你... ..."
我： “我会克服这些小问题，保证完成任务的，老板我先去忙了”
实现技术
虽然老板可能想要一个某B、某音、某斯拉，很明显这是在扯淡，好在我不是第一天上班了，我不会去仔细分析可行性和成本，然后写二十页密密麻麻的ppt，然后绞尽脑汁在字里行间塞满“你TMD是白痴”等励志话语。
很简单，打开搜索引擎，输入"树莓派" “webcam” “python”，选择一个看上去靠谱的项目就行。最终我参考这个方案，选择了技术栈
https://github.com/RuiSantosdotme/Random-Nerd-Tutorials/blob/master/Projects/rpi_camera_surveillance_system.py
- OpenCV 读取摄像头数据，转为JPEG
- 通过http + mjpeg 实现视频推流
- 通过浏览器原生支持播放视频
至于你说效果达不到怎么办，呵呵，PPT里还不是我说啥就是啥
代码实现
# import time
from threading import Condition
from http import server
PAGE = """\
<html>
<head>
<title>Camera</title>
</head>
<body>
<center><h1>Camera</h1></center>
<center><img src="stream.mjpg" width="640" height="480"></center>
</body>
</html>
"""
class StreamingOutput(object):
def __init__(self):
self.frame = None
self.condition = Condition()
def write(self, bs: bytes):
if bs.startswith(b'\xff\xd8'):
# New frame, copy the existing buffer's content and notify all
# clients it's available
with self.condition:
self.frame = bs
self.condition.notify_all()
class StreamingHandler(server.BaseHTTPRequestHandler):
def do_GET(self):
if self.path == '/':
self.send_response(301)
self.send_header('Location', '/index.html')
self.end_headers()
elif self.path == '/index.html':
content = PAGE.encode('utf-8')
self.send_response(200)
self.send_header('Content-Type', 'text/html')
self.send_header('Content-Length', str(len(content)))
self.end_headers()
self.wfile.write(content)
elif self.path == '/stream.mjpg':
self.send_response(200)
self.send_header('Age', "0")
self.send_header('Cache-Control', 'no-cache, private')
self.send_header('Pragma', 'no-cache')
self.send_header('Content-Type', 'multipart/x-mixed-replace; boundary=FRAME')
self.end_headers()
try:
while True:
with output.condition:
output.condition.wait()
frame = output.frame
self.wfile.write(b'--FRAME\r\n')
self.send_header('Content-Type', 'image/jpeg')
self.send_header('Content-Length', str(len(frame)))
self.end_headers()
self.wfile.write(frame)
self.wfile.write(b'\r\n')
except Exception as e:
print(
'Removed streaming client %s: %s',
self.client_address, str(e))
else:
self.send_error(404)
self.end_headers()
import cv2
from toolbox.parallel import thread
@thread
def read_cap(output: StreamingOutput):
camera = cv2.VideoCapture(0)
print(camera.get(cv2.CAP_PROP_FPS))
print(camera.get(cv2.CAP_PROP_FRAME_WIDTH))
print(camera.get(cv2.CAP_PROP_FRAME_HEIGHT))
# cap = cv2.VideoCapture(1)
print(camera.isOpened())
try:
while True:
success, frame = camera.read() # 读取
encoded, buffer = cv2.imencode('.jpg', frame)
output.write(buffer.tobytes())
# time.sleep(0.1)
finally:
camera.release()
ADDRESS = ('0.0.0.0', 8000)
if __name__ == "__main__":
output = StreamingOutput()
read_cap(output)
try:
print(f"http://{ADDRESS[0]}:{ADDRESS[1]}")
server = server.HTTPServer(ADDRESS, StreamingHandler)
server.serve_forever()
finally:
pass只依赖了cv2，from toolbox.parallel import thread是一个装饰器，可以快速地创建一个线程，并在其中运行函数，很容易实现，就不展开了。
实现效果
打开浏览器，就可以看到摄像头的视频直播了。
流量分析
直观感受上，延时还可以接收，就懒得仔细分析了。
480x360的分辨率（360P），30FPS居然吃掉了3M左右的网络带宽，果然mjpeg被替代是有原因的。
废物利用
于是我把这堆东西包装了一下，然后认真做了PPT，最后给老板吹得天花乱坠
老板虽然一脸懵逼，但是最后还是基于了高度赞扬，又说了一些我们大有可为的事情，并建议我们将这个推进开展和兄弟团队的合作，然后争取在下个月前整个更好的PPT，然后大家齐心协力推广给客户（实际上就是指空手套白狼，骗客户花钱让我们做研发）
大家在老板面前，一起表达了合作的意向，出了门大家又很默契的装作集体失忆，然后愉快地各干的去了。
本着不浪费的原则，于是我灵机一动，将笔记本摄像头方向调整了一下，恰好可以监看到我工位后面的区域，这样在摸鱼时候就不用担心被突然袭击了，完美。
