Server/services/hue/bridge.go

package hue

import (
	"context"
	"errors"
	lucifer3 "git.aiterp.net/lucifer3/server"
	"git.aiterp.net/lucifer3/server/device"
	"git.aiterp.net/lucifer3/server/events"
	"git.aiterp.net/lucifer3/server/internal/color"
	"git.aiterp.net/lucifer3/server/internal/gentools"
	"golang.org/x/sync/errgroup"
	"log"
	"math"
	"strings"
	"sync"
	"time"
)

func NewBridge(host string, client *Client) *Bridge {
	return &Bridge{
		client:                   client,
		host:                     host,
		ctx:                      context.Background(),
		cancel:                   func() {},
		resources:                map[string]*ResourceData{},
		activeStates:             map[string]device.State{},
		desiredStates:            map[string]device.State{},
		colorFlags:               map[string]device.ColorFlags{},
		hasSeen:                  map[string]bool{},
		triggerCongruenceCheckCh: make(chan struct{}, 2),
	}
}

type Bridge struct {
	mu sync.Mutex

	client *Client
	host   string
	ctx    context.Context
	cancel context.CancelFunc

	resources     map[string]*ResourceData
	activeStates  map[string]device.State
	desiredStates map[string]device.State
	colorFlags    map[string]device.ColorFlags
	reachable     map[string]bool
	hasSeen       map[string]bool

	triggerCongruenceCheckCh chan struct{}

	lastDiscoverCancel context.CancelFunc
}

func (b *Bridge) SearchDevices(timeout time.Duration) error {
	discoverCtx, cancel := context.WithCancel(b.ctx)

	b.mu.Lock()
	if b.lastDiscoverCancel != nil {
		b.lastDiscoverCancel()
	}
	b.lastDiscoverCancel = cancel
	b.mu.Unlock()

	if timeout <= time.Second*10 {
		timeout = time.Second * 10
	}

	// Spend half the time waiting for devices
	// TODO: Wait for v2 endpoint
	ctx, cancel := context.WithTimeout(discoverCtx, timeout/2)
	defer cancel()
	err := b.client.LegacyDiscover(ctx, "sensors")
	if err != nil {
		return err
	}
	<-ctx.Done()
	if discoverCtx.Err() != nil {
		return discoverCtx.Err()
	}

	// Spend half the time waiting for lights
	// TODO: Wait for v2 endpoint
	ctx, cancel = context.WithTimeout(discoverCtx, timeout/2)
	defer cancel()
	err = b.client.LegacyDiscover(ctx, "sensors")
	if err != nil {
		return err
	}
	<-ctx.Done()
	if discoverCtx.Err() != nil {
		return discoverCtx.Err()
	}

	// Let the main loop get the new light.
	return nil
}

func (b *Bridge) RefreshAll() ([]lucifer3.Event, error) {
	ctx, cancel := context.WithTimeout(b.ctx, time.Second*15)
	defer cancel()

	allResources, err := b.client.AllResources(ctx)
	if err != nil {
		return nil, err
	}

	resources := make(map[string]*ResourceData, len(allResources))
	for i := range allResources {
		resources[allResources[i].ID] = &allResources[i]
	}

	b.mu.Lock()
	hasSeen := b.hasSeen
	reachable := b.reachable
	b.mu.Unlock()
	hasSeen = gentools.CopyMap(hasSeen)
	reachable = gentools.CopyMap(reachable)

	colorFlags := make(map[string]device.ColorFlags)
	activeStates := make(map[string]device.State)

	newEvents := make([]lucifer3.Event, 0, 0)
	for id, res := range resources {
		if res.Type == "device" && res.Metadata.Archetype != "bridge_v2" {
			hwState, hwEvent := res.GenerateEvent(b.host, resources)
			if !hasSeen[id] {
				newEvents = append(newEvents, hwState, hwEvent, events.DeviceReady{ID: hwState.ID})
				hasSeen[id] = true
			} else {
				newEvents = append(newEvents, hwState)
			}

			activeStates[id] = hwState.State
			colorFlags[id] = hwState.ColorFlags
			reachable[id] = !hwState.Unreachable
		}
	}

	b.mu.Lock()
	b.resources = resources
	b.hasSeen = hasSeen
	b.colorFlags = colorFlags
	b.activeStates = activeStates
	b.reachable = reachable
	b.mu.Unlock()

	return newEvents, nil
}

