testing_deployment/code/server_lib/asr_system_servicer.py

__all__= ['ASRSystemServicer','init']

import grpc,threading,traceback
import logging as Log

import mllp_grpc.asr_system_pb2_grpc as asr_system_pb2_grpc
import mllp_grpc.asr_system_pb2 as asr_system_pb2
import mllp_grpc.asr_common_pb2 as asr_common_pb2
import google.protobuf.empty_pb2 as empty_pb2

def ii_(func,txt,context):
    Log.info('[%s] (%s) %s'%(context.peer(),func,txt))

class ASRSystemServicer(asr_system_pb2_grpc.ASRSystemServicer):

    def __init__(self,system):
        self._sys= system

    def GetSystemInfo(self,request,context):
        def ii(txt): ii_('GetSystemInfo',txt,context)
        ii("Retrieving info from ASR system")

        langs=[asr_common_pb2.SystemInfo.Lang(code=x[0],
                                              text=x[1])
               for x in self._sys.lang]
        date= asr_common_pb2.SystemInfo.Date(day=self._sys.date[0],
                                             month=self._sys.date[1],
                                             year=self._sys.date[2])
        info= asr_common_pb2.SystemInfo(id=self._sys.name,
                                        langs=langs,
                                        tag=self._sys.tag,
                                        date=date)
        num_recos= self._sys.num_recos_available
        ret= asr_system_pb2.GetSystemInfoResponse(info=info,
                                                  num_decoders=len(self._sys),
                                                  num_decoders_available=num_recos,
                                                  enabled=self._sys.enabled)
        return ret
        
    def Decode(self,request_iterator,context):
        def ii(txt): ii_('Decode',txt,context)
        def ee(code,txt):
            status= asr_common_pb2.DecodeResponse.Status(code=code,
                                                         details=txt)
            return asr_common_pb2.DecodeResponse(status=status)
        # Check system selected
        # Check recogniser available
        reco= self._sys.get_reco()
        if reco is None:
            return ee(asr_common_pb2.DecodeResponse.Status.Code.ERR_NO_RECO_AVAILABLE,
                      "no recogniser available for ASR system '%s'"%self._sys.name)
        code= asr_common_pb2.DecodeResponse.Status.Code.READY
        status= asr_common_pb2.DecodeResponse.Status(code=code)
        yield asr_common_pb2.DecodeResponse(status=status)
        ii('starting decoding with system: %s'%self._sys.name)
        # Create a thread to feed data
        def feed_data():
            try:
                for d in request_iterator:
                    if d.HasField('token'):
                        reco.feed(d.token)
                    else:
                        reco.feed(d.data)
            except Exception as e:
                traceback.print_exc()
                ii('[feed data] client closed connection, recognition stoped')
            reco.feed()
        t= threading.Thread(target=feed_data)
        t.start()
        # Process output
        try:
            for o in reco.output:
                if o.novar is None: # EOS or ERR
                    if o.err==True: # Err
                        yield ee(asr_common_pb2.DecodeResponse.Status.Code.ERR_RECO,
                                 "an unexpected error ocurred during recognition")
                    else: pass # nothing on EOS
                else:
                    val= asr_common_pb2.DecodeResponse.Status.Code.OK
                    status= asr_common_pb2.DecodeResponse.Status(code=val)
                    yield asr_common_pb2.DecodeResponse(status=status,
                                                        hyp_novar=o.novar,
                                                        hyp_var=o.var,
                                                        score=o.score,
                                                        nframes=o.nframes,
                                                        eos=o.eos)
        except:
            traceback.print_exc()
            ii('[read output] client closed connection, recognition stoped')
            for o in reco.output:
                pass
        # Wait thread (should be stopped at this point) and return reco
        t.join()
        self._sys.append(reco)
        ii('decoding finished')
    # end Decode

    def SetEnabled(self,request,context):
        def ii(txt): ii_('SetEnabled',txt,context)
        ii("system %s"%('enabled' if request.value else 'disabled'))
        self._sys.set_enabled(request.value)
        ret= empty_pb2.Empty()
        return ret
        
def init(server,system):
    asr_system_pb2_grpc.add_ASRSystemServicer_to_server(
        ASRSystemServicer(system),server)