[HOW-TO] make this rtsp stream stable?

[xucaiqin]

i have a example to push rtsp stream to my server raspberrypi 4b picamera2 0.3.18 python 3.11.2 camera.py

from picamera2 import Picamera2
from picamera2.encoders import H264Encoder
from picamera2.outputs import FfmpegOutput
from tool.log.log_uru import Logger

class Camera(object):
    def __init__(self, sn):
        self.rtsp = None
        self.lug = Logger().get_logger
        self.picam2 = Picamera2()
        # Picamera2.set_logging(Picamera2.DEBUG)
        # rtsp推流的编码
        self.encoder = None
        resolution = (1280, 720)
        self.picam2.configure(self.picam2.create_video_configuration(main={"size": resolution}))
        self.picam2.start()
        self.lug.info("Camera initialized")
        self.start_rtsp()

    def error_callback(self, e):
        """
        picamera2异常断开的回调函数
        """
        self.lug.error("Camera Error {}", e)
        # 重连
        # self.picam2.stop_encoder(self.encoder)
        # self.start_rtsp()
        # self.lug.warning("reconnecting...")

    def start_rtsp(self):
        # self.encoder = H264Encoder(bitrate=1000000)
        self.encoder = H264Encoder(bitrate=1000000, repeat=True, iperiod=15, framerate=15)
        self.rtsp = FfmpegOutput("-f rtsp rtsp://myserverip/live/test")
        self.rtsp.error_callback = self.error_callback
        self.encoder.output = self.rtsp
        self.picam2.start_encoder(self.encoder)

    def stop(self):
        """
        停止camera
        :return:
        """
        self.picam2.stop()

i run this camera. main.py

if __name__ == '__main__':
   Camera("camera")
   while True:
     pass

However, if he pushes the RTSP stream for a maximum of 10 minutes, a broken pipe exception will appear。i want to know this is the ffmpeg's error or picamera2's error;

[davidplowman]

Hi, thanks for the question. I'm afraid it's a bit difficult to know what's going wrong here. Are there any console messages? Anything in dmesg? Do the images show corruption before it fails? Does the RTSP server output any errors? Perhaps you could try sending the encoded video to a different kind of output, just to check it's not the camera system that's failing?

A couple of minor (and probably unrelated things).

