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)