4 Commits

Author SHA1 Message Date
Adrian Zuercher
3704edebd5 fix wrong call and missing send set 2025-07-27 13:07:16 +02:00
Adrian Zuercher
a503b71fb6 add read write lock to dmx data 2025-07-26 07:18:26 +02:00
Adrian Zuercher
05409c2544 change send data to dmx as 23 millisecond ticker when new data 2025-07-25 22:37:35 +02:00
Adrian Zuercher
b484745ed1 increase timeout of ping from 10 to 40 2025-07-25 20:59:17 +02:00
5 changed files with 75 additions and 40 deletions

View File

@@ -78,10 +78,10 @@ func (d *ArtNetDriver) SetValue(bus string, address uint, value uint8) error {
if _, ok := d.Buses[bus]; !ok { if _, ok := d.Buses[bus]; !ok {
return fmt.Errorf("no bus '%s' found", bus) return fmt.Errorf("no bus '%s' found", bus)
} }
if d.Buses[bus].Data == nil { if d.Buses[bus].DMX.Data == nil {
return fmt.Errorf("no dmx data on bus '%s' found", bus) return fmt.Errorf("no dmx data on bus '%s' found", bus)
} }
d.Buses[bus].Data.SetValue(address, value) d.Buses[bus].SetDMXData(address, value)
return nil return nil
} }
@@ -105,7 +105,6 @@ func (d *ArtNetDriver) Connect(ip, id string, port uint) error {
request := json_data.NewResponse() request := json_data.NewResponse()
err = json.Unmarshal(data, &request) err = json.Unmarshal(data, &request)
if err != nil { if err != nil {
return return
} }
@@ -141,9 +140,6 @@ func (d *ArtNetDriver) Connect(ip, id string, port uint) error {
// send data to all buses that the send flage is true // send data to all buses that the send flage is true
func (d *ArtNetDriver) SendData() { func (d *ArtNetDriver) SendData() {
for _, bus := range d.Buses { for _, bus := range d.Buses {
if !bus.Reachable { bus.Send = bus.Reachable
continue
}
bus.Send <- bus.Data
} }
} }

View File

@@ -6,6 +6,7 @@ import (
"fmt" "fmt"
"net" "net"
"net/http" "net/http"
"sync"
"time" "time"
"github.com/gin-gonic/gin" "github.com/gin-gonic/gin"
@@ -25,11 +26,12 @@ type Bus struct {
Name string `yaml:"name" json:"name"` Name string `yaml:"name" json:"name"`
Ip string `yaml:"ip" json:"ip"` Ip string `yaml:"ip" json:"ip"`
Port *int `yaml:"port" json:"port,omitempty"` Port *int `yaml:"port" json:"port,omitempty"`
Data *DMX `yaml:"-" json:"-"` DMX *DMX `yaml:"-" json:"-"`
Resubscribe *[]json_dataModels.Subscription `yaml:"-" json:"resubscribe"` Resubscribe *[]json_dataModels.Subscription `yaml:"-" json:"resubscribe"`
Watchdog context.CancelFunc `yaml:"-" json:"-"` Watchdog context.CancelFunc `yaml:"-" json:"-"`
mu sync.Mutex `yaml:"-" json:"-"`
Reachable bool `yaml:"-" json:"-"` Reachable bool `yaml:"-" json:"-"`
Send chan *DMX `yaml:"-" json:"-"` Send bool `yaml:"-" json:"-"`
} }
// adds new Art-Net interface to driver port 0 = 6454 (default art-net) // adds new Art-Net interface to driver port 0 = 6454 (default art-net)
@@ -42,7 +44,7 @@ func NewBus(name, ip string, port int) *Bus {
Name: name, Name: name,
Ip: ip, Ip: ip,
Port: &port, Port: &port,
Data: NewDMXUniverse(), DMX: NewDMXUniverse(),
} }
return &i return &i
} }
@@ -90,7 +92,7 @@ func (b *Bus) Poll(interval time.Duration) error {
} }
}() }()
_, err = conn.Write(NewArtNetPackage(b.Data)) _, err = conn.Write(NewArtNetPackage(b.DMX))
if err != nil { if err != nil {
return err return err
} }
@@ -107,8 +109,6 @@ func (b *Bus) Start(log *logging.Logger) error {
var ctx context.Context var ctx context.Context
ctx, b.Watchdog = context.WithCancel(context.Background()) ctx, b.Watchdog = context.WithCancel(context.Background())
b.Send = make(chan *DMX, 1024)
go func() { go func() {
var interval time.Duration = 10 * time.Second var interval time.Duration = 10 * time.Second
log.Info("bus.Start", fmt.Sprintf("device:%s ip:%s watchdog starting", b.Name, b.Ip)) log.Info("bus.Start", fmt.Sprintf("device:%s ip:%s watchdog starting", b.Name, b.Ip))
@@ -129,7 +129,7 @@ func (b *Bus) Start(log *logging.Logger) error {
} else { } else {
b.Reachable = true b.Reachable = true
// send data as a heartbeat for she ArtNet Protocol // send data as a heartbeat for she ArtNet Protocol
b.Send <- b.Data b.Send = true
log.Info("bus.Start", fmt.Sprintf("device:%s ip:%s watchdog running", b.Name, b.Ip)) log.Info("bus.Start", fmt.Sprintf("device:%s ip:%s watchdog running", b.Name, b.Ip))
interval = 30 * time.Second interval = 30 * time.Second
} }
@@ -148,18 +148,34 @@ func (b *Bus) Start(log *logging.Logger) error {
} }
go func() { go func() {
defer conn.Close() ticker := time.NewTicker(23 * time.Millisecond)
//close send channel
defer close(b.Send)
for send := range b.Send { defer func() {
_, err = conn.Write(NewArtNetPackage(send)) b.Send = false
conn.Close()
ticker.Stop()
}()
for range ticker.C {
if !b.Send {
continue
}
b.Send = false
b.mu.Lock()
data := NewDMXUniverse()
copy(data.Data, b.DMX.GetDMXData())
b.mu.Unlock()
_, err = conn.Write(NewArtNetPackage(data))
if err != nil { if err != nil {
log.Error("bus.Start", err) log.Error("bus.Start", err)
return return
} }
time.Sleep(23 * time.Millisecond)
} }
}() }()
return nil return nil
} }
@@ -169,8 +185,8 @@ func (b *Bus) Stop() {
if b.Watchdog != nil { if b.Watchdog != nil {
//cancels context //cancels context
b.Watchdog() b.Watchdog()
//close send channel
close(b.Send) b.Send = false
} }
} }
@@ -199,6 +215,12 @@ func (b *Bus) ParsePayload(c *gin.Context) error {
return nil return nil
} }
func (b *Bus) SetDMXData(channel uint, value uint8) error {
b.DMX.SetValue(channel, value)
b.Send = true
return nil
}
func isUDPReachable(ip string) (recieved bool, err error) { func isUDPReachable(ip string) (recieved bool, err error) {
p := fastping.NewPinger() p := fastping.NewPinger()
ra, err := net.ResolveIPAddr("ip4:icmp", ip) ra, err := net.ResolveIPAddr("ip4:icmp", ip)

View File

@@ -1,12 +1,29 @@
package models package models
type DMX []byte import (
"sync"
)
type DMX struct {
Data []byte
mu sync.Mutex
}
func NewDMXUniverse() *DMX { func NewDMXUniverse() *DMX {
dmx := make(DMX, 512) return &DMX{
return &dmx Data: make([]byte, 512),
}
}
func (d *DMX) GetDMXData() (data []byte) {
d.mu.Lock()
data = d.Data
d.mu.Unlock()
return
} }
func (d *DMX) SetValue(channel uint, value uint8) { func (d *DMX) SetValue(channel uint, value uint8) {
(*d)[channel] = value d.mu.Lock()
d.Data[channel] = value
d.mu.Unlock()
} }

View File

@@ -23,7 +23,7 @@ func NewArtNetPackage(data *DMX) Package {
packet.WriteByte(0x00) // Sequence packet.WriteByte(0x00) // Sequence
packet.WriteByte(0x00) // Physical packet.WriteByte(0x00) // Physical
binary.Write(packet, binary.BigEndian, uint16(0)) // Universe (net:subuni, usually 0) binary.Write(packet, binary.BigEndian, uint16(0)) // Universe (net:subuni, usually 0)
binary.Write(packet, binary.BigEndian, uint16(len(*data))) // Length binary.Write(packet, binary.BigEndian, uint16(len(data.Data))) // Length
packet.Write(*data) packet.Write(data.Data)
return packet.Bytes() return packet.Bytes()
} }

View File

@@ -12,10 +12,10 @@ import (
const ( const (
// Time allowed to write a message to the peer. // Time allowed to write a message to the peer.
writeWait = 10 * time.Second writeWait = 40 * time.Second
// Time allowed to read the next pong message from the peer. // Time allowed to read the next pong message from the peer.
pongWait = 10 * time.Second pongWait = 40 * time.Second
) )
type Client struct { type Client struct {
@@ -89,10 +89,10 @@ func (c *Client) Read() {
} }
c.conn.SetReadDeadline(time.Now().Add(pongWait)) c.conn.SetReadDeadline(time.Now().Add(pongWait))
for c.Connected {
c.conn.SetReadDeadline(time.Now().Add(pongWait))
msgType, msg, err := c.conn.ReadMessage()
for c.Connected {
msgType, msg, err := c.conn.ReadMessage()
c.conn.SetReadDeadline(time.Now().Add(pongWait))
if err != nil { if err != nil {
c.handleError(fmt.Errorf("read error (ip:%s): %w", c.ip, err)) c.handleError(fmt.Errorf("read error (ip:%s): %w", c.ip, err))
return return