lutron-control/device.go

package main

import (
	"bytes"
	"context"
	"fmt"
	"math"
	"strconv"
	"strings"
	"sync"
	"time"

	log "github.com/sirupsen/logrus"
)

// Lutron integration protocol constants (not user-configurable).
const (
	qseTerminator      = "\r\n"
	qseActionZoneLevel = 14  // Action number for a zone's light level (component 1-24).
	qseBtnAction       = 3   // "Press" action carried by the enable/disable signals.
	qseActionRelease   = 4   // "Release" action for phantom buttons (shades react on release).
	qseSceneController = 141 // Component number of the GRAFIK Eye QS scene controller.
	qseActionScene     = 7   // Action number for the current scene (set/get).
	qseMaxScene        = 16  // Highest scene number the GRAFIK Eye QS supports.

	// Zone movement actions (component 1-24). Raise/lower start motion and run
	//  until a stop; the panel reports the new level via zone monitoring.
	qseActionStartRaising  = 18
	qseActionStartLowering = 19
	qseActionStopMoving    = 20

	// Scene controller (141) actions (QS Standalone).
	qseActionZoneLock  = 15 // Zone lock: 0 off, 1 on.
	qseActionSceneLock = 16 // Scene lock: 0 off, 1 on.
	qseActionSequence  = 17 // Sequence: 0 off, 1 scenes 1-4, 2 scenes 5-16.

	// qseSceneOffButton is the Scene Off phantom-button component.
	qseSceneOffButton = 83

	// zoneStopGrace is how long zone writes stay suppressed after a raise/lower
	//  stop so the panel's final level report can update the target before the
	//  writer resumes, preventing it from yanking the zone back to its pre-move
	//  level.
	zoneStopGrace = 2 * time.Second

	// #MONITORING types (protocol p.12).
	qseMonitorZone  = 5  // Zone-level monitoring.
	qseMonitorScene = 8  // Scene monitoring.
	qseMonitorReply = 11 // Reply state (responses cease if disabled).

	// writeInterval is the zone-writer tick.
	writeInterval = 100 * time.Millisecond

	// Minimum inter-message delays mandated by the integration protocol: a command
	//  ("#") requires at least 100 ms before the next message, a query ("?") at
	//  least 1500 ms. The protocol also advises relying on monitoring rather than
	//  polling, which is why queries are only used for the connect-time sync and
	//  the single-query watchdog probe.
	commandInterMessageDelay = 100 * time.Millisecond
	queryInterMessageDelay   = 1500 * time.Millisecond

	// queryReplyWindow bounds how long after we query a zone its reply is treated
	//  as solicited. A solicited reply (notably the watchdog liveness probe) is
	//  used for liveness only and must not override the level we maintain; an
	//  unsolicited report past this window is a genuine external change to follow.
	queryReplyWindow = 5 * time.Second

	// zoneEchoWindow bounds how long after we write a zone its report is treated
	//  as an echo of that write rather than a genuine external change. The panel
	//  quantizes levels to its own scale, so its echo rarely byte-matches what we
	//  sent; adopting it back into the target would fight our own writes.
	zoneEchoWindow = 5 * time.Second
)

// qseErrorDescriptions maps ~ERROR,<n> codes from the Lutron integration
// protocol (doc 040249) to human-readable text.
var qseErrorDescriptions = map[string]string{
	"1": "parameter count mismatch",
	"2": "object does not exist (check integration ID)",
	"3": "invalid action number",
	"4": "parameter data out of range",
	"5": "parameter data malformed",
	"6": "unsupported command",
}

// shadeColumn holds the GRAFIK Eye QS phantom-button component numbers for one
// shade column. Open, close, and preset react on release (a press then release);
// raise and lower start motion on press and are halted by releasing the held
// component.
type shadeColumn struct{ open, preset, close, lower, raise int }

// qseShadeColumns maps a 1-indexed shade column to its component numbers.
var qseShadeColumns = map[int]shadeColumn{
	1: {open: 38, preset: 39, close: 40, lower: 41, raise: 47},
	2: {open: 44, preset: 45, close: 46, lower: 52, raise: 53},
	3: {open: 50, preset: 51, close: 56, lower: 57, raise: 58},
}

// sourceBinding tracks a control source's arbitration state on a device. Zone
// and scene activity are tracked separately so the device knows whether zones
// are being actively driven.
type sourceBinding struct {
	name            string
	priority        int
	hold            time.Duration
	fade            string // Zone fade time this source applies; empty uses the device default.
	lastZoneActive  time.Time
	lastSceneActive time.Time
}

// activeAt reports whether the source has applied zone or scene control within
// its hold window as of now.
func (b *sourceBinding) activeAt(now time.Time) bool {
	return now.Sub(b.lastZoneActive) < b.hold || now.Sub(b.lastSceneActive) < b.hold
}

// zoneFeedbackFn is called when the device reports a zone's current level.
type zoneFeedbackFn func(zone int, level byte)

// sceneFeedbackFn is called when the device reports the active scene.
type sceneFeedbackFn func(scene int)

// MonitorEvent is a parsed monitoring message reported by the panel. It carries
// every "~" feedback line generically so a source can relay arbitrary monitoring
// (zone, scene, button, occupancy, etc.) without the device modelling each type.
type MonitorEvent struct {
	Family string   // The "~KEYWORD": DEVICE, OUTPUT, GROUP, MONITORING, SYSTEM, ERROR.
	Fields []string // Comma-split fields after the keyword (integration id, component, action, params).
	Raw    string   // The original line, for passthrough consumers.
}

// monitorFn is called for every monitoring message the panel reports.
type monitorFn func(ev MonitorEvent)

