1
2
3
4
5
6
7
8
9

# Go 1.21+ must be installed
go version

# FFmpeg must be installed
ffmpeg -version

# (Optional) For OpenCV and GoCV
# Linux: apt-get install libopencv-dev
# macOS: brew install opencv

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

video-processor/
├── cmd/
│   └── main.go
├── internal/
│   ├── pipeline/
│   ├── detector/
│   ├── source/
│   └── publisher/
├── config/
│   └── config.yaml
├── go.mod
└── Dockerfile

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# Create the project
mkdir video-processor && cd video-processor
go mod init video-processor

# Add dependencies
go get github.com/eclipse/paho.mqtt.golang
go get gopkg.in/yaml.v3
go get github.com/sirupsen/logrus

# (Optional) GoCV
go get gocv.io/x/gocv

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# Install OpenCV dependencies
sudo apt-get update
sudo apt-get install -y \
    libopencv-dev \
    pkg-config \
    libavcodec-dev \
    libavformat-dev \
    libavutil-dev \
    libswscale-dev

# Install GoCV
go get gocv.io/x/gocv

1
2
3
4
5

# Install OpenCV with Homebrew
brew install opencv pkg-config

# Install GoCV
go get gocv.io/x/gocv

1
2
3
4
5
6
7
8
9

# Install OpenCV with Chocolatey (or manually)
choco install opencv

# Set environment variables
set CGO_CPPFLAGS=-IC:\opencv\build\include
set CGO_LDFLAGS=-LC:\opencv\build\x64\vc15\lib

# Install GoCV
go get gocv.io/x/gocv

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# Build stage
FROM golang:1.23 AS builder

