Added start of audio streaming

Author: djthorpe

pkg/whisper/api/logging.go

@@ -0,0 +1,26 @@
+package api
+
+import (
+	"log"
+	"net/http"
+	"sync/atomic"
+	"time"
+)
+
+var (
+	req int32
+)
+
+func wrapLogging(fn http.HandlerFunc) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		req := nextReq()
+		delta := time.Now()
+		log.Printf("R%d %s %s", req, r.Method, r.URL)
+		fn(w, r)
+		log.Printf("R%d Took %v", req, time.Since(delta).Truncate(time.Millisecond))
+	}
+}
+
+func nextReq() int32 {
+	return atomic.AddInt32(&req, 1)
+}

pkg/whisper/api/register.go

@@ -15,7 +15,7 @@ import (
 func RegisterEndpoints(base string, mux *http.ServeMux, whisper *whisper.Whisper) {
 	// Health: GET /v1/health
 	//   returns an empty OK response
-	mux.HandleFunc(joinPath(base, "health"), func(w http.ResponseWriter, r *http.Request) {
+	mux.HandleFunc(joinPath(base, "health"), wrapLogging(func(w http.ResponseWriter, r *http.Request) {
 		defer r.Body.Close()
 
 		switch r.Method {
@@ -24,14 +24,14 @@ func RegisterEndpoints(base string, mux *http.ServeMux, whisper *whisper.Whisper
 		default:
 			httpresponse.Error(w, http.StatusMethodNotAllowed)
 		}
-	})
+	}))
 
 	// List Models: GET /v1/models
 	//   returns available models
 	// Download Model: POST /v1/models?stream={bool}
 	//   downloads a model from the server
 	//   if stream is true then progress is streamed back to the client
-	mux.HandleFunc(joinPath(base, "models"), func(w http.ResponseWriter, r *http.Request) {
+	mux.HandleFunc(joinPath(base, "models"), wrapLogging(func(w http.ResponseWriter, r *http.Request) {
 		defer r.Body.Close()
 
 		switch r.Method {
@@ -42,13 +42,13 @@ func RegisterEndpoints(base string, mux *http.ServeMux, whisper *whisper.Whisper
 		default:
 			httpresponse.Error(w, http.StatusMethodNotAllowed)
 		}
-	})
+	}))
 
 	// Get: GET /v1/models/{id}
 	//   returns an existing model
 	// Delete: DELETE /v1/models/{id}
 	//   deletes an existing model
-	mux.HandleFunc(joinPath(base, "models/{id}"), func(w http.ResponseWriter, r *http.Request) {
+	mux.HandleFunc(joinPath(base, "models/{id}"), wrapLogging(func(w http.ResponseWriter, r *http.Request) {
 		defer r.Body.Close()
 
 		id := r.PathValue("id")
@@ -60,12 +60,12 @@ func RegisterEndpoints(base string, mux *http.ServeMux, whisper *whisper.Whisper
 		default:
 			httpresponse.Error(w, http.StatusMethodNotAllowed)
 		}
-	})
+	}))
 
 	// Translate: POST /v1/audio/translations
 	//   Translates audio into english or another language  - language parameter should be set to the
 	//   destination language of the audio. Will default to english if not set.
-	mux.HandleFunc(joinPath(base, "audio/translations"), func(w http.ResponseWriter, r *http.Request) {
+	mux.HandleFunc(joinPath(base, "audio/translations"), wrapLogging(func(w http.ResponseWriter, r *http.Request) {
 		defer r.Body.Close()
 
 		switch r.Method {
@@ -74,12 +74,12 @@ func RegisterEndpoints(base string, mux *http.ServeMux, whisper *whisper.Whisper
 		default:
 			httpresponse.Error(w, http.StatusMethodNotAllowed)
 		}
-	})
+	}))
 
 	// Transcribe: POST /v1/audio/transcriptions
 	//   Transcribes audio into the input language - language parameter should be set to the source
 	//   language of the audio
-	mux.HandleFunc(joinPath(base, "audio/transcriptions"), func(w http.ResponseWriter, r *http.Request) {
+	mux.HandleFunc(joinPath(base, "audio/transcriptions"), wrapLogging(func(w http.ResponseWriter, r *http.Request) {
 		defer r.Body.Close()
 
 		switch r.Method {
@@ -88,7 +88,21 @@ func RegisterEndpoints(base string, mux *http.ServeMux, whisper *whisper.Whisper
 		default:
 			httpresponse.Error(w, http.StatusMethodNotAllowed)
 		}
-	})
+	}))
+
+	// Transcribe: POST /v1/audio/transcriptions/{model-id}
+	//   Transcribes streamed media into the input language
+	mux.HandleFunc(joinPath(base, "audio/transcriptions/{model}"), wrapLogging(func(w http.ResponseWriter, r *http.Request) {
+		defer r.Body.Close()
+
+		model := r.PathValue("model")
+		switch r.Method {
+		case http.MethodPost:
+			TranscribeStream(r.Context(), whisper, w, r, model)
+		default:
+			httpresponse.Error(w, http.StatusMethodNotAllowed)
+		}
+	}))
 }
 
 /////////////////////////////////////////////////////////////////////////////

pkg/whisper/api/transcribe.go

@@ -122,7 +122,6 @@ func TranscribeFile(ctx context.Context, service *whisper.Whisper, w http.Respon
 		if translate {
 			result.Task = "translate"
 		}
-		result.Duration = schema.Timestamp(segmenter.Duration())
 		result.Language = task.Language()
 
 		// Output the header
@@ -142,6 +141,93 @@ func TranscribeFile(ctx context.Context, service *whisper.Whisper, w http.Respon
 			return err
 		}
 
+		// Set the language and duration
+		result.Language = task.Language()
+		result.Duration = schema.Timestamp(segmenter.Duration())
+
+		// Return success
+		return nil
+	}); err != nil {
+		if stream != nil {
+			stream.Write("error", err.Error())
+		} else {
+			httpresponse.Error(w, http.StatusInternalServerError, err.Error())
+		}
+		return
+	}
+
+	// Return transcription if not streaming
+	if stream == nil {
+		httpresponse.JSON(w, result, http.StatusOK, 2)
+	} else {
+		stream.Write("ok")
+	}
+}
+
+func TranscribeStream(ctx context.Context, service *whisper.Whisper, w http.ResponseWriter, r *http.Request, modelId string) {
+	var query queryTranscribe
+	if err := httprequest.Query(&query, r.URL.Query()); err != nil {
+		httpresponse.Error(w, http.StatusBadRequest, err.Error())
+		return
+	}
+
+	// Get the model
+	model := service.GetModelById(modelId)
+	if model == nil {
+		httpresponse.Error(w, http.StatusNotFound, "model not found")
+		return
+	}
+
+	// Create a segmenter - read segments based on 10 second segment size
+	segmenter, err := segmenter.New(r.Body, 10*time.Second, whisper.SampleRate)
+	if err != nil {
+		httpresponse.Error(w, http.StatusBadRequest, err.Error())
+		return
+	}
+
+	// Create a text stream
+	var stream *httpresponse.TextStream
+	if query.Stream {
+		if stream = httpresponse.NewTextStream(w); stream == nil {
+			httpresponse.Error(w, http.StatusInternalServerError, "Cannot create text stream")
+			return
+		}
+		defer stream.Close()
+	}
+
+	// Get context for the model, perform transcription
+	var result *schema.Transcription
+	if err := service.WithModel(model, func(task *task.Context) error {
+		// Set parameters for ttranslation, default to auto
+		task.SetTranslate(false)
+		if err := task.SetLanguage("auto"); err != nil {
+			return err
+		}
+
+		// TODO: Set temperature, etc
+
+		// Create response
+		result = task.Result()
+		result.Task = "transcribe"
+		result.Language = task.Language()
+
+		// Output the header
+		if stream != nil {
+			stream.Write("task", result)
+		}
+
+		// Read samples and transcribe them
+		if err := segmenter.Decode(ctx, func(ts time.Duration, buf []float32) error {
+			// Perform the transcription, output segments in realtime, return any errors
+			return task.Transcribe(ctx, ts, buf, stream != nil, func(segment *schema.Segment) {
+				if stream != nil {
+					stream.Write("segment", segment)
+				}
+			})
+		}); err != nil {
+			return err
+		}
+
 		// Set the language
 		result.Language = task.Language()