func (b *Bridge) ApplyPatches(resources []ResourceData) (events []lucifer3.Event, shouldRefresh bool) {
	b.mu.Lock()
	resourceMap := b.resources
	activeStates := b.activeStates
	reachable := b.reachable
	colorFlags := b.colorFlags
	b.mu.Unlock()

	mapCopy := gentools.CopyMap(resourceMap)
	activeStatesCopy := gentools.CopyMap(activeStates)
	reachableCopy := gentools.CopyMap(reachable)
	colorFlagsCopy := gentools.CopyMap(colorFlags)

	for _, resource := range resources {
		if mapCopy[resource.ID] != nil {
			mapCopy[resource.ID] = mapCopy[resource.ID].WithPatch(resource)
		} else {
			log.Println(resource.ID, resource.Type, "not seen!")
			shouldRefresh = true
		}
	}

	for _, resource := range resources {
		if resource.Owner != nil && resource.Owner.Kind == "device" {
			if parent, ok := mapCopy[resource.Owner.ID]; ok {
				hwState, _ := parent.GenerateEvent(b.host, mapCopy)
				events = append(events, hwState)

				activeStatesCopy[resource.Owner.ID] = hwState.State
				reachableCopy[resource.Owner.ID] = !hwState.Unreachable
				if hwState.ColorFlags != 0 {
					colorFlagsCopy[resource.Owner.ID] = hwState.ColorFlags
				}
			}
		}
	}

	b.mu.Lock()
	b.resources = mapCopy
	b.activeStates = activeStatesCopy
	b.reachable = reachableCopy
	b.colorFlags = colorFlagsCopy
	b.mu.Unlock()

	return
}

func (b *Bridge) SetStates(patch map[string]device.State) {
	b.mu.Lock()
	newStates := gentools.CopyMap(b.desiredStates)
	resources := b.resources
	b.mu.Unlock()

	prefix := "hue:" + b.host + ":"
	for id, state := range patch {
		if !strings.HasPrefix(id, prefix) {
			continue
		}
		id = id[len(prefix):]

		resource := resources[id]
		if resource == nil {
			continue
		}

		if !state.Empty() {
			newStates[id] = resource.FixState(state, resources)
		} else {
			delete(newStates, id)
		}
	}

	b.mu.Lock()
	b.desiredStates = newStates
	b.mu.Unlock()

	b.triggerCongruenceCheck()
}

func (b *Bridge) Run(ctx context.Context, bus *lucifer3.EventBus) interface{} {
	hwEvents, err := b.RefreshAll()
	if err != nil {
		return errors.New("failed to connect to bridge")
	}

	go b.makeCongruentLoop(ctx)

	bus.RunEvents(hwEvents)

	sse := b.client.SSE(ctx)
	step := time.NewTicker(time.Second * 30)
	defer step.Stop()

	for {
		select {
		case updates, ok := <-sse:
			{
				if !ok {
					return errors.New("SSE lost connection")
				}

				newEvents, shouldUpdate := b.ApplyPatches(
					gentools.Flatten(gentools.Map(updates, func(update SSEUpdate) []ResourceData {
						return update.Data
					})),
				)

				bus.RunEvents(newEvents)

				if shouldUpdate {
					hwEvents, err := b.RefreshAll()
					if err != nil {
						return errors.New("failed to refresh states")
					}

					bus.RunEvents(hwEvents)
				}

				b.triggerCongruenceCheck()
			}
		case <-step.C:
			hwEvents, err := b.RefreshAll()
			if err != nil {
				return nil
			}

			bus.RunEvents(hwEvents)
			b.triggerCongruenceCheck()
		case <-ctx.Done():
			{
				return nil
			}
		}
	}
}