# Install OpenCV dependencies
RUN apt-get update && apt-get install -y \
    libopencv-dev \
    pkg-config \
    && rm -rf /var/lib/apt/lists/*

WORKDIR /app
COPY go.mod go.sum ./
RUN go mod download

COPY . .
RUN CGO_ENABLED=1 go build -o video-processor ./cmd/main.go

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

// cmd/main.go
package main

import (
    "context"
    "log"
    "time"
)

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    
    // Test video source (from file)
    source := NewVideoSource("file://test.mp4", 640, 480)
    
    if err := source.Start(ctx); err != nil {
        log.Fatal(err)
    }
    
    // Read first 10 frames
    frameCount := 0
    for frame := range source.Frames() {
        log.Printf("Frame %d received: %dx%d", frameCount, frame.Width, frame.Height)
        frameCount++
        if frameCount >= 10 {
            break
        }
    }
    
    source.Stop()
    log.Println("Test completed!")
}

1
2
3
4
5
6

// Format selection in FFmpeg
// For RGB24:
args := []string{"-pix_fmt", "rgb24"}

// For YUV420 (smaller):
args := []string{"-pix_fmt", "yuv420p"}

1

ffmpeg -f v4l2 -i /dev/video0 -f rawvideo -pix_fmt rgb24 -s 640x480 pipe:1

1

ffmpeg -i test_video.mp4 -f rawvideo -pix_fmt rgb24 -s 640x480 pipe:1

1
2

ffmpeg -rtsp_transport tcp -i rtsp://username:password@camera-ip:554/stream \
       -f rawvideo -pix_fmt rgb24 -s 640x480 pipe:1

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153

package main

import (
    "bufio"
    "context"
    "fmt"
    "io"
    "log"
    "os/exec"
    "sync"
    "time"
)

type Frame struct {
    Data      []byte
    Width     int
    Height    int
    Timestamp time.Time
}

type VideoSource struct {
    sourceURL string
    width     int
    height    int
    cmd       *exec.Cmd
    stdout    io.ReadCloser
    frames    chan *Frame
    errors    chan error
    wg        sync.WaitGroup
}

func NewVideoSource(sourceURL string, width, height int) *VideoSource {
    return &VideoSource{
        sourceURL: sourceURL,
        width:     width,
        height:    height,
        frames:    make(chan *Frame, 10), // Buffer 10 frames
        errors:    make(chan error, 1),
    }
}

func (vs *VideoSource) Start(ctx context.Context) error {
    // Create FFmpeg command
    args := []string{
        "-rtsp_transport", "tcp",
        "-i", vs.sourceURL,
        "-f", "rawvideo",
        "-pix_fmt", "rgb24",
        "-s", fmt.Sprintf("%dx%d", vs.width, vs.height),
        "pipe:1",
    }
    
    vs.cmd = exec.CommandContext(ctx, "ffmpeg", args...)
    
    var err error
    vs.stdout, err = vs.cmd.StdoutPipe()
    if err != nil {
        return fmt.Errorf("stdout pipe: %w", err)
    }

    stderr, err := vs.cmd.StderrPipe()
    if err != nil {
        return fmt.Errorf("stderr pipe: %w", err)
    }

    if err := vs.cmd.Start(); err != nil {
        return fmt.Errorf("start ffmpeg: %w", err)
    }

    // Log FFmpeg stderr output for debugging
    go func() {
        scanner := bufio.NewScanner(stderr)
        for scanner.Scan() {
            log.Printf("FFmpeg: %s", scanner.Text())
        }
    }()
    
    // Frame reading goroutine
    vs.wg.Add(1)
    go vs.readFrames(ctx)
    
    return nil
}

func (vs *VideoSource) readFrames(ctx context.Context) {
    defer vs.wg.Done()
    
    frameSize := vs.width * vs.height * 3 // RGB24: 3 bytes per pixel
    
    frameBuffer := make([]byte, frameSize)
    
    for {
        select {
        case <-ctx.Done():
            // When context is cancelled, also terminate the FFmpeg process
            if vs.cmd != nil && vs.cmd.Process != nil {
                _ = vs.cmd.Process.Kill()
            }
            return
        default:
            // Read a complete frame
            n, err := io.ReadFull(vs.stdout, frameBuffer)
            if err != nil {
                if err != io.EOF && err != io.ErrUnexpectedEOF {
                    select {
                    case vs.errors <- fmt.Errorf("read frame: %w", err):
                    case <-ctx.Done():
                    }
                }
                return
            }
            
            if n != frameSize {
                continue // Incomplete frame, skip
            }
            
            // Copy new frame buffer (goroutine safe)
            frameData := make([]byte, frameSize)
            copy(frameData, frameBuffer)
            
            frame := &Frame{
                Data:      frameData,
                Width:     vs.width,
                Height:    vs.height,
                Timestamp: time.Now(),
            }
            
            select {
            case vs.frames <- frame:
            case <-ctx.Done():
                return
            }
        }
    }
}

func (vs *VideoSource) Frames() <-chan *Frame {
    return vs.frames
}

func (vs *VideoSource) Errors() <-chan error {
    return vs.errors
}

func (vs *VideoSource) Stop() error {
    if vs.cmd != nil && vs.cmd.Process != nil {
        vs.cmd.Process.Kill()
    }
    vs.wg.Wait()
    close(vs.frames)
    close(vs.errors)
    return nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86

package main

import (
    "context"
    "fmt"
    "sync"
    "time"
)

type Pipeline struct {
    source    *VideoSource
    processors []FrameProcessor
    wg        sync.WaitGroup
}

type FrameProcessor interface {
    Process(ctx context.Context, in <-chan *Frame, out chan<- *Frame)
    Name() string
}

func NewPipeline(source *VideoSource) *Pipeline {
    return &Pipeline{
        source:     source,
        processors: make([]FrameProcessor, 0),
    }
}

func (p *Pipeline) AddProcessor(processor FrameProcessor) {
    p.processors = append(p.processors, processor)
}

func (p *Pipeline) Run(ctx context.Context) error {
    // Start video source
    if err := p.source.Start(ctx); err != nil {
        return err
    }
    
    // Create channels
    channels := make([]chan *Frame, len(p.processors)+1)
    for i := range channels {
        channels[i] = make(chan *Frame, 10)
    }
    
    // Copy from source to first channel
    p.wg.Add(1)
    go func() {
        defer p.wg.Done()
        defer close(channels[0])
        for frame := range p.source.Frames() {
            select {
            case channels[0] <- frame:
            case <-ctx.Done():
                return
            }
        }
    }()
    
    // Run each processor
    for i, proc := range p.processors {
        p.wg.Add(1)
        go func(idx int, processor FrameProcessor) {
            defer p.wg.Done()
            defer func() {
                if r := recover(); r != nil {
                    fmt.Printf("Processor panic: %v\n", r)
                }
                close(channels[idx+1])
            }()
            processor.Process(ctx, channels[idx], channels[idx+1])
        }(i, proc)
    }
    
    // Error handling
    go func() {
        for err := range p.source.Errors() {
            // Log error (example)
            fmt.Printf("Video source error: %v\n", err)
        }
    }()
    
    return nil
}

func (p *Pipeline) Stop() error {
    return p.source.Stop()
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

type Frame struct {
    Data      []byte
    Width     int
    Height    int
    Timestamp time.Time
    Metadata  map[string]interface{} // For analysis results
}

func NewFrame(data []byte, width, height int) *Frame {
    return &Frame{
        Data:      data,
        Width:     width,
        Height:    height,
        Timestamp: time.Now(),
        Metadata:  make(map[string]interface{}),
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122

import (
    "image"
    "sync"
)

type ROIDetector struct {
    regions []image.Rectangle
    mutex   sync.RWMutex
}

func NewROIDetector() *ROIDetector {
    return &ROIDetector{
        regions: make([]image.Rectangle, 0),
    }
}

// Determine ROI based on motion detection results
func (rd *ROIDetector) UpdateROI(motionMap []bool, width, height int) {
    rd.mutex.Lock()
    defer rd.mutex.Unlock()
    
    // Detect motion regions and create bounding box
    // Simple implementation: bounding box of motion pixels
    minX, minY := width, height
    maxX, maxY := 0, 0
    
    for y := 0; y < height; y++ {
        for x := 0; x < width; x++ {
            idx := y*width + x
            if idx < len(motionMap) && motionMap[idx] {
                if x < minX {
                    minX = x
                }
                if x > maxX {
                    maxX = x
                }
                if y < minY {
                    minY = y
                }
                if y > maxY {
                    maxY = y
                }
            }
        }
    }
    
    if maxX > minX && maxY > minY {
        // Add padding
        padding := 20
        roi := image.Rect(
            maxInt(0, minX-padding),
            maxInt(0, minY-padding),
            minInt(width, maxX+padding),
            minInt(height, maxY+padding),
        )
        rd.regions = []image.Rectangle{roi}
    }
}

func maxInt(a, b int) int {
    if a > b {
        return a
    }
    return b
}

func minInt(a, b int) int {
    if a < b {
        return a
    }
    return b
}

// Extract only the ROI region from frame
func (rd *ROIDetector) ExtractROI(frame *Frame) []*Frame {
    rd.mutex.RLock()
    defer rd.mutex.RUnlock()
    
    roiFrames := make([]*Frame, 0, len(rd.regions))
    
    for _, roi := range rd.regions {
        roiData := extractRegion(frame.Data, frame.Width, frame.Height, roi)
        roiFrame := &Frame{
            Data:      roiData,
            Width:     roi.Dx(),
            Height:    roi.Dy(),
            Timestamp: frame.Timestamp,
            Metadata: map[string]interface{}{
                "roi":        roi,
                "original_width":  frame.Width,
                "original_height": frame.Height,
            },
        }
        roiFrames = append(roiFrames, roiFrame)
    }
    
    return roiFrames
}

func extractRegion(data []byte, width, height int, roi image.Rectangle) []byte {
    roiWidth := roi.Dx()
    roiHeight := roi.Dy()

    // Bounds check once (ROI'nin frame sınırları içinde olduğundan emin ol)
    if roi.Min.X < 0 || roi.Min.Y < 0 || roi.Max.X > width || roi.Max.Y > height {
        return nil
    }

    roiData := make([]byte, roiWidth*roiHeight*3)

    for y := 0; y < roiHeight; y++ {
        srcY := roi.Min.Y + y
        srcOffset := (srcY*width + roi.Min.X) * 3
        dstOffset := (y * roiWidth) * 3

        // Copy whole row at once (faster, fewer bounds checks)
        copy(roiData[dstOffset:dstOffset+roiWidth*3],
            data[srcOffset:srcOffset+roiWidth*3])
    }

    return roiData
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

package main

import (
    "context"
    "sync"
)

type MotionDetector struct {
    threshold     float64
    previousFrame []byte
    width, height int
    mutex         sync.RWMutex // thread-safety for previousFrame
}

func NewMotionDetector(threshold float64, width, height int) *MotionDetector {
    return &MotionDetector{
        threshold: threshold,
        width:     width,
        height:    height,
    }
}

func (md *MotionDetector) Process(ctx context.Context, in <-chan *Frame, out chan<- *Frame) {
    for frame := range in {
        select {
        case <-ctx.Done():
            return
        default:
            hasMotion := md.detectMotion(frame)
            frame.Metadata["motion_detected"] = hasMotion
            
            if hasMotion {
                frame.Metadata["motion_score"] = md.calculateMotionScore(frame)
            }
            
            select {
            case out <- frame:
            case <-ctx.Done():
                return
            }
        }
    }
}

func (md *MotionDetector) Name() string {
    return "MotionDetector"
}

func (md *MotionDetector) detectMotion(frame *Frame) bool {
    md.mutex.Lock()
    defer md.mutex.Unlock()
    
    if md.previousFrame == nil {
        // First frame, save and no motion
        md.previousFrame = make([]byte, len(frame.Data))
        copy(md.previousFrame, frame.Data)
        return false
    }
    
    diff := md.calculateDifference(frame.Data, md.previousFrame)
    motionRatio := float64(diff) / float64(len(frame.Data))
    
    // Reuse previous frame buffer (reallocate only if size changed)
    if md.previousFrame == nil || len(md.previousFrame) != len(frame.Data) {
        md.previousFrame = make([]byte, len(frame.Data))
    }
    copy(md.previousFrame, frame.Data)
    
    return motionRatio > md.threshold
}

func (md *MotionDetector) calculateDifference(current, previous []byte) int {
    diff := 0
    for i := 0; i < len(current) && i < len(previous); i++ {
        if current[i] > previous[i] {
            diff += int(current[i] - previous[i])
        } else {
            diff += int(previous[i] - current[i])
        }
    }
    return diff
}

func (md *MotionDetector) calculateMotionScore(frame *Frame) float64 {
    md.mutex.RLock()
    defer md.mutex.RUnlock()
    
    if md.previousFrame == nil {
        return 0.0
    }
    
    diff := md.calculateDifference(frame.Data, md.previousFrame)
    maxDiff := len(frame.Data) * 255
    return float64(diff) / float64(maxDiff)
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117

package main

import (
    "context"
    "fmt"
    "image"
    "os"
    "strings"
    "gocv.io/x/gocv"
)

type ObjectDetector struct {
    net    gocv.Net
    classes []string
}

func NewObjectDetector(modelPath, configPath, classesPath string) (*ObjectDetector, error) {
    net := gocv.ReadNet(modelPath, configPath)
    if net.Empty() {
        return nil, fmt.Errorf("failed to load model")
    }
    
    // Use GPU if available (optional)
    net.SetPreferableBackend(gocv.NetBackendOpenVINO)
    net.SetPreferableTarget(gocv.NetTargetCPU) // or NetTargetVPU
    
    // Read classes file
    classes, err := loadClasses(classesPath)
    if err != nil {
        return nil, fmt.Errorf("failed to load classes: %w", err)
    }
    
    return &ObjectDetector{
        net:     net,
        classes: classes,
    }, nil
}

func (od *ObjectDetector) Process(ctx context.Context, in <-chan *Frame, out chan<- *Frame) {
    for frame := range in {
        select {
        case <-ctx.Done():
            return
        default:
            detections := od.detect(frame)
            frame.Metadata["detections"] = detections
            
            select {
            case out <- frame:
            case <-ctx.Done():
                return
            }
        }
    }
}

func (od *ObjectDetector) Name() string {
    return "ObjectDetector"
}

func (od *ObjectDetector) detect(frame *Frame) []Detection {
    // Convert frame data to gocv.Mat
    img, err := gocv.NewMatFromBytes(
        frame.Height,
        frame.Width,
        gocv.MatTypeCV8UC3,
        frame.Data,
    )
    if err != nil {
        return nil
    }
    defer img.Close()
    
    // Create blob (model input format)
    blob := gocv.BlobFromImage(img, 1.0/255.0, image.Pt(416, 416), gocv.NewScalar(0, 0, 0, 0), true, false)
    defer blob.Close()
    
    // Inference
    od.net.SetInput(blob, "")
    prob := od.net.Forward("")
    defer prob.Close()
    
    // Parse results
    detections := parseDetections(prob, frame.Width, frame.Height)
    
    return detections
}

type Detection struct {
    Class      string
    Confidence float64
    BBox       image.Rectangle
}

func parseDetections(prob gocv.Mat, imgWidth, imgHeight int) []Detection {
    // Parse according to model output
    // This example is simplified, real implementation depends on model
    return []Detection{}
}

func loadClasses(path string) ([]string, error) {
    data, err := os.ReadFile(path)
    if err != nil {
        return nil, fmt.Errorf("failed to read classes file: %w", err)
    }
    
    lines := strings.Split(string(data), "\n")
    classes := make([]string, 0, len(lines))
    for _, line := range lines {
        line = strings.TrimSpace(line)
        if line != "" {
            classes = append(classes, line)
        }
    }
    
    return classes, nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76

package main

import (
    "bytes"
    "context"
    "encoding/base64"
    "encoding/json"
    "fmt"
    "net/http"
    "time"
)

type ExternalDetector struct {
    client  *http.Client
    apiURL  string
    timeout time.Duration
}

func (ed *ExternalDetector) Process(ctx context.Context, in <-chan *Frame, out chan<- *Frame) {
    for frame := range in {
        select {
        case <-ctx.Done():
            return
        default:
            detections, err := ed.callInferenceService(frame)
            if err == nil {
                frame.Metadata["detections"] = detections
            } else {
                frame.Metadata["detection_error"] = err.Error()
            }
            
            select {
            case out <- frame:
            case <-ctx.Done():
                return
            }
        }
    }
}

func (ed *ExternalDetector) callInferenceService(frame *Frame) ([]Detection, error) {
    // Base64 encode frame or send binary
    reqBody, err := json.Marshal(map[string]interface{}{
        "image": base64.StdEncoding.EncodeToString(frame.Data),
        "width": frame.Width,
        "height": frame.Height,
    })
    if err != nil {
        return nil, err
    }
    
    ctx, cancel := context.WithTimeout(context.Background(), ed.timeout)
    defer cancel()
    
    req, err := http.NewRequestWithContext(ctx, "POST", ed.apiURL, bytes.NewReader(reqBody))
    if err != nil {
        return nil, err
    }
    req.Header.Set("Content-Type", "application/json")
    
    resp, err := ed.client.Do(req)
    if err != nil {
        return nil, err
    }
    defer resp.Body.Close()
    
    var result struct {
        Detections []Detection `json:"detections"`
    }
    
    if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
        return nil, err
    }
    
    return result.Detections, nil
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150

package main

import (
    "context"
    "fmt"
    "sync"
    "time"
)

type EdgeDecisionLayer struct {
    minConfidence  float64
    eventPublisher EventPublisher
    lastEvents     map[string]time.Time
    mutex          sync.RWMutex
    cooldown       time.Duration
}

type EventPublisher interface {
    PublishWithContext(ctx context.Context, event *Event) error
}

type Event struct {
    Type       string                 `json:"type"`
    Timestamp  time.Time              `json:"timestamp"`
    Confidence float64                `json:"confidence"`
    Detections []Detection            `json:"detections,omitempty"`
    Metadata   map[string]interface{} `json:"metadata"`
}

func NewEdgeDecisionLayer(minConfidence float64, publisher EventPublisher, cooldown time.Duration) *EdgeDecisionLayer {
    return &EdgeDecisionLayer{
        minConfidence:  minConfidence,
        eventPublisher: publisher,
        lastEvents:     make(map[string]time.Time),
        cooldown:       cooldown,
    }
}

func (edl *EdgeDecisionLayer) Process(ctx context.Context, in <-chan *Frame, out chan<- *Frame) {
    for frame := range in {
        select {
        case <-ctx.Done():
            return
        default:
            edl.processFrame(frame)
            
            // Continue frame in pipeline
            select {
            case out <- frame:
            case <-ctx.Done():
                return
            }
        }
    }
}

func (edl *EdgeDecisionLayer) Name() string {
    return "EdgeDecisionLayer"
}

func (edl *EdgeDecisionLayer) processFrame(frame *Frame) {
    // Check if detections exist
    detections, ok := frame.Metadata["detections"].([]Detection)
    if !ok {
        return
    }
    
    // Filter high confidence detections
    significantDetections := edl.filterByConfidence(detections)
    if len(significantDetections) == 0 {
        return
    }
    
    // Determine event type
    eventType := edl.determineEventType(significantDetections)
    
    // Cooldown check (don't send same event too frequently)
    if !edl.shouldSendEvent(eventType) {
        return
    }
    
    // Create event
    event := &Event{
        Type:       eventType,
        Timestamp:  frame.Timestamp,
        Confidence: edl.calculateMaxConfidence(significantDetections),
        Detections: significantDetections,
        Metadata: map[string]interface{}{
            "frame_width":  frame.Width,
            "frame_height": frame.Height,
        },
    }

    // Publish asynchronously with timeout to avoid blocking the pipeline
    go func(evt *Event, evtType string) {
        ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
        defer cancel()

        if err := edl.eventPublisher.PublishWithContext(ctx, evt); err != nil {
            fmt.Printf("Failed to publish event: %v\n", err)
            return
        }

        // Update cooldown on successful publish
        edl.mutex.Lock()
        edl.lastEvents[evtType] = time.Now()
        edl.mutex.Unlock()
    }(event, eventType)
}

func (edl *EdgeDecisionLayer) filterByConfidence(detections []Detection) []Detection {
    filtered := make([]Detection, 0)
    for _, det := range detections {
        if det.Confidence >= edl.minConfidence {
            filtered = append(filtered, det)
        }
    }
    return filtered
}

func (edl *EdgeDecisionLayer) determineEventType(detections []Detection) string {
    // Simple logic: use first detection's class
    // Real application can have more complex logic
    if len(detections) > 0 {
        return detections[0].Class + "_detected"
    }
    return "unknown"
}

func (edl *EdgeDecisionLayer) shouldSendEvent(eventType string) bool {
    edl.mutex.RLock()
    lastTime, exists := edl.lastEvents[eventType]
    edl.mutex.RUnlock()
    
    if !exists {
        return true
    }
    
    return time.Since(lastTime) > edl.cooldown
}

func (edl *EdgeDecisionLayer) calculateMaxConfidence(detections []Detection) float64 {
    max := 0.0
    for _, det := range detections {
        if det.Confidence > max {
            max = det.Confidence
        }
    }
    return max
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

package main

import (
    "encoding/json"
    mqtt "github.com/eclipse/paho.mqtt.golang"
)

type MQTTPublisher struct {
    client mqtt.Client
    topic  string
}

func NewMQTTPublisher(brokerURL, topic, clientID string) (*MQTTPublisher, error) {
    opts := mqtt.NewClientOptions()
    opts.AddBroker(brokerURL)
    opts.SetClientID(clientID)
    
    client := mqtt.NewClient(opts)
    if token := client.Connect(); token.Wait() && token.Error() != nil {
        return nil, token.Error()
    }
    
    return &MQTTPublisher{
        client: client,
        topic:  topic,
    }, nil
}

func (mp *MQTTPublisher) PublishWithContext(ctx context.Context, event *Event) error {
    data, err := json.Marshal(event)
    if err != nil {
        return err
    }
    
    token := mp.client.Publish(mp.topic, 1, false, data)

    done := make(chan struct{})
    go func() {
        token.Wait()
        close(done)
    }()

    select {
    case <-ctx.Done():
        return ctx.Err()
    case <-done:
        return token.Error()
    }
}

func (mp *MQTTPublisher) Close() error {
    if mp.client != nil && mp.client.IsConnected() {
        // 250ms timeout for pending operations
        mp.client.Disconnect(250)
    }
    return nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

package main

import (
    "bytes"
    "encoding/json"
    "fmt"
    "net/http"
)

type WebhookPublisher struct {
    client  *http.Client
    webhookURL string
}

func (wp *WebhookPublisher) PublishWithContext(ctx context.Context, event *Event) error {
    data, err := json.Marshal(event)
    if err != nil {
        return err
    }
    
    req, err := http.NewRequestWithContext(ctx, "POST", wp.webhookURL, bytes.NewReader(data))
    if err != nil {
        return err
    }
    req.Header.Set("Content-Type", "application/json")
    
    resp, err := wp.client.Do(req)
    if err != nil {
        return err
    }
    defer resp.Body.Close()
    
    if resp.StatusCode >= 400 {
        return fmt.Errorf("webhook returned status %d", resp.StatusCode)
    }
    
    return nil
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107

import (
    "context"
    "fmt"
    "log"
    "sync"
)

type MultiCameraManager struct {
    cameras   map[string]*CameraPipeline
    mutex     sync.RWMutex
    ctx       context.Context
    cancel    context.CancelFunc
}

type CameraPipeline struct {
    ID       string
    Source   *VideoSource
    Pipeline *Pipeline
    Config   CameraConfig
}

type CameraConfig struct {
    URL           string
    Width         int
    Height        int
    MotionEnabled bool
    ObjectEnabled bool
}

func NewMultiCameraManager() *MultiCameraManager {
    ctx, cancel := context.WithCancel(context.Background())
    return &MultiCameraManager{
        cameras: make(map[string]*CameraPipeline),
        ctx:     ctx,
        cancel:  cancel,
    }
}

func (mcm *MultiCameraManager) AddCamera(id string, config CameraConfig) error {
    mcm.mutex.Lock()
    defer mcm.mutex.Unlock()
    
    source := NewVideoSource(config.URL, config.Width, config.Height)
    pipeline := NewPipeline(source)
    
    // Add motion detector
    if config.MotionEnabled {
        motionDetector := NewMotionDetector(0.05, config.Width, config.Height)
        pipeline.AddProcessor(motionDetector)
    }
    
    // Add object detector
    if config.ObjectEnabled {
        objectDetector, err := NewObjectDetector("model.onnx", "config.yaml", "classes.txt")
        if err == nil {
            pipeline.AddProcessor(objectDetector)
        }
    }
    
    cameraPipeline := &CameraPipeline{
        ID:       id,
        Source:   source,
        Pipeline: pipeline,
        Config:   config,
    }
    
    mcm.cameras[id] = cameraPipeline
    
    // Start pipeline
    go func() {
        if err := pipeline.Run(mcm.ctx); err != nil {
            log.Printf("Camera %s pipeline error: %v", id, err)
        }
    }()
    
    return nil
}

func (mcm *MultiCameraManager) RemoveCamera(id string) error {
    mcm.mutex.Lock()
    defer mcm.mutex.Unlock()
    
    if camera, exists := mcm.cameras[id]; exists {
        camera.Pipeline.Stop()
        delete(mcm.cameras, id)
    }
    
    return nil
}

func (mcm *MultiCameraManager) GetCameraStatus(id string) (map[string]interface{}, error) {
    mcm.mutex.RLock()
    defer mcm.mutex.RUnlock()
    
    camera, exists := mcm.cameras[id]
    if !exists {
        return nil, fmt.Errorf("camera %s not found", id)
    }
    
    return map[string]interface{}{
        "id":       camera.ID,
        "url":      camera.Config.URL,
        "status":   "running",
        "width":    camera.Config.Width,
        "height":   camera.Config.Height,
    }, nil
}