EE-status-v3/EagleEyev3/__init__.py

import json
import logging
import requests
from flask import Flask, request
from .settings import *

logging.basicConfig(level=logging.INFO)

class EagleEyev3():
    """
    Class representing the EagleEyev3 client.
    """

    def __init__(self):
        """
        Initializes the EagleEyev3 client object.
        """
        self.client_id = None
        self.client_secret = None
        self.access_token = None
        self.refresh_token = None
        self.redirect_uri = None

        self._load_vars_from_settings()

        self.user_base_url = None
        self.current_user = None
        self.users = []
        self.bridges = []
        self.cameras = []
        self.switches = []
        self.users = []
        self.accounts = []

        self.base_url = None

        self.lazy_login = True

        if self.lazy_login:
            self._load_access_token()

    def _load_vars_from_settings(self):
        """
        Load variables from the settings module.
        """
        self.client_id = settings.client_id
        self.client_secret = settings.client_secret
        self.server_host = settings.server_host
        self.server_port = settings.server_port

        # Combine server_protocol, server_host, and server_port to make the redirect_uri
        # Note: Please see the note in settings.py about trailing slashes and modify this line if needed
        self.redirect_uri = f"{settings.server_protocol}://{settings.server_host}:{settings.server_port}"

    def _save_access_token(self):
        with open(".lazy_login", "w") as json_file:
            json.dump({
                'access_token': self.access_token,
                'refresh_token': self.refresh_token,
                'current_user': self.current_user
            }, json_file)

    def _load_access_token(self):
        with open(".lazy_login", "r") as json_file:
            saved = json.load(json_file)
            if 'access_token' in saved:
                self.access_token = saved['access_token']
            if 'refresh_token' in saved:
                self.refresh_token = saved['refresh_token']

        self.get_base_url(cascade=True)


    def login_tokens(self, code=None, cascade=True):
        """
        Obtains login tokens using the authorization code.

        Args:
            code (str): The authorization code.
            cascade (bool): Indicates whether to cascade and get the base URL and current user information.

        Returns:
            dict: Dictionary containing the success status, response HTTP status code, data, and current user information.
        """
        baseUrl = "https://auth.eagleeyenetworks.com/oauth2/token"
        pathUrl = f"?grant_type=authorization_code&scope=vms.all&code={code}&redirect_uri={self.redirect_uri}" # Note the trailing slash, make sure it matches the whitelist
        url = baseUrl + pathUrl

        # Send a POST request to obtain login tokens
        response = requests.post(url, auth=(self.client_id, self.client_secret))
        response_json = response.json()

        logging.info(f"{response.status_code} in login_tokens")

        if response.status_code == 200:
            success = True
            self.access_token = response_json['access_token']
            self.refresh_token = response_json['refresh_token']

            if self.lazy_login:
                self._save_access_token()

            if cascade:
                self.get_base_url()
        else:
            success = False

        return {
            "success": success,
            "response_http_status": response.status_code,
            "data": response_json,
            'current_user': self.current_user
        }

    def get_base_url(self, cascade=True):
        """
        Obtains the base URL for the user.

        Args:
            cascade (bool): Indicates whether to cascade and get the current user information.

        Returns:
            dict: Dictionary containing the success status, response HTTP status code, and data.
        """
        url = "https://api.eagleeyenetworks.com/api/v3.0/clientSettings"
        headers = { 
                    "Authorization": f"Bearer {self.access_token}", 
                    "Accept": "application/json"
                }

        # Send a GET request to obtain the base URL
        response = requests.get(url, headers=headers)
        response_json = response.json()

        logging.info(f"{response.status_code} in get_base_url")

        if response.status_code == 200:
            success = True
            if 'httpsBaseUrl' in response_json and 'hostname' in response_json['httpsBaseUrl']:
                self.user_base_url = response_json['httpsBaseUrl']['hostname']
                if cascade:
                    self.get_current_user()
        else:
            success = False

        return {
            "success": success,
            "response_http_status": response.status_code,
            "data": response_json
        }

    def get_current_user(self):
        """
        Obtains the information of the current user.

        Returns:
            dict: Dictionary containing the success status, response HTTP status code, and data.
        """
        url = f"https://{self.user_base_url}/api/v3.0/users/self"
        headers = { 
                    "Authorization": f"Bearer {self.access_token}", 
                    "Accept": "application/json"
                }

        # Send a GET request to obtain the current user information
        response = requests.get(url, headers=headers)
        response_json = response.json()

        logging.info(f"{response.status_code} in get_current_user")

        if response.status_code == 200:
            success = True
            self.current_user = response_json
        else:
            success = False

        return {
            "success": success,
            "response_http_status": response.status_code,
            "data": response_json
        }

    def get_list_of_users(self):
        """
        Obtains the list of users.

        Returns:
            dict: Dictionary containing the success status, response HTTP status code, and data.
        """
        url = f"https://{self.user_base_url}/api/v3.0/users"
        headers = { 
                    "Authorization": f"Bearer {self.access_token}", 
                    "Accept": "application/json"
                }

        response = requests.get(url, headers=headers)
        response_json = response.json()

        logging.info(f"{response.status_code} in get_list_of_users")

        if response.status_code == 200:
            success = True
            self.users = [i for i in response_json['results']]
        else:
            success = False

        return {
            "success": success,
            "response_http_status": response.status_code,
            "data": response_json
        }

    def get_list_of_cameras(self):
        """
        Obtains the list of cameras.

        Returns:
            dict: Dictionary containing the success status, response HTTP status code, and data.
        """
        url = f"https://{self.user_base_url}/api/v3.0/cameras"
        headers = { 
                    "Authorization": f"Bearer {self.access_token}", 
                    "Accept": "application/json"
                }

        response = requests.get(url, headers=headers)
        response_json = response.json()

        logging.info(f"{response.status_code} in get_list_of_cameras")

        if response.status_code == 200:
            success = True
            self.cameras = [i for i in response_json['results']]
        else:
            success = False

        return {
            "success": success,
            "response_http_status": response.status_code,
            "data": response_json
        }

    def get_list_of_bridges(self):
        """
        Obtains the list of bridges.

        Returns:
            dict: Dictionary containing the success status, response HTTP status code, and data.
        """
        url = f"https://{self.user_base_url}/api/v3.0/bridges"
        headers = { 
                    "Authorization": f"Bearer {self.access_token}", 
                    "Accept": "application/json"
                }

        response = requests.get(url, headers=headers)
        response_json = response.json()

        logging.info(f"{response.status_code} in get_list_of_bridges")

        if response.status_code == 200:
            success = True
            self.bridges = [i for i in response_json['results']]
        else:
            success = False

        return {
            "success": success,
            "response_http_status": response.status_code,
            "data": response_json
        }

    def get_list_of_switches(self):
        """
        Obtains the list of switches.

        Returns:
            dict: Dictionary containing the success status, response HTTP status code, and data.
        """
        url = f"https://{self.user_base_url}/api/v3.0/switches"
        headers = { 
                    "Authorization": f"Bearer {self.access_token}", 
                    "Accept": "application/json"
                }

        response = requests.get(url, headers=headers)
        response_json = response.json()

        logging.info(f"{response.status_code} in get_list_of_switches")

        if response.status_code == 200:
            success = True
            self.switches = [i for i in response_json['results']]
        else:
            success = False

        return {
            "success": success,
            "response_http_status": response.status_code,
            "data": response_json
        }

    def get_list_of_available_devices(self, deviceType__in="camera"):
        """
        Obtains the list of available devices.

        Returns:
            dict: Dictionary containing the success status, response HTTP status code, and data.
        """
        url = f"https://{self.user_base_url}/api/v3.0/availableDevices?deviceType__in={deviceType__in}"
        headers = { 
                    "Authorization": f"Bearer {self.access_token}", 
                    "Accept": "application/json"
                }

        response = requests.get(url, headers=headers)
        response_json = response.json()

        logging.info(f"{response.status_code} in get_list_of_available_devices")

        if response.status_code == 200:
            success = True
        else:
            success = False

        return {
            "success": success,
            "response_http_status": response.status_code,
            "data": response_json
        }

    def get_list_of_multi_cameras(self):
        """
        Obtains the list of multi-cameras.

        Returns:
            dict: Dictionary containing the success status, response HTTP status code, and data.
        """
        url = f"https://{self.user_base_url}/api/v3.0/multiCameras"
        headers = { 
                    "Authorization": f"Bearer {self.access_token}", 
                    "Accept": "application/json"
                }

        response = requests.get(url, headers=headers)
        response_json = response.json()

        logging.info(f"{response.status_code} in get_list_of_multi_cameras")

        if response.status_code == 200:
            success = True
        else:
            success = False

        return {
            "success": success,
            "response_http_status": response.status_code,
            "data": response_json
        }

    def get_list_of_feeds(self):
        """
        Obtains the list of feeds.

        Returns:
            dict: Dictionary containing the success status, response HTTP status code, and data.
        """
        url = f"https://{self.user_base_url}/api/v3.0/feeds"
        headers = { 
                    "Authorization": f"Bearer {self.access_token}", 
                    "Accept": "application/json"
                }

        response = requests.get(url, headers=headers)
        response_json = response.json()

        logging.info(f"{response.status_code} in get_list_of_feeds")

        if response.status_code == 200:
            success = True
        else:
            success = False

        return {
            "success": success,
            "response_http_status": response.status_code,
            "data": response_json
        }


class Device():

    def __init__(self, id=None, name=None, status=None, account_id=None):

        self.id = id
        self.name = name
        self.status = status
        self.account_id = account_id

    def get_id(self):
        return self.id

    def get_status(self):
        return self.status



class Camera(Device):

    def __init__(self, id=None, name=None, status=None, account_id=None, bridge_id=None):
        super().__init__(id=id, name=name, status=status, account_id=account_id)
        self.bridge_id = bridge_id
        self.previews = []
        self.videos = []