groundstation/internal/osd/osd.go

// Package osd is the shared flight OSD snapshot contract between the Ground
// Station supervisor and the kiosk display process. The supervisor owns the
// RC, arming, video and operator-configuration state and publishes immutable
// snapshots; the display renders the latest fresh snapshot on the transparent
// OSD overlay. The snapshot is a plain value: everything published is copied,
// so a reader can never observe or mutate a half-written state.
//
// The adaptive-link contract is modeled as four distinct values that must
// never be collapsed into one another: the operator configuration (selected
// power table plus the temporary CH6 maxPower ceiling), the ground-station
// policy (requested runtime level), the aircraft VTX acknowledgment (applied
// level and lifecycle) and the observed receiver link. Their producers —
// adaptive control, the VTX ack transport and WFB RX metric ingestion — do
// not exist yet and stay out of scope; until they arrive the snapshot carries
// explicit pending/unknown states instead of fabricated data. CH6 is operator
// intent only: it constrains what may be requested and is never proof of
// applied VTX power, and static configuration is never rendered or serialized
// as applied VTX state.
package osd

import (
	"encoding/binary"
	"errors"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"sync"
	"sync/atomic"
	"time"

	"git.piksar.eu/piotr/groundstation/internal/protocol"
)

const (
	Magic0  byte = 'G'
	Magic1  byte = 'O'
	Version byte = 2

	// Size is the fixed encoded snapshot length: magic and version, frame
	// sequence and timestamp, all 16 final RC channels, one flags byte,
	// the publish time, the four adaptive-link contract values and the CRC.
	Size = 3 + 2 + 4 + protocol.Channels*2 + 1 + 8 + // 50: v2 header block
		1 + // operator maxPower ceiling
		4 + 8 + // operator table generation + hash
		1 + 8 + 8 + // policy level + reason + timestamp
		4 + 8 + 1 + 1 + // ack table generation + hash + level + lifecycle
		8 + // link freshness age
		2 // CRC

	// DefaultStatePath is where the supervisor publishes snapshots and
	// where the display reads them unless configured otherwise. It lives
	// in tmpfs: snapshots are runtime telemetry, not persistent state.
	DefaultStatePath = "/run/groundstation/osd-state.bin"

	// Heartbeat is how often an unchanged snapshot is republished, so the
	// display keeps seeing it as fresh while the UART stays healthy.
	Heartbeat = 500 * time.Millisecond

	// writeInterval is the minimum gap between two state-file writes. It
	// bounds the write rate when RC channels change on every frame without
	// delaying an arm or power transition noticeably.
	writeInterval = 50 * time.Millisecond

	// Freshness is the display-side maximum age of a rendered snapshot: a
	// dead UART must end OSD rendering instead of freezing stale telemetry
	// over live video.
	Freshness = 2 * time.Second

	// MaxVTXLevel is the upper bound of the VTX power level domain shared
	// by the operator ceiling, the ground-station policy and the applied
	// acknowledgment.
	MaxVTXLevel uint8 = 127
)

// VTX acknowledgment lifecycle states.
const (
	// VTXLifecyclePending means no acknowledgment has arrived yet.
	VTXLifecyclePending uint8 = 0
	// VTXLifecycleApplied means the VTX acknowledged applying the level.
	VTXLifecycleApplied uint8 = 1
	// VTXLifecycleError means the acknowledgment reported a lifecycle error.
	VTXLifecycleError uint8 = 2
)

// Snapshot flag bits in the encoded record.
const (
	flagArmed byte = 1 << iota
	flagVideo
	flagTableSelected
	flagPolicyPending
	flagAckPending
	flagLinkKnown
)

// Fixed field offsets of the encoded record.
const (
	publishedOffset    = 42
	opMaxPowerOffset   = 50
	opTableGenOffset   = 51
	opTableHashOffset  = 55
	policyLevelOffset  = 63
	policyReasonOffset = 64
	policyTimeOffset   = 72
	ackTableGenOffset  = 80
	ackTableHashOffset = 84
	ackLevelOffset     = 92
	ackLifecycleOffset = 93
	linkAgeOffset      = 94
	crcOffset          = Size - 2
)

var (
	ErrLength  = errors.New("groundstation/osd: invalid snapshot length")
	ErrMagic   = errors.New("groundstation/osd: invalid snapshot magic")
	ErrVersion = errors.New("groundstation/osd: unsupported snapshot version")
	ErrCRC     = errors.New("groundstation/osd: invalid snapshot crc")
)

// Table identifies a selectable VTX power table by generation and content
// hash. Zero generation and hash mean no table is selected yet. The
// identifiers are the operator's opaque selection of a table target — the
// table generator (future work) owns real identity — and a statically
// configured selection is operator intent: it is never applied VTX state.
type Table struct {
	Gen  uint32
	Hash uint64
}