// Device supervises a single Lutron interface: it owns the connection, reads
// integration responses, writes changed zone levels, and arbitrates between the
// control sources that want to drive its zones.
type Device struct {
	cfg   *DeviceConfig
	zones int

	// Connection ownership. connMu guards conn and serializes writes.
	connMu sync.Mutex
	conn   transport

	// Outbound pacing. paceMu serializes all outbound messages and enforces the
	//  protocol's minimum inter-message delays; nextSendAt is the earliest time
	//  the next message may be sent. lastWasZoneOp records whether the previous
	//  send was a zone-value operation so a run of them can burst together.
	paceMu        sync.Mutex
	nextSendAt    time.Time
	lastWasZoneOp bool

	// Zone and arbitration state.
	dataMu          sync.Mutex
	target          []byte      // Desired zone levels (0-255).
	sent            []byte      // Last levels written to the panel.
	reported        []byte      // Last levels reported by the panel.
	zoneFade        []string    // Per-zone fade time set by the last source to drive each zone.
	zoneRamping     []bool      // Per-zone raise/lower in progress; writes are suppressed until stop.
	zoneSuppress    []time.Time // Per-zone write-suppression deadline (raise/lower stop grace).
	shadeHeld       map[int]int // Per-column held shade component, released on stop.
	lastZoneQuery   []time.Time // When each zone was last queried, to spot solicited replies.
	lastZoneWrite   []time.Time // When each zone was last written, to spot reports echoing our own writes.
	sendAll         bool        // Force a full resend on the next write pass.
	controlDisabled bool        // Panel signalled integration control is disabled.
	resyncing       bool        // Connect-time level sync in progress; suppress zone writes.
	pendingResync   int         // Zone reports still awaited before the resync is complete.
	bindings        []*sourceBinding
	feedbackFns     []zoneFeedbackFn

	// Generic monitoring fan-out. monitorFns receive every "~" feedback line;
	//  requestedMonitoring is the union of #MONITORING types sources have asked the
	//  panel to report, enabled on connect alongside the built-in defaults.
	monitorFns          []monitorFn
	requestedMonitoring map[int]bool

	// Scene state.
	sceneFeedbackFns []sceneFeedbackFn
	sceneMonitoring  bool      // A scene source is attached; enable scene monitoring.
	sceneSuppress    time.Time // Suppress zone writes until this time (a scene is holding).

	// Watchdog timestamps (Unix nanoseconds).
	lastResponse atomicTime
	lastWrite    atomicTime
	lastReset    atomicTime

	// Lifecycle.
	ctx       context.Context
	reconnect chan string
	wg        sync.WaitGroup
}

// NewDevice constructs a device from its configuration.
func NewDevice(cfg *DeviceConfig) *Device {
	d := &Device{
		cfg:           cfg,
		zones:         cfg.Zones,
		target:        make([]byte, cfg.Zones),
		sent:          make([]byte, cfg.Zones),
		reported:      make([]byte, cfg.Zones),
		zoneFade:      make([]string, cfg.Zones),
		zoneRamping:   make([]bool, cfg.Zones),
		zoneSuppress:  make([]time.Time, cfg.Zones),
		shadeHeld:     make(map[int]int),
		lastZoneQuery: make([]time.Time, cfg.Zones),
		lastZoneWrite: make([]time.Time, cfg.Zones),
		sendAll:       true,
		reconnect:     make(chan string, 1),

		requestedMonitoring: make(map[int]bool),
	}
	return d
}

// Name returns the device's configured name.
func (d *Device) Name() string { return d.cfg.Name }

// Zones returns the number of controllable zones.
func (d *Device) Zones() int { return d.zones }

// RegisterSource registers a control source for arbitration and returns its
// binding handle. It must be called before Start.
func (d *Device) RegisterSource(name string, priority int, hold time.Duration, fade string) *sourceBinding {
	b := &sourceBinding{name: name, priority: priority, hold: hold, fade: fade}
	d.dataMu.Lock()
	d.bindings = append(d.bindings, b)
	d.dataMu.Unlock()
	return b
}

// OnZoneFeedback registers a callback invoked when the panel reports a zone's
// level. It must be called before Start.
func (d *Device) OnZoneFeedback(fn zoneFeedbackFn) {
	d.dataMu.Lock()
	d.feedbackFns = append(d.feedbackFns, fn)
	d.dataMu.Unlock()
}

// OnSceneFeedback registers a callback invoked when the panel reports the active
// scene. It must be called before Start.
func (d *Device) OnSceneFeedback(fn sceneFeedbackFn) {
	d.dataMu.Lock()
	d.sceneFeedbackFns = append(d.sceneFeedbackFns, fn)
	d.dataMu.Unlock()
}

// NoteSceneControl marks that a scene source is attached so scene monitoring is
// enabled on connect. It must be called before Start.
func (d *Device) NoteSceneControl() {
	d.dataMu.Lock()
	d.sceneMonitoring = true
	d.dataMu.Unlock()
}

// OnMonitor registers a callback invoked for every monitoring message the panel
// reports. It must be called before Start.
func (d *Device) OnMonitor(fn monitorFn) {
	d.dataMu.Lock()
	d.monitorFns = append(d.monitorFns, fn)
	d.dataMu.Unlock()
}

// NoteMonitoring records #MONITORING types a source wants the panel to report so
// the union is enabled on connect. Type 255 requests all monitoring. It must be
// called before Start.
func (d *Device) NoteMonitoring(types ...int) {
	d.dataMu.Lock()
	for _, t := range types {
		if t > 0 {
			d.requestedMonitoring[t] = true
		}
	}
	d.dataMu.Unlock()
}

// lockedOut reports whether a higher-priority source is currently active,
// blocking this binding from taking control. Caller must hold dataMu.
func (d *Device) lockedOut(b *sourceBinding, now time.Time) bool {
	for _, other := range d.bindings {
		if other == b {
			continue
		}
		if other.priority > b.priority && other.activeAt(now) {
			return true
		}
	}
	return false
}

