Add reset functionality successfully to the streaming process

[ywangwxd]

Hello, If you are like me and want to have a reset functionality, I would like to share my experiece here.

By reset functionality I mean this effect: 1, start the websocket server 2, send audio to server using diart.client or any other front-end inerface, e.g., a gradio app with microphone recording 3, now you want to process another audio by not interruptting the server

If you do not have a reset function, all the audio contents are assumed to be consecutive parts of a single large file. The timestamps and speaker labelling will be consistent. But having a reset function, you can use the websocket service to process audios which are independent from each other. To do so, I have added special logic to process a reset signal in the on_message_recevied callback function. When the server received a reset signal, it will do the following. This is the piece of code in sources.py

    def _on_message_received(
        self,
        client: Dict[Text, Any],
        server: WebsocketServer,
        message: AnyStr,
    ):
      ....#other logic
      if message=="reset":

            #get the StreamingInference object, which is passed to the source object 
            #when initializing the StreamingInference object.                
            inf=self.inf 
            if inf is None:
                print("StreamingInference object is None, reset failed")
                return

             # pipeline has buffer data internally, so you need to reset it
            inf.pipeline.reset()   

            # Redo the subscribe operation to make the stream pipe as a brand new one. 
            # Otherwise  the ops `buffer_with_count` will keep some buffer data
            inf.reset_subscriber() 

            return

I hope this can help you.

[juanmc2005]

I think this is solved with multi-client support (#263). Instead of resetting the pipeline with a message, you would create a new client.