Java ws client support (#651)

zhaomingwork

2023-06-19 293e212128a6d3da9ed95294fe843a20d7da759d

Java ws client support (#651)

* add java websocket support

* make little changes

 funasr/runtime/java/FunasrWsClient.java

New file
@@ -0,0 +1,344 @@
//

// Copyright FunASR (https://github.com/alibaba-damo-academy/FunASR). All Rights

// Reserved. MIT License  (https://opensource.org/licenses/MIT)

//

/*

 * // 2022-2023 by zhaomingwork@qq.com

 */

// java FunasrWsClient

// usage:  FunasrWsClient [-h] [--port PORT] [--host HOST] [--audio_in AUDIO_IN] [--num_threads NUM_THREADS]

//                 [--chunk_size CHUNK_SIZE] [--chunk_interval CHUNK_INTERVAL] [--mode MODE]

package websocket;



import java.io.*;

import java.net.URI;

import java.net.URISyntaxException;

import java.nio.*;

import java.util.Map;

import net.sourceforge.argparse4j.ArgumentParsers;

import net.sourceforge.argparse4j.inf.ArgumentParser;

import net.sourceforge.argparse4j.inf.ArgumentParserException;

import net.sourceforge.argparse4j.inf.Namespace;

import org.java_websocket.client.WebSocketClient;

import org.java_websocket.drafts.Draft;

import org.java_websocket.handshake.ServerHandshake;

import org.json.simple.JSONArray;

import org.json.simple.JSONObject;

import org.json.simple.parser.JSONParser;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;



/** This example demonstrates how to connect to websocket server. */

public class FunasrWsClient extends WebSocketClient {



  public class RecWavThread extends Thread {

    private FunasrWsClient funasrClient;



    public RecWavThread(FunasrWsClient funasrClient) {

      this.funasrClient = funasrClient;

    }



    public void run() {

      this.funasrClient.recWav();

    }

  }



  private static final Logger logger = LoggerFactory.getLogger(FunasrWsClient.class);



  public FunasrWsClient(URI serverUri, Draft draft) {

    super(serverUri, draft);

  }



  public FunasrWsClient(URI serverURI) {

    super(serverURI);

  }



  public FunasrWsClient(URI serverUri, Map<String, String> httpHeaders) {

    super(serverUri, httpHeaders);

  }



  public void getSslContext(String keyfile, String certfile) {

    // TODO

    return;

  }



  // send json at first time

  public void sendJson(

      String mode, String strChunkSize, int chunkInterval, String wavName, boolean isSpeaking) {

    try {



      JSONObject obj = new JSONObject();

      obj.put("mode", mode);

      JSONArray array = new JSONArray();

      String[] chunkList = strChunkSize.split(",");

      for (int i = 0; i < chunkList.length; i++) {

        array.add(Integer.valueOf(chunkList[i].trim()));

      }



      obj.put("chunk_size", array);

      obj.put("chunk_interval", new Integer(chunkInterval));

      obj.put("wav_name", wavName);

      if (isSpeaking) {

        obj.put("is_speaking", new Boolean(true));

      } else {

        obj.put("is_speaking", new Boolean(false));

      }

      logger.info("sendJson: " + obj);

      // return;



      send(obj.toString());



      return;

    } catch (Exception e) {

      e.printStackTrace();

    }

  }



  // send json at end of wav

  public void sendEof() {

    try {

      JSONObject obj = new JSONObject();



      obj.put("is_speaking", new Boolean(false));



      logger.info("sendEof: " + obj);

      // return;



      send(obj.toString());

      iseof = true;

      return;

    } catch (Exception e) {

      e.printStackTrace();

    }

  }



  // function for rec wav file

  public void recWav() {

    sendJson(mode, strChunkSize, chunkInterval, wavName, true);

    File file = new File(FunasrWsClient.wavPath);



    int chunkSize = sendChunkSize;

    byte[] bytes = new byte[chunkSize];



    int readSize = 0;

    try (FileInputStream fis = new FileInputStream(file)) {

      if (FunasrWsClient.wavPath.endsWith(".wav")) {

        fis.read(bytes, 0, 44); //skip first 44 wav header

      }

      readSize = fis.read(bytes, 0, chunkSize);

      while (readSize > 0) {

        // send when it is chunk size

        if (readSize == chunkSize) {

          send(bytes); // send buf to server



        } else {

          // send when at last or not is chunk size

          byte[] tmpBytes = new byte[readSize];

          for (int i = 0; i < readSize; i++) {

            tmpBytes[i] = bytes[i];

          }

          send(tmpBytes);

        }

        // if not in offline mode, we simulate online stream by sleep

        if (!mode.equals("offline")) {

          Thread.sleep(Integer.valueOf(chunkSize / 32));

        }



        readSize = fis.read(bytes, 0, chunkSize);

      }



      if (!mode.equals("offline")) {

        // if not offline, we send eof and wait for 3 seconds to close

        Thread.sleep(2000);

        sendEof();

        Thread.sleep(3000);

        close();

      } else {

        // if offline, just send eof

        sendEof();

      }



    } catch (Exception e) {

      e.printStackTrace();

    }

  }



  @Override

  public void onOpen(ServerHandshake handshakedata) {



    RecWavThread thread = new RecWavThread(this);

    thread.start();

  }



  @Override

  public void onMessage(String message) {

    JSONObject jsonObject = new JSONObject();

    JSONParser jsonParser = new JSONParser();

    logger.info("received: " + message);

    try {

      jsonObject = (JSONObject) jsonParser.parse(message);

      logger.info("text: " + jsonObject.get("text"));

    } catch (org.json.simple.parser.ParseException e) {

      e.printStackTrace();

    }

    if (iseof && mode.equals("offline")) {

      close();

    }

  }



  @Override

  public void onClose(int code, String reason, boolean remote) {



    logger.info(

        "Connection closed by "

            + (remote ? "remote peer" : "us")

            + " Code: "

            + code

            + " Reason: "

            + reason);

  }



  @Override

  public void onError(Exception ex) {

    logger.info("ex: " + ex);

    ex.printStackTrace();

    // if the error is fatal then onClose will be called additionally

  }



  private boolean iseof = false;

  public static String wavPath;

  static String mode = "online";

  static String strChunkSize = "5,10,5";

  static int chunkInterval = 10;

  static int sendChunkSize = 1920;



  String wavName = "javatest";



  public static void main(String[] args) throws URISyntaxException {

    ArgumentParser parser = ArgumentParsers.newArgumentParser("ws client").defaultHelp(true);

    parser

        .addArgument("--port")

        .help("Port on which to listen.")

        .setDefault("8889")

        .type(String.class)

        .required(false);

    parser

        .addArgument("--host")

        .help("the IP address of server.")

        .setDefault("127.0.0.1")

        .type(String.class)

        .required(false);

    parser

        .addArgument("--audio_in")

        .help("wav path for decoding.")

        .setDefault("asr_example.wav")

        .type(String.class)

        .required(false);

    parser

        .addArgument("--num_threads")

        .help("num of threads for test.")

        .setDefault(1)

        .type(Integer.class)

        .required(false);

    parser

        .addArgument("--chunk_size")

        .help("chunk size for asr.")

        .setDefault("5, 10, 5")

        .type(String.class)

        .required(false);

    parser

        .addArgument("--chunk_interval")

        .help("chunk for asr.")

        .setDefault(10)

        .type(Integer.class)

        .required(false);



    parser

        .addArgument("--mode")

        .help("mode for asr.")

        .setDefault("offline")

        .type(String.class)

        .required(false);

    String srvIp = "";

    String srvPort = "";

    String wavPath = "";

    int numThreads = 1;

    String chunk_size = "";

    int chunk_interval = 10;

    String strmode = "offline";



    try {

      Namespace ns = parser.parseArgs(args);

      srvIp = ns.get("host");

      srvPort = ns.get("port");

      wavPath = ns.get("audio_in");

      numThreads = ns.get("num_threads");

      chunk_size = ns.get("chunk_size");

      chunk_interval = ns.get("chunk_interval");

      strmode = ns.get("mode");

      System.out.println(srvPort);



    } catch (ArgumentParserException ex) {

      ex.getParser().handleError(ex);

      return;

    }



    FunasrWsClient.strChunkSize = chunk_size;

    FunasrWsClient.chunkInterval = chunk_interval;

    FunasrWsClient.wavPath = wavPath;

    FunasrWsClient.mode = strmode;

    System.out.println(

        "serIp="

            + srvIp

            + ",srvPort="

            + srvPort

            + ",wavPath="

            + wavPath

            + ",strChunkSize"

            + strChunkSize);



    class ClientThread implements Runnable {



      String srvIp;

      String srvPort;



      ClientThread(String srvIp, String srvPort, String wavPath) {

        this.srvIp = srvIp;

        this.srvPort = srvPort;

      }



      public void run() {

        try {



          int RATE = 16000;

          String[] chunkList = strChunkSize.split(",");

          int int_chunk_size = 60 * Integer.valueOf(chunkList[1].trim()) / chunkInterval;

          int CHUNK = Integer.valueOf(RATE / 1000 * int_chunk_size);

          int stride =

              Integer.valueOf(

                  60 * Integer.valueOf(chunkList[1].trim()) / chunkInterval / 1000 * 16000 * 2);

          System.out.println("chunk_size:" + String.valueOf(int_chunk_size));

          System.out.println("CHUNK:" + CHUNK);

          System.out.println("stride:" + String.valueOf(stride));

          FunasrWsClient.sendChunkSize = CHUNK * 2;



          String wsAddress = "ws://" + srvIp + ":" + srvPort;



          FunasrWsClient c = new FunasrWsClient(new URI(wsAddress));



          c.connect();



          System.out.println("wsAddress:" + wsAddress);

        } catch (Exception e) {

          e.printStackTrace();

          System.out.println("e:" + e);

        }

      }

    }

    for (int i = 0; i < numThreads; i++) {

      System.out.println("Thread1 is running...");

      Thread t = new Thread(new ClientThread(srvIp, srvPort, wavPath));

      t.start();

    }

  }

}


 funasr/runtime/java/Makefile

New file
@@ -0,0 +1,76 @@


ENTRY_POINT = ./



 




WEBSOCKET_DIR:= ./

WEBSOCKET_FILES = \

    $(WEBSOCKET_DIR)/FunasrWsClient.java \

 




LIB_BUILD_DIR = ./lib





 


JAVAC = javac



BUILD_DIR = build





RUNJFLAGS = -Dfile.encoding=utf-8

	
	
vpath %.class $(BUILD_DIR)

vpath %.java src

	
	
 
	
rebuild: clean all

	
.PHONY:  clean run downjar



downjar:

    wget https://repo1.maven.org/maven2/org/slf4j/slf4j-api/1.7.25/slf4j-api-1.7.25.jar -P ./lib/

    wget https://repo1.maven.org/maven2/org/slf4j/slf4j-simple/1.7.25/slf4j-simple-1.7.25.jar -P ./lib/

    #wget https://github.com/TooTallNate/Java-WebSocket/releases/download/v1.5.3/Java-WebSocket-1.5.3.jar -P ./lib/

    wget https://repo1.maven.org/maven2/org/java-websocket/Java-WebSocket/1.5.3/Java-WebSocket-1.5.3.jar -P ./lib/

    wget https://storage.googleapis.com/google-code-archive-downloads/v2/code.google.com/json-simple/json-simple-1.1.1.jar -P ./lib/

    wget https://github.com/argparse4j/argparse4j/releases/download/argparse4j-0.9.0/argparse4j-0.9.0.jar -P ./lib/

    rm -frv build

    mkdir build

clean:

    rm -frv  $(BUILD_DIR)/*

    rm -frv  $(LIB_BUILD_DIR)/*

    mkdir -p $(BUILD_DIR)

    mkdir -p ./lib



	
 


 


runclient:

        java   -cp $(BUILD_DIR):lib/Java-WebSocket-1.5.3.jar:lib/slf4j-simple-1.7.25.jar:lib/slf4j-api-1.7.25.jar:lib/json-simple-1.1.1.jar:lib/argparse4j-0.9.0.jar   $(RUNJFLAGS)    websocket.FunasrWsClient --host localhost --port 8889 --audio_in ./asr_example.wav --num_threads 1 --mode 2pass



 


buildwebsocket: $(WEBSOCKET_FILES:.java=.class)

	


%.class: %.java



                $(JAVAC) -cp $(BUILD_DIR):lib/slf4j-simple-1.7.25.jar:lib/slf4j-api-1.7.25.jar:lib/Java-WebSocket-1.5.3.jar:lib/json-simple-1.1.1.jar:lib/argparse4j-0.9.0.jar -d $(BUILD_DIR) -encoding UTF-8 $<



packjar:

                jar cvfe lib/funasrclient.jar . -C $(BUILD_DIR) .

				
all: clean buildlib packjar buildfile buildmic downjar buildwebsocket



 







 funasr/runtime/java/readme.md

New file
@@ -0,0 +1,66 @@
# Client for java websocket example



 


## Building for Linux/Unix



### install java environment

```shell

#  in ubuntu

apt-get install openjdk-11-jdk

```



 


### Build and run by make





```shell

cd funasr/runtime/java

# download java lib

make downjar

# compile 
make buildwebsocket

# run client

make runclient



```



## Run java websocket client by shell



```shell

# full command refer to Makefile runclient

usage:  FunasrWsClient [-h] [--port PORT] [--host HOST] [--audio_in AUDIO_IN] [--num_threads NUM_THREADS]

                 [--chunk_size CHUNK_SIZE] [--chunk_interval CHUNK_INTERVAL] [--mode MODE]



Where:

   --host <string>

     (required)  server-ip



   --port <int>

     (required)  port



   --audio_in <string>

     (required)  the wav or pcm file path



   --num_threads <int>

     thread number for test

   
   --mode

     asr mode, support "offline" "online" "2pass"



 


example:

FunasrWsClient --host localhost --port 8889 --audio_in ./asr_example.wav --num_threads 1 --mode 2pass



result json, example like:

{"mode":"offline","text":"欢迎大家来体验达摩院推出的语音识别模型","wav_name":"javatest"}

```





## Acknowledge

1. This project is maintained by [FunASR community](https://github.com/alibaba-damo-academy/FunASR).

2. We acknowledge [zhaoming](https://github.com/zhaomingwork/FunASR/tree/java-ws-client-support/funasr/runtime/java) for contributing the java websocket client example.