// ApplyZoneLevels sets the target level for a set of 1-indexed zones on behalf of
// a source, leaving every other zone untouched (a source only ever drives the
// zones it owns). It returns false when the caller can't take control — either a
// higher-priority source is currently active, or the panel has signalled that
// integration control is disabled — in which case the levels are ignored and the
// caller should reflect the device's real state instead of its request.
func (d *Device) ApplyZoneLevels(b *sourceBinding, levels map[int]byte) bool {
	d.dataMu.Lock()
	defer d.dataMu.Unlock()

	now := time.Now()
	if d.controlDisabled || d.lockedOut(b, now) {
		return false
	}

	b.lastZoneActive = now
	// Zone control takes over from any active scene.
	d.sceneSuppress = time.Time{}
	for zone, level := range levels {
		if zone >= 1 && zone <= d.zones {
			d.target[zone-1] = level
			d.zoneFade[zone-1] = b.fade
			// An absolute level cancels any raise/lower in progress for the zone.
			d.zoneRamping[zone-1] = false
			d.zoneSuppress[zone-1] = time.Time{}
		}
	}
	return true
}

// RefreshZoneActivity re-asserts a source's zone activity without changing any
// levels, keeping its arbitration lock alive between value changes. A
// continuously streaming source (e.g. sACN holding a static look) needs this
// because its receiver only reports changed frames, so a steady stream would
// otherwise look idle once the hold window elapsed and let a lower-priority
// source take over. It is a no-op while integration control is disabled or a
// higher-priority source holds control, matching ApplyZoneLevels.
func (d *Device) RefreshZoneActivity(b *sourceBinding) {
	d.dataMu.Lock()
	defer d.dataMu.Unlock()

	now := time.Now()
	if d.controlDisabled || d.lockedOut(b, now) {
		return
	}
	b.lastZoneActive = now
}

// ApplyScene activates a GRAFIK Eye scene on behalf of a source. It returns false
// when a higher-priority source is active. While the scene holds, zone writes are
// suppressed so the scene isn't immediately overwritten by a resend.
func (d *Device) ApplyScene(b *sourceBinding, scene int) bool {
	d.dataMu.Lock()
	now := time.Now()
	if d.controlDisabled || d.lockedOut(b, now) {
		d.dataMu.Unlock()
		return false
	}
	b.lastSceneActive = now
	d.sceneSuppress = now.Add(b.hold)
	d.dataMu.Unlock()

	// Activate the scene via the Scene Controller component (141), action 7.
	cmd := fmt.Sprintf("#DEVICE,%d,%d,%d,%d", d.cfg.IntegrationID, qseSceneController, qseActionScene, scene)
	log.Infof("[%s] Activating scene %d", d.cfg.Name, scene)
	log.Debugf("[%s] TX %s", d.cfg.Name, cmd)
	d.write(cmd + qseTerminator)
	return true
}

// deviceCommand formats a "#DEVICE,<id>,<component>,<action>[,<params>...]" line,
// terminated for transmission.
func (d *Device) deviceCommand(component, action int, params ...string) string {
	cmd := fmt.Sprintf("#DEVICE,%d,%d,%d", d.cfg.IntegrationID, component, action)
	if len(params) > 0 {
		cmd += "," + strings.Join(params, ",")
	}
	return cmd + qseTerminator
}

// RaiseZone starts raising a zone's level on behalf of a source and holds until a
// matching StopZone. Zone writes for the zone are suppressed while it moves so the
// resend loop can't interrupt the ramp; the new level is followed via zone
// monitoring. It returns false when the caller can't take control (a higher-
// priority source is active or integration control is disabled).
func (d *Device) RaiseZone(b *sourceBinding, zone int) bool {
	return d.moveZone(b, zone, qseActionStartRaising)
}

// LowerZone starts lowering a zone's level; see RaiseZone.
func (d *Device) LowerZone(b *sourceBinding, zone int) bool {
	return d.moveZone(b, zone, qseActionStartLowering)
}

// moveZone begins a raise or lower on a zone, marking it ramping so the writer
// leaves it alone until StopZone.
func (d *Device) moveZone(b *sourceBinding, zone, action int) bool {
	if zone < 1 || zone > d.zones {
		return false
	}
	d.dataMu.Lock()
	now := time.Now()
	if d.controlDisabled || d.lockedOut(b, now) {
		d.dataMu.Unlock()
		return false
	}
	b.lastZoneActive = now
	d.sceneSuppress = time.Time{}
	d.zoneRamping[zone-1] = true
	d.dataMu.Unlock()

	d.write(d.deviceCommand(zone, action))
	return true
}

// StopZone halts a zone started by RaiseZone or LowerZone. Zone writes stay
// suppressed for a short grace period so the panel's final level report updates
// the target before the writer resumes, keeping the zone at its resting level.
// Stop is always sent (even when locked out) so a moving zone can be halted.
func (d *Device) StopZone(b *sourceBinding, zone int) bool {
	if zone < 1 || zone > d.zones {
		return false
	}
	d.dataMu.Lock()
	now := time.Now()
	b.lastZoneActive = now
	d.zoneRamping[zone-1] = false
	d.zoneSuppress[zone-1] = now.Add(zoneStopGrace)
	d.dataMu.Unlock()

	d.write(d.deviceCommand(zone, qseActionStopMoving))
	return true
}

