libpyvivotek/vivotek.py
"""A python implementation of the Vivotek IB8369A"""
from textwrap import wrap
import requests
from requests.auth import HTTPBasicAuth
from requests.auth import HTTPDigestAuth
DEFAULT_EVENT_0_KEY = "event_i0_enable"
CGI_BASE_PATH = "/cgi-bin"
API_PATHS = {
"get": "/getparam.cgi",
"get_anon": "/getparam.cgi",
"set": "/setparam.cgi",
"still": "/video.jpg",
}
SECURITY_LEVELS = {
"anonymous": 0,
"viewer": 1,
"operator": 4,
"admin": 6
}
class VivotekCameraError(Exception):
"""Custom Error class for VivotekCamera"""
class VivotekCamera():
"""A Vivotek IB8369A camera object"""
# pylint: disable=too-many-instance-attributes
# pylint: disable=too-many-arguments
def __init__(self, host, sec_lvl, port=None, usr=None, pwd=None, digest_auth=False, ssl=None,
verify_ssl=True):
"""
Initialize a camera.
"""
self.host = host
if port is None:
self._port = 443 if ssl else 80
else:
self._port = port
if ssl or (self._port == 443 and ssl is None):
self._ssl = True
else:
self._ssl = False
if self._ssl is False:
self.verify_ssl = False
else:
self.verify_ssl = verify_ssl
if sec_lvl not in SECURITY_LEVELS:
raise VivotekCameraError(f"Invalid security level: {sec_lvl}")
if usr is None or sec_lvl == 'anonymous':
self._requests_auth = None
self._security_level = 'anonymous'
else:
self._security_level = sec_lvl
if digest_auth:
self._requests_auth = HTTPDigestAuth(usr, pwd)
else:
self._requests_auth = HTTPBasicAuth(usr, pwd)
self._model_name = None
_protocol = 'https' if self._ssl else 'http'
self._url_base = _protocol + "://" + self.host
if (ssl is None and port != 80) or (ssl and port != 443):
self._url_base += ":" + str(self._port)
self._cgi_url_base = self._url_base + CGI_BASE_PATH + "/" + self._security_level
self._get_param_url = self._cgi_url_base + API_PATHS["get"]
self._set_param_url = self._cgi_url_base + API_PATHS["set"]
self._still_image_url = self._url_base + CGI_BASE_PATH + "/viewer" + API_PATHS["still"]
def event_enabled(self, event_key):
"""Return true if event for the provided key is enabled."""
response = self.get_param(event_key)
return int(response.replace("'", "")) == 1
def snapshot(self, quality=3):
"""Return the bytes of current still image."""
try:
response = requests.get(
self._still_image_url,
params={"quality": quality},
auth=self._requests_auth,
timeout=10,
verify=self.verify_ssl,
)
return response.content
except requests.exceptions.RequestException as error:
raise VivotekCameraError from error
def get_mac(self):
"""Return the MAC address with colons"""
return ":".join(wrap(self.get_serial(), 2))
def get_serial(self):
"""Return the serial number which is also the MAC address."""
return self.get_param('system_info_serialnumber')
def get_param(self, param):
"""Return the value of the provided key."""
request_args = {
"params": param,
"timeout": 10,
"verify": self.verify_ssl
}
if self._requests_auth is not None:
request_args['auth'] = self._requests_auth
try:
response = requests.get(self._get_param_url, **request_args)
return self.__parse_response_value(response)
except requests.exceptions.RequestException as error:
raise VivotekCameraError from error
def set_param(self, param, value):
"""Set and return the value of the provided key."""
if SECURITY_LEVELS[self._security_level] < 4:
raise VivotekCameraError(
f"Security level {self._security_level} is too low to set parameters.")
try:
response = requests.post(
self._set_param_url,
auth=self._requests_auth,
data={param: value},
timeout=10,
verify=self.verify_ssl,
)
return self.__parse_response_value(response)
except requests.exceptions.RequestException as error:
raise VivotekCameraError from error
@property
def model_name(self):
"""Return the model name of the camera."""
if self._model_name is not None:
return self._model_name
self._model_name = self.get_param("system_info_modelname")
return self._model_name
@staticmethod
def __parse_response_value(response):
"""
Parse the response from an API call and return the value only.
This assumes the response is in the key='value' format.
An error will be raised when ERROR is found in the body of the response.
"""
if 'ERROR' in response.text:
raise VivotekCameraError(response.text)
if response.status_code == 401:
raise VivotekCameraError('Unauthorized. Credentials may be invalid.')
return response.text.strip().split('=')[1].replace("'", "")