stream.go

package gogige

import (
	"context"
	"errors"
	"fmt"
	"net"
	"sync"
	"sync/atomic"
	"time"

	"github.com/aaronmurniadi/gogige/gvcp"
	"github.com/aaronmurniadi/gogige/gvsp"
	"github.com/aaronmurniadi/gogige/internal/color"
)

var errStreamNotOpen = errors.New("gige: stream not open")

// Session is a persistent GVSP stream session. It implements Grabber.
type Session struct {
	cam        *Camera
	stream     *gvsp.Stream
	opened     atomic.Bool
	mu         sync.Mutex
	ownCam     bool // if true, Close() closes cam
	ip         string
	packetSize int
	component  Component

	hb *gvcp.Heartbeat
}

// NewSession returns an empty GVSP session (Connects on Open if no camera set).
func NewSession() *Session { return &Session{component: ComponentColor} }

// NewFromCamera returns a Session that streams using cam without taking ownership.
func NewFromCamera(cam *Camera) *Session {
	return &Session{cam: cam, component: ComponentColor}
}

// SetComponent selects which BSCF/SFNC component Grab returns (color/depth/mono).
func (s *Session) SetComponent(c Component) {
	if s == nil || c == ComponentUnknown {
		return
	}
	s.mu.Lock()
	s.component = c
	s.mu.Unlock()
}

// Open connects (if needed), starts GVSP, and begins acquisition.
func (s *Session) Open(ip string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.opened.Load() {
		return nil
	}
	s.ip = ip
	if s.cam == nil {
		cam, err := Connect(ip)
		if err != nil {
			return err
		}
		s.cam = cam
		s.ownCam = true
	}
	stream, err := gvsp.ListenStream(0)
	if err != nil {
		return err
	}
	s.stream = stream
	if g := s.cam.GVCP(); g != nil {
		stream.SetResender(func(blockID uint64, first, last uint32, extended bool) {
			_ = g.RequestResend(0, blockID, first, last, extended)
		})
	}
	if got := stream.RecvBuffer(); got > 0 && got < gvsp.MinRecvBufSize {
		s.cam.Logger().Warn("gvsp SO_RCVBUF below recommended",
			"got", got, "min", gvsp.MinRecvBufSize, "want", gvsp.DefaultRecvBufSize)
	}

	local := s.cam.GVCP().LocalAddr()
	if local == nil || local.IP == nil {
		stream.Close()
		s.stream = nil
		return errors.New("gige: no local address for stream destination")
	}
	destIP := local.IP
	if destIP.IsUnspecified() {
		destIP = firstIPv4()
	}
	camIP := net.ParseIP(ip)
	if camIP == nil {
		camIP = net.ParseIP(s.cam.IP)
	}
	mtu := gvsp.PathMTU(camIP)
	want := gvsp.PacketSizeForMTU(mtu)
	actual, err := gvcp.StartAcquisition(s.cam.GVCP(), s.cam, destIP, stream.Port(), want)
	if err != nil {
		stream.Close()
		s.stream = nil
		return err
	}
	s.packetSize = actual
	if actual < want {
		s.cam.Logger().Warn("GevSCPSPacketSize negotiated lower than path MTU",
			"want", want, "actual", actual, "path_mtu", mtu)
	}
	s.cam.Logger().Info("gvsp stream setup",
		"mtu", mtu, "scps", s.packetSize, "rcvbuf", stream.RecvBuffer())
	s.startHeartbeatLocked()
	s.opened.Store(true)
	return nil
}

func firstIPv4() net.IP {
	ifaces, _ := net.Interfaces()
	for _, iface := range ifaces {
		if iface.Flags&net.FlagUp == 0 || iface.Flags&net.FlagLoopback != 0 {
			continue
		}
		addrs, _ := iface.Addrs()
		for _, a := range addrs {
			var ip net.IP
			switch v := a.(type) {
			case *net.IPNet:
				ip = v.IP
			case *net.IPAddr:
				ip = v.IP
			}
			if ip4 := ip.To4(); ip4 != nil {
				return ip4
			}
		}
	}
	return net.IPv4(127, 0, 0, 1)
}

// Close stops acquisition and releases stream resources.
// The Camera is closed only when the Session owns it (not when Device owns it).
func (s *Session) Close() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.opened.Store(false)
	s.stopHeartbeatLocked()
	if s.cam != nil {
		_ = gvcp.StopAcquisition(s.cam)
	}
	if s.stream != nil {
		_ = s.stream.Close()
		s.stream = nil
	}
	if s.ownCam && s.cam != nil {
		s.cam.Close()
		s.cam = nil
	}
	return nil
}

