Real time notification of Event

[NickWaterton]

I recently just implemented a realtime event notifier - for VideoMotion events, but you could use any event, or multiple events, it's a list.

Is anyone interested in implementing it here?

This is the basic code (part of a camera class):

def stream_motion_data(self):
        text = ''
        chars = 0
        content_len = 0
        api_URL = 'http://%s:%s/cgi-bin/eventManager.cgi?action=attach&codes=[VideoMotion]' % (self.ip, self.port)

        r=requests.get(api_URL, auth=HTTPDigestAuth(self.user, self.password), stream=True)
            if r.status_code != requests.codes.ok:
                self.log.error('bad connection: %s' % r.status_code)
                r.raise_for_status()

            for line in r.iter_lines():
                # filter out keep-alive new lines
                if line:
                    decoded_line = line.decode('utf-8')
                    if decoded_line.startswith('--'):  # start of new chunk
                        text = ''
                        chars = -1
                    else:
                        if 'Content' not in decoded_line:
                            text += '%s\n' % decoded_line
                            chars = len(text)
                            self.log.debug('current content length: %s' % chars)
                        elif 'Content-Length:' in decoded_line:
                            content_len = int(decoded_line.replace('Content-Length:','').strip())
                            self.log.debug('content length: %s' % content_len)
                        if chars == content_len:
                            self.log.debug('decoding lines')
                            self.decode_line(text)

def decode_line(self, lines):  
        self.log.info(lines)
        data = lines.split(';')
        try:
            regions = json.loads(data[-1].replace('data=',''))
            regions = regions.get('RegionName', [])
        except json.JSONDecodeError:
            regions = []

        self.log.debug(' '.join(data))
        if 'VideoMotion' in data[0]:
            if 'Start' in data[1]:
                self.log.info('Motion Detected, regions: %s' % (regions))

It's a bit rough, as I just hacked it together to trigger a light when motion is detected via the camera.

The output looks like this:

[I 2020-01-09 10:39:49,528] (MainThread) Motion Detected, regions: ['Region1', 'Region2']

you'd have to run it in a thread if you wanted to do anything else with the camera while it's running.

I don't want to submit a PR if no-one is interested.

[dougsland]

Yeah, I think everyone is interested in improvements please sent the pull request :) @NickWaterton

[NickWaterton]

Added the pull request #140

[pnbruckner]

Will be addressed in #146.

[pnbruckner]

146 has been merged.