您可以使用烧瓶框架来做到这一点。
这不仅是mjpeg。
我适应从这里的一些代码：https://blog.miguelgrinberg.com/post/video-streaming-with-flask
APP.py
#!/usr/bin/env python
from importlib import import_module
import os
from flask import Flask, render_template, Response
# import camera driver
if os.environ.get('CAMERA'):
Camera = import_module('camera_' + os.environ['CAMERA']).Camera
else:
from camera import Camera
# Raspberry Pi camera module (requires picamera package)
# from camera_pi import Camera
app = Flask(__name__)
@app.route('/')
def index():
"""Video streaming home page."""
return render_template('index.html')
def gen(camera):
"""Video streaming generator function."""
while True:
frame = camera.get_frame()
yield (b'--frame\r\n'
b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n')
@app.route('/video_feed')
def video_feed():
"""Video streaming route. Put this in the src attribute of an img tag."""
return Response(gen(Camera()),
mimetype='multipart/x-mixed-replace; boundary=frame')
if __name__ == '__main__':
app.run(host='0.0.0.0', threaded=True)
base_camera.py
import time
import threading
try:
from greenlet import getcurrent as get_ident
except ImportError:
try:
from thread import get_ident
except ImportError:
from _thread import get_ident
class CameraEvent(object):
"""An Event-like class that signals all active clients when a new frame is
available.
"""
def __init__(self):
self.events = {}
def wait(self):
"""Invoked from each client's thread to wait for the next frame."""
ident = get_ident()
if ident not in self.events:
# this is a new client
# add an entry for it in the self.events dict
# each entry has two elements, a threading.Event() and a timestamp
self.events[ident] = [threading.Event(), time.time()]
return self.events[ident][0].wait()
def set(self):
"""Invoked by the camera thread when a new frame is available."""
now = time.time()
remove = None
for ident, event in self.events.items():
if not event[0].isSet():
# if this client's event is not set, then set it
# also update the last set timestamp to now
event[0].set()
event[1] = now
else:
# if the client's event is already set, it means the client
# did not process a previous frame
# if the event stays set for more than 5 seconds, then assume
# the client is gone and remove it
if now - event[1] > 5:
remove = ident
if remove:
del self.events[remove]
def clear(self):
"""Invoked from each client's thread after a frame was processed."""
self.events[get_ident()][0].clear()
class BaseCamera(object):
thread = None # background thread that reads frames from camera
frame = None # current frame is stored here by background thread
last_access = 0 # time of last client access to the camera
event = CameraEvent()
def __init__(self):
"""Start the background camera thread if it isn't running yet."""
if BaseCamera.thread is None:
BaseCamera.last_access = time.time()
# start background frame thread
BaseCamera.thread = threading.Thread(target=self._thread)
BaseCamera.thread.start()
# wait until frames are available
while self.get_frame() is None:
time.sleep(0)
def get_frame(self):
"""Return the current camera frame."""
BaseCamera.last_access = time.time()
# wait for a signal from the camera thread
BaseCamera.event.wait()
BaseCamera.event.clear()
return BaseCamera.frame
@staticmethod
def frames():
""""Generator that returns frames from the camera."""
raise RuntimeError('Must be implemented by subclasses.')
@classmethod
def _thread(cls):
"""Camera background thread."""
print('Starting camera thread.')
frames_iterator = cls.frames()
for frame in frames_iterator:
BaseCamera.frame = frame
BaseCamera.event.set() # send signal to clients
time.sleep(0)
# if there hasn't been any clients asking for frames in
# the last 10 seconds then stop the thread
if time.time() - BaseCamera.last_access > 10:
frames_iterator.close()
print('Stopping camera thread due to inactivity.')
break
BaseCamera.thread = None
camera.py
#D:\gstreamer\1.0\x86\bin>gst-launch-1.0.exe multifilesrc loop=true start-index=0 stop-index=0 location=d:/python/temp.png ! decodebin ! identity sleep-time=1000000 ! videoconvert ! autovideosink
import shutil
import time
import os,sys
from PIL import Image, ImageFont, ImageDraw, ImageFile
from io import BytesIO
from base_camera import BaseCamera
im = Image.new("RGB", (300, 30), (220, 180, 180))
#im.format'JPEG'
dr = ImageDraw.Draw(im)
font = ImageFont.truetype(os.path.join("fonts", "msyh.ttf"), 16)
text =time.strftime("%m/%d %H:%M:%S") +u"这是一段测试文本。"
dr.text((10, 5), text, font=font, fill="#000000")
im.save("d://python/temp.jpg")
dr.rectangle((0,0,300,500),fill="#FFFFFF")
text =time.strftime("%m/%d %H:%M:%S") +u"这是一段测试文本。"
dr.text((10, 5),text, font=font, fill="#000000")
f = BytesIO()
f.name="sdf.jpg"
im.save(f,"JPEG")
f.seek(0)
f.close()
class Camera(BaseCamera):
"""An emulated camera implementation that streams a repeated sequence of
files 1.jpg, 2.jpg and 3.jpg at a rate of one frame penr second."""
imgs = [open(f + '.jpg', 'rb').read() for f in ['1', '2', '3']]
@staticmethod
def frames():
while True:
text =time.strftime("%m/%d %H:%M:%S") +u"这是一段测试文本。"
dr.rectangle((0,0,300,500),fill="#FFFFFF")
dr.text((10, 5), text, font=font, fill="#000000")
f = BytesIO()
im.save(f,'JPEG')
try :
im.save("d:/python/temp.jpg")
except :
print("Unexpected error:", sys.exc_info()[0])
pass
# shutil.copy("d:/python/temp2.png","d:/python/temp.png")
f.seek(0)
time.sleep(1)
yield f.read() #Camera.imgs[int(time.time()) % 3]
这是多个问题，最他们实际上并没有在StackOverflow上工作。第一个是“分享你的经验”，第二个是图书馆购物，第三个是一般的，第四个只是要求一般的建议。这可能更适合在邮件列表或论坛上。（如果你有一个更具体的问题，更关系到编写代码或解释协议，当然，这将是一个完美的问题。） – abarnert
谢谢你的建议，我必须学习发布好问题。但是，我发现自己有一个工作概念验证。我想我会在这里发布它作为我自己问题的答案，这是一件好事，还是应该删除这个问题？ – Damien
我认为它不是一个有用的答案，不会有人会问。特别是因为它只是一个链接（以及将来可能会改变的代码链接）。如果您希望人们查看您的代码，请使用CodeReview而不是StackOverflow。如果您希望人们使用您的代码，请将其包装并放在PyPI上。 – abarnert