// ApplyShade drives a shade column on behalf of a source. The action is one of
// "open", "close", "preset", "raise", "lower", or "stop". Open, close, and preset
// are momentary (press then release); raise and lower start motion and are halted
// by a subsequent "stop". It returns false for an unknown column or action.
func (d *Device) ApplyShade(b *sourceBinding, column int, action string) bool {
	col, ok := qseShadeColumns[column]
	if !ok {
		return false
	}
	switch action {
	case "open":
		d.pressRelease(col.open)
	case "close":
		d.pressRelease(col.close)
	case "preset":
		d.pressRelease(col.preset)
	case "raise":
		d.shadeStart(column, col.raise)
	case "lower":
		d.shadeStart(column, col.lower)
	case "stop":
		d.shadeStop(column)
	default:
		return false
	}
	return true
}

// pressRelease issues a press then release on a phantom-button component, which
// the panel acts on at the release.
func (d *Device) pressRelease(component int) {
	d.write(d.deviceCommand(component, qseBtnAction))
	d.write(d.deviceCommand(component, qseActionRelease))
}

// shadeStart presses a shade raise/lower component to begin motion, recording the
// held component so a later stop releases it. A held opposite direction is
// released first.
func (d *Device) shadeStart(column, component int) {
	d.dataMu.Lock()
	prev := d.shadeHeld[column]
	d.shadeHeld[column] = component
	d.dataMu.Unlock()
	if prev != 0 && prev != component {
		d.write(d.deviceCommand(prev, qseActionRelease))
	}
	d.write(d.deviceCommand(component, qseBtnAction))
}

// shadeStop releases the held shade component for a column, halting motion.
func (d *Device) shadeStop(column int) {
	d.dataMu.Lock()
	component := d.shadeHeld[column]
	delete(d.shadeHeld, column)
	d.dataMu.Unlock()
	if component != 0 {
		d.write(d.deviceCommand(component, qseActionRelease))
	}
}

// SetZoneLock enables or disables the zone lock (QS Standalone).
func (d *Device) SetZoneLock(on bool) {
	d.write(d.deviceCommand(qseSceneController, qseActionZoneLock, boolParam(on)))
}

// SetSceneLock enables or disables the scene lock (QS Standalone).
func (d *Device) SetSceneLock(on bool) {
	d.write(d.deviceCommand(qseSceneController, qseActionSceneLock, boolParam(on)))
}

// SetSequence sets the scene sequence state: 0 off, 1 scenes 1-4, 2 scenes 5-16.
func (d *Device) SetSequence(mode int) bool {
	if mode < 0 || mode > 2 {
		return false
	}
	d.write(d.deviceCommand(qseSceneController, qseActionSequence, strconv.Itoa(mode)))
	return true
}

// SceneOff activates the Scene Off button on behalf of a source, turning all
// zones off. Like ApplyScene it suppresses zone writes while the off look holds
// so the writer doesn't immediately re-light the zones.
func (d *Device) SceneOff(b *sourceBinding) bool {
	d.dataMu.Lock()
	now := time.Now()
	if d.controlDisabled || d.lockedOut(b, now) {
		d.dataMu.Unlock()
		return false
	}
	b.lastSceneActive = now
	d.sceneSuppress = now.Add(b.hold)
	d.dataMu.Unlock()

	d.pressRelease(qseSceneOffButton)
	return true
}

// boolParam renders a Lutron 0/1 on/off parameter.
func boolParam(on bool) string {
	if on {
		return "1"
	}
	return "0"
}

// ZoneTargets returns a snapshot of the current target zone levels.
func (d *Device) ZoneTargets() []byte {
	d.dataMu.Lock()
	defer d.dataMu.Unlock()
	out := make([]byte, d.zones)
	copy(out, d.target)
	return out
}

// Start launches the supervisor and worker loops.
func (d *Device) Start(ctx context.Context) {
	d.ctx = ctx
	d.wg.Add(1)
	go d.supervisor(ctx)
	d.wg.Add(1)
	go d.writeLoop(ctx)
	d.wg.Add(1)
	go d.resendLoop(ctx)
	d.wg.Add(1)
	go d.watchdog(ctx)
}

// Stop closes the connection and waits for the loops to exit.
func (d *Device) Stop() {
	d.closeConn()
	d.wg.Wait()
}

// supervisor owns the (re)connect path. Other loops request a reconnect and the
// supervisor reopens the link, then resyncs zone levels from the panel.
func (d *Device) supervisor(ctx context.Context) {
	defer d.wg.Done()
	for {
		if ctx.Err() != nil {
			return
		}

		// Reset per-connection state before the connection is exposed. Writes are
		//  suppressed until the connect-time query sync populates the current zone
		//  levels, so the writer can't push its zero-initialized targets and
		//  momentarily drive every zone to off. Setting the gate before connect
		//  closes the window where the writer could see the new connection but not
		//  yet the resync flag.
		d.dataMu.Lock()
		d.sendAll = true
		d.resyncing = true
		d.pendingResync = d.zones
		d.dataMu.Unlock()

		conn := d.connect(ctx)
		if conn == nil {
			return // Shutting down.
		}
		d.lastResponse.set(time.Now())

		// Configure monitoring and resync zone levels concurrently with the
		//  reader. The zone queries are paced 1500 ms apart, so running them inline
		//  would block reading for the whole burst and leave the panel's replies
		//  sitting unread in the port buffer; spawning them lets responses be
		//  processed as they arrive.
		syncDone := make(chan struct{})
		go func() {
			defer close(syncDone)
			d.setupMonitoring()
			d.queryZoneLevels()
			// Failsafe: the resync gate normally clears as each zone reports its
			//  level. If a zone never replies, give the last query's response time
			//  to arrive, then lift the gate so writes aren't suppressed forever.
			select {
			case <-ctx.Done():
			case <-time.After(queryInterMessageDelay):
			}
			d.dataMu.Lock()
			d.resyncing = false
			d.dataMu.Unlock()
		}()

		// Read until the connection fails or a reconnect is requested.
		d.readLoop(ctx, conn)

		// Let the sync goroutine unwind before reconnecting so it can't write to
		//  the next connection. Once the link is down its paced writes fail fast.
		<-syncDone

		select {
		case <-ctx.Done():
			return
		default:
		}
	}
}

