reolink_api/reolinkapi/mixins/ptz.py

from typing import Dict
import time

class PtzAPIMixin:
    """
    API for PTZ functions.
    """
    def _send_operation(self, operation: str, speed: float, index: float = None) -> Dict:
        # Refactored to reduce redundancy
        param = {"channel": 0, "op": operation, "speed": speed}
        if index is not None:
            param['id'] = index
        data = [{"cmd": "PtzCtrl", "action": 0, "param": param}]
        return self._execute_command('PtzCtrl', data)

    def _send_noparm_operation(self, operation: str) -> Dict:
        data = [{"cmd": "PtzCtrl", "action": 0, "param": {"channel": 0, "op": operation}}]
        return self._execute_command('PtzCtrl', data)

    def _send_set_preset(self, enable: float, preset: float = 1, name: str = 'pos1') -> Dict:
        data = [{"cmd": "SetPtzPreset", "action": 0, "param": {
            "channel": 0, "enable": enable, "id": preset, "name": name}}]
        return self._execute_command('PtzCtrl', data)

    def go_to_preset(self, speed: float = 60, index: float = 1) -> Dict:
        """
        Move the camera to a preset location
        :return: response json
        """
        return self._send_operation('ToPos', speed=speed, index=index)

    def add_preset(self, preset: float = 1, name: str = 'pos1') -> Dict:
        """
        Adds the current camera position to the specified preset.
        :return: response json
        """
        return self._send_set_preset(enable=1, preset=preset, name=name)

    def remove_preset(self, preset: float = 1, name: str = 'pos1') -> Dict:
        """
        Removes the specified preset
        :return: response json
        """
        return self._send_set_preset(enable=0, preset=preset, name=name)

    def move_right(self, speed: float = 25) -> Dict:
        """
        Move the camera to the right
        The camera moves self.stop_ptz() is called.
        :return: response json
        """
        return self._send_operation('Right', speed=speed)

    def move_right_up(self, speed: float = 25) -> Dict:
        """
        Move the camera to the right and up
        The camera moves self.stop_ptz() is called.
        :return: response json
        """
        return self._send_operation('RightUp', speed=speed)

    def move_right_down(self, speed: float = 25) -> Dict:
        """
        Move the camera to the right and down
        The camera moves self.stop_ptz() is called.
        :return: response json
        """
        return self._send_operation('RightDown', speed=speed)

    def move_left(self, speed: float = 25) -> Dict:
        """
        Move the camera to the left
        The camera moves self.stop_ptz() is called.
        :return: response json
        """
        return self._send_operation('Left', speed=speed)

    def move_left_up(self, speed: float = 25) -> Dict:
        """
        Move the camera to the left and up
        The camera moves self.stop_ptz() is called.
        :return: response json
        """
        return self._send_operation('LeftUp', speed=speed)

    def move_left_down(self, speed: float = 25) -> Dict:
        """
        Move the camera to the left and down
        The camera moves self.stop_ptz() is called.
        :return: response json
        """
        return self._send_operation('LeftDown', speed=speed)

    def move_up(self, speed: float = 25) -> Dict:
        """
        Move the camera up.
        The camera moves self.stop_ptz() is called.
        :return: response json
        """
        return self._send_operation('Up', speed=speed)

    def move_down(self, speed: float = 25) -> Dict:
        """
        Move the camera down.
        The camera moves self.stop_ptz() is called.
        :return: response json
        """
        return self._send_operation('Down', speed=speed)

    def stop_ptz(self) -> Dict:
        """
        Stops the cameras current action.
        :return: response json
        """
        return self._send_noparm_operation('Stop')

    def auto_movement(self, speed: float = 25) -> Dict:
        """
        Move the camera in a clockwise rotation.
        The camera moves self.stop_ptz() is called.
        :return: response json
        """
        return self._send_operation('Auto', speed=speed)

    def get_zoom_focus_pos(self) -> Dict:
        body = [{"cmd": "GetZoomFocus", "action": 0, "param": {'channel':0}}]
        return self._execute_command('GetZoomFocus',body)

    def set_auto_focus_state(self, state: bool = True) -> Dict:
        body = [{"cmd": "SetAutoFocus", "action": 0, "param": {'AutoFocus':{'channel':0,'disable':int(state)}}}]
        return self._execute_command('SetAutoFocus',body)

    def get_auto_focus_state(self) -> bool:
        body = [{"cmd": "GetAutoFocus", "action": 1, "param": {'channel':0}}]
        return bool(self._execute_command('GetAutoFocus',body)[0]['value']['AutoFocus']['disable'])


    def set_zoom_pos(self, pos: int) -> Dict:
        body = [{"cmd": "StartZoomFocus", "action": 0, "param": {'ZoomFocus':{'channel':0,'pos':pos,'op':'ZoomPos'}}}]
        self._execute_command('StartZoomFocus',body)


    def set_focus_pos(self, pos: int) -> Dict:
        body = [{"cmd": "StartZoomFocus", "action": 0, "param": {'ZoomFocus':{'channel':0,'pos':pos,'op':'FocusPos'}}}]
        self._execute_command('StartZoomFocus',body)


    def generic_set_block(self, pos:int, get_func, set_func) -> int:
        

        change_index = 0
        last_pos = get_func()
        pos = min([pos,254])
        pos = max([pos, 0])

        target_pos = pos
        set_func(pos)
        for i in range(100):
            cpos = get_func()
            if cpos != last_pos:
                change_index = i

            if cpos == target_pos:
                break


            if (i - change_index) > 8:
                break

            last_pos = cpos
            time.sleep(0.25)

        return cpos

    def set_focus_pos_block(self, pos: int) -> int:
        return self.generic_set_block(pos, self.get_focus_pos, self.set_focus_pos)

    def set_zoom_pos_block(self, pos: int) -> int:
        return self.generic_set_block(pos, self.get_zoom_pos, self.set_zoom_pos)

    def get_zoom_pos(self) -> int:
        return self.get_zoom_focus_pos()[0]['value']['ZoomFocus']['zoom']['pos']

    def get_focus_pos(self) -> int:
        return self.get_zoom_focus_pos()[0]['value']['ZoomFocus']['focus']['pos']