Examples

The following examples illustrate the basics of writing a plugin, as well as the use of classes and methods.

  1. If a detected face contains a label ‘emo’, this plugin will request facen and emotions data extraction and then log the received data.

    import logging
    
    from ntech import sfapi_client
    
    from facerouter.plugin import Plugin
    
    logger = logging.getLogger(__name__)
    
    
    class LogEmoPlugin(Plugin):
        async def preprocess(self, request, labels):
            if labels.get('emo'):
                return ('facen', 'emotions')
    
        async def process(self, request, photo, bbox, event_id, detection: sfapi_client.DetectFace):
            logger.info('%r: %r', bbox, detection.features.get('emotions')[0]['emotion'])
            logger.info('%r: params: ', bbox)
            for param in request.params._fields:
                param_repr = repr(getattr(request.params, param))
                if len(param_repr) > 100:
                    param_repr = param_repr[:97] + "..."
                logger.info("%r: %s", param, param_repr)
    
    
    def activate(app, ctx, plugin_name, plugin_source):
        return LogEmoPlugin(ctx=ctx)
    
  2. This plugin requests facen extraction, after that it saves a face in the 'ppl' gallery of the biometric database. If such a gallery doesn’t exist, it will be created.

    import logging
    import PIL.Image
    import time
    from io import BytesIO
    
    from ntech import sfapi_client
    
    from facerouter.plugin import Plugin
    
    logger = logging.getLogger(__name__)
    
    
    class EnrollPlugin(Plugin):
        async def preprocess(self, request, labels):
            if labels.get('lol') == 'kek':
                return ('facen',)
    
        async def process(self, request, photo, bbox, event_id, detection: sfapi_client.DetectFace):
            img = PIL.Image.open(BytesIO(photo))
            thumb = img.crop(bbox)
            fname = '/tmp/%x.jpeg' % (event_id,)
            thumb.save(fname)
            while True:
                try:
                    await self.ctx.sfapi['ppl'].add(event_id, detection, meta={
                        'timestamp': int(time.time()),
                        'photo_hash': fname,
                    })
                except sfapi_client.SFApiRemoteError as e:
                    if e.code == "GALLERY_NOT_FOUND":
                        await self.ctx.sfapi['ppl'].create()
                    else:
                        raise
                else:
                    break
            logger.info('%r: %r %r', bbox, event_id, fname)
    
    
    def activate(app, ctx, plugin_name, plugin_source):
        return EnrollPlugin(ctx=ctx)