// connect opens the transport, retrying with exponential backoff until it
// succeeds or the context is cancelled (in which case it returns nil).
func (d *Device) connect(ctx context.Context) transport {
	backoff := time.Duration(d.cfg.Reliability.ReconnectBackoffMinSec) * time.Second
	maxBackoff := time.Duration(d.cfg.Reliability.ReconnectBackoffMaxSec) * time.Second
	// Guard against a zero/negative configured backoff, which would spin the retry
	//  loop and never grow (0*2 stays 0).
	if backoff <= 0 {
		backoff = time.Second
	}
	if maxBackoff < backoff {
		maxBackoff = backoff
	}
	for {
		if ctx.Err() != nil {
			return nil
		}
		log.Infof("[%s] Opening %s connection", d.cfg.Name, d.cfg.Transport)
		conn, err := openTransport(d.cfg)
		if err == nil {
			d.connMu.Lock()
			d.conn = conn
			d.connMu.Unlock()
			// Drain any stale reconnect request from a prior connection.
			select {
			case <-d.reconnect:
			default:
			}
			log.Infof("[%s] Connection established", d.cfg.Name)
			return conn
		}
		log.Errorf("[%s] Connect failed: %s; retry in %s", d.cfg.Name, err, backoff)

		select {
		case <-ctx.Done():
			return nil
		case <-time.After(backoff):
		}
		backoff = min(backoff*2, maxBackoff)
	}
}

// requestReconnect closes the current connection and wakes the supervisor.
func (d *Device) requestReconnect(reason string) {
	log.Warnf("[%s] Reconnect requested: %s", d.cfg.Name, reason)
	d.closeConn()
	select {
	case d.reconnect <- reason:
	default:
	}
}

// closeConn closes the current connection if any.
func (d *Device) closeConn() {
	d.connMu.Lock()
	if d.conn != nil {
		d.conn.Close()
		d.conn = nil
	}
	d.connMu.Unlock()
}

// write sends a payload to the panel, serialized and paced to honor the
// protocol's minimum inter-message delays, and requests a reconnect on failure.
// It returns true on success.
func (d *Device) write(payload string) bool {
	// Queries ("?") and commands ("#") carry different post-send delays.
	isQuery := len(payload) > 0 && payload[0] == '?'
	zoneOp := isZoneValueOp(payload)

	// Serialize all outbound messages. A run of zone-value operations bursts so
	//  every zone changes at once instead of cascading; the inter-message delay is
	//  only owed when this message differs in kind from the previous send.
	d.paceMu.Lock()
	defer d.paceMu.Unlock()
	if !(zoneOp && d.lastWasZoneOp) {
		d.paceWait()
	}

	d.connMu.Lock()
	conn := d.conn
	if conn == nil {
		d.connMu.Unlock()
		return false
	}
	_, err := conn.Write([]byte(payload))
	d.connMu.Unlock()

	// Schedule the earliest time a differing next message may be sent.
	delay := commandInterMessageDelay
	if isQuery {
		delay = queryInterMessageDelay
	}
	d.nextSendAt = time.Now().Add(delay)
	d.lastWasZoneOp = zoneOp

	if err != nil {
		log.Errorf("[%s] Write failed: %s", d.cfg.Name, err)
		d.requestReconnect("write error")
		return false
	}
	d.lastWrite.set(time.Now())
	return true
}

// isZoneValueOp reports whether a payload is a zone-level set ("#DEVICE,...,14,...").
// Consecutive zone-level sets burst without the inter-message delay so all zones
// change together rather than cascading. Zone-level queries ("?DEVICE,...,14") are
// deliberately excluded: the protocol mandates a 1500 ms gap between queries, so
// they must pace normally rather than burst.
func isZoneValueOp(payload string) bool {
	body := strings.TrimRight(payload, qseTerminator)
	if !strings.HasPrefix(body, "#DEVICE,") {
		return false
	}
	fields := strings.Split(body, ",")
	return len(fields) >= 4 && fields[3] == strconv.Itoa(qseActionZoneLevel)
}

// paceWait blocks until the minimum inter-message delay owed from the previous
// send has elapsed, returning early on shutdown. It must be called with paceMu
// held.
func (d *Device) paceWait() {
	wait := time.Until(d.nextSendAt)
	if wait <= 0 {
		return
	}
	timer := time.NewTimer(wait)
	defer timer.Stop()
	select {
	case <-d.ctx.Done():
	case <-timer.C:
	}
}

// readLoop reads CRLF-terminated integration responses from conn until it
// fails, dispatching each line. It returns so the supervisor can reconnect.
func (d *Device) readLoop(ctx context.Context, conn transport) {
	buf := make([]byte, 0, 256)
	tmp := make([]byte, 256)
	for {
		n, err := conn.Read(tmp)
		if err != nil {
			if ctx.Err() == nil {
				log.Errorf("[%s] Read failed: %s", d.cfg.Name, err)
				d.requestReconnect("read error")
			}
			return
		}
		if n == 0 {
			continue
		}
		buf = append(buf, tmp[:n]...)

		// Extract and handle each complete line.
		for {
			i := bytes.IndexByte(buf, '\n')
			if i < 0 {
				break
			}
			line := strings.TrimRight(string(buf[:i]), "\r")
			buf = buf[i+1:]
			d.handleLine(line)
		}

		// Guard against an unterminated flood growing the buffer without bound.
		if len(buf) > 4096 {
			buf = buf[:0]
		}
	}
}

