From d0413d659cfc0f68bca3bb877603586ab6df7964 Mon Sep 17 00:00:00 2001 From: Ayush Pratap Date: Thu, 9 Jan 2020 17:50:29 +0530 Subject: [PATCH 1/4] Added support for : get_device_status get_device_time get_get_upnp_ports_status --- pyhik/hikvision.py | 178 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 178 insertions(+) diff --git a/pyhik/hikvision.py b/pyhik/hikvision.py index 79bcd03..ed8ed5a 100755 --- a/pyhik/hikvision.py +++ b/pyhik/hikvision.py @@ -18,6 +18,7 @@ try: import xml.etree.cElementTree as ET + from bs4 import BeautifulSoup except ImportError: import xml.etree.ElementTree as ET @@ -430,6 +431,183 @@ def get_device_info(self): _LOGGING.error('There was a problem: %s', err) return None + # Get the status of the device : /ISAPI/System/status + def get_device_status(self): + """Parse deviceInfo into dictionary.""" + device_status = {} + url = '%s/ISAPI/System/status' % self.root_url + using_digest = False + + try: + response = self.hik_request.get(url, timeout=CONNECT_TIMEOUT) + if response.status_code == requests.codes.unauthorized: + _LOGGING.debug('Basic authentication failed. Using digest.') + self.hik_request.auth = HTTPDigestAuth(self.usr, self.pwd) + using_digest = True + response = self.hik_request.get(url) + + if response.status_code == requests.codes.not_found: + # Try alternate URL for deviceInfo + _LOGGING.debug('Using alternate deviceInfo URL.') + url = '%s/System/status' % self.root_url + response = self.hik_request.get(url) + # Seems to be difference between camera and nvr, they can't seem to + # agree if they should 404 or 401 first + if not using_digest and response.status_code == requests.codes.unauthorized: + _LOGGING.debug('Basic authentication failed. Using digest.') + self.hik_request.auth = HTTPDigestAuth(self.usr, self.pwd) + using_digest = True + response = self.hik_request.get(url) + + except (requests.exceptions.RequestException, + requests.exceptions.ConnectionError) as err: + _LOGGING.error('Unable to fetch status of device, error: %s', err) + return None + + if response.status_code == requests.codes.unauthorized: + _LOGGING.error('Authentication failed') + return None + + if response.status_code != requests.codes.ok: + # If we didn't receive 200, abort + _LOGGING.debug('Unable to fetch status of device.') + return None + + try: + soup = BeautifulSoup(response.text, 'lxml') + + device_status['currentdevicetime'] = soup.devicestatus.currentdevicetime.string.strip() + device_status['deviceuptime'] = soup.devicestatus.deviceuptime.string.strip() + device_status['cpudescription'] = soup.devicestatus.cpulist.cpu.cpudescription.string.strip() + device_status['cpuutilization'] = soup.devicestatus.cpulist.cpu.cpuutilization.string.strip() + device_status['memorydescription'] = soup.devicestatus.memorylist.memory.memorydescription.string.strip() + device_status['memoryusage'] = soup.devicestatus.memorylist.memory.memoryusage.string.strip() + device_status['memoryavailable'] = soup.devicestatus.memorylist.memory.memoryavailable.string.strip() + + return device_status + + except AttributeError as err: + _LOGGING.error('Entire response: %s', response.text) + _LOGGING.error('There was a problem: %s', err) + return None + + # Get the time from the device : /ISAPI/System/time + def get_device_time(self): + """ + Parse device time into dictionary + """ + device_time = {} + url = "%s/ISAPI/System/time" % self.root_url + using_digest = False + + try: + response = self.hik_request.get(url, timeout=CONNECT_TIMEOUT) + if response.status_code == requests.codes.unauthorized: + _LOGGING.debug('Basic authentication failed. Using digest.') + self.hik_request.auth = HTTPDigestAuth(self.usr, self.pwd) + using_digest = True + response = self.hik_request.get(url) + + if response.status_code == requests.codes.not_found: + """ + Tyr alternate URL for time from device + """ + _LOGGING.debug('Using alternate URL for device from time') + url = '%s/System/time' % self.root_url + response = self.hik_request.get(url) + + if not using_digest and response.status_code == requests.codes.unauthorized: + _LOGGING.debug('Basic authentication failed. Using digest.') + self.hik_request.auth = HTTPDigestAuth(self.usr, self.pwd) + using_digest = True + response = self.hik_request.get(url) + + except (requests.exceptions.RequestException, requests.exceptions.ConnectionError) as err: + _LOGGING.error('Unable to fetch time of device, error: %s', err) + return None + + if response.status_code == requests.codes.unauthorized: + _LOGGING.error('Authentication failed') + return None + + if response.status_code != requests.codes.ok: + # If we didn't receive 200, abort + _LOGGING.debug('Unable to fetch time of device') + return None + try: + soup = BeautifulSoup(response.text, 'lxml') + + device_time['timemode'] = soup.time.timemode.string + device_time['localtime'] = soup.time.localtime.string + device_time['timezone'] = soup.time.timezone.string + + return device_time + + except AttributeError as err: + _LOGGING.error('Entire response: %s', response.text) + _LOGGING.error('There was a problem: %s', err) + + # Get the upnp ports : /ISAPI/System/Network/UPnP/ports/status + # Currently user can get information about , `http` , `rtsp`, `https` ports + def get_upnp_ports_status(self): + """ + Parse the upnp port status into dictionary + """ + upnp_port = {} + url = "%s/ISAPI/System/Network/UPnP/ports/status" % self.root_url + using_digest = False + + try: + response = self.hik_request.get(url, timeout=CONNECT_TIMEOUT) + if response.status_code == requests.codes.unauthorized: + _LOGGING.debug('basic authentication failed. Using digest.') + self.hik_request.auth = HTTPDigestAuth(self.usr, self.pwd) + using_digest = True + response = self.hik_request.get(url) + + if response.status_code == requests.codes.not_found: + """ + Try alternate URL + """ + _LOGGING.debug('Using alternate URL for Upnp ports') + url = '%s/System/time' % self.root_url + response = self.hik_request.get(url) + + if not using_digest and response.status_code == requests.codes.unauthorized: + _LOGGING.debug('Basic authentication failed. Using digest') + self.hik_request.auth = HTTPDigestAuth(self.usr, self.pwd) + using_digest = True + response = self.hik_request.get(url) + except (requests.exceptions.RequestException, requests.exceptions.ConnectionError) as err: + _LOGGING.error('Unable to fetch upnp ports, error: %s', err) + return None + + if response.status_code == requests.codes.unauthorized: + _LOGGING.error('Authentication failed') + return None + + if response.status_code != requests.codes.ok: + # If we didn't receive 200, abort + _LOGGING.debug('Unable to fetch upnp ports') + return None + try: + soup = BeautifulSoup(response.text, 'lxml') + soup.prettify() + portstatus = soup.find_all('portstatus') + # Parse portstatus + for t in portstatus: + if t.internalport.text == "http": + upnp_port['httpPort'] = t.externalport.text + if t.internalport.text == "rtsp": + upnp_port['rtspPort'] = t.externalport.text + if t.internalport.text == 'https': + upnp_port['httpsPort'] = t.externalport.text + + return upnp_port + except AttributeError as err: + _LOGGING.error('Entire response: %s', response.text) + _LOGGING.error('There was a problem: %s', err) + def watchdog_handler(self): """Take care of threads if wachdog expires.""" _LOGGING.debug('%s Watchdog expired. Resetting connection.', self.name) From 110463e3fc0451571551fdfa607cce7a94a4d7c4 Mon Sep 17 00:00:00 2001 From: Ayush Pratap Date: Fri, 17 Jan 2020 05:02:32 +0530 Subject: [PATCH 2/4] Update .gitignore --- .gitignore | 1 + 1 file changed, 1 insertion(+) diff --git a/.gitignore b/.gitignore index c86a0fb..883d474 100755 --- a/.gitignore +++ b/.gitignore @@ -68,3 +68,4 @@ target/ bin/ local/ share/ +.idea/* From 0b3b900c65e7fc7141df98f769aed1fa5ff2ffd7 Mon Sep 17 00:00:00 2001 From: Ayush Pratap Date: Fri, 17 Jan 2020 05:02:59 +0530 Subject: [PATCH 3/4] Removed BeautifulSoup dependency --- pyhik/hikvision.py | 54 +++++++++++++++++++++++----------------------- 1 file changed, 27 insertions(+), 27 deletions(-) diff --git a/pyhik/hikvision.py b/pyhik/hikvision.py index ed8ed5a..2263801 100755 --- a/pyhik/hikvision.py +++ b/pyhik/hikvision.py @@ -18,7 +18,6 @@ try: import xml.etree.cElementTree as ET - from bs4 import BeautifulSoup except ImportError: import xml.etree.ElementTree as ET @@ -243,6 +242,7 @@ def initialize(self): device_info = self.get_device_info() if device_info is None: + print('Hello this is it') self.name = None self.cam_id = None self.event_states = None @@ -402,6 +402,7 @@ def get_device_info(self): except (requests.exceptions.RequestException, requests.exceptions.ConnectionError) as err: _LOGGING.error('Unable to fetch deviceInfo, error: %s', err) + print('This is test') return None if response.status_code == requests.codes.unauthorized: @@ -431,7 +432,11 @@ def get_device_info(self): _LOGGING.error('There was a problem: %s', err) return None - # Get the status of the device : /ISAPI/System/status + """ + This function is added by Ayush Pratap Singh (ayushs56@gmail.com) + This function is used to get the status of the device + """ + def get_device_status(self): """Parse deviceInfo into dictionary.""" device_status = {} @@ -474,15 +479,15 @@ def get_device_status(self): return None try: - soup = BeautifulSoup(response.text, 'lxml') + tree = ET.fromstring(response.text) - device_status['currentdevicetime'] = soup.devicestatus.currentdevicetime.string.strip() - device_status['deviceuptime'] = soup.devicestatus.deviceuptime.string.strip() - device_status['cpudescription'] = soup.devicestatus.cpulist.cpu.cpudescription.string.strip() - device_status['cpuutilization'] = soup.devicestatus.cpulist.cpu.cpuutilization.string.strip() - device_status['memorydescription'] = soup.devicestatus.memorylist.memory.memorydescription.string.strip() - device_status['memoryusage'] = soup.devicestatus.memorylist.memory.memoryusage.string.strip() - device_status['memoryavailable'] = soup.devicestatus.memorylist.memory.memoryavailable.string.strip() + device_status['currentdevicetime'] = tree[0].text.strip() + device_status['deviceuptime'] = tree[1].text.strip() + device_status['cpudescription'] = tree[2][0][0].text.strip() + device_status['cpuutilization'] = tree[2][0][1].text.strip() + device_status['memorydescription'] = tree[3][0][0].text.strip() + device_status['memoryusage'] = tree[3][0][1].text.strip() + device_status['memoryavailable'] = tree[3][0][2].text.strip() return device_status @@ -491,7 +496,6 @@ def get_device_status(self): _LOGGING.error('There was a problem: %s', err) return None - # Get the time from the device : /ISAPI/System/time def get_device_time(self): """ Parse device time into dictionary @@ -535,11 +539,11 @@ def get_device_time(self): _LOGGING.debug('Unable to fetch time of device') return None try: - soup = BeautifulSoup(response.text, 'lxml') + tree = ET.fromstring(response.text) - device_time['timemode'] = soup.time.timemode.string - device_time['localtime'] = soup.time.localtime.string - device_time['timezone'] = soup.time.timezone.string + device_time['timemode'] = tree[0].text.strip() + device_time['localtime'] = tree[1].text.strip() + device_time['timezone'] = tree[2].text.strip() return device_time @@ -547,8 +551,10 @@ def get_device_time(self): _LOGGING.error('Entire response: %s', response.text) _LOGGING.error('There was a problem: %s', err) - # Get the upnp ports : /ISAPI/System/Network/UPnP/ports/status - # Currently user can get information about , `http` , `rtsp`, `https` ports + """ + Get the Ports using the endpoint : /ISAPI/System/Network/UPnP/ports/status + """ + #TODO : This needs to be implemented def get_upnp_ports_status(self): """ Parse the upnp port status into dictionary @@ -591,17 +597,11 @@ def get_upnp_ports_status(self): _LOGGING.debug('Unable to fetch upnp ports') return None try: - soup = BeautifulSoup(response.text, 'lxml') - soup.prettify() - portstatus = soup.find_all('portstatus') + tree = ET.fromstring(response.text) # Parse portstatus - for t in portstatus: - if t.internalport.text == "http": - upnp_port['httpPort'] = t.externalport.text - if t.internalport.text == "rtsp": - upnp_port['rtspPort'] = t.externalport.text - if t.internalport.text == 'https': - upnp_port['httpsPort'] = t.externalport.text + upnp_port['httpPort'] = tree[3][0][3].text.strip() + upnp_port['rtspPort'] = tree[3][2][3].text.strip() + upnp_port['httpsPort'] = tree[3][3][3].text.strip() return upnp_port except AttributeError as err: From 301fa1cdef802965770b3c3325e7463edb7c174e Mon Sep 17 00:00:00 2001 From: Ayush Pratap Date: Thu, 23 Jan 2020 15:43:32 +0530 Subject: [PATCH 4/4] Remove unwanted statements --- pyhik/hikvision.py | 11 ----------- 1 file changed, 11 deletions(-) diff --git a/pyhik/hikvision.py b/pyhik/hikvision.py index 2263801..ce04e13 100755 --- a/pyhik/hikvision.py +++ b/pyhik/hikvision.py @@ -242,7 +242,6 @@ def initialize(self): device_info = self.get_device_info() if device_info is None: - print('Hello this is it') self.name = None self.cam_id = None self.event_states = None @@ -402,7 +401,6 @@ def get_device_info(self): except (requests.exceptions.RequestException, requests.exceptions.ConnectionError) as err: _LOGGING.error('Unable to fetch deviceInfo, error: %s', err) - print('This is test') return None if response.status_code == requests.codes.unauthorized: @@ -432,11 +430,6 @@ def get_device_info(self): _LOGGING.error('There was a problem: %s', err) return None - """ - This function is added by Ayush Pratap Singh (ayushs56@gmail.com) - This function is used to get the status of the device - """ - def get_device_status(self): """Parse deviceInfo into dictionary.""" device_status = {} @@ -551,10 +544,6 @@ def get_device_time(self): _LOGGING.error('Entire response: %s', response.text) _LOGGING.error('There was a problem: %s', err) - """ - Get the Ports using the endpoint : /ISAPI/System/Network/UPnP/ports/status - """ - #TODO : This needs to be implemented def get_upnp_ports_status(self): """ Parse the upnp port status into dictionary