441 lines
14 KiB
Python
441 lines
14 KiB
Python
import json
|
||
import logging
|
||
import requests
|
||
from .settings import *
|
||
|
||
from datetime import datetime, timedelta
|
||
from pytz import timezone
|
||
|
||
|
||
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.lazy_login = True
|
||
|
||
if self.lazy_login:
|
||
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")
|
||
|
||
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?include=timeZone"
|
||
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?include=timeZone"
|
||
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?include=status"
|
||
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 = [Camera(id=i['id'], name=i['name'], status=i['status'], account_id=i['accountId'], bridge_id=i['bridgeId']) 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['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}"
|
||
|
||
|
||
|
||
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 = []
|
||
|
||
|
||
|