rtsp-stream 0.1.0

Lightweight, thread-based RTSP video stream reader with automatic reconnection.

rtsp-stream

Lightweight, thread-based RTSP video stream reader with automatic reconnection — drop the VideoCapture boilerplate and let the library handle feed drops for you.

Install

pip install rtsp-stream

Note: opencv-python is listed as a dependency. If your project already uses opencv-contrib-python or a headless variant, install with --no-deps and manage OpenCV separately.

Quick start

from rtsp_stream import VideoStream

stream = VideoStream("rtsp://camera-ip/stream")
stream.start()

while True:
    frame = stream.read()   # returns None until first frame arrives
    if frame is not None:
        # do something with the numpy array
        pass

stream.stop()

Context manager

with VideoStream("rtsp://camera-ip/stream").start() as stream:
    frame = stream.read()

---

Configuration

Parameter	Type	Default	Description
rtsp_url	str	—	Any URL accepted by cv2.VideoCapture
reconnect_delay	float	1.0	Seconds between reconnection attempts
max_reconnect_attempts	int | None	None	Max consecutive attempts before giving up (None = forever)
reconnect_backoff	float	1.0	Multiply delay by this after each failure (e.g. 2.0 = exponential back-off)
max_reconnect_delay	float	30.0	Upper bound on back-off delay
on_connect	Callable	None	Called every time a connection succeeds
on_disconnect	Callable	None	Called every time the feed drops
on_reconnect_failure	Callable	None	Called when max_reconnect_attempts is exhausted
cap_backend	int	cv2.CAP_ANY	OpenCV backend flag
cap_buffer_size	int | None	None	Sets CAP_PROP_BUFFERSIZE right after opening (reduces RTSP latency)

Exponential back-off example

stream = VideoStream(
    "rtsp://camera-ip/stream",
    reconnect_delay=1.0,
    reconnect_backoff=2.0,       # 1s → 2s → 4s → 8s …
    max_reconnect_delay=30.0,    # capped at 30s
    max_reconnect_attempts=10,
)
stream.start()

Callbacks

def on_connect():
    print("Camera back online")

def on_disconnect():
    print("Feed dropped — waiting to reconnect")

def on_failure():
    print("Gave up reconnecting")
    # trigger an alert, restart a service, etc.

stream = VideoStream(
    "rtsp://camera-ip/stream",
    on_connect=on_connect,
    on_disconnect=on_disconnect,
    on_reconnect_failure=on_failure,
)
stream.start()

---

Properties

stream.fps            # float
stream.frame_width    # int
stream.frame_height   # int
stream.is_connected   # bool
stream.reconnect_count  # int — total reconnect attempts so far

---

Logging

The library uses Python's standard logging module under the rtsp_stream logger. To see reconnect events:

import logging
logging.basicConfig(level=logging.INFO)

---

Running tests

python3 -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"
pytest -q

License

MIT