2023-05-18 15:40:02 +00:00
import json
import logging
import requests
from flask import Flask , request
from . settings import *
logging . basicConfig ( level = logging . INFO )
class EagleEyev3 ( ) :
2023-05-29 21:33:44 +00:00
"""
Class representing the EagleEyev3 client .
"""
2023-05-18 15:40:02 +00:00
def __init__ ( self ) :
2023-05-29 21:33:44 +00:00
"""
Initializes the EagleEyev3 client object .
"""
2023-05-18 15:40:02 +00:00
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 = [ ]
2023-05-29 21:33:44 +00:00
self . switches = [ ]
2023-05-18 15:40:02 +00:00
self . users = [ ]
self . accounts = [ ]
self . base_url = None
def _load_vars_from_settings ( self ) :
2023-05-29 21:33:44 +00:00
"""
Load variables from the settings module .
"""
2023-05-18 15:40:02 +00:00
self . client_id = settings . client_id
self . client_secret = settings . client_secret
self . server_host = settings . server_host
self . server_port = settings . server_port
2023-05-18 15:43:13 +00:00
# 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
2023-05-18 15:40:02 +00:00
self . redirect_uri = f " { settings . server_protocol } :// { settings . server_host } : { settings . server_port } "
def login_tokens ( self , code = None , cascade = True ) :
2023-05-29 21:33:44 +00:00
"""
Obtains login tokens using the authorization code .
2023-05-18 15:43:13 +00:00
Args :
code ( str ) : The authorization code .
cascade ( bool ) : Indicates whether to cascade and get the base URL and current user information .
2023-05-29 21:33:44 +00:00
2023-05-18 15:43:13 +00:00
Returns :
dict : Dictionary containing the success status , response HTTP status code , data , and current user information .
"""
2023-05-18 15:40:02 +00:00
baseUrl = " https://auth.eagleeyenetworks.com/oauth2/token "
2023-05-18 15:43:13 +00:00
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
2023-05-18 15:40:02 +00:00
url = baseUrl + pathUrl
2023-05-18 15:43:13 +00:00
# Send a POST request to obtain login tokens
2023-05-18 15:40:02 +00:00
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 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 ) :
2023-05-29 21:33:44 +00:00
"""
Obtains the base URL for the user .
2023-05-18 15:43:13 +00:00
Args :
cascade ( bool ) : Indicates whether to cascade and get the current user information .
2023-05-29 21:33:44 +00:00
2023-05-18 15:43:13 +00:00
Returns :
dict : Dictionary containing the success status , response HTTP status code , and data .
"""
2023-05-18 15:40:02 +00:00
url = " https://api.eagleeyenetworks.com/api/v3.0/clientSettings "
headers = {
" Authorization " : f " Bearer { self . access_token } " ,
" Accept " : " application/json "
}
2023-05-18 15:43:13 +00:00
# Send a GET request to obtain the base URL
2023-05-18 15:40:02 +00:00
response = requests . get ( url , headers = headers )
response_json = response . json ( )
logging . info ( f " { response . status_code } in get_base_url " )
2023-05-18 15:43:13 +00:00
2023-05-18 15:40:02 +00:00
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 ) :
2023-05-29 21:33:44 +00:00
"""
Obtains the information of the current user .
2023-05-18 15:43:13 +00:00
Returns :
dict : Dictionary containing the success status , response HTTP status code , and data .
"""
2023-05-18 15:40:02 +00:00
url = f " https:// { self . user_base_url } /api/v3.0/users/self "
headers = {
" Authorization " : f " Bearer { self . access_token } " ,
" Accept " : " application/json "
}
2023-05-18 15:43:13 +00:00
# Send a GET request to obtain the current user information
2023-05-18 15:40:02 +00:00
response = requests . get ( url , headers = headers )
response_json = response . json ( )
logging . info ( f " { response . status_code } in get_current_user " )
2023-05-18 15:43:13 +00:00
2023-05-18 15:40:02 +00:00
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
}
2023-05-29 21:33:44 +00:00
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
}