diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 72c78c729..05ccbe449 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -18,6 +18,7 @@ import ( "sync/atomic" "time" + "github.com/sipeed/picoclaw/pkg/asr" "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/channels" "github.com/sipeed/picoclaw/pkg/commands" @@ -31,7 +32,6 @@ import ( "github.com/sipeed/picoclaw/pkg/state" "github.com/sipeed/picoclaw/pkg/tools" "github.com/sipeed/picoclaw/pkg/utils" - "github.com/sipeed/picoclaw/pkg/voice" ) type AgentLoop struct { @@ -51,7 +51,7 @@ type AgentLoop struct { fallback *providers.FallbackChain channelManager *channels.Manager mediaStore media.MediaStore - transcriber voice.Transcriber + transcriber asr.Transcriber cmdRegistry *commands.Registry mcp mcpRuntime hookRuntime hookRuntime @@ -1040,7 +1040,7 @@ func (al *AgentLoop) SetMediaStore(s media.MediaStore) { } // SetTranscriber injects a voice transcriber for agent-level audio transcription. -func (al *AgentLoop) SetTranscriber(t voice.Transcriber) { +func (al *AgentLoop) SetTranscriber(t asr.Transcriber) { al.transcriber = t } diff --git a/pkg/voice/transcriber.go b/pkg/asr/asr.go similarity index 99% rename from pkg/voice/transcriber.go rename to pkg/asr/asr.go index 15f7fc0d0..1e2b73500 100644 --- a/pkg/voice/transcriber.go +++ b/pkg/asr/asr.go @@ -1,4 +1,4 @@ -package voice +package asr import ( "bytes" diff --git a/pkg/voice/transcriber_test.go b/pkg/asr/asr_test.go similarity index 99% rename from pkg/voice/transcriber_test.go rename to pkg/asr/asr_test.go index 20ba5388b..c6d04c08b 100644 --- a/pkg/voice/transcriber_test.go +++ b/pkg/asr/asr_test.go @@ -1,4 +1,4 @@ -package voice +package asr import ( "testing" diff --git a/pkg/audio/ogg.go b/pkg/audio/ogg.go new file mode 100644 index 000000000..49a2b93f9 --- /dev/null +++ b/pkg/audio/ogg.go @@ -0,0 +1,55 @@ +package audio + +import ( + "bytes" + "fmt" + "io" +) + +// DecodeOggOpus reads an Ogg format stream and extracts individual Opus payloads. +// It calls onFrame for every complete Opus frame found in the stream. +func DecodeOggOpus(r io.Reader, onFrame func([]byte) error) error { + var packet []byte + header := make([]byte, 27) + + for { + if _, err := io.ReadFull(r, header); err != nil { + if err == io.EOF || err == io.ErrUnexpectedEOF { + return nil + } + return fmt.Errorf("failed to read ogg header: %w", err) + } + if string(header[:4]) != "OggS" { + return fmt.Errorf("invalid ogg magic string") + } + + pageSegments := int(header[26]) + segmentTable := make([]byte, pageSegments) + if _, err := io.ReadFull(r, segmentTable); err != nil { + return fmt.Errorf("failed to read segment table: %w", err) + } + + for _, lacing := range segmentTable { + segment := make([]byte, lacing) + if _, err := io.ReadFull(r, segment); err != nil { + return fmt.Errorf("failed to read segment data: %w", err) + } + + packet = append(packet, segment...) + + // If lacing is less than 255, the packet is complete + if lacing < 255 { + if len(packet) > 0 { + // Ignore Ogg Opus headers + if !bytes.HasPrefix(packet, []byte("OpusHead")) && !bytes.HasPrefix(packet, []byte("OpusTags")) { + if err := onFrame(packet); err != nil { + return err + } + } + // Start new packet + packet = nil + } + } + } + } +} diff --git a/pkg/bus/types.go b/pkg/bus/types.go index 9c637f3e7..e036b4ede 100644 --- a/pkg/bus/types.go +++ b/pkg/bus/types.go @@ -68,6 +68,6 @@ type AudioChunk struct { // VoiceControl represents state or commands for voice sessions. type VoiceControl struct { SessionID string `json:"session_id"` - Type string `json:"type"` // "state", "command" + Type string `json:"type"` // "state", "command" Action string `json:"action"` // "idle", "listening", "start", "stop" } diff --git a/pkg/channels/discord/discord.go b/pkg/channels/discord/discord.go index 06a0f175f..ebdbd1437 100644 --- a/pkg/channels/discord/discord.go +++ b/pkg/channels/discord/discord.go @@ -20,8 +20,8 @@ import ( "github.com/sipeed/picoclaw/pkg/identity" "github.com/sipeed/picoclaw/pkg/logger" "github.com/sipeed/picoclaw/pkg/media" + "github.com/sipeed/picoclaw/pkg/tts" "github.com/sipeed/picoclaw/pkg/utils" - "github.com/sipeed/picoclaw/pkg/voice" ) const ( @@ -44,7 +44,7 @@ type DiscordChannel struct { typingStop map[string]chan struct{} // chatID → stop signal botUserID string // stored for mention checking bus *bus.MessageBus - tts voice.TTSProvider + tts tts.TTSProvider } func NewDiscordChannel(cfg config.DiscordConfig, bus *bus.MessageBus) (*DiscordChannel, error) { @@ -93,7 +93,7 @@ func (c *DiscordChannel) Start(ctx context.Context) error { c.botUserID = botUser.ID c.session.AddHandler(c.handleMessage) - + go c.listenVoiceControl(c.ctx) if err := c.session.Open(); err != nil { diff --git a/pkg/channels/discord/init.go b/pkg/channels/discord/init.go index 13e4fbc91..e6be8ff36 100644 --- a/pkg/channels/discord/init.go +++ b/pkg/channels/discord/init.go @@ -4,14 +4,14 @@ import ( "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/channels" "github.com/sipeed/picoclaw/pkg/config" - "github.com/sipeed/picoclaw/pkg/voice" + "github.com/sipeed/picoclaw/pkg/tts" ) func init() { channels.RegisterFactory("discord", func(cfg *config.Config, b *bus.MessageBus) (channels.Channel, error) { ch, err := NewDiscordChannel(cfg.Channels.Discord, b) if err == nil { - ch.tts = voice.DetectTTS(cfg) + ch.tts = tts.DetectTTS(cfg) } return ch, err }) diff --git a/pkg/gateway/gateway.go b/pkg/gateway/gateway.go index f9d1c824b..ad7591e3a 100644 --- a/pkg/gateway/gateway.go +++ b/pkg/gateway/gateway.go @@ -12,6 +12,7 @@ import ( "time" "github.com/sipeed/picoclaw/pkg/agent" + "github.com/sipeed/picoclaw/pkg/asr" "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/channels" _ "github.com/sipeed/picoclaw/pkg/channels/dingtalk" @@ -303,7 +304,7 @@ func setupAndStartServices( agentLoop.SetChannelManager(runningServices.ChannelManager) agentLoop.SetMediaStore(runningServices.MediaStore) - if transcriber := voice.DetectTranscriber(cfg); transcriber != nil { + if transcriber := asr.DetectTranscriber(cfg); transcriber != nil { agentLoop.SetTranscriber(transcriber) logger.InfoCF("voice", "Transcription enabled (agent-level)", map[string]any{"provider": transcriber.Name()}) @@ -538,7 +539,7 @@ func restartServices( fmt.Println(" ✓ Device event service restarted") } - transcriber := voice.DetectTranscriber(cfg) + transcriber := asr.DetectTranscriber(cfg) al.SetTranscriber(transcriber) if transcriber != nil { logger.InfoCF("voice", "Transcription re-enabled (agent-level)", map[string]any{"provider": transcriber.Name()}) diff --git a/pkg/voice/tts.go b/pkg/tts/tts.go similarity index 99% rename from pkg/voice/tts.go rename to pkg/tts/tts.go index 8de0bbc9c..63b4ecd24 100644 --- a/pkg/voice/tts.go +++ b/pkg/tts/tts.go @@ -1,4 +1,4 @@ -package voice +package tts import ( "bytes" diff --git a/pkg/voice/agent.go b/pkg/voice/agent.go index 6a2abd24f..6a330d3d4 100644 --- a/pkg/voice/agent.go +++ b/pkg/voice/agent.go @@ -11,6 +11,7 @@ import ( "github.com/pion/rtp" "github.com/pion/webrtc/v3/pkg/media/oggwriter" + "github.com/sipeed/picoclaw/pkg/asr" "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/logger" ) @@ -61,13 +62,13 @@ func (a *speechAccumulator) Close() { type Agent struct { bus *bus.MessageBus - transcriber Transcriber + transcriber asr.Transcriber mu sync.Mutex sessions map[string]*speechAccumulator // keyed by sessionID_speakerID } -func NewAgent(mb *bus.MessageBus, t Transcriber) *Agent { +func NewAgent(mb *bus.MessageBus, t asr.Transcriber) *Agent { return &Agent{ bus: mb, transcriber: t,