// handleLine parses one integration response and updates state.
func (d *Device) handleLine(line string) {
	// The QSE echoes a "QSE>" prompt; strip it and trim whitespace.
	line = strings.TrimSpace(strings.ReplaceAll(line, "QSE>", ""))
	if line == "" {
		return
	}
	d.lastResponse.set(time.Now())
	log.Debugf("[%s] RX %s", d.cfg.Name, line)

	upper := strings.ToUpper(line)
	switch {
	case strings.HasPrefix(upper, "~ERROR"):
		d.handleError(line)
	case d.cfg.DisableComponent != 0 && line == d.disableSignal():
		log.Infof("[%s] Received disable signal", d.cfg.Name)
		d.dataMu.Lock()
		d.controlDisabled = true
		d.dataMu.Unlock()
	case d.cfg.EnableComponent != 0 && line == d.enableSignal():
		log.Infof("[%s] Received enable signal", d.cfg.Name)
		d.dataMu.Lock()
		d.controlDisabled = false
		d.sendAll = true
		d.dataMu.Unlock()
	case strings.HasPrefix(line, d.devicePrefix()):
		d.handleDeviceFeedback(line)
	}

	// Relay every "~" report to generic monitor subscribers (OSC/MQTT), in
	//  addition to the typed handling above which drives arbitration and HA state.
	d.dispatchMonitor(line)
}

// dispatchMonitor parses a "~" feedback line into a MonitorEvent and fans it out
// to monitor subscribers. Non-"~" lines (echoes, prompts) are ignored.
func (d *Device) dispatchMonitor(line string) {
	if !strings.HasPrefix(line, "~") {
		return
	}
	d.dataMu.Lock()
	if len(d.monitorFns) == 0 {
		d.dataMu.Unlock()
		return
	}
	fns := d.monitorFns
	d.dataMu.Unlock()

	fields := strings.Split(line, ",")
	ev := MonitorEvent{
		Family: strings.ToUpper(strings.TrimPrefix(fields[0], "~")),
		Fields: fields[1:],
		Raw:    line,
	}
	for _, fn := range fns {
		fn(ev)
	}
}

// handleDeviceFeedback dispatches a "~DEVICE,<id>,..." response to the zone-level
// or scene handler based on its component and action.
func (d *Device) handleDeviceFeedback(line string) {
	fields := strings.Split(line, ",")
	if len(fields) < 4 || fields[1] != strconv.Itoa(d.cfg.IntegrationID) {
		return
	}
	switch {
	case fields[3] == strconv.Itoa(qseActionZoneLevel):
		d.handleZoneFeedback(fields)
	case fields[2] == strconv.Itoa(qseSceneController) && fields[3] == strconv.Itoa(qseActionScene):
		d.handleSceneFeedback(fields)
	}
}

// handleZoneFeedback parses a "~DEVICE,<id>,<zone>,14,<level%>" response, records
// the reported level, and mirrors it to feedback subscribers. The target is
// synced to the reported level during the connect-time resync and when an
// unsolicited report reflects a genuine external change (a keypad or scene), so
// the writer follows it instead of fighting it. A report that merely echoes one
// of our own recent writes or query probes is not adopted: the panel quantizes
// levels to its own scale, so its echo rarely byte-matches what we sent, and
// adopting it back would create a self-sustaining write loop (and re-populate
// zones mid-fade that a release is trying to drive to 0).
func (d *Device) handleZoneFeedback(fields []string) {
	if len(fields) < 5 {
		return
	}
	zone, err := strconv.Atoi(fields[2])
	if err != nil || zone < 1 || zone > d.zones {
		return
	}
	pct, err := strconv.ParseFloat(fields[4], 64)
	if err != nil {
		return
	}
	level := byte(math.Round(pct / 100.0 * 255.0))

	d.dataMu.Lock()
	d.reported[zone-1] = level
	// Classify the report: a solicited reply follows our own query, an echo follows
	//  our own write. Neither is an external change, so neither should move the
	//  target. Adopt only a resync reply or a genuine unsolicited report.
	solicited := time.Since(d.lastZoneQuery[zone-1]) < queryReplyWindow
	echo := time.Since(d.lastZoneWrite[zone-1]) < zoneEchoWindow
	resyncing := d.resyncing
	adopt := resyncing || (!solicited && !echo)
	if adopt {
		d.target[zone-1] = level
		d.sent[zone-1] = level
	}
	// Lift the connect-time resync gate once every zone has reported, so the
	//  writer resumes with targets that mirror the panel rather than zeros.
	if resyncing && d.pendingResync > 0 {
		d.pendingResync--
		if d.pendingResync == 0 {
			d.resyncing = false
		}
	}
	// Mirror genuine reports to subscribers (e.g. Home Assistant), but not a
	//  solicited probe reply, which is for liveness only and must not flip
	//  subscribers to a value that disagrees with the level we maintain.
	var fns []zoneFeedbackFn
	if resyncing || !solicited {
		fns = d.feedbackFns
	}
	d.dataMu.Unlock()

	for _, fn := range fns {
		fn(zone, level)
	}
}

// handleSceneFeedback parses a "~DEVICE,<id>,141,7,<scene>" response and notifies
// scene subscribers.
func (d *Device) handleSceneFeedback(fields []string) {
	if len(fields) < 5 {
		return
	}
	scene, err := strconv.Atoi(fields[4])
	if err != nil {
		return
	}
	d.dataMu.Lock()
	fns := d.sceneFeedbackFns
	d.dataMu.Unlock()

	for _, fn := range fns {
		fn(scene)
	}
}

