How to use pyHik - 10 common examples

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 device info.')
            return None

        try:
            tree = ET.fromstring(response.text)
            # Try to fetch namespace from XML
            nmsp = tree.tag.split('}')[0].strip('{')
            self.namespace = nmsp if nmsp.startswith('http') else XML_NAMESPACE
            _LOGGING.debug('Using Namespace: %s', self.namespace)

            for item in tree:
                tag = item.tag.split('}')[1]
                device_info[tag] = item.text

            return device_info

        except AttributeError as err:
            _LOGGING.error('Entire response: %s', response.text)
            _LOGGING.error('There was a problem: %s', err)
            return None

def get_event_triggers(self):
        """
        Returns dict of supported events.
        Key = Event Type
        List = Channels that have that event activated
        """
        events = {}
        nvrflag = False
        event_xml = []

        url = '%s/ISAPI/Event/triggers' % self.root_url

        try:
            response = self.hik_request.get(url, timeout=CONNECT_TIMEOUT)
            if response.status_code == requests.codes.not_found:
                # Try alternate URL for triggers
                _LOGGING.debug('Using alternate triggers URL.')
                url = '%s/Event/triggers' % self.root_url
                response = self.hik_request.get(url)

        except (requests.exceptions.RequestException,
                requests.exceptions.ConnectionError) as err:
            _LOGGING.error('Unable to fetch events, error: %s', err)
            return None

        if response.status_code != 200:
            # If we didn't recieve 200, abort
            return None

        # pylint: disable=too-many-nested-blocks

def get_device_info(self):
        """Parse deviceInfo into dictionary."""
        device_info = {}
        url = '%s/ISAPI/System/deviceInfo' % 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/deviceInfo' % 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)

"""
                            If we got this far we found an event that we want
                            to track.
                            """
                            events.setdefault(ettype.text, []) \
                                .append(etchannel_num)

        except (AttributeError, ET.ParseError) as err:
            _LOGGING.error(
                'There was a problem finding an element: %s', err)
            return None

        if nvrflag:
            self.device_type = NVR_DEVICE
        else:
            self.device_type = CAM_DEVICE
        _LOGGING.debug('Processed %s as %s Device.',
                       self.cam_id, self.device_type)

        _LOGGING.debug('Found events: %s', events)
        self.hik_request.close()

        # Change back namespace if needed
        if self.temp_namespace is not None:
            self.namespace = self.temp_namespace
            _LOGGING.debug('Changing Namespace: %s', self.namespace)

        return events

self.usr = usr
        self.pwd = pwd
        self.cam_id = 0
        self.name = ''
        self.device_type = None
        self.motion_detection = None
        self._motion_detection_xml = None

        self.root_url = '{}:{}'.format(host, port)

        # Build requests session for main thread calls
        # Default to basic authentication. It will change to digest inside
        # get_device_info if basic fails
        self.hik_request = requests.Session()
        self.hik_request.auth = (usr, pwd)
        self.hik_request.headers.update(DEFAULT_HEADERS)

        # Define event stream processing thread
        self.kill_thrd = threading.Event()
        self.reset_thrd = threading.Event()
        self.thrd = threading.Thread(
            target=self.alert_stream, args=(self.reset_thrd, self.kill_thrd,))
        self.thrd.daemon = False

        # Callbacks
        self._updateCallbacks = []

        self.initialize()

    def __init__(self, host=None, port=DEFAULT_PORT,
                 usr=None, pwd=None):
        """Initialize device."""

        _LOGGING.debug("pyHik %s initializing new hikvision device at: %s",
                       __version__, host)

        self.event_states = {}

        self.watchdog = Watchdog(300.0, self.watchdog_handler)

        self.namespace = XML_NAMESPACE
        self.temp_namespace = None

        if not host:
            _LOGGING.error('Host not specified! Cannot continue.')
            return

self.element_query('notificationMethod'))
                        if ntype.text == 'center' or ntype.text == 'HTTP':
                            """
                            If we got this far we found an event that we want
                            to track.
                            """
                            events.setdefault(ettype.text, []) \
                                .append(etchannel_num)

        except (AttributeError, ET.ParseError) as err:
            _LOGGING.error(
                'There was a problem finding an element: %s', err)
            return None

        if nvrflag:
            self.device_type = NVR_DEVICE
        else:
            self.device_type = CAM_DEVICE
        _LOGGING.debug('Processed %s as %s Device.',
                       self.cam_id, self.device_type)

        _LOGGING.debug('Found events: %s', events)
        self.hik_request.close()

        # Change back namespace if needed
        if self.temp_namespace is not None:
            self.namespace = self.temp_namespace
            _LOGGING.debug('Changing Namespace: %s', self.namespace)

        return events

def alert_stream(self, reset_event, kill_event):
        """Open event stream."""
        _LOGGING.debug('Stream Thread Started: %s, %s', self.name, self.cam_id)
        start_event = False
        parse_string = ""
        fail_count = 0

        url = '%s/ISAPI/Event/notification/alertStream' % self.root_url

        # pylint: disable=too-many-nested-blocks
        while True:

            try:
                stream = self.hik_request.get(url, stream=True,
                                              timeout=(CONNECT_TIMEOUT,
                                                       READ_TIMEOUT))
                if stream.status_code == requests.codes.not_found:
                    # Try alternate URL for stream
                    url = '%s/Event/notification/alertStream' % self.root_url
                    stream = self.hik_request.get(url, stream=True)

                if stream.status_code != requests.codes.ok:
                    raise ValueError('Connection unsucessful.')
                else:
                    _LOGGING.debug('%s Connection Successful.', self.name)
                    fail_count = 0
                    self.watchdog.start()

                for line in stream.iter_lines():
                    # _LOGGING.debug('Processing line from %s', self.name)
                    # filter out keep-alive new lines
                    if line:

for key in device_info:
            if key == 'deviceName':
                self.name = device_info[key]
            elif key == 'deviceID':
                if len(device_info[key]) > 10:
                    self.cam_id = device_info[key]
                else:
                    self.cam_id = uuid.uuid4()

        events_available = self.get_event_triggers()
        if events_available:
            for event, channel_list in events_available.items():
                for channel in channel_list:
                    try:
                        self.event_states.setdefault(
                            SENSOR_MAP[event.lower()], []).append(
                                [False, channel, 0, datetime.datetime.now()])
                    except KeyError:
                        # Sensor type doesn't have a known friendly name
                        # We can't reliably handle it at this time...
                        _LOGGING.warning(
                            'Sensor type "%s" is unsupported.', event)

            _LOGGING.debug('Initialized Dictionary: %s', self.event_states)
        else:
            _LOGGING.debug('No Events available in dictionary.')

        self.get_motion_detection()

def __init__(self, host=None, port=DEFAULT_PORT,
                 usr=None, pwd=None):
        """Initialize device."""

        _LOGGING.debug("pyHik %s initializing new hikvision device at: %s",
                       __version__, host)

        self.event_states = {}

        self.watchdog = Watchdog(300.0, self.watchdog_handler)

        self.namespace = XML_NAMESPACE
        self.temp_namespace = None

        if not host:
            _LOGGING.error('Host not specified! Cannot continue.')
            return

        self.host = host
        self.usr = usr
        self.pwd = pwd
        self.cam_id = 0