// Selected reports whether a table has been selected at all.
func (t Table) Selected() bool { return t.Gen != 0 || t.Hash != 0 }

// OperatorConfig is the operator's configuration intent: the selected power
// table plus the temporary CH6 maxPower ceiling. It is never proof of
// applied VTX power: the ceiling constrains what the system may request,
// while the applied state comes only from a VTX acknowledgment.
type OperatorConfig struct {
	Table    Table
	MaxPower uint8 // maxPower ceiling in the 0..127 level domain
}

// GSPolicy is the ground station's requested runtime VTX power level with
// the reason and time of the request. No policy producer exists yet (the
// adaptive controller is future work), so published policies stay Pending
// rather than fabricating a level. Within the 0..127 domain, level 0 is an
// explicit off / no-runtime-command request; only Pending marks the absent
// producer.
type GSPolicy struct {
	Pending     bool
	Level       uint8   // requested level, 0..127; 0 is explicit off
	Reason      [8]byte // producer-defined short reason tag
	TimestampMS int64
}

// VTXAck is the aircraft VTX's acknowledgment: the table it accepted, the
// level it applied and its lifecycle/error state. No acknowledgment
// transport exists yet, so published acks stay Pending. Applied level 0 is
// an explicit applied-off state; only Pending marks the absent transport.
type VTXAck struct {
	Pending   bool
	Table     Table
	Level     uint8 // applied level, 0..127; 0 is explicit off
	Lifecycle uint8
}

// GSObservedLink is the ground-station receiver's link observation with its
// freshness. WFB RX metrics are out of scope until that producer exists, so
// the observation stays explicitly UNKNOWN instead of carrying fabricated
// numbers.
type GSObservedLink struct {
	Known bool  // false = explicitly unknown
	AgeMS int64 // freshness of the metrics at publication; 0 while unknown
}

// Snapshot is one immutable flight state: the final RC channels, the arm
// state derived from the configured arm channel, the supervisor's
// decoded-video presence knowledge and the four adaptive-link contract
// values, plus the provenance of the RC frame it was built from. The display
// never renders from the Video field: it gates on the video lifecycle it
// observes on the display server itself. PublishedMS is stamped at
// publication and is what the display uses to reject stale snapshots.
type Snapshot struct {
	Sequence    uint16
	TimestampMS uint32
	Channels    [protocol.Channels]int16
	Armed       bool
	Video       bool
	OpConfig    OperatorConfig
	Policy      GSPolicy
	Ack         VTXAck
	Link        GSObservedLink
	PublishedMS int64
}

// CeilingFromChannel maps the final CH6 channel onto the temporary operator
// maxPower ceiling: the full channel range -1024..1024 scales linearly onto
// the 0..127 level domain with clamping. The ceiling is operator intent
// only and never proof of applied VTX power.
func CeilingFromChannel(channel int16) uint8 {
	if channel <= -1024 {
		return 0
	}
	if channel >= 1024 {
		return MaxVTXLevel
	}
	return uint8((int32(channel) + 1024) * int32(MaxVTXLevel) / 2048)
}

func Encode(snapshot Snapshot) []byte {
	buf := make([]byte, Size)
	buf[0], buf[1], buf[2] = Magic0, Magic1, Version
	binary.LittleEndian.PutUint16(buf[3:5], snapshot.Sequence)
	binary.LittleEndian.PutUint32(buf[5:9], snapshot.TimestampMS)
	for i, value := range snapshot.Channels {
		binary.LittleEndian.PutUint16(buf[9+i*2:9+i*2+2], uint16(value))
	}
	var flags byte
	if snapshot.Armed {
		flags |= flagArmed
	}
	if snapshot.Video {
		flags |= flagVideo
	}
	if snapshot.OpConfig.Table.Selected() {
		flags |= flagTableSelected
	}
	if snapshot.Policy.Pending {
		flags |= flagPolicyPending
	}
	if snapshot.Ack.Pending {
		flags |= flagAckPending
	}
	if snapshot.Link.Known {
		flags |= flagLinkKnown
	}
	buf[41] = flags
	binary.LittleEndian.PutUint64(buf[publishedOffset:opMaxPowerOffset], uint64(snapshot.PublishedMS))
	buf[opMaxPowerOffset] = snapshot.OpConfig.MaxPower
	binary.LittleEndian.PutUint32(buf[opTableGenOffset:opTableHashOffset], snapshot.OpConfig.Table.Gen)
	binary.LittleEndian.PutUint64(buf[opTableHashOffset:policyLevelOffset], snapshot.OpConfig.Table.Hash)
	buf[policyLevelOffset] = snapshot.Policy.Level
	copy(buf[policyReasonOffset:policyTimeOffset], snapshot.Policy.Reason[:])
	binary.LittleEndian.PutUint64(buf[policyTimeOffset:ackTableGenOffset], uint64(snapshot.Policy.TimestampMS))
	binary.LittleEndian.PutUint32(buf[ackTableGenOffset:ackTableHashOffset], snapshot.Ack.Table.Gen)
	binary.LittleEndian.PutUint64(buf[ackTableHashOffset:ackLevelOffset], snapshot.Ack.Table.Hash)
	buf[ackLevelOffset] = snapshot.Ack.Level
	buf[ackLifecycleOffset] = snapshot.Ack.Lifecycle
	binary.LittleEndian.PutUint64(buf[linkAgeOffset:crcOffset], uint64(snapshot.Link.AgeMS))
	binary.LittleEndian.PutUint16(buf[crcOffset:], protocol.CRC16CCITT(buf[:crcOffset]))
	return buf
}

