2023-05-18 15:40:02 +00:00
import json
import logging
import requests
from . settings import *
2023-05-30 20:24:23 +00:00
from datetime import datetime , timedelta
from pytz import timezone
2023-05-18 15:40:02 +00:00
logging . basicConfig ( level = logging . INFO )
2023-05-30 20:24:23 +00:00
2023-05-18 15:40:02 +00:00
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 = [ ]
2023-06-01 15:28:19 +00:00
self . user_tz_obj = None
2023-05-18 15:40:02 +00:00
2023-05-30 01:28:23 +00:00
self . lazy_login = True
if self . lazy_login :
2023-05-30 13:14:59 +00:00
try :
self . _load_access_token ( )
except FileNotFoundError as e :
logging . warn ( " self.lazy_login is set to {self.lazy_login} but could not find .lazy_login file to load " )
2023-05-30 01:28:23 +00:00
2023-05-18 15:40:02 +00:00
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 } "
2023-05-30 01:28:23 +00:00
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 )
2023-06-01 15:28:19 +00:00
def time_now ( self ) :
return datetime . now ( tz = self . user_tz_obj ) . isoformat ( timespec = ' milliseconds ' )
def time_before ( self , ts = None , hours = 6 ) :
if ts == None :
ts = datetime . now ( tz = self . user_tz_obj )
if type ( ts ) == str :
ts = datetime . fromisoformat ( ts )
return ( ts - timedelta ( hours = hours ) ) . isoformat ( timespec = ' milliseconds ' )
2023-05-18 15:40:02 +00:00
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 ' ]
2023-05-30 01:28:23 +00:00
if self . lazy_login :
self . _save_access_token ( )
2023-05-18 15:40:02 +00:00
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-30 20:24:23 +00:00
url = f " https:// { self . user_base_url } /api/v3.0/users/self?include=timeZone "
2023-05-18 15:40:02 +00:00
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 .
"""
2023-05-30 20:24:23 +00:00
url = f " https:// { self . user_base_url } /api/v3.0/users?include=timeZone "
2023-05-29 21:33:44 +00:00
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 .
"""
2023-05-30 01:54:14 +00:00
url = f " https:// { self . user_base_url } /api/v3.0/cameras?include=status "
2023-05-29 21:33:44 +00:00
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
2023-06-01 04:16:11 +00:00
self . cameras = [
Camera ( id = i [ ' id ' ] , \
name = i [ ' name ' ] , \
status = i [ ' status ' ] , \
account_id = i [ ' accountId ' ] , \
bridge_id = i [ ' bridgeId ' ] , \
user_base_url = self . user_base_url , \
een_instance = self )
for i in response_json [ ' results ' ] ]
for camera in self . cameras :
camera . user_base_url = self . user_base_url
2023-05-29 21:33:44 +00:00
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
}
2023-05-30 01:28:23 +00:00
2023-06-01 04:16:11 +00:00
2023-05-30 01:28:23 +00:00
class Device ( ) :
2023-06-01 04:16:11 +00:00
def __init__ ( self , id = None , name = None , status = None , account_id = None , user_base_url = None , een_instance = None ) :
2023-05-30 01:28:23 +00:00
self . id = id
self . name = name
self . status = status
self . account_id = account_id
2023-06-01 04:16:11 +00:00
self . user_base_url = user_base_url ,
self . een_instance = een_instance
2023-05-30 01:28:23 +00:00
def get_id ( self ) :
return self . id
def get_status ( self ) :
2023-05-30 01:54:14 +00:00
return self . status [ ' connectionStatus ' ]
def is_online ( self ) :
return self . status [ ' connectionStatus ' ] == " online "
def is_offline ( self ) :
return self . status [ ' connectionStatus ' ] == " deviceOffline " or self . status [ ' connectionStatus ' ] == " bridgeOffline "
def __repr__ ( self ) :
if self . is_online ( ) :
online = ' ✅ '
elif self . is_offline ( ) :
online = ' ❌ '
else :
online = ' ? '
return f " { online } [ { self . id } ] - { self . name } "
2023-05-30 01:28:23 +00:00
class Camera ( Device ) :
2023-06-01 04:16:11 +00:00
def __init__ ( self , id = None , name = None , status = None , account_id = None , bridge_id = None , user_base_url = None , een_instance = None ) :
super ( ) . __init__ ( id = id , name = name , status = status , account_id = account_id , user_base_url = user_base_url , een_instance = een_instance )
2023-05-30 01:28:23 +00:00
self . bridge_id = bridge_id
self . previews = [ ]
self . videos = [ ]
2023-06-01 04:16:11 +00:00
self . events = {
' status ' : [ ] ,
' motion ' : [ ]
}
2023-05-30 01:54:14 +00:00
2023-06-01 04:16:11 +00:00
def get_list_of_events ( self , start_timestamp = None , end_timestamp = None ) :
"""
Obtains the list of events .
Returns :
dict : Dictionary containing the success status , response HTTP status code , and data .
"""
if start_timestamp == None or end_timestamp == None :
return {
" success " : False ,
" response_http_status " : None ,
" data " : { ' msg ' : ' get_list_of_events called without required args, needs start_timestamp, end_timestamp ' }
}
url = f " https:// { self . user_base_url } /api/v3.0/events?pageSize=100&include=data.een.deviceCloudStatusUpdate.v1&startTimestamp__gte= { start_timestamp } &endTimestamp__lte= { end_timestamp } &actor=camera%3A { self . id } "
headers = {
" Authorization " : f " Bearer { self . een_instance . access_token } " ,
" Accept " : " application/json "
}
response = requests . get ( url , headers = headers )
response_json = response . json ( )
logging . debug ( f " { response . status_code } returned from { url } with { headers } " )
logging . info ( f " { response . status_code } in get_list_of_events " )
if response . status_code == 200 :
success = True
# filter events by type
[ self . events [ ' status ' ] . append ( i ) for i in response . json ( ) [ ' results ' ] if i [ ' type ' ] == ' een.deviceCloudStatusUpdateEvent.v1 ' ]
[ self . events [ ' motion ' ] . append ( i ) for i in response . json ( ) [ ' results ' ] if i [ ' type ' ] == ' een.motionDetectionEvent.v1 ' ]
# remove duplicates
seen = set ( )
self . events [ ' status ' ] = [ event for event in self . events [ ' status ' ] if event [ ' id ' ] not in seen and not seen . add ( event [ ' id ' ] ) ]
seen = set ( )
self . events [ ' motion ' ] = [ event for event in self . events [ ' motion ' ] if event [ ' id ' ] not in seen and not seen . add ( event [ ' id ' ] ) ]
else :
success = False
2023-05-30 01:28:23 +00:00
2023-06-01 04:16:11 +00:00
return {
" success " : success ,
" response_http_status " : response . status_code ,
" data " : response_json
}