// handleError logs a ~ERROR response and, for the NWK-lockup error, resets the
// interface. The reset is rate-limited because a wedged NWK floods the error.
func (d *Device) handleError(line string) {
	code := ""
	if _, after, ok := strings.Cut(line, ","); ok {
		code = strings.TrimSpace(after)
	}
	desc := qseErrorDescriptions[code]
	if desc == "" {
		desc = "unknown error"
	}

	// Error 6 ("unsupported command") is also the symptom of the NWK lockup that
	//  only #RESET,0 clears, so we recover from it; other errors are logged only.
	if code == "6" {
		cooldown := time.Duration(d.cfg.Reliability.ResetCooldownSec) * time.Second
		if time.Since(d.lastReset.get()) >= cooldown {
			d.lastReset.set(time.Now())
			log.Warnf("[%s] NWK returned %s (%s); sending #RESET,0", d.cfg.Name, line, desc)
			d.write("#RESET,0" + qseTerminator)
		}
		return
	}
	log.Warnf("[%s] NWK returned %s (%s)", d.cfg.Name, line, desc)
}

// writeLoop periodically pushes changed zone levels out to the panel.
func (d *Device) writeLoop(ctx context.Context) {
	defer d.wg.Done()
	ticker := time.NewTicker(writeInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			d.flushZones()
		}
	}
}

// flushZones writes any zone levels that changed since the last pass.
func (d *Device) flushZones() {
	d.dataMu.Lock()
	// Don't push zone levels while the connect-time resync is still in progress,
	//  while integration control is disabled, or while a scene is holding (so the
	//  scene isn't immediately overwritten).
	if d.resyncing || d.controlDisabled || time.Now().Before(d.sceneSuppress) {
		d.dataMu.Unlock()
		return
	}
	now := time.Now()
	targets := make([]byte, d.zones)
	copy(targets, d.target)
	fades := make([]string, d.zones)
	copy(fades, d.zoneFade)
	ramping := make([]bool, d.zones)
	copy(ramping, d.zoneRamping)
	suppress := make([]time.Time, d.zones)
	copy(suppress, d.zoneSuppress)
	// Snapshot the last-sent levels under the lock too; the read loop writes them
	//  from handleZoneFeedback, so comparing against d.sent directly would race.
	sent := make([]byte, d.zones)
	copy(sent, d.sent)
	resendAll := d.sendAll
	d.sendAll = false
	d.dataMu.Unlock()

	// Skip writing entirely if the link is down; the supervisor is reconnecting.
	d.connMu.Lock()
	connected := d.conn != nil
	d.connMu.Unlock()
	if !connected {
		return
	}

	for zone := range d.zones {
		// Skip zones being raised/lowered and those still in their post-stop grace
		//  so the writer can't interrupt a ramp or yank a zone back to its
		//  pre-move level before the panel reports the resting level.
		if ramping[zone] || now.Before(suppress[zone]) {
			continue
		}
		if !resendAll && targets[zone] == sent[zone] {
			continue
		}
		if !d.sendZoneLevel(zone+1, targets[zone], fades[zone]) {
			// Write failed: mark a full resend for next pass and let the supervisor
			//  reopen the port.
			d.dataMu.Lock()
			d.sendAll = true
			d.dataMu.Unlock()
			return
		}
		// Record the write so the panel's echo of it isn't mistaken for an
		//  external change in handleZoneFeedback.
		d.dataMu.Lock()
		d.sent[zone] = targets[zone]
		d.lastZoneWrite[zone] = time.Now()
		d.dataMu.Unlock()
	}
}

// sendZoneLevel writes a single zone's level as a percentage with the given fade
// time, falling back to the device default when the source set none.
func (d *Device) sendZoneLevel(zone int, value byte, fade string) bool {
	if fade == "" {
		fade = d.cfg.Fade
	}
	pct := math.Round(float64(value)/255.0*100.0*100.0) / 100.0
	cmd := fmt.Sprintf("#DEVICE,%d,%d,%d,%.2f,%s",
		d.cfg.IntegrationID, zone, qseActionZoneLevel, pct, fade)
	log.Debugf("[%s] TX %s", d.cfg.Name, cmd)
	return d.write(cmd + qseTerminator)
}

// setupMonitoring configures which integration messages the panel reports. The
// protocol advises enabling only what is needed; we ensure zone-level and reply
// monitoring are on (so feedback works regardless of the unit's programmed
// defaults) plus scene monitoring when a scene source is attached, then apply any
// configured overrides.
func (d *Device) setupMonitoring() {
	m := d.cfg.Monitoring
	if m.Manage == nil || *m.Manage {
		d.setMonitoring(qseMonitorZone, true)
		d.setMonitoring(qseMonitorReply, true)
		d.dataMu.Lock()
		wantScene := d.sceneMonitoring
		// Snapshot the source-requested types, skipping the built-ins already sent.
		requested := make([]int, 0, len(d.requestedMonitoring))
		for t := range d.requestedMonitoring {
			if t != qseMonitorZone && t != qseMonitorReply {
				requested = append(requested, t)
			}
		}
		d.dataMu.Unlock()
		if wantScene {
			d.setMonitoring(qseMonitorScene, true)
		}
		// Enable the union of monitoring types sources asked to relay.
		for _, t := range requested {
			if t == qseMonitorScene && wantScene {
				continue // Already enabled above.
			}
			d.setMonitoring(t, true)
		}
	}
	for _, t := range m.Enable {
		d.setMonitoring(t, true)
	}
	for _, t := range m.Disable {
		d.setMonitoring(t, false)
	}
}

// setMonitoring enables or disables one monitoring type via #MONITORING.
func (d *Device) setMonitoring(monitorType int, enable bool) {
	action := 2 // Disable.
	if enable {
		action = 1 // Enable.
	}
	cmd := fmt.Sprintf("#MONITORING,%d,%d", monitorType, action)
	log.Debugf("[%s] TX %s", d.cfg.Name, cmd)
	d.write(cmd + qseTerminator)
}