func Decode(buf []byte) (Snapshot, error) {
	var snapshot Snapshot
	if len(buf) != Size {
		return snapshot, ErrLength
	}
	if buf[0] != Magic0 || buf[1] != Magic1 {
		return snapshot, ErrMagic
	}
	if buf[2] != Version {
		return snapshot, ErrVersion
	}
	if got, want := binary.LittleEndian.Uint16(buf[crcOffset:]), protocol.CRC16CCITT(buf[:crcOffset]); got != want {
		return snapshot, fmt.Errorf("%w: got %04x want %04x", ErrCRC, got, want)
	}
	snapshot.Sequence = binary.LittleEndian.Uint16(buf[3:5])
	snapshot.TimestampMS = binary.LittleEndian.Uint32(buf[5:9])
	for i := range snapshot.Channels {
		snapshot.Channels[i] = int16(binary.LittleEndian.Uint16(buf[9+i*2 : 9+i*2+2]))
	}
	flags := buf[41]
	snapshot.Armed = flags&flagArmed != 0
	snapshot.Video = flags&flagVideo != 0
	snapshot.OpConfig.MaxPower = buf[opMaxPowerOffset]
	snapshot.OpConfig.Table.Gen = binary.LittleEndian.Uint32(buf[opTableGenOffset:opTableHashOffset])
	snapshot.OpConfig.Table.Hash = binary.LittleEndian.Uint64(buf[opTableHashOffset:policyLevelOffset])
	snapshot.Policy.Level = buf[policyLevelOffset]
	copy(snapshot.Policy.Reason[:], buf[policyReasonOffset:policyTimeOffset])
	snapshot.Policy.TimestampMS = int64(binary.LittleEndian.Uint64(buf[policyTimeOffset:ackTableGenOffset]))
	snapshot.Ack.Table.Gen = binary.LittleEndian.Uint32(buf[ackTableGenOffset:ackTableHashOffset])
	snapshot.Ack.Table.Hash = binary.LittleEndian.Uint64(buf[ackTableHashOffset:ackLevelOffset])
	snapshot.Ack.Level = buf[ackLevelOffset]
	snapshot.Ack.Lifecycle = buf[ackLifecycleOffset]
	snapshot.Link.AgeMS = int64(binary.LittleEndian.Uint64(buf[linkAgeOffset:crcOffset]))
	snapshot.Policy.Pending = flags&flagPolicyPending != 0
	snapshot.Ack.Pending = flags&flagAckPending != 0
	snapshot.Link.Known = flags&flagLinkKnown != 0
	snapshot.PublishedMS = int64(binary.LittleEndian.Uint64(buf[publishedOffset:opMaxPowerOffset]))
	return snapshot, nil
}

// DecodeFile reads and decodes one snapshot from its state file.
func DecodeFile(path string) (Snapshot, error) {
	data, err := os.ReadFile(path)
	if err != nil {
		return Snapshot{}, err
	}
	return Decode(data)
}

// Publisher publishes immutable snapshots latest-wins. Publish never blocks:
// it stores the snapshot atomically and only wakes the writer, so the UART,
// WFB and video paths never wait on state-file I/O. The writer persists the
// latest snapshot whenever it changes and republishes it as a heartbeat so
// the display keeps seeing it as fresh.
type Publisher struct {
	latest        atomic.Pointer[Snapshot]
	wake          chan struct{}
	done          chan struct{}
	stopped       chan struct{}
	closeOnce     sync.Once
	path          string
	heartbeat     time.Duration
	writeInterval time.Duration
	write         func(path string, data []byte) error
	logger        *log.Logger
}

// NewPublisher builds an in-memory publisher without a writer goroutine.
func NewPublisher() *Publisher {
	return newFilePublisher("", nil, Heartbeat, writeInterval, nil)
}