// PauseStreaming stops image transfer but keeps CCP, heartbeat, and GVSP socket.
func (s *Session) PauseStreaming() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if !s.opened.Load() {
		return nil
	}
	if s.cam != nil {
		_ = gvcp.StopAcquisition(s.cam)
		if g := s.cam.GVCP(); g != nil {
			_ = g.WriteReg(gvcp.Stream0Port, 0)
		}
	}
	return nil
}

// ResumeStreaming re-programs the stream channel and starts acquisition again.
func (s *Session) ResumeStreaming() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if !s.opened.Load() {
		return errors.New("gige: stream not open")
	}
	if s.cam == nil || s.stream == nil {
		return errors.New("gige: stream not open")
	}
	local := s.cam.GVCP().LocalAddr()
	if local == nil || local.IP == nil {
		return errors.New("gige: no local address for stream destination")
	}
	destIP := local.IP
	if destIP.IsUnspecified() {
		destIP = firstIPv4()
	}
	ps := s.packetSize
	if ps <= 0 {
		ps = gvsp.PacketSizeForMTU(gvsp.PathMTU(net.ParseIP(s.ip)))
		s.packetSize = ps
	}
	actual, err := gvcp.StartAcquisition(s.cam.GVCP(), s.cam, destIP, s.stream.Port(), ps)
	if err != nil {
		return err
	}
	s.packetSize = actual
	return nil
}

// Pause implements Grabber.
func (s *Session) Pause(ctx context.Context) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	return s.PauseStreaming()
}

// Resume implements Grabber.
func (s *Session) Resume(ctx context.Context) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	return s.ResumeStreaming()
}

func (s *Session) startHeartbeatLocked() {
	s.stopHeartbeatLocked()
	if s.cam == nil || s.cam.GVCP() == nil {
		return
	}
	s.hb = s.cam.GVCP().StartHeartbeat()
}

func (s *Session) stopHeartbeatLocked() {
	if s.hb == nil {
		return
	}
	s.hb.Stop()
	s.hb = nil
}

// Opened reports whether the session is open.
func (s *Session) Opened() bool { return s.opened.Load() }

// Grab implements Grabber: receives one GVSP frame, parses BSCF, returns JPEG Sample.
func (s *Session) Grab(ctx context.Context) (Sample, error) {
	sample := Sample{PackCount: -1}
	data, meta, err := s.recvFrame(ctx)
	if err != nil {
		return sample, err
	}
	s.mu.Lock()
	kind := s.component
	s.mu.Unlock()
	if kind == ComponentUnknown {
		kind = ComponentColor
	}
	sample, err = SampleFromBSCFComponent(data, kind)
	if err != nil {
		if gvsp.IsBSCF(data) {
			return sample, err
		}
		// Non-BSCF payload.
		sample.RawColor = data
		sample.PixelWidth = meta.width
		sample.PixelHeight = meta.height
		sample.PixelFormat = meta.pixelFormat
		sample.Component = kind
		if sample.PixelWidth == 0 || sample.PixelHeight == 0 {
			return sample, fmt.Errorf("gige: grab: %w", err)
		}
	}
	jpeg, jerr := color.EncodeJPEG(sample.RawColor, sample.PixelWidth, sample.PixelHeight, sample.PixelFormat, 60)
	if jerr != nil {
		return sample, jerr
	}
	sample.JPEG = jpeg
	return sample, nil
}

// GrabAll receives one GVSP frame and returns a JPEG Sample for every BSCF component
// (color, depth, mono, …). Non-BSCF payloads yield a single sample.
func (s *Session) GrabAll(ctx context.Context) ([]Sample, error) {
	data, meta, err := s.recvFrame(ctx)
	if err != nil {
		return nil, err
	}
	var samples []Sample
	if gvsp.IsBSCF(data) {
		samples, err = SampleAllFromBSCF(data)
		if err != nil {
			return nil, err
		}
	} else {
		samples = []Sample{{
			RawColor:    data,
			PixelWidth:  meta.width,
			PixelHeight: meta.height,
			PixelFormat: meta.pixelFormat,
			Component:   ComponentUnknown,
			PackCount:   -1,
		}}
	}
	out := make([]Sample, 0, len(samples))
	var errs []error
	for _, sample := range samples {
		if sample.PixelWidth <= 0 || sample.PixelHeight <= 0 || len(sample.RawColor) == 0 {
			continue
		}
		jpeg, jerr := color.EncodeJPEG(sample.RawColor, sample.PixelWidth, sample.PixelHeight, sample.PixelFormat, 60)
		if jerr != nil {
			errs = append(errs, fmt.Errorf("%s: %w", sample.Component, jerr))
			continue
		}
		sample.JPEG = jpeg
		out = append(out, sample)
	}
	if len(out) == 0 {
		if len(errs) > 0 {
			return nil, fmt.Errorf("gige: grab all: %w", errors.Join(errs...))
		}
		return nil, errors.New("gige: grab all: no usable components")
	}
	return out, nil
}

