1
2
3
4
5
6
7
8
9

# Go 1.21+ kurulu olmalı
go version

# FFmpeg kurulu olmalı
ffmpeg -version

# (Opsiyonel) OpenCV ve GoCV için
# 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

# Projeyi oluştur
mkdir video-processor && cd video-processor
go mod init video-processor

# Bağımlılıkları ekle
go get github.com/eclipse/paho.mqtt.golang
go get gopkg.in/yaml.v3
go get github.com/sirupsen/logrus

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

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

# OpenCV bağımlılıklarını kur
sudo apt-get update
sudo apt-get install -y \
    libopencv-dev \
    pkg-config \
    libavcodec-dev \
    libavformat-dev \
    libavutil-dev \
    libswscale-dev

# GoCV'yi kur
go get gocv.io/x/gocv

1
2
3
4
5

# Homebrew ile OpenCV kur
brew install opencv pkg-config

# GoCV'yi kur
go get gocv.io/x/gocv

1
2
3
4
5
6
7
8
9

# Chocolatey ile OpenCV kur (veya manuel)
choco install opencv

# Environment variable ayarla
set CGO_CPPFLAGS=-IC:\opencv\build\include
set CGO_LDFLAGS=-LC:\opencv\build\x64\vc15\lib

# GoCV'yi kur
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

# OpenCV bağımlılıklarını kur
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 (dosyadan)
    source := NewVideoSource("file://test.mp4", 640, 480)
    
    if err := source.Start(ctx); err != nil {
        log.Fatal(err)
    }
    
    // İlk 10 frame'i oku
    frameCount := 0
    for frame := range source.Frames() {
        log.Printf("Frame %d alındı: %dx%d", frameCount, frame.Width, frame.Height)
        frameCount++
        if frameCount >= 10 {
            break
        }
    }
    
    source.Stop()
    log.Println("Test tamamlandı!")
}

1
2
3
4
5
6

// FFmpeg'de format seçimi
// RGB24 için:
args := []string{"-pix_fmt", "rgb24"}

// YUV420 için (daha küçük):
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 {
    // FFmpeg komutunu oluştur
    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)
    }

    // FFmpeg stderr loglarını ayrı goroutine'de oku
    go func() {
        scanner := bufio.NewScanner(stderr)
        for scanner.Scan() {
            log.Printf("FFmpeg: %s", scanner.Text())
        }
    }()
    
    // Frame okuma goroutine'i
    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: her pixel 3 byte
    
    frameBuffer := make([]byte, frameSize)
    
    for {
        select {
        case <-ctx.Done():
            // Context iptal olduğunda FFmpeg process'ini de sonlandır
            if vs.cmd != nil && vs.cmd.Process != nil {
                _ = vs.cmd.Process.Kill()
            }
            return
        default:
            // Bir frame'in tamamını oku
            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 // Eksik frame, atla
            }
            
            // Yeni frame buffer'ı kopyala (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 {
    // Video kaynağını başlat
    if err := p.source.Start(ctx); err != nil {
        return err
    }
    
    // Channel'ları oluştur
    channels := make([]chan *Frame, len(p.processors)+1)
    for i := range channels {
        channels[i] = make(chan *Frame, 10)
    }
    
    // Source'dan ilk channel'a kopyala
    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
            }
        }
    }()
    
    // Her processor'ı çalıştır
    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)
    }
    
    // Hata kontrolü
    go func() {
        for err := range p.source.Errors() {
            // Log error (örnek)
            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{} // Analiz sonuçları için
}

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),
    }
}

// Motion detection sonuçlarına göre ROI belirle
func (rd *ROIDetector) UpdateROI(motionMap []bool, width, height int) {
    rd.mutex.Lock()
    defer rd.mutex.Unlock()
    
    // Motion olan bölgeleri tespit et ve bounding box oluştur
    // Basit implementasyon: motion olan pixel'lerin bounding box'ı
    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 {
        // Padding ekle
        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
}

// Frame'in sadece ROI bölgesini çıkar
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()

    // ROI'nin frame sınırları içinde olduğundan emin ol (bounds check)
    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

        // Her satırı tek seferde kopyala (daha hızlı, daha az bounds check)
        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 // previousFrame için thread-safety
}

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 {
        // İlk frame, sakla ve hareket yok
        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))
    
    // Önceki frame buffer'ını yeniden kullan (gerekiyorsa yeniden allocate et)
    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")
    }
    
    // GPU varsa kullan (opsiyonel)
    net.SetPreferableBackend(gocv.NetBackendOpenVINO)
    net.SetPreferableTarget(gocv.NetTargetCPU) // veya NetTargetVPU
    
    // Classes dosyasını oku
    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 {
    // Frame data'yı gocv.Mat'e dönüştür
    img, err := gocv.NewMatFromBytes(
        frame.Height,
        frame.Width,
        gocv.MatTypeCV8UC3,
        frame.Data,
    )
    if err != nil {
        return nil
    }
    defer img.Close()
    
    // Blob oluştur (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()
    
    // Sonuçları parse et
    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 {
    // Model output'una göre parse et
    // Bu örnek basitleştirilmiş, gerçek implementasyon model'e bağlı
    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) {
    // Frame'i base64 encode et veya binary gönder
    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)
            
            // Frame'i pipeline'a devam ettir
            select {
            case out <- frame:
            case <-ctx.Done():
                return
            }
        }
    }
}

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

func (edl *EdgeDecisionLayer) processFrame(frame *Frame) {
    // Detections varsa kontrol et
    detections, ok := frame.Metadata["detections"].([]Detection)
    if !ok {
        return
    }
    
    // Yüksek confidence'lı detections'ları filtrele
    significantDetections := edl.filterByConfidence(detections)
    if len(significantDetections) == 0 {
        return
    }
    
    // Event tipini belirle
    eventType := edl.determineEventType(significantDetections)
    
    // Cooldown kontrolü (aynı event'i çok sık gönderme)
    if !edl.shouldSendEvent(eventType) {
        return
    }
    
    // Event oluştur
    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,
        },
    }
    
    // Pipeline'ı bloklamamak için event'i async ve timeout'lu publish et
    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
        }

        // Başarılı publish sonrası cooldown güncelle
        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 {
    // Basit logic: ilk detection'ın class'ını kullan
    // Gerçek uygulamada daha karmaşık logic olabilir
    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() {
        // Bekleyen operasyonlar için 250ms timeout ile disconnect
        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)
    
    // Motion detector ekle
    if config.MotionEnabled {
        motionDetector := NewMotionDetector(0.05, config.Width, config.Height)
        pipeline.AddProcessor(motionDetector)
    }
    
    // Object detector ekle
    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
    
    // Pipeline'ı başlat
    go func() {
        if err := pipeline.Run(mcm.ctx); err != nil {
            log.Printf("Camera %s pipeline hatası: %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
}