// NewFilePublisher builds a publisher that persists snapshots to path for
// the display process. Failures are logged and retried; they never block
// Publish.
func NewFilePublisher(path string, logger *log.Logger) *Publisher {
	return newFilePublisher(path, writeStateFile, Heartbeat, writeInterval, logger)
}

func newFilePublisher(path string, write func(string, []byte) error, heartbeat, interval time.Duration, logger *log.Logger) *Publisher {
	p := &Publisher{
		wake:          make(chan struct{}, 1),
		done:          make(chan struct{}),
		stopped:       make(chan struct{}),
		path:          path,
		heartbeat:     heartbeat,
		writeInterval: interval,
		write:         write,
		logger:        logger,
	}
	if path != "" {
		go p.loop()
	}
	return p
}

// Publish records the snapshot and wakes the writer. The published value is
// stamped with the publication time and stored as an immutable copy.
func (p *Publisher) Publish(snapshot Snapshot) {
	snapshot.PublishedMS = time.Now().UnixMilli()
	p.latest.Store(&snapshot)
	if p.path == "" {
		return
	}
	select {
	case p.wake <- struct{}{}:
	default:
	}
}

// Latest reports the most recently published snapshot, if any.
func (p *Publisher) Latest() (Snapshot, bool) {
	snapshot := p.latest.Load()
	if snapshot == nil {
		return Snapshot{}, false
	}
	return *snapshot, true
}

// Close stops the writer and removes the state file, so a stopped
// supervisor never leaves stale telemetry for the display to render.
// It is safe to call more than once and on an in-memory publisher.
func (p *Publisher) Close() {
	p.closeOnce.Do(func() {
		if p.path == "" {
			return
		}
		close(p.done)
		<-p.stopped
		_ = os.Remove(p.path)
	})
}

func (p *Publisher) loop() {
	defer close(p.stopped)
	ticker := time.NewTicker(p.writeInterval)
	defer ticker.Stop()

	var last Snapshot
	have := false
	var lastWrite time.Time
	var lastErrLog time.Time
	for {
		select {
		case <-p.wake:
		case <-ticker.C:
		case <-p.done:
			return
		}

		snapshot := p.latest.Load()
		if snapshot == nil {
			continue
		}
		// Changed states go out as soon as the write floor allows; an
		// unchanged state goes out on the heartbeat so the display keeps
		// seeing it as fresh.
		now := time.Now()
		if have {
			if *snapshot == last {
				if now.Sub(lastWrite) < p.heartbeat {
					continue
				}
			} else if now.Sub(lastWrite) < p.writeInterval {
				continue
			}
		}
		if err := p.write(p.path, Encode(*snapshot)); err != nil {
			// Failures are retried on every tick, so a persistently
			// broken state path is logged at most once per heartbeat
			// instead of flooding the journal.
			if p.logger != nil && now.Sub(lastErrLog) >= p.heartbeat {
				p.logger.Printf("osd state write %s: %v", p.path, err)
				lastErrLog = now
			}
			continue
		}
		last, have, lastWrite = *snapshot, true, now
	}
}

// writeStateFile replaces the state file atomically: readers either see the
// previous complete snapshot or the new one, never a partial write. The file
// stays world-readable so the unprivileged display user can read it; it
// carries only flight telemetry.
func writeStateFile(path string, data []byte) error {
	dir := filepath.Dir(path)
	if err := os.MkdirAll(dir, 0755); err != nil {
		return err
	}
	temp, err := os.CreateTemp(dir, filepath.Base(path)+".tmp-*")
	if err != nil {
		return err
	}
	tmpPath := temp.Name()
	defer os.Remove(tmpPath)
	if _, err := temp.Write(data); err != nil {
		temp.Close()
		return err
	}
	if err := temp.Chmod(0644); err != nil {
		temp.Close()
		return err
	}
	if err := temp.Close(); err != nil {
		return err
	}
	return os.Rename(tmpPath, path)
}

// FileSource reads the supervisor-published snapshot from its state file.
// It is the display-side end of the snapshot contract: reads that fail or
// return stale snapshots report no snapshot, so the OSD stays transparent
// instead of rendering telemetry from a dead UART.
type FileSource struct {
	path string
	// now is the clock used for freshness checks; tests replace it.
	now func() time.Time
}

// NewFileSource builds a source reading the given state file.
func NewFileSource(path string) *FileSource {
	return &FileSource{path: path, now: time.Now}
}

// Latest reports the current snapshot while it is fresh.
func (s *FileSource) Latest() (Snapshot, bool) {
	snapshot, err := DecodeFile(s.path)
	if err != nil {
		return Snapshot{}, false
	}
	if age := s.now().Sub(time.UnixMilli(snapshot.PublishedMS)); age > Freshness {
		return Snapshot{}, false
	}
	return snapshot, true
}