fyah/AI-tianyan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add dynamic config reload, unique device identity (UUID), and MQTT support

Browse Source 
- New StreamManager for dynamic RTSP/FLV stream lifecycle
- Dynamic worker scaling for inference
- Device UUID generation and persistence
- Telegraf config and NPU monitoring scripts
- .gitignore for build artifacts
This commit is contained in:
master
2026-05-08 16:26:10 +08:00
parent 2e7245d62e
commit 6e88d2392f
12 changed files with 433 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
internal/config/config.go
View File

@@ -1,14 +1,21 @@
package config
import (
"crypto/rand"
"encoding/hex"
"os"
"path/filepath"
"gopkg.in/yaml.v3"
)
type Config struct {
DeviceUUID string `yaml:"device_uuid"`
EdgeID string `yaml:"edge_id"`
CloudURL string `yaml:"cloud_url"`
MqttBroker string `yaml:"mqtt_broker"`
MqttUser string `yaml:"mqtt_user"`
MqttPass string `yaml:"mqtt_pass"`
EdgeToken string `yaml:"edge_token"`
RTSPURLs []string `yaml:"rtsp_urls"`
InferSocket string `yaml:"infer_socket"`
@@ -16,12 +23,14 @@ type Config struct {
InferWorkers int `yaml:"infer_workers"`
ConfThreshold float64 `yaml:"conf_threshold"`
DedupWindowSec int `yaml:"dedup_window_sec"`
OTAUrl string `yaml:"ota_url"`
Version string `yaml:"version"`
configPath string `json:"-"`
}
func Load(path string) *Config {
cfg := &Config{
EdgeID: "edge-demo-001",
EdgeID: "edge-unknown",
CloudURL: "http://localhost:8004",
InferSocket: "/tmp/edge-infer.sock",
InferFPS: 5,
@@ -29,12 +38,15 @@ func Load(path string) *Config {
ConfThreshold: 0.5,
DedupWindowSec: 30,
Version: "1.0.0",
configPath: path,
}
data, err := os.ReadFile(path)
if err != nil {
return cfg
cfg.loadUUID()
if data, err := os.ReadFile(path); err == nil {
_ = yaml.Unmarshal(data, cfg)
}
_ = yaml.Unmarshal(data, cfg)
if cfg.InferWorkers <= 0 {
cfg.InferWorkers = 1
}
@@ -43,3 +55,31 @@ func Load(path string) *Config {
}
return cfg
}
func (c *Config) loadUUID() {
uuidPath := "/opt/tianyan-edge/device.uuid"
if data, err := os.ReadFile(uuidPath); err == nil {
c.DeviceUUID = string(data)
} else {
b := make([]byte, 16)
rand.Read(b)
c.DeviceUUID = hex.EncodeToString(b)
os.MkdirAll(filepath.Dir(uuidPath), 0755)
os.WriteFile(uuidPath, []byte(c.DeviceUUID), 0644)
}
}
func (c *Config) Save() error {
data, err := yaml.Marshal(c)
if err != nil {
return err
}
return os.WriteFile(c.configPath, data, 0644)
}
func (c *Config) GetDeviceIdentity() string {
if c.DeviceUUID != "" {
return c.DeviceUUID
}
return c.EdgeID
}

123
internal/control/mqtt_manager.go Normal file
View File

@@ -0,0 +1,123 @@
package control
import (
"context"
"encoding/json"
"fmt"
"log"
"time"
mqtt "github.com/eclipse/paho.mqtt.golang"
"tianyan-edge/internal/config"
)
type MqttManager struct {
client mqtt.Client
cfg *config.Config
updates chan *config.Config // Channel to broadcast config updates
}
func NewMqttManager(cfg *config.Config) *MqttManager {
return &MqttManager{
cfg: cfg,
updates: make(chan *config.Config, 10),
}
}
func (m *MqttManager) Run(ctx context.Context) {
if m.cfg.MqttBroker == "" {
log.Println("mqtt: no broker configured, skipping")
return
}
opts := mqtt.NewClientOptions()
opts.AddBroker(m.cfg.MqttBroker)
opts.SetClientID(m.cfg.GetDeviceIdentity())
opts.SetUsername(m.cfg.MqttUser)
opts.SetPassword(m.cfg.MqttPass)
opts.SetAutoReconnect(true)
opts.SetMaxReconnectInterval(1 * time.Minute)
// LWT
willTopic := fmt.Sprintf("tianyan/edge/%s/status", m.cfg.GetDeviceIdentity())
opts.SetWill(willTopic, "offline", 1, true)
opts.SetOnConnectHandler(func(c mqtt.Client) {
log.Println("mqtt: connected to broker")
c.Publish(willTopic, 1, true, "online")
configTopic := fmt.Sprintf("tianyan/edge/%s/config", m.cfg.GetDeviceIdentity())
token := c.Subscribe(configTopic, 1, m.handleConfigUpdate)
token.Wait()
if token.Error() != nil {
log.Printf("mqtt: subscribe failed: %v", token.Error())
} else {
log.Printf("mqtt: subscribed to %s", configTopic)
}
})
m.client = mqtt.NewClient(opts)
if token := m.client.Connect(); token.Wait() && token.Error() != nil {
log.Printf("mqtt: connect error: %v", token.Error())
}
<-ctx.Done()
m.client.Disconnect(1000)
log.Println("mqtt: disconnected")
}
func (m *MqttManager) handleConfigUpdate(c mqtt.Client, msg mqtt.Message) {
log.Printf("mqtt: received config update on %s", msg.Topic())
var newCfg map[string]interface{}
if err := json.Unmarshal(msg.Payload(), &newCfg); err != nil {
log.Printf("mqtt: config parse error: %v", err)
return
}
// Update local config struct
if v, ok := newCfg["infer_fps"].(float64); ok {
m.cfg.InferFPS = int(v)
}
if v, ok := newCfg["conf_threshold"].(float64); ok {
m.cfg.ConfThreshold = v
}
if v, ok := newCfg["infer_workers"].(float64); ok {
m.cfg.InferWorkers = int(v)
}
if v, ok := newCfg["rtsp_urls"]; ok {
if urls, ok := v.([]interface{}); ok {
var s []string
for _, u := range urls {
if str, ok := u.(string); ok {
s = append(s, str)
}
}
m.cfg.RTSPURLs = s
}
}
// Save to disk
if err := m.cfg.Save(); err != nil {
log.Printf("mqtt: failed to save config: %v", err)
} else {
log.Println("mqtt: config saved to disk")
}
// Broadcast update to other components
select {
case m.updates <- m.cfg:
default:
log.Println("mqtt: update channel full, dropping update")
}
}
func (m *MqttManager) GetUpdates() <-chan *config.Config {
return m.updates
}
func (m *MqttManager) Publish(topic string, payload interface{}) {
if m.client == nil || !m.client.IsConnected() {
return
}
data, _ := json.Marshal(payload)
m.client.Publish(topic, 1, false, data)
}

77
internal/infer/client.go
View File

@@ -46,31 +46,75 @@ type dedupState struct {
}
type Client struct {
cfg *config.Config
frames <-chan stream.Frame
events chan<- event.SuspectedEvent
dedup *dedupState
cfg *config.Config
frames <-chan stream.Frame
events chan<- event.SuspectedEvent
dedup *dedupState
updates <-chan *config.Config
}
func NewClient(cfg *config.Config, frames <-chan stream.Frame, events chan<- event.SuspectedEvent) *Client {
func NewClient(cfg *config.Config, frames <-chan stream.Frame, events chan<- event.SuspectedEvent, updates <-chan *config.Config) *Client {
return &Client{
cfg: cfg,
frames: frames,
events: events,
dedup: &dedupState{seen: make(map[string]float64)},
cfg: cfg,
frames: frames,
events: events,
dedup: &dedupState{seen: make(map[string]float64)},
updates: updates,
}
}
func (c *Client) Run(ctx context.Context) {
var wg sync.WaitGroup
for i := 0; i < c.cfg.InferWorkers; i++ {
// Worker management loop
numWorkers := c.cfg.InferWorkers
activeWorkers := make([]context.CancelFunc, numWorkers)
for i := 0; i < numWorkers; i++ {
workerCtx, cancel := context.WithCancel(ctx)
activeWorkers[i] = cancel
wg.Add(1)
go func(workerID int) {
go func(id int, wCtx context.Context) {
defer wg.Done()
c.workerLoop(ctx, workerID)
}(i)
c.workerLoop(wCtx, id)
}(i, workerCtx)
}
for {
select {
case <-ctx.Done():
for _, cancel := range activeWorkers {
cancel()
}
wg.Wait()
return
case newCfg := <-c.updates:
c.cfg = newCfg
target := c.cfg.InferWorkers
if target < 0 {
target = 0
}
// Scale up
for i := len(activeWorkers); i < target; i++ {
workerCtx, cancel := context.WithCancel(ctx)
activeWorkers = append(activeWorkers, cancel)
wg.Add(1)
go func(id int, wCtx context.Context) {
defer wg.Done()
c.workerLoop(wCtx, id)
}(i, workerCtx)
}
// Scale down
if target < len(activeWorkers) {
for i := target; i < len(activeWorkers); i++ {
activeWorkers[i]()
}
activeWorkers = activeWorkers[:target]
}
log.Printf("infer: workers scaled to %d", target)
}
}
wg.Wait()
}
func (c *Client) workerLoop(ctx context.Context, workerID int) {
@@ -89,10 +133,9 @@ func (c *Client) workerLoop(ctx context.Context, workerID int) {
log.Printf("infer[w%d]: socket connected", workerID)
if !c.loop(ctx, conn, workerID) {
conn.Close()
return
time.Sleep(1 * time.Second)
}
conn.Close()
time.Sleep(500 * time.Millisecond)
}
}
@@ -103,12 +146,10 @@ func (c *Client) loop(ctx context.Context, conn net.Conn, workerID int) bool {
return false
case f := <-c.frames:
if err := c.send(conn, f); err != nil {
log.Printf("infer[w%d]: send error: %v", workerID, err)
return true
}
res, err := c.recv(conn)
if err != nil {
log.Printf("infer[w%d]: recv error: %v", workerID, err)
return true
}
c.emitEvents(f, res)

129
internal/stream/ingestor.go
View File

@@ -1,7 +1,6 @@
package stream
import (
"bytes"
"context"
"fmt"
"io"
@@ -10,6 +9,8 @@ import (
"sync"
"time"
"bytes"
"tianyan-edge/internal/config"
)
@@ -21,63 +22,109 @@ type Frame struct {
TS float64
}
type Ingestor struct {
cfg *config.Config
frames chan<- Frame
type StreamManager struct {
cfg *config.Config
frames chan<- Frame
processes map[string]*exec.Cmd
mu sync.Mutex
updates <-chan *config.Config
}
func NewIngestor(cfg *config.Config, frames chan<- Frame) *Ingestor {
return &Ingestor{cfg: cfg, frames: frames}
}
func (ing *Ingestor) Run(ctx context.Context) {
if len(ing.cfg.RTSPURLs) == 0 {
log.Println("stream: no RTSP URLs configured, idle")
<-ctx.Done()
return
func NewStreamManager(cfg *config.Config, frames chan<- Frame, updates <-chan *config.Config) *StreamManager {
return &StreamManager{
cfg: cfg,
frames: frames,
processes: make(map[string]*exec.Cmd),
updates: updates,
}
var wg sync.WaitGroup
for i, url := range ing.cfg.RTSPURLs {
wg.Add(1)
go func(idx int, u string) {
defer wg.Done()
ing.streamLoop(ctx, idx, u)
}(i, url)
}
wg.Wait()
}
func (ing *Ingestor) streamLoop(ctx context.Context, idx int, url string) {
func (sm *StreamManager) Run(ctx context.Context) {
// Initial start
sm.applyStreams(ctx, sm.cfg.RTSPURLs)
for {
select {
case <-ctx.Done():
sm.stopAll()
return
case newCfg := <-sm.updates:
log.Printf("stream: applying new config, urls=%d", len(newCfg.RTSPURLs))
sm.applyStreams(ctx, newCfg.RTSPURLs)
}
}
}
func (sm *StreamManager) applyStreams(ctx context.Context, urls []string) {
sm.mu.Lock()
defer sm.mu.Unlock()
// Identify URLs to remove (present in sm.processes but not in new urls)
toRemove := map[string]*exec.Cmd{}
for url, cmd := range sm.processes {
found := false
for _, u := range urls {
if u == url {
found = true
break
}
}
if !found {
toRemove[url] = cmd
}
}
// Stop removed streams
for url, cmd := range toRemove {
log.Printf("stream: stopping %s", url)
cmd.Process.Kill()
cmd.Wait()
delete(sm.processes, url)
}
// Start new streams
for i, url := range urls {
if _, exists := sm.processes[url]; !exists {
go sm.streamLoop(ctx, i, url)
}
}
}
func (sm *StreamManager) streamLoop(ctx context.Context, idx int, url string) {
deviceID := fmt.Sprintf("cam-%03d", idx)
fps := ing.cfg.InferFPS
if fps <= 0 {
fps = 5
}
for {
select {
case <-ctx.Done():
return
default:
}
cmd := exec.CommandContext(ctx, "ffmpeg",
"-rtsp_transport", "tcp", "-i", url,
"-vf", fmt.Sprintf("fps=%d", fps),
"-i", url,
"-vf", fmt.Sprintf("fps=%d", sm.cfg.InferFPS),
"-f", "image2pipe", "-vcodec", "mjpeg", "-q:v", "5", "-",
)
stdout, err := cmd.StdoutPipe()
if err != nil {
log.Printf("stream[%d] stdout pipe failed: %v, retry 10s", idx, err)
time.Sleep(10 * time.Second)
log.Printf("stream[%d] stdout pipe failed: %v", idx, err)
time.Sleep(5 * time.Second)
continue
}
sm.mu.Lock()
sm.processes[url] = cmd
sm.mu.Unlock()
if err := cmd.Start(); err != nil {
log.Printf("stream[%d] ffmpeg start failed: %v, retry 10s", idx, err)
time.Sleep(10 * time.Second)
log.Printf("stream[%d] start failed: %v", idx, err)
time.Sleep(5 * time.Second)
continue
}
log.Printf("stream[%d] connected url=%s", idx, url)
ing.readFrames(ctx, stdout, idx, deviceID, url)
sm.readFrames(ctx, stdout, idx, deviceID, url)
cmd.Wait()
select {
case <-ctx.Done():
return
@@ -88,7 +135,7 @@ func (ing *Ingestor) streamLoop(ctx context.Context, idx int, url string) {
}
}
func (ing *Ingestor) readFrames(ctx context.Context, r io.Reader, idx int, deviceID, url string) {
func (sm *StreamManager) readFrames(ctx context.Context, r io.Reader, idx int, deviceID, url string) {
soi, eoi := []byte{0xFF, 0xD8}, []byte{0xFF, 0xD9}
buf, tmp := make([]byte, 0, 1<<20), make([]byte, 32768)
for {
@@ -116,7 +163,7 @@ func (ing *Ingestor) readFrames(ctx context.Context, r io.Reader, idx int, devic
copy(frame, buf[:end])
buf = buf[end:]
select {
case ing.frames <- Frame{StreamID: idx, DeviceID: deviceID, URL: url,
case sm.frames <- Frame{StreamID: idx, DeviceID: deviceID, URL: url,
JPEG: frame, TS: float64(time.Now().UnixMilli()) / 1000.0}:
default:
}
@@ -127,3 +174,13 @@ func (ing *Ingestor) readFrames(ctx context.Context, r io.Reader, idx int, devic
}
}
}
func (sm *StreamManager) stopAll() {
sm.mu.Lock()
defer sm.mu.Unlock()
for url, cmd := range sm.processes {
log.Printf("stream: stopping all %s", url)
cmd.Process.Kill()
delete(sm.processes, url)
}
}