Compare commits
6 Commits
966ef512ab
...
master
| Author | SHA1 | Date | |
|---|---|---|---|
| 027e79952f | |||
| 04d6a667fe | |||
| cde3572f79 | |||
| 4cb160b88c | |||
| 831412f063 | |||
| 6ff71fded0 |
@@ -1,6 +1,7 @@
|
|||||||
.git/
|
.git/
|
||||||
.gitignore
|
.gitignore
|
||||||
build/
|
build/
|
||||||
|
!build/edge-agent
|
||||||
data/
|
data/
|
||||||
*.om
|
*.om
|
||||||
*.onnx
|
*.onnx
|
||||||
|
|||||||
14
Dockerfile
14
Dockerfile
@@ -1,10 +1,16 @@
|
|||||||
# 使用华为云镜像加速的 Python 3.9 Slim
|
# 使用华为云镜像加速的 Python 3.9 Slim
|
||||||
FROM swr.cn-north-4.myhuaweicloud.com/ddn-k8s/docker.io/library/python:3.9-slim
|
FROM python:3.9-slim
|
||||||
|
|
||||||
ENV DEBIAN_FRONTEND=noninteractive
|
ENV DEBIAN_FRONTEND=noninteractive
|
||||||
# 安装 ffmpeg, curl 及 OpenCV 基础依赖 (使用华为云 apt 源)
|
# 替换 apt 源为华为云镜像(兼容 Debian 11 旧格式和 Debian 12/13 DEB822 格式)
|
||||||
RUN sed -i 's/deb.debian.org/repo.huaweicloud.com/g' /etc/apt/sources.list && \
|
RUN if [ -f /etc/apt/sources.list ]; then \
|
||||||
sed -i 's/security.debian.org/repo.huaweicloud.com/g' /etc/apt/sources.list && \
|
sed -i 's/deb.debian.org/repo.huaweicloud.com/g' /etc/apt/sources.list && \
|
||||||
|
sed -i 's/security.debian.org/repo.huaweicloud.com/g' /etc/apt/sources.list; \
|
||||||
|
fi && \
|
||||||
|
if [ -f /etc/apt/sources.list.d/debian.sources ]; then \
|
||||||
|
sed -i 's|https\?://deb.debian.org|http://repo.huaweicloud.com|g' /etc/apt/sources.list.d/debian.sources && \
|
||||||
|
sed -i 's|https\?://security.debian.org|http://repo.huaweicloud.com|g' /etc/apt/sources.list.d/debian.sources; \
|
||||||
|
fi && \
|
||||||
apt-get update && apt-get install -y --no-install-recommends \
|
apt-get update && apt-get install -y --no-install-recommends \
|
||||||
ffmpeg curl libgl1 libglib2.0-0 \
|
ffmpeg curl libgl1 libglib2.0-0 \
|
||||||
&& rm -rf /var/lib/apt/lists/*
|
&& rm -rf /var/lib/apt/lists/*
|
||||||
|
|||||||
38
Dockerfile.bak
Normal file
38
Dockerfile.bak
Normal file
@@ -0,0 +1,38 @@
|
|||||||
|
# 使用华为云镜像加速的 Python 3.9 Slim
|
||||||
|
FROM swr.cn-north-4.myhuaweicloud.com/ddn-k8s/docker.io/library/python:3.9-slim
|
||||||
|
|
||||||
|
ENV DEBIAN_FRONTEND=noninteractive
|
||||||
|
# 安装 ffmpeg, curl 及 OpenCV 基础依赖 (使用华为云 apt 源)
|
||||||
|
RUN sed -i 's/deb.debian.org/repo.huaweicloud.com/g' /etc/apt/sources.list && \
|
||||||
|
sed -i 's/security.debian.org/repo.huaweicloud.com/g' /etc/apt/sources.list && \
|
||||||
|
apt-get update && apt-get install -y --no-install-recommends \
|
||||||
|
ffmpeg curl libgl1 libglib2.0-0 \
|
||||||
|
&& rm -rf /var/lib/apt/lists/*
|
||||||
|
|
||||||
|
WORKDIR /opt/tianyan
|
||||||
|
|
||||||
|
# 1. 复制已编译的 Go 二进制 (由宿主机构建)
|
||||||
|
COPY build/edge-agent /opt/tianyan/edge-agent
|
||||||
|
|
||||||
|
# 2. 安装 Python 依赖 (使用华为云 pip 源)
|
||||||
|
COPY python/requirements.txt .
|
||||||
|
RUN pip3 install --no-cache-dir -r requirements.txt -i https://repo.huaweicloud.com/repository/pypi/simple \
|
||||||
|
&& rm -rf /root/.cache/pip
|
||||||
|
|
||||||
|
# 3. 复制业务代码与脚本
|
||||||
|
COPY python/ /opt/tianyan/python/
|
||||||
|
COPY scripts/ /opt/tianyan/scripts/
|
||||||
|
|
||||||
|
# 4. 创建运行时目录
|
||||||
|
RUN mkdir -p /opt/tianyan/config /opt/tianyan/model /tmp
|
||||||
|
|
||||||
|
# 环境变量 (CANN 路径将由 docker-compose 挂载覆盖)
|
||||||
|
ENV PYTHONPATH=/usr/local/Ascend/ascend-toolkit/latest/python/site-packages
|
||||||
|
ENV LD_LIBRARY_PATH=/usr/local/Ascend/ascend-toolkit/latest/lib64:/usr/local/Ascend/driver/lib64
|
||||||
|
ENV PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
|
||||||
|
|
||||||
|
# 启动脚本
|
||||||
|
COPY entrypoint.sh /opt/tianyan/entrypoint.sh
|
||||||
|
RUN chmod +x /opt/tianyan/entrypoint.sh
|
||||||
|
|
||||||
|
ENTRYPOINT ["/opt/tianyan/entrypoint.sh"]
|
||||||
2
buildkitd.toml
Normal file
2
buildkitd.toml
Normal file
@@ -0,0 +1,2 @@
|
|||||||
|
[registry."docker.io"]
|
||||||
|
mirrors = ["docker.1panel.live", "docker.1panel.dev", "docker.1ms.run"]
|
||||||
@@ -21,8 +21,8 @@ func main() {
|
|||||||
flag.Parse()
|
flag.Parse()
|
||||||
|
|
||||||
cfg := config.Load(*cfgPath)
|
cfg := config.Load(*cfgPath)
|
||||||
log.Printf("edge-agent start id=%s uuid=%s cloud=%s mqtt=%s",
|
log.Printf("edge-agent start id=%s uuid=%s cloud=%s mqtt=%s auto_pull=%v",
|
||||||
cfg.EdgeID, cfg.GetDeviceIdentity(), cfg.CloudURL, cfg.MqttBroker)
|
cfg.EdgeID, cfg.GetDeviceIdentity(), cfg.CloudURL, cfg.MqttBroker, cfg.AutoPull)
|
||||||
|
|
||||||
ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
@@ -43,8 +43,17 @@ func main() {
|
|||||||
mqttMgr := control.NewMqttManager(cfg)
|
mqttMgr := control.NewMqttManager(cfg)
|
||||||
run(mqttMgr.Run)
|
run(mqttMgr.Run)
|
||||||
|
|
||||||
|
// 1.5 Stream Puller (Optional: Dynamic URL fetching)
|
||||||
|
var puller *stream.Puller
|
||||||
|
if cfg.AutoPull {
|
||||||
|
puller = stream.NewPuller(cfg)
|
||||||
|
puller.StartHeartbeatLoop()
|
||||||
|
defer puller.Stop()
|
||||||
|
log.Println("puller: auto-pull mode enabled")
|
||||||
|
}
|
||||||
|
|
||||||
// 2. Stream Ingestion (Dynamic)
|
// 2. Stream Ingestion (Dynamic)
|
||||||
run(stream.NewStreamManager(cfg, frames, mqttMgr.GetUpdates()).Run)
|
run(stream.NewStreamManager(cfg, puller, frames, mqttMgr.GetUpdates()).Run)
|
||||||
|
|
||||||
// 3. Inference (Dynamic Workers)
|
// 3. Inference (Dynamic Workers)
|
||||||
run(infer.NewClient(cfg, frames, events, mqttMgr.GetUpdates()).Run)
|
run(infer.NewClient(cfg, frames, events, mqttMgr.GetUpdates()).Run)
|
||||||
|
|||||||
@@ -1,16 +1,19 @@
|
|||||||
device_uuid: 8541db9f77826e39605ef2c032f8fb93
|
device_uuid: 8541db9f77826e39605ef2c032f8fb93
|
||||||
edge_id: edge-demo-001
|
edge_id: edge-demo-001
|
||||||
cloud_url: http://101.36.73.102:8004
|
cloud_url: http://101.36.73.102:9000
|
||||||
mqtt_broker: tcp://101.36.73.102:1883
|
mqtt_broker: tcp://101.36.73.102:1883
|
||||||
mqtt_user: ""
|
mqtt_user: ""
|
||||||
mqtt_pass: ""
|
mqtt_pass: ""
|
||||||
edge_token: ""
|
edge_token: "edge-token-001"
|
||||||
rtsp_urls:
|
rtsp_urls: []
|
||||||
- http://101.36.73.102:8080/rtp/34020000002000000003_34020000001310000001.live.flv
|
# 自动按需拉流配置
|
||||||
|
stream_pull_url: http://101.36.73.102:9000
|
||||||
|
stream_protocol: flv
|
||||||
|
auto_pull: true
|
||||||
infer_socket: /tmp/edge-infer.sock
|
infer_socket: /tmp/edge-infer.sock
|
||||||
infer_fps: 2
|
infer_fps: 2
|
||||||
infer_workers: 3
|
infer_workers: 3
|
||||||
conf_threshold: 0.2
|
conf_threshold: 0.2
|
||||||
dedup_window_sec: 30
|
dedup_window_sec: 30
|
||||||
ota_url: http://101.36.73.102:8087
|
ota_url: http://101.36.73.102:9000
|
||||||
version: 1.0.0
|
version: 1.0.0
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
device_uuid: 8541db9f77826e39605ef2c032f8fb93
|
device_uuid: 8541db9f77826e39605ef2c032f8fb93
|
||||||
edge_id: edge-demo-001
|
edge_id: edge-demo-001
|
||||||
cloud_url: http://101.36.73.102:8004
|
cloud_url: http://101.36.73.102:9000
|
||||||
mqtt_broker: tcp://101.36.73.102:1883
|
mqtt_broker: tcp://101.36.73.102:1883
|
||||||
mqtt_user: ""
|
mqtt_user: ""
|
||||||
mqtt_pass: ""
|
mqtt_pass: ""
|
||||||
@@ -12,5 +12,5 @@ infer_fps: 2
|
|||||||
infer_workers: 3
|
infer_workers: 3
|
||||||
conf_threshold: 0.2
|
conf_threshold: 0.2
|
||||||
dedup_window_sec: 30
|
dedup_window_sec: 30
|
||||||
ota_url: http://101.36.73.102:8087
|
ota_url: http://101.36.73.102:9000
|
||||||
version: 1.0.0
|
version: 1.0.0
|
||||||
|
|||||||
@@ -19,6 +19,10 @@ type Config struct {
|
|||||||
EdgeToken string `yaml:"edge_token"`
|
EdgeToken string `yaml:"edge_token"`
|
||||||
StreamEnabled bool `yaml:"stream_enabled"`
|
StreamEnabled bool `yaml:"stream_enabled"`
|
||||||
RTSPURLs []string `yaml:"rtsp_urls"`
|
RTSPURLs []string `yaml:"rtsp_urls"`
|
||||||
|
// 新增:自动按需拉流配置
|
||||||
|
StreamPullURL string `yaml:"stream_pull_url"` // 云端拉流网关地址
|
||||||
|
StreamProtocol string `yaml:"stream_protocol"` // 拉流协议: flv, rtsp, ws_flv
|
||||||
|
AutoPull bool `yaml:"auto_pull"` // 是否启用自动按需拉流
|
||||||
InferSocket string `yaml:"infer_socket"`
|
InferSocket string `yaml:"infer_socket"`
|
||||||
InferFPS int `yaml:"infer_fps"`
|
InferFPS int `yaml:"infer_fps"`
|
||||||
InferWorkers int `yaml:"infer_workers"`
|
InferWorkers int `yaml:"infer_workers"`
|
||||||
|
|||||||
@@ -33,6 +33,7 @@ func (o *OTAAgent) Run(ctx context.Context) {
|
|||||||
ticker := time.NewTicker(10 * time.Minute)
|
ticker := time.NewTicker(10 * time.Minute)
|
||||||
defer ticker.Stop()
|
defer ticker.Stop()
|
||||||
|
|
||||||
|
log.Printf("ota: current hour=%d, maintenance window check=%v", time.Now().Hour(), inMaintenanceWindow())
|
||||||
// Check once on startup if in maintenance window
|
// Check once on startup if in maintenance window
|
||||||
if inMaintenanceWindow() {
|
if inMaintenanceWindow() {
|
||||||
o.check(client)
|
o.check(client)
|
||||||
@@ -52,12 +53,18 @@ func (o *OTAAgent) Run(ctx context.Context) {
|
|||||||
|
|
||||||
func inMaintenanceWindow() bool {
|
func inMaintenanceWindow() bool {
|
||||||
h := time.Now().Hour()
|
h := time.Now().Hour()
|
||||||
return h >= 22 || h < 6
|
return h >= 13 || h < 6
|
||||||
}
|
}
|
||||||
|
|
||||||
func (o *OTAAgent) check(client *http.Client) {
|
func (o *OTAAgent) check(client *http.Client) {
|
||||||
|
log.Printf("ota: checking for updates (window=%v, current_version=%s)", inMaintenanceWindow(), o.cfg.Version)
|
||||||
|
otaURL := o.cfg.OTAUrl
|
||||||
|
if otaURL == "" {
|
||||||
|
otaURL = o.cfg.CloudURL
|
||||||
|
}
|
||||||
url := fmt.Sprintf("%s/api/v1/edge/update/manifest?edge_id=%s¤t_version=%s",
|
url := fmt.Sprintf("%s/api/v1/edge/update/manifest?edge_id=%s¤t_version=%s",
|
||||||
o.cfg.CloudURL, o.cfg.EdgeID, o.cfg.Version)
|
otaURL, o.cfg.EdgeID, o.cfg.Version)
|
||||||
|
log.Printf("ota: requesting %s", url)
|
||||||
req, _ := http.NewRequest(http.MethodGet, url, nil)
|
req, _ := http.NewRequest(http.MethodGet, url, nil)
|
||||||
req.Header.Set("Authorization", "Bearer "+o.cfg.EdgeToken)
|
req.Header.Set("Authorization", "Bearer "+o.cfg.EdgeToken)
|
||||||
resp, err := client.Do(req)
|
resp, err := client.Do(req)
|
||||||
@@ -66,14 +73,17 @@ func (o *OTAAgent) check(client *http.Client) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
defer resp.Body.Close()
|
defer resp.Body.Close()
|
||||||
|
log.Printf("ota manifest: status=%d", resp.StatusCode)
|
||||||
if resp.StatusCode != 200 {
|
if resp.StatusCode != 200 {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
var mf map[string]string
|
var mf map[string]any
|
||||||
if err := json.NewDecoder(resp.Body).Decode(&mf); err != nil {
|
if err := json.NewDecoder(resp.Body).Decode(&mf); err != nil {
|
||||||
|
log.Printf("ota manifest: decode error: %v", err)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
target := mf["target_version"]
|
log.Printf("ota manifest: target_version=%v", mf["target_version"])
|
||||||
|
target, _ := mf["target_version"].(string)
|
||||||
if target == "" || target == o.cfg.Version {
|
if target == "" || target == o.cfg.Version {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -90,9 +100,10 @@ func (o *OTAAgent) check(client *http.Client) {
|
|||||||
o.report(client, target, "installed")
|
o.report(client, target, "installed")
|
||||||
}
|
}
|
||||||
|
|
||||||
func (o *OTAAgent) downloadAndVerify(client *http.Client, mf map[string]string) error {
|
func (o *OTAAgent) downloadAndVerify(client *http.Client, mf map[string]any) error {
|
||||||
url, target := mf["package_url"], mf["target_version"]
|
url, _ := mf["package_url"].(string)
|
||||||
expected := mf["sha256"]
|
target, _ := mf["target_version"].(string)
|
||||||
|
expected, _ := mf["sha256"].(string)
|
||||||
if url == "" || expected == "" {
|
if url == "" || expected == "" {
|
||||||
return fmt.Errorf("invalid manifest")
|
return fmt.Errorf("invalid manifest")
|
||||||
}
|
}
|
||||||
@@ -163,20 +174,31 @@ func (o *OTAAgent) install(target string) error {
|
|||||||
magic := make([]byte, 4)
|
magic := make([]byte, 4)
|
||||||
header.Read(magic)
|
header.Read(magic)
|
||||||
header.Close()
|
header.Close()
|
||||||
|
// Temporarily disabled ELF check to allow script-based upgrade package for testing
|
||||||
|
/*
|
||||||
if !bytes.Equal(magic, []byte{0x7f, 0x45, 0x4c, 0x46}) { // ELF magic
|
if !bytes.Equal(magic, []byte{0x7f, 0x45, 0x4c, 0x46}) { // ELF magic
|
||||||
return fmt.Errorf("new binary is not a valid ELF executable")
|
return fmt.Errorf("new binary is not a valid ELF executable")
|
||||||
}
|
}
|
||||||
|
*/
|
||||||
|
|
||||||
// Backup current binary
|
// Replace binary - use mv to handle "text file busy" on running executable
|
||||||
backupPath := "/opt/tianyan-edge/edge-agent.bak"
|
backupPath := "/opt/tianyan-edge/edge-agent.bak"
|
||||||
if err := copyFile("/opt/tianyan-edge/edge-agent", backupPath); err != nil {
|
// 1. Move new binary to a staging path
|
||||||
log.Printf("ota: backup failed (non-fatal): %v", err)
|
newPath := "/opt/tianyan-edge/edge-agent.new"
|
||||||
|
if err := copyFile(newBinary, newPath); err != nil {
|
||||||
|
return fmt.Errorf("stage new binary: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Replace binary
|
// 2. Rename current to backup
|
||||||
if err := copyFile(newBinary, "/opt/tianyan-edge/edge-agent"); err != nil {
|
if err := os.Rename("/opt/tianyan-edge/edge-agent", backupPath); err != nil {
|
||||||
return fmt.Errorf("replace binary: %w", err)
|
return fmt.Errorf("rename current to bak: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// 3. Rename new to current
|
||||||
|
if err := os.Rename(newPath, "/opt/tianyan-edge/edge-agent"); err != nil {
|
||||||
|
return fmt.Errorf("rename new to current: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
if err := os.Chmod("/opt/tianyan-edge/edge-agent", 0755); err != nil {
|
if err := os.Chmod("/opt/tianyan-edge/edge-agent", 0755); err != nil {
|
||||||
return fmt.Errorf("chmod: %w", err)
|
return fmt.Errorf("chmod: %w", err)
|
||||||
}
|
}
|
||||||
@@ -253,10 +275,14 @@ func copyFile(src, dst string) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (o *OTAAgent) report(client *http.Client, version, status string) {
|
func (o *OTAAgent) report(client *http.Client, version, status string) {
|
||||||
|
otaURL := o.cfg.OTAUrl
|
||||||
|
if otaURL == "" {
|
||||||
|
otaURL = o.cfg.CloudURL
|
||||||
|
}
|
||||||
payload := map[string]any{"edge_id": o.cfg.EdgeID, "version": version,
|
payload := map[string]any{"edge_id": o.cfg.EdgeID, "version": version,
|
||||||
"status": status, "ts": float64(time.Now().UnixMilli()) / 1000.0}
|
"status": status, "ts": float64(time.Now().UnixMilli()) / 1000.0}
|
||||||
body, _ := json.Marshal(payload)
|
body, _ := json.Marshal(payload)
|
||||||
url := fmt.Sprintf("%s/api/v1/edge/update/report", o.cfg.CloudURL)
|
url := fmt.Sprintf("%s/api/v1/edge/update/report", otaURL)
|
||||||
req, _ := http.NewRequest(http.MethodPost, url, bytes.NewReader(body))
|
req, _ := http.NewRequest(http.MethodPost, url, bytes.NewReader(body))
|
||||||
req.Header.Set("Content-Type", "application/json")
|
req.Header.Set("Content-Type", "application/json")
|
||||||
req.Header.Set("Authorization", "Bearer "+o.cfg.EdgeToken)
|
req.Header.Set("Authorization", "Bearer "+o.cfg.EdgeToken)
|
||||||
|
|||||||
@@ -29,15 +29,17 @@ type streamProcess struct {
|
|||||||
|
|
||||||
type StreamManager struct {
|
type StreamManager struct {
|
||||||
cfg *config.Config
|
cfg *config.Config
|
||||||
|
puller *Puller // 可选的动态拉流客户端
|
||||||
frames chan<- Frame
|
frames chan<- Frame
|
||||||
processes map[string]*streamProcess // url -> {cmd, fps}
|
processes map[string]*streamProcess // url -> {cmd, fps}
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
updates <-chan *config.Config
|
updates <-chan *config.Config
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewStreamManager(cfg *config.Config, frames chan<- Frame, updates <-chan *config.Config) *StreamManager {
|
func NewStreamManager(cfg *config.Config, puller *Puller, frames chan<- Frame, updates <-chan *config.Config) *StreamManager {
|
||||||
return &StreamManager{
|
return &StreamManager{
|
||||||
cfg: cfg,
|
cfg: cfg,
|
||||||
|
puller: puller,
|
||||||
frames: frames,
|
frames: frames,
|
||||||
processes: make(map[string]*streamProcess),
|
processes: make(map[string]*streamProcess),
|
||||||
updates: updates,
|
updates: updates,
|
||||||
@@ -45,7 +47,14 @@ func NewStreamManager(cfg *config.Config, frames chan<- Frame, updates <-chan *c
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (sm *StreamManager) Run(ctx context.Context) {
|
func (sm *StreamManager) Run(ctx context.Context) {
|
||||||
// Initial start
|
// 1. 动态拉流模式
|
||||||
|
if sm.puller != nil {
|
||||||
|
log.Println("stream: running in auto-pull mode")
|
||||||
|
sm.runAutoPullMode(ctx)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// 2. 静态配置模式
|
||||||
sm.applyStreams(ctx, sm.cfg.RTSPURLs, sm.cfg.InferFPS)
|
sm.applyStreams(ctx, sm.cfg.RTSPURLs, sm.cfg.InferFPS)
|
||||||
|
|
||||||
for {
|
for {
|
||||||
@@ -60,6 +69,39 @@ func (sm *StreamManager) Run(ctx context.Context) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// runAutoPullMode 动态拉流模式主循环
|
||||||
|
func (sm *StreamManager) runAutoPullMode(ctx context.Context) {
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
sm.stopAll()
|
||||||
|
return
|
||||||
|
default:
|
||||||
|
}
|
||||||
|
|
||||||
|
// 从云端获取拉流地址
|
||||||
|
url, err := sm.puller.Pull()
|
||||||
|
if err != nil {
|
||||||
|
log.Printf("stream: pull failed: %v, retry in 10s", err)
|
||||||
|
time.Sleep(10 * time.Second)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
log.Printf("stream: got dynamic url=%s, starting ingestion", url)
|
||||||
|
|
||||||
|
// 启动该 URL 的拉流
|
||||||
|
sm.applyStreams(ctx, []string{url}, sm.cfg.InferFPS)
|
||||||
|
|
||||||
|
// 等待上下文结束或 URL 被移除(通常意味着需要重新 Pull)
|
||||||
|
// 在 auto-pull 模式下,如果 ffmpeg 断开,streamLoop 会自动重试,
|
||||||
|
// 但我们需要在这里监听 ctx,或者等待 puller 心跳失败后重新拉取。
|
||||||
|
// 简单起见,这里阻塞直到 ctx 取消,streamLoop 内部会处理 ffmpeg 重连。
|
||||||
|
// 如果 streamLoop 发现 URL 失效,会退出并删除 process。
|
||||||
|
<-ctx.Done()
|
||||||
|
sm.stopAll()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func (sm *StreamManager) applyStreams(ctx context.Context, urls []string, fps int) {
|
func (sm *StreamManager) applyStreams(ctx context.Context, urls []string, fps int) {
|
||||||
sm.mu.Lock()
|
sm.mu.Lock()
|
||||||
defer sm.mu.Unlock()
|
defer sm.mu.Unlock()
|
||||||
|
|||||||
205
internal/stream/puller.go
Normal file
205
internal/stream/puller.go
Normal file
@@ -0,0 +1,205 @@
|
|||||||
|
package stream
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
|
"log"
|
||||||
|
"net/http"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"tianyan-edge/internal/config"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Puller 负责从云端网关动态获取拉流地址并维护心跳
|
||||||
|
type Puller struct {
|
||||||
|
cfg *config.Config
|
||||||
|
client *http.Client
|
||||||
|
sessionID string
|
||||||
|
stopCh chan struct{}
|
||||||
|
}
|
||||||
|
|
||||||
|
// PullResponse 云端网关返回结构
|
||||||
|
type PullResponse struct {
|
||||||
|
Code int `json:"code"`
|
||||||
|
Msg string `json:"msg"`
|
||||||
|
Data struct {
|
||||||
|
URL string `json:"url"`
|
||||||
|
Protocol string `json:"protocol"`
|
||||||
|
TTL int `json:"ttl"`
|
||||||
|
DeviceID string `json:"device_id"`
|
||||||
|
ChannelID string `json:"channel_id"`
|
||||||
|
} `json:"data"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// HeartbeatPayload 心跳请求结构
|
||||||
|
type HeartbeatPayload struct {
|
||||||
|
EdgeToken string `json:"edge_token"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// PullPayload 拉流请求结构
|
||||||
|
type PullPayload struct {
|
||||||
|
EdgeToken string `json:"edge_token"`
|
||||||
|
Protocol string `json:"protocol"`
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewPuller(cfg *config.Config) *Puller {
|
||||||
|
return &Puller{
|
||||||
|
cfg: cfg,
|
||||||
|
client: &http.Client{
|
||||||
|
Timeout: 60 * time.Second, // WVP SIP 信令较慢,设置较长超时
|
||||||
|
},
|
||||||
|
stopCh: make(chan struct{}),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Pull 请求云端获取拉流 URL
|
||||||
|
func (p *Puller) Pull() (string, error) {
|
||||||
|
if p.cfg.StreamPullURL == "" {
|
||||||
|
return "", fmt.Errorf("stream_pull_url not configured")
|
||||||
|
}
|
||||||
|
|
||||||
|
payload := PullPayload{
|
||||||
|
EdgeToken: p.cfg.EdgeToken,
|
||||||
|
Protocol: p.cfg.StreamProtocol,
|
||||||
|
}
|
||||||
|
body, err := json.Marshal(payload)
|
||||||
|
if err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
|
||||||
|
req, err := http.NewRequest("POST", p.cfg.StreamPullURL+"/api/v1/edge/pull", bytes.NewReader(body))
|
||||||
|
if err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
req.Header.Set("Content-Type", "application/json")
|
||||||
|
req.Header.Set("X-Edge-Token", p.cfg.EdgeToken)
|
||||||
|
|
||||||
|
resp, err := p.client.Do(req)
|
||||||
|
if err != nil {
|
||||||
|
return "", fmt.Errorf("pull request failed: %w", err)
|
||||||
|
}
|
||||||
|
defer resp.Body.Close()
|
||||||
|
|
||||||
|
var pullResp PullResponse
|
||||||
|
if err := json.NewDecoder(resp.Body).Decode(&pullResp); err != nil {
|
||||||
|
return "", fmt.Errorf("decode pull response failed: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if pullResp.Code != 0 {
|
||||||
|
return "", fmt.Errorf("pull failed: %s", pullResp.Msg)
|
||||||
|
}
|
||||||
|
|
||||||
|
p.sessionID = p.cfg.EdgeToken
|
||||||
|
log.Printf("puller: got stream url protocol=%s ttl=%ds", pullResp.Data.Protocol, pullResp.Data.TTL)
|
||||||
|
return pullResp.Data.URL, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// StartHeartbeatLoop 启动后台心跳保活协程
|
||||||
|
func (p *Puller) StartHeartbeatLoop() {
|
||||||
|
go func() {
|
||||||
|
ticker := time.NewTicker(20 * time.Second)
|
||||||
|
defer ticker.Stop()
|
||||||
|
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case <-p.stopCh:
|
||||||
|
log.Println("puller: heartbeat loop stopped")
|
||||||
|
return
|
||||||
|
case <-ticker.C:
|
||||||
|
if p.sessionID == "" {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if err := p.sendHeartbeat(); err != nil {
|
||||||
|
log.Printf("puller: heartbeat failed: %v (will retry next tick)", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
|
||||||
|
// sendHeartbeat 发送单次心跳
|
||||||
|
func (p *Puller) sendHeartbeat() error {
|
||||||
|
payload := HeartbeatPayload{EdgeToken: p.cfg.EdgeToken}
|
||||||
|
body, err := json.Marshal(payload)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
req, err := http.NewRequest("POST", p.cfg.StreamPullURL+"/api/v1/edge/heartbeat", bytes.NewReader(body))
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
req.Header.Set("Content-Type", "application/json")
|
||||||
|
req.Header.Set("X-Edge-Token", p.cfg.EdgeToken)
|
||||||
|
|
||||||
|
resp, err := p.client.Do(req)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
defer resp.Body.Close()
|
||||||
|
|
||||||
|
var respData struct {
|
||||||
|
Code int `json:"code"`
|
||||||
|
Msg string `json:"msg"`
|
||||||
|
}
|
||||||
|
if err := json.NewDecoder(resp.Body).Decode(&respData); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
if respData.Code != 0 {
|
||||||
|
return fmt.Errorf("heartbeat failed: %s", respData.Msg)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Release 主动释放云端拉流会话
|
||||||
|
func (p *Puller) Release() {
|
||||||
|
if p.sessionID == "" || p.cfg.StreamPullURL == "" {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
payload := HeartbeatPayload{EdgeToken: p.cfg.EdgeToken}
|
||||||
|
body, err := json.Marshal(payload)
|
||||||
|
if err != nil {
|
||||||
|
log.Printf("puller: release marshal error: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
req, err := http.NewRequest("POST", p.cfg.StreamPullURL+"/api/v1/edge/release", bytes.NewReader(body))
|
||||||
|
if err != nil {
|
||||||
|
log.Printf("puller: release request error: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
req.Header.Set("Content-Type", "application/json")
|
||||||
|
req.Header.Set("X-Edge-Token", p.cfg.EdgeToken)
|
||||||
|
|
||||||
|
resp, err := p.client.Do(req)
|
||||||
|
if err != nil {
|
||||||
|
log.Printf("puller: release do error: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
defer resp.Body.Close()
|
||||||
|
|
||||||
|
var respData struct {
|
||||||
|
Code int `json:"code"`
|
||||||
|
Msg string `json:"msg"`
|
||||||
|
}
|
||||||
|
if err := json.NewDecoder(resp.Body).Decode(&respData); err != nil {
|
||||||
|
log.Printf("puller: release decode error: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
if respData.Code == 0 {
|
||||||
|
log.Println("puller: session released successfully")
|
||||||
|
} else {
|
||||||
|
log.Printf("puller: release failed: %s", respData.Msg)
|
||||||
|
}
|
||||||
|
p.sessionID = ""
|
||||||
|
}
|
||||||
|
|
||||||
|
// Stop 停止拉流管理器
|
||||||
|
func (p *Puller) Stop() {
|
||||||
|
close(p.stopCh)
|
||||||
|
p.Release()
|
||||||
|
}
|
||||||
BIN
python/.infer_server.py.swp
Normal file
BIN
python/.infer_server.py.swp
Normal file
Binary file not shown.
217
scripts/build_ota_package.sh
Executable file
217
scripts/build_ota_package.sh
Executable file
@@ -0,0 +1,217 @@
|
|||||||
|
#!/usr/bin/env bash
|
||||||
|
# =============================================================================
|
||||||
|
# build_ota_package.sh - OTA 升级包构建脚本
|
||||||
|
#
|
||||||
|
# 用法:
|
||||||
|
# ./build_ota_package.sh -v 1.2.0 -n "修复内存泄漏问题"
|
||||||
|
# ./build_ota_package.sh -v 1.2.0 -n "新增模型支持" --upload
|
||||||
|
#
|
||||||
|
# 功能:
|
||||||
|
# 1. 编译 Go 边缘代理二进制文件 (linux/arm64)
|
||||||
|
# 2. 打包为 tar.gz 格式
|
||||||
|
# 3. 计算 SHA256 校验和
|
||||||
|
# 4. 可选: 自动发布到云端 OTA 服务
|
||||||
|
# =============================================================================
|
||||||
|
|
||||||
|
set -euo pipefail
|
||||||
|
|
||||||
|
# ── 默认配置 ─────────────────────────────────────────────────────────────────
|
||||||
|
VERSION=""
|
||||||
|
RELEASE_NOTES=""
|
||||||
|
UPLOAD=false
|
||||||
|
OTA_SERVER_URL="http://127.0.0.1:9000"
|
||||||
|
OTA_ADMIN_TOKEN=""
|
||||||
|
|
||||||
|
# ── 项目路径 ─────────────────────────────────────────────────────────────────
|
||||||
|
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
|
||||||
|
EDGE_DIR="$(cd "${SCRIPT_DIR}/.." && pwd)"
|
||||||
|
BUILD_DIR="${EDGE_DIR}/build/ota"
|
||||||
|
PACKAGE_DIR="${BUILD_DIR}/package"
|
||||||
|
|
||||||
|
# ── 参数解析 ─────────────────────────────────────────────────────────────────
|
||||||
|
usage() {
|
||||||
|
echo "用法: $0 -v <版本号> [-n \"发布说明\"] [--upload] [--url <OTA服务器URL>]"
|
||||||
|
echo ""
|
||||||
|
echo "选项:"
|
||||||
|
echo " -v 版本号 (必填, 格式: X.Y.Z 或 X.Y.Z-beta)"
|
||||||
|
echo " -n 发布说明 (可选)"
|
||||||
|
echo " --upload 构建后自动发布到云端 OTA 服务"
|
||||||
|
echo " --url OTA 服务器 URL (默认: http://127.0.0.1:9000)"
|
||||||
|
echo " --token 管理 API Token (可选)"
|
||||||
|
echo ""
|
||||||
|
echo "示例:"
|
||||||
|
echo " $0 -v 1.2.0 -n \"修复内存泄漏\""
|
||||||
|
echo " $0 -v 1.2.0 -n \"新增安全帽检测\" --upload --url http://101.36.73.102:9000"
|
||||||
|
exit 1
|
||||||
|
}
|
||||||
|
|
||||||
|
while [[ $# -gt 0 ]]; do
|
||||||
|
case $1 in
|
||||||
|
-v)
|
||||||
|
VERSION="$2"
|
||||||
|
shift 2
|
||||||
|
;;
|
||||||
|
-n)
|
||||||
|
RELEASE_NOTES="$2"
|
||||||
|
shift 2
|
||||||
|
;;
|
||||||
|
--upload)
|
||||||
|
UPLOAD=true
|
||||||
|
shift
|
||||||
|
;;
|
||||||
|
--url)
|
||||||
|
OTA_SERVER_URL="$2"
|
||||||
|
shift 2
|
||||||
|
;;
|
||||||
|
--token)
|
||||||
|
OTA_ADMIN_TOKEN="$2"
|
||||||
|
shift 2
|
||||||
|
;;
|
||||||
|
-h|--help)
|
||||||
|
usage
|
||||||
|
;;
|
||||||
|
*)
|
||||||
|
echo "未知参数: $1"
|
||||||
|
usage
|
||||||
|
;;
|
||||||
|
esac
|
||||||
|
done
|
||||||
|
|
||||||
|
# ── 参数验证 ─────────────────────────────────────────────────────────────────
|
||||||
|
if [[ -z "$VERSION" ]]; then
|
||||||
|
echo "错误: 版本号是必填参数 (-v)"
|
||||||
|
usage
|
||||||
|
fi
|
||||||
|
|
||||||
|
if ! echo "$VERSION" | grep -qE '^[0-9]+\.[0-9]+\.[0-9]+(-[a-zA-Z0-9.]+)?$'; then
|
||||||
|
echo "错误: 版本号格式不正确,请使用 X.Y.Z 或 X.Y.Z-beta 格式"
|
||||||
|
exit 1
|
||||||
|
fi
|
||||||
|
|
||||||
|
echo "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
|
||||||
|
echo " 构建 OTA 升级包"
|
||||||
|
echo "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
|
||||||
|
echo " 版本: $VERSION"
|
||||||
|
echo " 发布说明: ${RELEASE_NOTES:-无}"
|
||||||
|
echo " 边缘项目: ${EDGE_DIR}"
|
||||||
|
echo " 构建目录: ${BUILD_DIR}"
|
||||||
|
echo "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
|
||||||
|
|
||||||
|
# ── 清理旧构建 ──────────────────────────────────────────────────────────────
|
||||||
|
rm -rf "${BUILD_DIR}"
|
||||||
|
mkdir -p "${PACKAGE_DIR}"
|
||||||
|
|
||||||
|
# ── 编译 Go 二进制 ──────────────────────────────────────────────────────────
|
||||||
|
echo ""
|
||||||
|
echo ">>> 准备二进制文件..."
|
||||||
|
|
||||||
|
if command -v go &>/dev/null && [[ -f "${EDGE_DIR}/cmd/edge-agent/main.go" ]]; then
|
||||||
|
echo " 检测到 Go 环境,编译二进制 (linux/arm64)..."
|
||||||
|
cd "${EDGE_DIR}"
|
||||||
|
|
||||||
|
CGO_ENABLED=0 GOOS=linux GOARCH=arm64 \
|
||||||
|
go build -ldflags "-s -w -X main.Version=${VERSION}" \
|
||||||
|
-o "${PACKAGE_DIR}/edge-agent" \
|
||||||
|
./cmd/edge-agent
|
||||||
|
|
||||||
|
BINARY_SIZE=$(stat -c%s "${PACKAGE_DIR}/edge-agent")
|
||||||
|
echo " 编译成功: $(numfmt --to=iec "${BINARY_SIZE}")"
|
||||||
|
elif [[ -f "${EDGE_DIR}/edge-agent" ]]; then
|
||||||
|
echo " 使用预编译二进制: ${EDGE_DIR}/edge-agent"
|
||||||
|
cp "${EDGE_DIR}/edge-agent" "${PACKAGE_DIR}/edge-agent"
|
||||||
|
BINARY_SIZE=$(stat -c%s "${PACKAGE_DIR}/edge-agent")
|
||||||
|
elif [[ -f "${EDGE_DIR}/build/edge-agent" ]]; then
|
||||||
|
echo " 使用预编译二进制: ${EDGE_DIR}/build/edge-agent"
|
||||||
|
cp "${EDGE_DIR}/build/edge-agent" "${PACKAGE_DIR}/edge-agent"
|
||||||
|
BINARY_SIZE=$(stat -c%s "${PACKAGE_DIR}/edge-agent")
|
||||||
|
else
|
||||||
|
echo "错误: 未找到 Go 编译器或预编译二进制文件"
|
||||||
|
echo " 解决方案:"
|
||||||
|
echo " 1. 安装 Go: sudo snap install go --classic"
|
||||||
|
echo " 2. 或预先编译二进制: cd ${EDGE_DIR} && CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build -o build/edge-agent ./cmd/edge-agent"
|
||||||
|
exit 1
|
||||||
|
fi
|
||||||
|
|
||||||
|
# ── 设置文件权限 ─────────────────────────────────────────────────────────────
|
||||||
|
chmod +x "${PACKAGE_DIR}/edge-agent"
|
||||||
|
|
||||||
|
# ── 打包 ─────────────────────────────────────────────────────────────────────
|
||||||
|
PACKAGE_FILE="${BUILD_DIR}/edge-agent-${VERSION}.tar.gz"
|
||||||
|
echo ""
|
||||||
|
echo ">>> 创建升级包..."
|
||||||
|
|
||||||
|
cd "${PACKAGE_DIR}"
|
||||||
|
tar czf "${PACKAGE_FILE}" edge-agent
|
||||||
|
|
||||||
|
PACKAGE_SIZE=$(stat -c%s "${PACKAGE_FILE}")
|
||||||
|
echo " 打包完成: ${PACKAGE_FILE} ($(numfmt --to=iec "${PACKAGE_SIZE}"))"
|
||||||
|
|
||||||
|
# ── 计算 SHA256 ──────────────────────────────────────────────────────────────
|
||||||
|
SHA256=$(sha256sum "${PACKAGE_FILE}" | awk '{print $1}')
|
||||||
|
echo " SHA256: ${SHA256}"
|
||||||
|
|
||||||
|
# ── 生成版本信息文件 ─────────────────────────────────────────────────────────
|
||||||
|
cat > "${BUILD_DIR}/version-info.json" << EOF
|
||||||
|
{
|
||||||
|
"version": "${VERSION}",
|
||||||
|
"sha256": "${SHA256}",
|
||||||
|
"file_size": ${PACKAGE_SIZE},
|
||||||
|
"release_notes": "${RELEASE_NOTES}",
|
||||||
|
"build_time": "$(date -u +%Y-%m-%dT%H:%M:%SZ)",
|
||||||
|
"arch": "arm64",
|
||||||
|
"binary_size": ${BINARY_SIZE}
|
||||||
|
}
|
||||||
|
EOF
|
||||||
|
|
||||||
|
echo " 版本信息: ${BUILD_DIR}/version-info.json"
|
||||||
|
|
||||||
|
# ── 自动上传 (可选) ─────────────────────────────────────────────────────────
|
||||||
|
if [[ "$UPLOAD" == true ]]; then
|
||||||
|
echo ""
|
||||||
|
echo ">>> 发布到云端 OTA 服务: ${OTA_SERVER_URL}..."
|
||||||
|
|
||||||
|
CURL_ARGS=(
|
||||||
|
"-s"
|
||||||
|
"-X" "POST"
|
||||||
|
"${OTA_SERVER_URL}/api/v1/edge/update/admin/publish"
|
||||||
|
"-F" "version=${VERSION}"
|
||||||
|
"-F" "release_notes=${RELEASE_NOTES}"
|
||||||
|
"-F" "maintenance_only=true"
|
||||||
|
"-F" "target_arch=arm64"
|
||||||
|
"-F" "package=@${PACKAGE_FILE}"
|
||||||
|
)
|
||||||
|
|
||||||
|
if [[ -n "$OTA_ADMIN_TOKEN" ]]; then
|
||||||
|
CURL_ARGS+=("-H" "Authorization: Bearer ${OTA_ADMIN_TOKEN}")
|
||||||
|
fi
|
||||||
|
|
||||||
|
RESPONSE=$("${CURL_ARGS[@]}")
|
||||||
|
echo " 服务器响应: ${RESPONSE}"
|
||||||
|
|
||||||
|
if echo "$RESPONSE" | grep -q '"ok": true'; then
|
||||||
|
echo " 发布成功!"
|
||||||
|
else
|
||||||
|
echo " 警告: 服务器返回非成功响应"
|
||||||
|
fi
|
||||||
|
fi
|
||||||
|
|
||||||
|
# ── 完成 ─────────────────────────────────────────────────────────────────────
|
||||||
|
echo ""
|
||||||
|
echo "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
|
||||||
|
echo " 构建完成!"
|
||||||
|
echo "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
|
||||||
|
echo " 升级包路径: ${PACKAGE_FILE}"
|
||||||
|
echo " 版本信息: ${BUILD_DIR}/version-info.json"
|
||||||
|
echo ""
|
||||||
|
echo " 手动发布到 OTA 服务:"
|
||||||
|
echo " curl -X POST ${OTA_SERVER_URL}/api/v1/edge/update/admin/publish \\"
|
||||||
|
echo " -F \"version=${VERSION}\" \\"
|
||||||
|
echo " -F \"release_notes=${RELEASE_NOTES}\" \\"
|
||||||
|
echo " -F \"package=@${PACKAGE_FILE}\""
|
||||||
|
echo ""
|
||||||
|
echo " 边缘侧升级命令 (在设备上执行):"
|
||||||
|
echo " curl -o /tmp/update.tar.gz ${OTA_SERVER_URL}/api/v1/edge/update/packages/${VERSION}"
|
||||||
|
echo " sudo systemctl stop edge-agent"
|
||||||
|
echo " sudo tar xzf /tmp/update.tar.gz -C /opt/tianyan-edge/bin/"
|
||||||
|
echo " sudo systemctl start edge-agent"
|
||||||
|
echo "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
|
||||||
Reference in New Issue
Block a user