You shouldn't do anything in the error_callback other than signal your main thread, which means you can't stop and restart your RTSP session there (printing or logging is OK, of course). Instead you need to signal your event loop to do that. (Is this error_callback actually being invoked? That would suggest it's the FFmpeg process that's dying, though I couldn't say why.)

The infinite loop without a "sleep" in it worried me a little bit - is that busy waiting? In any case, that's probably where you want to wait for an "error" signal and restart the RTSP session.

[xucaiqin]

Here's the full code main.py

import signal
import time
import sys
import RPi.GPIO as GPIO

from manager.device.index import DeviceManager
from manager.wireless.index import WirelessManager
from device.bluetooth.index import BluetoothDevice
from manager.protocol.mqtt.index import Mqtt
from manager.protocol.index import ProtocolManager
from tool.log.log_uru import Logger
from device.dht11.index import Dht11
from device.sgp30.index import Sgp30
from device.camera.index import Camera

lug = Logger().get_logger

# 定义信号处理函数
def signal_handler(sig, frame):
    lug.warning('收到信号:{} 退出程序中......', sig)
    ProtocolManager.terminate()
    DeviceManager.terminate()
    WirelessManager.terminate()
    GPIO.cleanup()
    lug.info("退出程序完成......")
    # 退出程序
    sys.exit(0)

# 注册信号处理函数
signal.signal(signal.SIGTERM, signal_handler)

if __name__ == '__main__':

    try:
        lug.info("项目启动中...")
        # 实例化mqtt协议
        Mqtt()
        # 初始化协议管理器
        ProtocolManager.init()
        # # 实例化蓝牙设备
        BluetoothDevice()
        # 初始化无线协议管理器
        # WirelessManager.init()

        # 温湿度计 5minutes每次
        dht11 = Dht11(18, "dht11_0001")
        lug.info("温湿度计启动...")

        # 环境传感器
        Sgp30(dht11, "SGP30_0001")
        lug.info("环境传感器启动...")

        # 摄像头 半小时采集一次图片
        Camera("camera")
        lug.info("摄像头启动...")

        # 开始定时任务
        # job = Job(mq)
        # job.add_job(data_job, 60 * 30, (mq,))
        # job.start()
        # lug.info("启动定时任务...")

        lug.info("项目启动完成......")
        while True:
            time.sleep(1)
            pass
    except KeyboardInterrupt:
        lug.error("捕获到 KeyboardInterrupt 异常，正在退出...")
        ProtocolManager.terminate()
        DeviceManager.terminate()
        sys.exit(0)
    except Exception as e:
        lug.error("异常: {}", e)
        ProtocolManager.terminate()
        DeviceManager.terminate()
        GPIO.cleanup()
        sys.exit(0)

camera index.py

import io
import threading
import time
from picamera2 import Picamera2
from picamera2.encoders import H264Encoder
from picamera2.outputs import FfmpegOutput
# from third.Out2 import FfmpegOutput
from device.models import Cam
from manager.device.index import Device, Topic
from manager.protocol.index import ProtocolManager
from tool.base64_tool import bytes_to_base64
from tool.log.log_uru import Logger
from tool.time_tool import ymd_hms_time

class Camera(Device):
    def get_protocol(self):
        return "Mqtt"

    def __init__(self, sn):
        super().__init__(sn, {}, {}, {})
        self.rtsp = None
        self.topic = Topic.camera_post
        self.lug = Logger().get_logger
        self.picam2 = Picamera2()
        # Picamera2.set_logging(Picamera2.DEBUG)
        # rtsp推流的编码
        self.encoder = None
        resolution = (1280, 720)
        self.picam2.configure(self.picam2.create_video_configuration(main={"size": resolution}))
        self.picam2.start()
        self.lug.info("Camera initialized")
        self.start_rtsp()
        self.push_task()

    def error_callback(self, e):
        """
        picamera2异常断开的回调函数
        """
        self.lug.error("Camera Error {}", e)
        # 重连
        # self.picam2.stop_encoder(self.encoder)
        # self.start_rtsp()
        # self.lug.warning("reconnecting...")

    def start_rtsp(self):
        # self.encoder = H264Encoder(bitrate=1000000)
        self.encoder = H264Encoder(bitrate=1000000, repeat=True, iperiod=15, framerate=15)
        self.rtsp = FfmpegOutput("-f rtsp rtsp://x.x.x.x/live/test")
        self.rtsp.error_callback = self.error_callback
        self.encoder.output = self.rtsp
        self.picam2.start_encoder(self.encoder)

    def __capture_pic(self):
        """
        抓取图片
        :return:
        """
        time.sleep(1)
        try:
            while True:
                data = io.BytesIO()
                time.sleep(1)
                self.picam2.capture_file(data, format='jpeg')
                b = data.getvalue()
                data.close()
                base_64 = bytes_to_base64(b)
                self.lug.debug("推送图片大小：{}", len(b))
                ProtocolManager.send_msg("Mqtt", self.topic,
                                         Cam(sn=self.sn, action="capture", val=base_64, time=ymd_hms_time()))
                time.sleep(60 * 5)
        except Exception as e:
            self.lug.error("抓取图片异常：{}", e)

    def push_task(self):
        """
        开启定时任务推送数据
        """
        threading.Thread(target=self.__capture_pic, daemon=True).start()

    def stop(self):
        """
        停止camera
        :return:
        """
        # self.rtsp.stop()
        # self.picam2.stop_encoder()
        self.picam2.stop()

This is a screenshot of the project repository structure。

This anomaly will appear regularly for about 10 minutes after the camera is activated。