// GrabComponents receives one GVSP frame and returns its BSCF components without
// JPEG encoding (raw Data, dimensions, pixel format). Non-BSCF payloads yield a
// single Sample with ComponentUnknown and no JPEG. Useful to probe what a camera
// is actually streaming beyond the JPEG color path.
func (s *Session) GrabComponents(ctx context.Context) ([]Sample, error) {
	data, meta, err := s.recvFrame(ctx)
	if err != nil {
		return nil, err
	}
	if gvsp.IsBSCF(data) {
		return gvsp.SampleAllFromBSCF(data)
	}
	return []Sample{{
		RawColor:    data,
		PixelWidth:  meta.width,
		PixelHeight: meta.height,
		PixelFormat: meta.pixelFormat,
		Component:   ComponentUnknown,
		PackCount:   -1,
	}}, nil
}

type frameMeta struct {
	width, height int
	pixelFormat   uint32
}

// recvFrame takes one GVSP frame and returns an owned copy of the payload.
func (s *Session) recvFrame(ctx context.Context) ([]byte, frameMeta, error) {
	var meta frameMeta
	if err := ctx.Err(); err != nil {
		return nil, meta, err
	}
	timeout := time.Second
	if dl, ok := ctx.Deadline(); ok {
		timeout = time.Until(dl)
		if timeout <= 0 {
			return nil, meta, context.DeadlineExceeded
		}
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if !s.opened.Load() || s.stream == nil {
		return nil, meta, errStreamNotOpen
	}
	frame, err := s.stream.Recv(timeout)
	if err != nil {
		return nil, meta, err
	}
	defer frame.Release()
	meta = frameMeta{
		width:       int(frame.Width),
		height:      int(frame.Height),
		pixelFormat: frame.PixelFormat,
	}
	return append([]byte(nil), frame.Data...), meta, nil
}

// recvFramePtr returns the pooled GVSP frame directly (no copy). Caller owns it
// and must Release() it. Used by the channel-based Stream API.
func (s *Session) recvFramePtr(ctx context.Context) (*gvsp.Frame, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	timeout := time.Second
	if dl, ok := ctx.Deadline(); ok {
		timeout = time.Until(dl)
		if timeout <= 0 {
			return nil, context.DeadlineExceeded
		}
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if !s.opened.Load() || s.stream == nil {
		return nil, errStreamNotOpen
	}
	return s.stream.Recv(timeout)
}

// Stream is a live GVSP acquisition channel (Phase 4). Create one via
// Camera.StartStream, consume pooled frames from Frames(), and Release() each
// frame back to the pre-allocated buffer pool.
type Stream struct {
	s      *Session
	cancel context.CancelFunc
	done   chan struct{}
	frames chan *Frame
	once   sync.Once
}

// StartStream opens a GVSP channel on c and begins acquisition. Frames are
// delivered over Stream.Frames() until the context is cancelled or Stop is
// called. The Camera stays open; close it separately when done.
func (c *Camera) StartStream(ctx context.Context) (*Stream, error) {
	if c == nil || c.GVCP() == nil {
		return nil, errStreamNotOpen
	}
	s := NewFromCamera(c)
	if err := s.Open(c.IP); err != nil {
		return nil, err
	}
	if ctx == nil {
		ctx = context.Background()
	}
	child, cancel := context.WithCancel(ctx)
	st := &Stream{
		s:      s,
		cancel: cancel,
		done:   make(chan struct{}),
		frames: make(chan *Frame, 4),
	}
	go st.loop(child)
	return st, nil
}

// Frames returns the channel of pooled frames. The channel closes when the
// stream ends (context cancelled, Stop called, or the session goes away).
func (st *Stream) Frames() <-chan *Frame {
	if st == nil {
		return nil
	}
	return st.frames
}

// Stop ends acquisition, closes the stream socket, and stops the background
// goroutine. Safe to call more than once.
func (st *Stream) Stop() {
	if st == nil {
		return
	}
	st.once.Do(func() {
		st.cancel()
		<-st.done
		_ = st.s.Close()
	})
}

// Pause stops image transfer but keeps the control channel and socket open.
func (st *Stream) Pause() error {
	if st == nil {
		return nil
	}
	return st.s.PauseStreaming()
}

// Resume re-programs the stream channel and restarts acquisition.
func (st *Stream) Resume() error {
	if st == nil {
		return errStreamNotOpen
	}
	return st.s.ResumeStreaming()
}

func (st *Stream) loop(ctx context.Context) {
	defer close(st.done)
	defer close(st.frames)
	for {
		frame, err := st.s.recvFramePtr(ctx)
		if err != nil {
			if ctx.Err() != nil {
				return
			}
			if err == errStreamNotOpen {
				return
			}
			continue
		}
		select {
		case st.frames <- frame:
		case <-ctx.Done():
			frame.Release()
			return
		}
	}
}