For anyone else here is a rough draft of a Python script that allows you to capture uncompressed images. I did have to install the API library https://github.com/fwestenberg/reolink

import asyncio import json import os import time from datetime import datetime as dt import cv2 # https://github.com/fwestenberg/reolink from reolink.camera_api import Api os.environ["OPENCV_FFMPEG_CAPTURE_OPTIONS"] = "rtsp_transport;tcp" params = {"ip": "100.101.102.103", "uname": "admin", "pass": "myverystrongpassword875"} # replace above with actual settings class MyReolink: class LoginContextManager: """Handle login logout""" async def __aenter__(self): self.api = Api(params["ip"], 80, params["uname"], params["pass"]) await self.api.get_settings() await self.api.get_states() return self.api async def __aexit__(self, exc_type, exc_val, traceback): if exc_type: print(exc_type, exc_val, traceback) await self.api.logout() async def get_presets(self): """Get PTZ presets Returns: dict: name an id of saved presets """ async with self.LoginContextManager() as api: # await api.get_settings() await api.get_states() return api.ptz_presets.copy() async def get_image_data(self): """Print current zoom and focus values""" async with self.LoginContextManager() as api: setts = {} foc = api.get_focus() zoom = api.get_zoom() setts = {"cur_focus": foc, "cur_zoom": zoom} print(setts) return foc, zoom async def wait_settle(self): """After moving camera to next PTZ location allow time for zoom and focus to settle Returns: tuple: (last_focus, last_zoom) - values """ async with self.LoginContextManager() as api: last_focus = [-1, -2, -3, -4] last_zoom = [-1, -2, -3, -4] while len(set(last_focus)) != 1 or len(set(last_zoom)) != 1: await api.get_settings() last_focus.pop(0) last_zoom.pop(0) last_focus.append(api.get_focus()) last_zoom.append(api.get_zoom()) time.sleep(0.5) print(f"{last_focus} - {last_zoom}") return last_focus[0], last_zoom[0] async def take_photo(self, preset, dir="./"): """Extract an uncompressed photo from RTSP stream Args: preset (str): preset name dir (str, optional): save path - Defaults to "./". """ # create timestamp ts = lambda: dt.now().strftime("%m%d%Y_%H%M%S") print("reading image") # using video stream from IP Webcam url = f"rtsp://{params['uname']}:{params['pass']}@{params['ip']}:554/h264Preview_01_main" frame = None while frame is None: cap = cv2.VideoCapture(url) frame = cap.read()[1] cv2.imwrite(f"{dir}hires_{ts()}_{preset}.png", frame) print(f"Image Saved") return await self.get_image_data() async def goto_preset(self, idx, vals): """Move to a saved PTZ point Args: idx (str): index of PTZ point vals (dict): dict of final zoom/focus """ async with self.LoginContextManager() as api: await api.get_settings() await api.get_states() await api.set_autofocus(True) await api.set_ptz_command("ToPos", idx, speed=64) if vals: await api.set_focus(vals["focus"]) await api.set_zoom(vals["zoom"]) async def ptz_photos(self, dir="./", saved_zoom=None): """Take a photo at each PTZ point""" old_ptz_zoom_focus = None if saved_zoom: with open("03152023_162243_ptz_zoom_focus.json", "r") as f: old_ptz_zoom_focus = json.load(f) presets = await self.get_presets() ptz_zoom_focus = {} for preset, idx in presets.items(): ptz_zoom_focus[preset] = {} print("\n", preset) if old_ptz_zoom_focus: await self.goto_preset(idx, old_ptz_zoom_focus[preset]) else: await self.goto_preset(idx, None) print("done") focus, zoom = await self.take_photo(preset, dir) ptz_zoom_focus[preset]["focus"] = focus ptz_zoom_focus[preset]["zoom"] = zoom with open("ptz_zoom_focus.json", "w") as f: json.dump(ptz_zoom_focus, f) print({(x["focus"], x["zoom"]) for x in ptz_zoom_focus.values()}) async def hires(self, dir): """Save a high res photo. Press enter at prompt or q to quit Args: dir (str): save directory """ while True: if input("Press ENTER to take a photo (q to quit): ") == "q": return await self.take_photo("rtsp", dir) async def snap(self, dir): """Save a compressed photo. Press enter at prompt or q to quit Args: dir (str): save directory """ ts = lambda: dt.now().strftime("%m%d%Y_%H%M%S") async with self.LoginContextManager() as api: while True: if input("Press ENTER to take a photo (q to quit): ") == "q": return img = await api.get_snapshot() with open(f"{dir}snap_{ts()}.jpg", "wb") as f: f.write(img) reo = MyReolink() highres = False use_saved_zoom_settings = False save_dir = "./repeatability test/" if highres: asyncio.run(reo.hires(save_dir)) else: asyncio.run(reo.snap(save_dir)) if use_saved_zoom_settings: asyncio.run(reo.ptz_photos(saved_zoom="03152023_162243_ptz_zoom_focus.json", dir=save_dir)) else: asyncio.run(reo.ptz_photos(dir=save_dir))

S3 Object Storage