// queryZoneLevels asks the panel for each zone's current level so our view
// matches reality after a (re)connect. Replies are handled in handleLine. The
// queries are paced (1500 ms apart) per the protocol, so this is used only for
// the one-time connect sync; ongoing updates arrive via zone monitoring.
func (d *Device) queryZoneLevels() {
	for zone := 1; zone <= d.zones; zone++ {
		d.queryZone(zone)
	}
}

// queryZone asks the panel for one zone's current level. It records the query
// time so the solicited reply is used for liveness only and does not override the
// level we maintain (see handleZoneFeedback).
func (d *Device) queryZone(zone int) {
	query := fmt.Sprintf("?DEVICE,%d,%d,%d", d.cfg.IntegrationID, zone, qseActionZoneLevel)
	d.dataMu.Lock()
	d.lastZoneQuery[zone-1] = time.Now()
	d.dataMu.Unlock()
	log.Debugf("[%s] TX %s", d.cfg.Name, query)
	d.write(query + qseTerminator)
}

// resendLoop forces a periodic full resend so the panel can't drift out of sync.
// A zero interval disables the periodic refresh: zones are then sent only when a
// target changes (reconnect and recovery resyncs still re-assert full state).
func (d *Device) resendLoop(ctx context.Context) {
	defer d.wg.Done()
	interval := time.Duration(d.cfg.Reliability.SendAllIntervalSec) * time.Second
	if interval <= 0 {
		return
	}
	rxTimeout := time.Duration(d.cfg.Reliability.RxTimeoutSec) * time.Second
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			// Skip resends when the link looks unhealthy; the watchdog handles it.
			if time.Since(d.lastResponse.get()) > rxTimeout {
				continue
			}
			d.dataMu.Lock()
			d.sendAll = true
			d.dataMu.Unlock()
		}
	}
}

// watchdog detects a stale link and reconnects it. The panel only
// replies to a #DEVICE write when a level actually changes, so a quiet RX is not
// itself a fault: we first probe the link (which gets a reply on a healthy link)
// and only reconnect if that probe also goes unanswered.
func (d *Device) watchdog(ctx context.Context) {
	defer d.wg.Done()
	interval := time.Duration(d.cfg.Reliability.WatchdogIntervalSec) * time.Second
	rxTimeout := time.Duration(d.cfg.Reliability.RxTimeoutSec) * time.Second
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	probed := false
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}

		now := time.Now()
		wroteRecently := now.Sub(d.lastWrite.get()) < rxTimeout
		staleFor := now.Sub(d.lastResponse.get())
		if wroteRecently && staleFor > rxTimeout {
			if !probed {
				d.probeLink(staleFor)
				probed = true
			} else {
				d.requestReconnect(fmt.Sprintf("no reply to probe; silent for %.0fs", staleFor.Seconds()))
				probed = false
			}
		} else {
			probed = false
		}
	}
}

// probeLink elicits a reply to confirm the link is alive and re-asserts state on
// recovery. It prefers re-asserting monitoring: the panel acks with ~MONITORING
// (proving liveness) and re-enabling the reply path recovers a NWK whose poll
// state has gone stale after idle — a bare ?DEVICE poll returns a stale level
// (e.g. 0.01) on such a link while monitoring stays accurate, so we lean on
// monitoring rather than polling. When monitoring is left to the unit's
// programming, fall back to a zone query whose reply (value ignored) still proves
// the link is alive. staleFor is how long the link has been silent, for logging.
func (d *Device) probeLink(staleFor time.Duration) {
	// Re-drive every zone's target on recovery. The probe itself touches only one
	//  channel, but the resend loop pauses while RX is stale, so all zones — not
	//  just the probed one — must be re-asserted once the link is back.
	d.dataMu.Lock()
	d.sendAll = true
	d.dataMu.Unlock()

	m := d.cfg.Monitoring
	if m.Manage == nil || *m.Manage {
		log.Infof("[%s] No reply for %.0fs while sending; re-asserting monitoring to test the link and restore the reply path",
			d.cfg.Name, staleFor.Seconds())
		d.setMonitoring(qseMonitorReply, true)
		d.setMonitoring(qseMonitorZone, true)
		// Re-assert source-requested monitoring too so the relay resumes after a
		//  NWK reply-path reset, not just the built-in zone/reply path.
		d.dataMu.Lock()
		requested := make([]int, 0, len(d.requestedMonitoring))
		for t := range d.requestedMonitoring {
			if t != qseMonitorZone && t != qseMonitorReply {
				requested = append(requested, t)
			}
		}
		d.dataMu.Unlock()
		for _, t := range requested {
			d.setMonitoring(t, true)
		}
		return
	}
	log.Infof("[%s] No reply for %.0fs while sending; querying zone 1 to test the link",
		d.cfg.Name, staleFor.Seconds())
	d.queryZone(1)
}

// devicePrefix is the "~DEVICE,<id>" prefix for this device's feedback lines.
func (d *Device) devicePrefix() string {
	return fmt.Sprintf("~DEVICE,%d", d.cfg.IntegrationID)
}

// disableSignal is the integration-control-disabled signal line.
func (d *Device) disableSignal() string {
	return fmt.Sprintf("~DEVICE,%d,%d,%d", d.cfg.IntegrationID, d.cfg.DisableComponent, qseBtnAction)
}

// enableSignal is the integration-control-enabled signal line.
func (d *Device) enableSignal() string {
	return fmt.Sprintf("~DEVICE,%d,%d,%d", d.cfg.IntegrationID, d.cfg.EnableComponent, qseBtnAction)
}

// atomicTime is a small mutex-guarded time value shared between loops.
type atomicTime struct {
	mu sync.Mutex
	t  time.Time
}

func (a *atomicTime) set(t time.Time) {
	a.mu.Lock()
	a.t = t
	a.mu.Unlock()
}

func (a *atomicTime) get() time.Time {
	a.mu.Lock()
	defer a.mu.Unlock()
	return a.t
}