lucifer/new-server
2
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

this does not work, it needs massive changes.
continuous-integration/drone/push Build is passing
Details

Browse Source
This commit is contained in:
gisle
2021-09-26 16:20:02 +02:00
parent b4fd6df454
commit f87c69efc7
15 changed files with 777 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/devices.go
+59 -3
View File
@@ -3,11 +3,14 @@ package api
import (
"context"
"git.aiterp.net/lucifer/new-server/app/config"
"git.aiterp.net/lucifer/new-server/app/services/scene"
"git.aiterp.net/lucifer/new-server/models"
"github.com/gin-gonic/gin"
"golang.org/x/sync/errgroup"
"log"
"strconv"
"strings"
"time"
)
func fetchDevices(ctx context.Context, fetchStr string) ([]models.Device, error) {
@@ -71,12 +74,15 @@ func Devices(r gin.IRoutes) {
return nil, err
}
_ = scene.GlobalManager().UpdateDevice(ctxOf(c), &devices[i], nil)
set[devices[i].ID] = true
changed = append(changed, devices[i])
}
}
config.PublishChannel <- changed
config.PublishChannel <- scene.GlobalManager().FilterUnassigned(changed)
go func() {
for _, device := range changed {
@@ -139,9 +145,11 @@ func Devices(r gin.IRoutes) {
if err != nil {
return nil, err
}
_ = scene.GlobalManager().UpdateDevice(ctxOf(c), &devices[i], nil)
}
config.PublishChannel <- devices
config.PublishChannel <- scene.GlobalManager().FilterUnassigned(devices)
go func() {
for _, device := range devices {
@@ -212,4 +220,52 @@ func Devices(r gin.IRoutes) {
return devices, nil
}))
}
r.PUT("/:fetch/scene", handler(func(c *gin.Context) (interface{}, error) {
var body models.DeviceSceneAssignment
err := parseBody(c, &body)
if err != nil {
return nil, err
}
devices, err := fetchDevices(ctxOf(c), c.Param("fetch"))
if err != nil {
return nil, err
}
if len(devices) == 0 {
return []models.Device{}, nil
}
assignedScene, err := config.SceneRepository().Find(ctxOf(c), body.SceneID)
if err != nil {
return nil, err
}
if body.DurationMS < 0 {
body.DurationMS = 0
}
body.StartTime = time.Now()
for i := range devices {
devices[i].SceneAssignments = []models.DeviceSceneAssignment{body}
err := scene.GlobalManager().UpdateDevice(ctxOf(c), &devices[i], assignedScene)
if err != nil {
return nil, err
}
}
eg := errgroup.Group{}
for i := range devices {
eg.Go(func() error {
return config.DeviceRepository().Save(ctxOf(c), &devices[i], 0)
})
}
err = eg.Wait()
if err != nil {
return nil, err
}
return devices, nil
}))
}
app/api/scenes.go
+37
View File
@@ -0,0 +1,37 @@
package api
import (
"git.aiterp.net/lucifer/new-server/app/config"
"git.aiterp.net/lucifer/new-server/models"
"github.com/gin-gonic/gin"
)
func Scenes(r gin.IRoutes) {
r.GET("", handler(func(c *gin.Context) (interface{}, error) {
return config.SceneRepository().FetchAll(ctxOf(c))
}))
r.GET("/:id", handler(func(c *gin.Context) (interface{}, error) {
return config.SceneRepository().Find(ctxOf(c), intParam(c, "id"))
}))
r.POST("", handler(func(c *gin.Context) (interface{}, error) {
var body models.Scene
err := parseBody(c, &body)
if err != nil {
return nil, err
}
err = body.Validate()
if err != nil {
return nil, err
}
err = config.SceneRepository().Save(ctxOf(c), &body)
if err != nil {
return nil, err
}
return body, nil
}))
}
app/config/db.go
+3 -2
View File
@@ -24,8 +24,9 @@ func DBX() *sqlx.DB {
MySqlSchema(),
))
dbx.SetMaxIdleConns(50)
dbx.SetMaxOpenConns(100)
dbx.SetMaxIdleConns(20)
dbx.SetMaxOpenConns(40)
dbx.SetConnMaxLifetime(0)
}
return dbx
app/config/repo.go
+4
View File
@@ -20,3 +20,7 @@ func DeviceRepository() models.DeviceRepository {
func EventHandlerRepository() models.EventHandlerRepository {
return &mysql.EventHandlerRepo{DBX: DBX()}
}
func SceneRepository() models.SceneRepository {
return &mysql.SceneRepo{DBX: DBX()}
}
app/server.go
+5
View File
@@ -1,10 +1,12 @@
package app
import (
"context"
"fmt"
"git.aiterp.net/lucifer/new-server/app/api"
"git.aiterp.net/lucifer/new-server/app/config"
"git.aiterp.net/lucifer/new-server/app/services"
"git.aiterp.net/lucifer/new-server/app/services/scene"
"github.com/gin-gonic/gin"
"log"
)
@@ -15,6 +17,8 @@ func StartServer() {
services.ConnectToBridges()
services.CheckNewDevices()
go scene.GlobalManager().Run(context.Background())
gin.SetMode(gin.ReleaseMode)
ginny := gin.New()
@@ -25,6 +29,7 @@ func StartServer() {
api.DriverKinds(apiGin.Group("/driver-kinds"))
api.Events(apiGin.Group("/events"))
api.EventHandlers(apiGin.Group("/event-handlers"))
api.Scenes(apiGin.Group("/scenes"))
log.Fatal(ginny.Run(fmt.Sprintf("0.0.0.0:%d", config.ServerPort())))
}
app/services/events.go
+18 -7
View File
@@ -4,6 +4,7 @@ import (
"context"
"fmt"
"git.aiterp.net/lucifer/new-server/app/config"
"git.aiterp.net/lucifer/new-server/app/services/scene"
"git.aiterp.net/lucifer/new-server/models"
"log"
"math"
@@ -150,21 +151,31 @@ func handleEvent(event models.Event) (responses []models.Event) {
}
}
config.PublishChannel <- allDevices
config.PublishChannel <- scene.GlobalManager().FilterUnassigned(allDevices)
wg := sync.WaitGroup{}
for _, device := range allDevices {
wg.Add(1)
go func(device models.Device) {
err := config.DeviceRepository().Save(context.Background(), &device, models.SMState)
if len(allDevices) > 0 {
wg.Add(1)
go func() {
err := config.DeviceRepository().SaveMany(context.Background(), models.SMState, allDevices)
if err != nil {
log.Println("Failed to save device for state:", err)
log.Println("Failed to save devices' state:", err)
}
wg.Done()
}(device)
}()
for _, device := range allDevices {
go func(device models.Device) {
err := scene.GlobalManager().UpdateDevice(context.Background(), &device, nil)
if err != nil {
log.Println("Failed to update devices' scene in resposne to event:", err)
}
}(device)
}
}
for _, handler := range deadHandlers {
wg.Add(1)
app/services/publish.go
+5
View File
@@ -47,6 +47,11 @@ func StartPublisher() {
}
if bridge.ID == 0 {
log.Println("Unknown bridge")
return
}
if bridge.Driver == models.DTLIFX {
return
}
bridge, err := config.BridgeRepository().Find(ctx, devices[0].BridgeID)
app/services/scene/manager.go
+207 -5
View File
@@ -1,27 +1,84 @@
package scene
import (
"context"
"git.aiterp.net/lucifer/new-server/app/config"
"git.aiterp.net/lucifer/new-server/models"
"log"
"sync"
"time"
)
type Manager struct {
mu sync.Mutex
assignments map[int]*Scene
allocations map[int]*Scene
devices map[int]*models.Device
scenes []*Scene
notifyCh chan struct{}
}
func (mgr *Manager) UpdateDevice(device *models.Device) {
func (mgr *Manager) UpdateDevice(ctx context.Context, device *models.Device, data *models.Scene) error {
mgr.mu.Lock()
defer mgr.mu.Unlock()
scene, err := mgr.findScene(ctx, device.SceneAssignments, data)
if err != nil {
return err
}
if mgr.allocations[device.ID] != nil && mgr.allocations[device.ID] != scene {
mgr.allocations[device.ID].RemoveDevice(*device)
}
if mgr.allocations[device.ID] != scene {
if mgr.allocations == nil {
mgr.allocations = make(map[int]*Scene, 8)
}
mgr.allocations[device.ID] = scene
if scene != nil {
scene.due = true
}
}
if scene != nil {
scene.UpsertDevice(*device)
if mgr.devices == nil {
mgr.devices = make(map[int]*models.Device)
}
mgr.devices[device.ID] = device
} else {
config.PublishChannel <- []models.Device{*device}
}
if mgr.notifyCh != nil {
close(mgr.notifyCh)
mgr.notifyCh = nil
}
return nil
}
func (mgr *Manager) GetUnassigned(devices []models.Device) []models.Device {
func (mgr *Manager) UpdateScene(data *models.Scene) {
mgr.mu.Lock()
for _, scene := range mgr.scenes {
if scene.data.ID == data.ID {
scene.UpdateScene(scene.data)
}
}
mgr.mu.Unlock()
}
func (mgr *Manager) FilterUnassigned(devices []models.Device) []models.Device {
mgr.mu.Lock()
defer mgr.mu.Unlock()
res := make([]models.Device, 0, len(devices))
for _, device := range devices {
if _, ok := mgr.assignments[device.ID]; !ok {
if mgr.allocations[device.ID] == nil {
res = append(res, device)
}
}
@@ -29,8 +86,153 @@ func (mgr *Manager) GetUnassigned(devices []models.Device) []models.Device {
return res
}
func (mgr *Manager) Run(ctx context.Context) {
devices, err := config.DeviceRepository().FetchByReference(ctx, models.RKAll, "")
if err == nil {
for _, device := range devices {
_ = mgr.UpdateDevice(ctx, &device, nil)
}
}
ticker := time.NewTicker(time.Millisecond * 101)
defer ticker.Stop()
mgr.mu.Lock()
mgr.notifyCh = make(chan struct{})
mgr.mu.Unlock()
for {
mgr.mu.Lock()
if mgr.notifyCh == nil {
mgr.notifyCh = make(chan struct{})
}
notifyCh := mgr.notifyCh
mgr.mu.Unlock()
select {
case <-ticker.C:
case <-notifyCh:
case <-ctx.Done():
return
}
timeout, cancel := context.WithTimeout(ctx, time.Millisecond*100)
var updates []models.Device
mgr.mu.Lock()
for _, device := range mgr.devices {
if device == nil {
continue
}
if mgr.reScene(timeout, device) {
updates = append(updates, *device)
}
}
for _, scene := range mgr.scenes {
if scene.Due() {
scene.Run()
}
updates = append(updates, scene.Flush()...)
}
mgr.mu.Unlock()
cancel()
if len(updates) > 0 {
log.Println("Updating", len(updates), "devices")
config.PublishChannel <- updates
}
}
}
func (mgr *Manager) reScene(ctx context.Context, device *models.Device) bool {
scene, err := mgr.findScene(ctx, device.SceneAssignments, nil)
if err != nil {
return false
}
if mgr.allocations[device.ID] != nil && mgr.allocations[device.ID] != scene {
mgr.allocations[device.ID].RemoveDevice(*device)
}
if mgr.allocations[device.ID] != scene {
if mgr.allocations == nil {
mgr.allocations = make(map[int]*Scene, 8)
}
mgr.allocations[device.ID] = scene
if scene == nil {
return true
}
mgr.allocations[device.ID].UpsertDevice(*device)
}
return false
}
func (mgr *Manager) findScene(ctx context.Context, assignments []models.DeviceSceneAssignment, data *models.Scene) (*Scene, error) {
if len(assignments) == 0 {
return nil, nil
}
var selected *models.DeviceSceneAssignment
for _, assignment := range assignments {
pos := int64(time.Since(assignment.StartTime) / time.Millisecond)
if assignment.DurationMS == 0 || (pos >= -1 && pos < assignment.DurationMS) {
selected = &assignment
}
}
if selected == nil {
return nil, nil
}
for _, scene := range mgr.scenes {
if scene.group == selected.Group && scene.startTime.Equal(selected.StartTime) && selected.DurationMS == scene.duration.Milliseconds() {
return scene, nil
}
}
if data == nil {
for i := range mgr.scenes {
if mgr.scenes[i].data.ID == selected.SceneID {
data = &mgr.scenes[i].data
break
}
}
}
if data == nil {
var err error
data, err = config.SceneRepository().Find(ctx, selected.SceneID)
if err != nil {
return nil, err
}
}
scene := &Scene{
startTime: selected.StartTime,
duration: time.Duration(selected.DurationMS) * time.Millisecond,
lastRun: 0,
due: true,
group: selected.Group,
data: *data,
roleList: make([][]models.Device, len(data.Roles)),
deviceMap: make(map[int]*models.Device, 8),
roleMap: make(map[int]int, 8),
changes: make(map[int]models.NewDeviceState, 8),
}
mgr.scenes = append(mgr.scenes, scene)
return scene, nil
}
var globalManager = &Manager{}
func Get() *Manager {
func GlobalManager() *Manager {
return globalManager
}
app/services/scene/scene.go
+114 -26
View File
@@ -1,14 +1,19 @@
package scene
import (
"context"
"git.aiterp.net/lucifer/new-server/app/config"
"git.aiterp.net/lucifer/new-server/models"
"math"
"time"
)
type Scene struct {
startTime time.Time
duration time.Duration
tag string
lastRun int64
due bool
group string
data models.Scene
roleList [][]models.Device
deviceMap map[int]*models.Device
@@ -21,32 +26,110 @@ func (s *Scene) UpdateScene(scene models.Scene) {
s.roleMap = make(map[int]int)
for _, device := range s.deviceMap {
s.moveDevice(*device)
s.moveDevice(*device, false)
}
for i, role := range s.data.Roles {
role.ApplyOrder(s.roleList[i])
}
s.Run()
}
func (s *Scene) UpsertDevice(device models.Device) {
s.deviceMap[device.ID] = &device
s.moveDevice(device)
s.moveDevice(device, true)
}
func (s *Scene) RemoveDevice(device models.Device) {
roleIndex, hasRole := s.roleMap[device.ID]
delete(s.deviceMap, device.ID)
delete(s.roleMap, device.ID)
if hasRole {
s.runRole(roleIndex)
}
}
func (s *Scene) Due() bool {
return s.due || s.intervalNumber() != s.lastRun
}
func (s *Scene) Flush() []models.Device {
res := make([]models.Device, 0, 8)
for key, change := range s.changes {
if s.deviceMap[key] != nil {
device := *s.deviceMap[key]
err := device.SetState(change)
if err != nil {
continue
}
res = append(res, device)
}
delete(s.changes, key)
}
return res
}
func (s *Scene) Run() {
s.lastRun = s.intervalNumber()
s.due = false
for index := range s.roleList {
s.runRole(index)
}
}
func (s *Scene) runRole(index int) {
role := s.data.Roles[index]
devices := s.roleList[index]
intervalNumber := s.intervalNumber()
intervalMax := s.intervalMax()
for i, device := range devices {
newState, changed := role.ApplyEffect(&device, models.SceneRunContext{
Index: i,
Length: len(devices),
IntervalNumber: intervalNumber,
IntervalMax: intervalMax,
})
if !changed {
go func() {
_ = config.DeviceRepository().Save(context.Background(), &device, models.SMState)
}()
}
s.changes[device.ID] = newState
}
}
func (s *Scene) moveDevice(device models.Device, run bool) {
oldRole, hasOldRole := s.roleMap[device.ID]
newRole := s.data.RoleIndex(&device)
// If it doesn't move between roles, just update and reorder it.
if hasOldRole && oldRole == newRole {
for i, device2 := range s.roleList[newRole] {
if device2.ID == device.ID {
s.roleList[newRole][i] = device
}
}
if run {
s.data.Roles[newRole].ApplyOrder(s.roleList[newRole])
s.runRole(newRole)
}
return
}
// Take it out of the old role.
if hasOldRole {
for i, device2 := range s.roleList[oldRole] {
if device2.ID == device.ID {
@@ -55,33 +138,38 @@ func (s *Scene) moveDevice(device models.Device, run bool) {
}
}
s.runRole(oldRole)
if run {
s.runRole(oldRole)
}
}
s.roleMap[device.ID] = newRole
s.runRole(newRole)
if newRole != -1 {
s.roleMap[device.ID] = newRole
s.roleList[newRole] = append(s.roleList[newRole], device)
if run {
s.data.Roles[newRole].ApplyOrder(s.roleList[newRole])
s.runRole(newRole)
}
}
s.due = true
}
func (s *Scene) intervalNumber() int64 {
intervalDuration := time.Duration(s.data.IntervalMS) * time.Millisecond
return int64(time.Since(s.startTime) / intervalDuration)
}
func (s *Scene) intervalFac() float64 {
return clamp(float64(time.Since(s.startTime)) / float64(s.duration))
}
func (s *Scene) positionFac(i, len int) float64 {
return clamp(float64(i) / float64(len))
}
func clamp(fac float64) float64 {
if fac > 1.00 {
return 1.00
} else if fac < 0.00 {
return 0.00
} else {
return fac
if intervalDuration == 0 {
return 0
}
}
return int64(math.Round(float64(time.Since(s.startTime)) / float64(intervalDuration)))
}
func (s *Scene) intervalMax() int64 {
intervalDuration := time.Duration(s.data.IntervalMS) * time.Millisecond
if intervalDuration == 0 {
return 1
}
return int64(s.duration / intervalDuration)
}