Create a mjpeg stream from jpeg images in python

Question

I need to serve real-time graphs and I would like to deliver a mjpeg stream over http (so that it is easy to include the graphs in a web-page by using a plain tag).

Is it possible to create an mjpeg stream from multiple jpeg images, in realtime ?

My strategy is:

1. Output the correct http headers:

   Cache-Control:no-store, no-cache, must-revalidate, pre-check=0, post-check=0, max-age=0
   Connection:close
   Content-Type:multipart/x-mixed-replace;boundary=boundarydonotcross
   Expires:Mon, 3 Jan 2000 12:34:56 GMT
   Pragma:no-cache
   Server:MJPG-Streamer/0.2
   

   (got it from a curl -I {on a mjpeg-streamer instance}, but this seems strange)

2. Simply yield the successive jpeg images binaries, taking care to:

   • prepend the correct headers at the beginning of the stream (as mjpeg-streamer does):

     Content-Type: image/jpeg
     Content-Length: 5427
     X-Timestamp: 3927662.086099
     

   • append the boundary string at the end of each jpeg streams.

     --boudary--
     

---

Questions:

Have you done that,

do you know a python module that does that,

do you think it would work,

have you got any advice ?

Answer 1

You may use Flask framework to do this.
It is not only for mjpeg.
I adapted some code from here: 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]

Answer 2

I got it working as a proof-of-concept: https://github.com/damiencorpataux/pymjpeg

For memory:

import os, time
from glob import glob
import sys
from BaseHTTPServer import HTTPServer, BaseHTTPRequestHandler

boundary = '--boundarydonotcross'

def request_headers():
    return {
        'Cache-Control': 'no-store, no-cache, must-revalidate, pre-check=0, post-check=0, max-age=0',
        'Connection': 'close',
        'Content-Type': 'multipart/x-mixed-replace;boundary=%s' % boundary,
        'Expires': 'Mon, 3 Jan 2000 12:34:56 GMT',
        'Pragma': 'no-cache',
    }

def image_headers(filename):
    return {
        'X-Timestamp': time.time(),
        'Content-Length': os.path.getsize(filename),
        #FIXME: mime-type must be set according file content
        'Content-Type': 'image/jpeg',
    }

# FIXME: should take a binary stream
def image(filename):
    with open(filename, "rb") as f:
        # for byte in f.read(1) while/if byte ?
        byte = f.read(1)
        while byte:
            yield byte
            # Next byte
            byte = f.read(1)

# Basic HTTP server
class MyHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        self.send_response(200)
        # Response headers (multipart)
        for k, v in pymjpeg.request_headers().items():
            self.send_header(k, v) 
        # Multipart content
        for filename in glob('img/*'):
            # Part boundary string
            self.end_headers()
            self.wfile.write(pymjpeg.boundary)
            self.end_headers()
            # Part headers
            for k, v in pymjpeg.image_headers(filename).items():
                self.send_header(k, v) 
            self.end_headers()
            # Part binary
            for chunk in pymjpeg.image(filename):
                self.wfile.write(chunk)
    def log_message(self, format, *args):
        return

httpd = HTTPServer(('', 8001), MyHandler)
httpd.serve_forever()