MXL Audio Flow reader

internal/source/source.go

@@ -164,3 +164,109 @@ func (s *Source) Width() uint32          { return s.width }
 func (s *Source) Height() uint32         { return s.height }
 func (s *Source) GrainCount() uint32     { return s.info.Config.Discrete.GrainCount }
 func (s *Source) Format() mxl.DataFormat { return s.info.Config.Common.Format }
+
+type AudioSource struct {
+	inst  *mxl.Instance
+	r     *mxl.Reader
+	info  mxl.FlowInfo
+	rate  mxl.Rational
+	chans uint64
+	idx   uint64
+}
+
+type AudioFrame struct {
+	Index       uint64
+	SampleCount uint64
+	Channels    uint64
+	Samples     [][]byte // per-channel byte slices (F32, deinterleaved)
+}
+
+func OpenAudio(domain, flowID string) (*AudioSource, error) {
+	inst, err := mxl.NewInstance(domain, "")
+	if err != nil {
+		return nil, fmt.Errorf("NewInstance: %w", err)
+	}
+	r, err := inst.NewReader(flowID)
+	if err != nil {
+		inst.Close()
+		return nil, fmt.Errorf("NewReader: %w", err)
+	}
+	info, err := r.Info()
+	if err != nil {
+		r.Close()
+		inst.Close()
+		return nil, fmt.Errorf("Info: %w", err)
+	}
+	if info.Config.Common.Format.IsDiscrete() {
+		r.Close()
+		inst.Close()
+		return nil, fmt.Errorf("flow is discrete (not audio)")
+	}
+	idx := info.Runtime.HeadIndex
+	if idx == 0 {
+		r.Close()
+		inst.Close()
+		return nil, fmt.Errorf("flow has no head yet (no producer?)")
+	}
+	return &AudioSource{
+		inst:  inst,
+		r:     r,
+		info:  info,
+		rate:  info.Config.Common.GrainRate,
+		chans: uint64(info.Config.Continuous.ChannelCount),
+		idx:   idx,
+	}, nil
+}
+
+func (s *AudioSource) NextAudio(ctx context.Context, batch uint64, timeout time.Duration) (AudioFrame, error) {
+	for {
+		select {
+		case <-ctx.Done():
+			return AudioFrame{}, ctx.Err()
+		default:
+		}
+		v, err := s.r.GetSamples(s.idx, int(batch), timeout)
+		switch {
+		case err == nil:
+			samples := make([][]byte, s.chans)
+			for ch := uint64(0); ch < s.chans; ch++ {
+				f1, f2, _ := v.ChannelFragments(ch)
+				if len(f2) > 0 {
+					samples[ch] = append(f1, f2...)
+				} else {
+					samples[ch] = f1
+				}
+			}
+			f := AudioFrame{
+				Index:       s.idx,
+				SampleCount: batch,
+				Channels:    s.chans,
+				Samples:     samples,
+			}
+			s.idx += batch
+			return f, nil
+		case errors.Is(err, mxl.ErrOutOfRangeEarly):
+			select {
+			case <-time.After(10 * time.Millisecond):
+			case <-ctx.Done():
+				return AudioFrame{}, ctx.Err()
+			}
+		case errors.Is(err, mxl.ErrOutOfRangeLate):
+			rt, rerr := s.r.Runtime()
+			if rerr != nil {
+				return AudioFrame{}, fmt.Errorf("Runtime: %w", rerr)
+			}
+			s.idx = rt.HeadIndex
+		default:
+			return AudioFrame{}, fmt.Errorf("GetSamples: %w", err)
+		}
+	}
+}
+
+func (s *AudioSource) Close() error {
+	_ = s.r.Close()
+	return s.inst.Close()
+}
+
+func (s *AudioSource) Rate() mxl.Rational { return s.rate }
+func (s *AudioSource) Channels() uint64   { return s.chans }