受信した音声データを Ogg ファイルで出力できる機能を追加する

amazon_transcribe_handler.go

@@ -96,10 +96,13 @@ func (h *AmazonTranscribeHandler) ResetRetryCount() int {
 	return h.RetryCount
 }
 
-func (h *AmazonTranscribeHandler) Handle(ctx context.Context, opusCh chan opusChannel) (*io.PipeReader, error) {
+func (h *AmazonTranscribeHandler) Handle(ctx context.Context, opusCh chan opusChannel, header soraHeader) (*io.PipeReader, error) {
 	at := NewAmazonTranscribe(h.Config, h.LanguageCode, int64(h.SampleRate), int64(h.ChannelCount))
 
-	packetReader := opus2ogg(ctx, opusCh, h.SampleRate, h.ChannelCount, h.Config)
+	packetReader, err := opus2ogg(ctx, opusCh, h.SampleRate, h.ChannelCount, h.Config, header)
+	if err != nil {
+		return nil, err
+	}
 
 	stream, err := at.Start(ctx, packetReader)
 	if err != nil {

config.go

@@ -65,6 +65,9 @@ type Config struct {
 	SampleRate   int `ini:"audio_sample_rate"`
 	ChannelCount int `ini:"audio_channel_count"`
 
+	EnableOggFileOutput bool   `ini:"enable_ogg_file_output"`
+	OggDir              string `ini:"ogg_dir"`
+
 	DumpFile string `ini:"dump_file"`
 
 	LogDir              string `ini:"log_dir"`
@@ -173,6 +176,10 @@ func setDefaultsConfig(config *Config) {
 	if config.RetryIntervalMs == 0 {
 		config.RetryIntervalMs = DefaultRetryIntervalMs
 	}
+
+	if config.OggDir == "" {
+		config.OggDir = "."
+	}
 }
 
 func validateConfig(config *Config) error {

config_example.ini

@@ -55,6 +55,10 @@ retry_interval_ms = 100
 # aws の場合は IsPartial が false, gcp の場合は IsFinal が true の場合の最終的な結果のみを返す指定
 final_result_only = true
 
+# 受信した音声データを Ogg ファイルで保存するかどうかです
+enable_ogg_file_output = false
+# Ogg ファイルの保存先ディレクトリです
+ogg_dir = "."
 
 # 採用する結果の信頼スコアの最小値です（aws 指定時のみ有効）
 # minimum_confidence_score が 0.0 の場合は信頼スコアによるフィルタリングは無効です

handler.go

@@ -8,6 +8,8 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"os"
+	"path"
 	"strings"
 	"time"
 
@@ -41,6 +43,16 @@ func NewSuzuErrorResponse(err error) TranscriptionResult {
 	}
 }
 
+type soraHeader struct {
+	SoraChannelID string `header:"Sora-Channel-Id"`
+	SoraSessionID string `header:"sora-session-id"`
+	// SoraClientID        string `header:"sora-client-id"`
+	SoraConnectionID string `header:"sora-connection-id"`
+	// SoraAudioCodecType  string `header:"sora-audio-codec-type"`
+	// SoraAudioSampleRate int64  `header:"sora-audio-sample-rate"`
+	SoraAudioStreamingLanguageCode string `header:"sora-audio-streaming-language-code"`
+}
+
 func getServiceHandler(serviceType string, config Config, channelID, connectionID string, sampleRate uint32, channelCount uint16, languageCode string, onResultFunc any) (serviceHandlerInterface, error) {
 	newHandlerFunc, err := NewServiceHandlerFuncs.get(serviceType)
 	if err != nil {
@@ -65,15 +77,7 @@ func (s *Server) createSpeechHandler(serviceType string, onResultFunc func(conte
 			return echo.NewHTTPError(http.StatusBadRequest)
 		}
 
-		h := struct {
-			SoraChannelID string `header:"Sora-Channel-Id"`
-			// SoraSessionID       string `header:"sora-session-id"`
-			// SoraClientID        string `header:"sora-client-id"`
-			SoraConnectionID string `header:"sora-connection-id"`
-			// SoraAudioCodecType  string `header:"sora-audio-codec-type"`
-			// SoraAudioSampleRate int64  `header:"sora-audio-sample-rate"`
-			SoraAudioStreamingLanguageCode string `header:"sora-audio-streaming-language-code"`
-		}{}
+		h := soraHeader{}
 		if err := (&echo.DefaultBinder{}).BindHeaders(c, &h); err != nil {
 			zlog.Error().
 				Err(err).
@@ -153,7 +157,7 @@ func (s *Server) createSpeechHandler(serviceType string, onResultFunc func(conte
 			serviceHandlerCtx, cancelServiceHandler := context.WithCancel(ctx)
 			defer cancelServiceHandler()
 
-			reader, err := serviceHandler.Handle(serviceHandlerCtx, opusCh)
+			reader, err := serviceHandler.Handle(serviceHandlerCtx, opusCh, h)
 			if err != nil {
 				zlog.Error().
 					Err(err).
@@ -459,11 +463,26 @@ func readOpus(ctx context.Context, reader io.Reader) chan opusChannel {
 	return opusCh
 }
 
-func opus2ogg(ctx context.Context, opusCh chan opusChannel, sampleRate uint32, channelCount uint16, c Config) io.ReadCloser {
+func opus2ogg(ctx context.Context, opusCh chan opusChannel, sampleRate uint32, channelCount uint16, c Config, header soraHeader) (io.ReadCloser, error) {
 	oggReader, oggWriter := io.Pipe()
 
+	writers := []io.Writer{}
+	writers = append(writers, oggWriter)
+
+	if c.EnableOggFileOutput {
+		fileName := fmt.Sprintf("%s_%s.ogg", header.SoraSessionID, header.SoraConnectionID)
+		filePath := path.Join(c.OggDir, fileName)
+		f, err := os.Create(filePath)
+		if err != nil {
+			return nil, err
+		}
+		writers = append(writers, f)
+	}
+
+	multiWriter := io.MultiWriter(writers...)
+
 	go func() {
-		o, err := NewWith(oggWriter, sampleRate, channelCount)
+		o, err := NewWith(multiWriter, sampleRate, channelCount)
 		if err != nil {
 			oggWriter.CloseWithError(err)
 			return
@@ -501,7 +520,7 @@ func opus2ogg(ctx context.Context, opusCh chan opusChannel, sampleRate uint32, c
 		}
 	}()
 
-	return oggReader
+	return oggReader, nil
 }
 
 type opusRequest struct {

packet_dump_handler.go

@@ -67,7 +67,7 @@ func (h *PacketDumpHandler) ResetRetryCount() int {
 	return h.RetryCount
 }
 
-func (h *PacketDumpHandler) Handle(ctx context.Context, opusCh chan opusChannel) (*io.PipeReader, error) {
+func (h *PacketDumpHandler) Handle(ctx context.Context, opusCh chan opusChannel, header soraHeader) (*io.PipeReader, error) {
 	c := h.Config
 	filename := c.DumpFile
 	channelID := h.ChannelID

service_handler.go

@@ -15,7 +15,7 @@ var (
 )
 
 type serviceHandlerInterface interface {
-	Handle(context.Context, chan opusChannel) (*io.PipeReader, error)
+	Handle(context.Context, chan opusChannel, soraHeader) (*io.PipeReader, error)
 	UpdateRetryCount() int
 	GetRetryCount() int
 	ResetRetryCount() int

speech_to_text_handler.go

@@ -91,10 +91,13 @@ func (h *SpeechToTextHandler) ResetRetryCount() int {
 	return h.RetryCount
 }
 
-func (h *SpeechToTextHandler) Handle(ctx context.Context, opusCh chan opusChannel) (*io.PipeReader, error) {
+func (h *SpeechToTextHandler) Handle(ctx context.Context, opusCh chan opusChannel, header soraHeader) (*io.PipeReader, error) {
 	stt := NewSpeechToText(h.Config, h.LanguageCode, int32(h.SampleRate), int32(h.ChannelCount))
 
-	packetReader := opus2ogg(ctx, opusCh, h.SampleRate, h.ChannelCount, h.Config)
+	packetReader, err := opus2ogg(ctx, opusCh, h.SampleRate, h.ChannelCount, h.Config, header)
+	if err != nil {
+		return nil, err
+	}
 
 	stream, err := stt.Start(ctx, packetReader)
 	if err != nil {

test_handler.go

@@ -73,7 +73,7 @@ func (h *TestHandler) ResetRetryCount() int {
 	return h.RetryCount
 }
 
-func (h *TestHandler) Handle(ctx context.Context, opusCh chan opusChannel) (*io.PipeReader, error) {
+func (h *TestHandler) Handle(ctx context.Context, opusCh chan opusChannel, header soraHeader) (*io.PipeReader, error) {
 	r, w := io.Pipe()
 
 	reader := opusChannelToIOReadCloser(ctx, opusCh)