Allow users to disable jpeg compression or use png

hidara

Hi,
Currently I have found two ways to capture still images

The inbuild default image capture(snapshot) - highly compressed
A python script that extracts a high resolution image from the RTSP stream

The first approach adds compression that I do not want. The second approach is perfect, but it means I have to set up a separate Linux machine that uses Python to extract stills from the RTSP stream.
Is there any way to disable jpeg compression for FTP images? I am not worried about the increased file size but the compression used for default snapshots makes this camera not fit for purpose.

See image below that shows the difference between the two approaches named above.
undefined

hidara

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://[censored]/fwestenberg/reolink

import asyncio
import json
import os
import time
from datetime import datetime as dt


import cv2


# https://[censored]/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))