func (b *Bridge) makeCongruentLoop(ctx context.Context) {
	for range b.triggerCongruenceCheckCh {
		if ctx.Err() != nil {
			break
		}

		// Make sure this loop takes half a second
		rateLimit := time.After(time.Second / 2)

		// Take states
		b.mu.Lock()
		resources := b.resources
		desiredStates := b.desiredStates
		activeStates := b.activeStates
		reachable := b.reachable
		colorFlags := b.colorFlags
		b.mu.Unlock()

		newActiveStates := make(map[string]device.State, 0)

		updates := make(map[string]ResourceUpdate)
		for id, desired := range desiredStates {
			active, activeOK := activeStates[id]
			lightID := resources[id].ServiceID("light")
			if !reachable[id] || !activeOK || lightID == nil {
				log.Println("No light", !reachable[id], !activeOK, lightID == nil)
				continue
			}

			light := resources[*lightID]
			if light == nil {
				log.Println("No light", *lightID)
				continue
			}

			// Handle power first
			if desired.Power != nil && active.Power != nil && *desired.Power != *active.Power {
				updates["light/"+*lightID] = ResourceUpdate{Power: gentools.Ptr(*active.Power)}

				newActiveState := activeStates[id]
				newActiveState.Power = gentools.Ptr(*desired.Power)
				newActiveStates[id] = newActiveState

				continue
			}
			if active.Power != nil && *active.Power == false {
				// Don't do more with shut-off-light.
				continue
			}

			updated := false
			update := ResourceUpdate{}
			newActiveState := activeStates[id]

			if active.Color != nil && desired.Color != nil {
				ac := *active.Color
				dc := *desired.Color

				if !dc.IsKelvin() || !colorFlags[id].IsWarmWhite() {
					dc, _ = dc.ToXY()
					dc.XY = gentools.Ptr(light.Color.Gamut.Conform(*dc.XY))
				}

				if dc.XY != nil {
					if ac.K != nil {
						ac.K = gentools.Ptr(1000000 / (1000000 / *ac.K))
					}

					acXY, _ := ac.ToXY()
					dist := dc.XY.DistanceTo(*acXY.XY)
					if dist > 0.0002 {
						update.ColorXY = gentools.Ptr(*dc.XY)
						updated = true
					}
				} else {
					dcMirek := 1000000 / *dc.K
					if dcMirek < light.ColorTemperature.MirekSchema.MirekMinimum {
						dcMirek = light.ColorTemperature.MirekSchema.MirekMinimum
					} else if dcMirek > light.ColorTemperature.MirekSchema.MirekMaximum {
						dcMirek = light.ColorTemperature.MirekSchema.MirekMaximum
					}
					acMirek := 0
					if ac.K != nil {
						acMirek = 1000000 / *ac.K
					}
					if acMirek != dcMirek {
						newActiveState.Color = &color.Color{K: gentools.Ptr(*dc.K)}

						update.Mirek = &dcMirek
						updated = true
					}
				}
			}

			if active.Intensity != nil && desired.Intensity != nil {
				if math.Abs(*active.Intensity-*desired.Intensity) >= 0.01 {
					update.Brightness = gentools.Ptr(*desired.Intensity * 100)
					newActiveState.Intensity = gentools.Ptr(*desired.Intensity)
					updated = true
				}
			}

			if updated {
				update.TransitionDuration = gentools.Ptr(time.Millisecond * 101)
				updates["light/"+*lightID] = update
				newActiveStates[id] = newActiveState
			}
		}

		if len(updates) > 0 {
			timeout, cancel := context.WithTimeout(ctx, time.Second)

			eg, ctx := errgroup.WithContext(timeout)
			for key := range updates {
				update := updates[key]
				split := strings.SplitN(key, "/", 2)
				link := ResourceLink{Kind: split[0], ID: split[1]}

				eg.Go(func() error {
					return b.client.UpdateResource(ctx, link, update)
				})
			}

			err := eg.Wait()
			if err != nil {
				log.Println("Failed to run update", err)
			}

			b.mu.Lock()
			activeStates = b.activeStates
			b.mu.Unlock()
			activeStates = gentools.CopyMap(activeStates)
			for id, state := range newActiveStates {
				activeStates[id] = state
			}
			b.mu.Lock()
			b.activeStates = activeStates
			b.mu.Unlock()

			cancel()
		}

		// Wait the remaining time for the rate limit
		<-rateLimit
	}
}

func (b *Bridge) triggerCongruenceCheck() {
	select {
	case b.triggerCongruenceCheckCh <- struct{}{}:
	default:
	}
}