go-mxl-player/internal/source/source.go

package source

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	mxl "github.com/qvest-digital/go-mxl/mxl"
)

type flowDef struct {
	FrameWidth  int    `json:"frame_width"`
	FrameHeight int    `json:"frame_height"`
	MediaType   string `json:"media_type"`
	Colorspace  string `json:"colorspace"`
	GrainRate   struct {
		Numerator   int64 `json:"numerator"`
		Denominator int64 `json:"denominator"`
	} `json:"grain_rate"`
}

type Frame struct {
	Index   uint64
	Width   uint32
	Height  uint32
	Stride  uint32
	Size    uint32
	Invalid bool
	Payload []byte
}

type Source struct {
	inst   *mxl.Instance
	reader *mxl.Reader
	info   mxl.FlowInfo
	def    string
	rate   mxl.Rational
	stride uint32
	width  uint32
	height uint32
	idx    uint64
}

func Open(domain, flowID string) (*Source, 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)
	}
	def, err := inst.FlowDef(flowID)
	if err != nil {
		r.Close()
		inst.Close()
		return nil, fmt.Errorf("FlowDef: %w", err)
	}
	var fd flowDef
	if err := json.Unmarshal([]byte(def), &fd); err != nil {
		r.Close()
		inst.Close()
		return nil, fmt.Errorf("parse flow def: %w", err)
	}
	rate := info.Config.Common.GrainRate
	idx := mxl.CurrentIndex(rate)
	if idx == mxl.UndefinedIndex {
		r.Close()
		inst.Close()
		return nil, fmt.Errorf("invalid grain rate: %d/%d", rate.Num, rate.Den)
	}
	return &Source{
		inst:   inst,
		reader: r,
		info:   info,
		def:    def,
		rate:   info.Config.Common.GrainRate,
		stride: info.Config.Discrete.SliceSizes[0],
		width:  uint32(fd.FrameWidth),
		height: uint32(fd.FrameHeight),
		idx:    idx,
	}, nil
}

func (s *Source) Close() error {
	_ = s.reader.Close()
	return s.inst.Close()
}

func (s *Source) Next(timeout time.Duration) (Frame, error) {
	for {
		g, err := s.reader.GetGrain(s.idx, timeout)
		switch {
		case err == nil:
			f := Frame{
				Index:   g.Index,
				Width:   s.width,
				Height:  s.height,
				Stride:  s.stride,
				Size:    g.GrainSize,
				Invalid: g.Invalid(),
				Payload: g.Payload,
			}
			s.idx++
			return f, nil
		case errors.Is(err, mxl.ErrTimeout):
			s.idx = mxl.CurrentIndex(s.rate)
		case errors.Is(err, mxl.ErrOutOfRangeEarly):
			time.Sleep(10 * time.Millisecond)
		case errors.Is(err, mxl.ErrOutOfRangeLate):
			s.idx = mxl.CurrentIndex(s.rate)
		default:
			return Frame{}, fmt.Errorf("GetGrain: %w", err)
		}
	}
}

func (s *Source) NextCtx(ctx context.Context, timeout time.Duration) (Frame, error) {
	for {
		select {
		case <-ctx.Done():
			return Frame{}, ctx.Err()
		default:
		}
		g, err := s.reader.GetGrain(s.idx, timeout)
		switch {
		case err == nil:
			f := Frame{
				Index:   g.Index,
				Width:   s.width,
				Height:  s.height,
				Stride:  s.stride,
				Size:    g.GrainSize,
				Invalid: g.Invalid(),
				Payload: g.Payload,
			}
			s.idx++
			return f, nil
		case errors.Is(err, mxl.ErrTimeout):
			s.idx = mxl.CurrentIndex(s.rate)
		case errors.Is(err, mxl.ErrOutOfRangeEarly):
			time.Sleep(10 * time.Millisecond)
		case errors.Is(err, mxl.ErrOutOfRangeLate):
			s.idx = mxl.CurrentIndex(s.rate)
		default:
			return Frame{}, fmt.Errorf("GetGrain: %w", err)
		}
	}
}

func (s *Source) FlowDef() string        { return s.def }
func (s *Source) Rate() mxl.Rational     { return s.rate }
func (s *Source) Stride() uint32         { return s.stride }
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 }