gilles/go2rtc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add motion detection feature with EMA-based P-frame size analysis

Browse Source 
- Implemented MotionDetector for detecting motion based on H.264 P-frame sizes.
- Introduced adjustable sensitivity threshold for motion detection.
- Added tests for various scenarios including motion detection, hold time, cooldown, and baseline adaptation.
- Created hksvSession to manage HDS DataStream connections for HKSV recording.
- Updated schema.json to include a new speaker option for 2-way audio support.
This commit is contained in:
Sergey Krashevich
2026-03-06 19:58:15 +03:00
parent 593dce6eb9
commit c567831c91
13 changed files with 2135 additions and 1178 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/hksv/README.md
+636
View File
@@ -0,0 +1,636 @@
# hksv - HomeKit Secure Video Library for Go
`hksv` is a standalone Go library that implements HomeKit Secure Video (HKSV) recording, motion detection, and HAP (HomeKit Accessory Protocol) camera server functionality. It can be used independently of go2rtc in any Go project that needs HKSV support.
## Author
Sergei "svk" Krashevich <svk@svk.su>
## Features
- **HKSV Recording** - Fragmented MP4 (fMP4) muxing with GOP-based buffering, sent over HDS (HomeKit DataStream)
- **Motion Detection** - P-frame size analysis using EMA (Exponential Moving Average) baseline with configurable threshold
- **HAP Server** - Full HomeKit pairing (SRP), encrypted communication, accessory management
- **Proxy Mode** - Transparent proxy for existing HomeKit cameras
- **Live Streaming** - Pluggable interface for RTP/SRTP live view (bring your own implementation)
- **Zero internal dependencies** - Only depends on `pkg/` packages, never on `internal/`
## Architecture
```
pkg/hksv/
hksv.go - Server, Config, interfaces (StreamProvider, PairingStore, etc.)
consumer.go - HKSVConsumer: fMP4 muxer + GOP buffer + HDS sender
session.go - hksvSession: HDS DataStream lifecycle management
motion.go - MotionDetector: P-frame based motion detection
helpers.go - Helper functions for ID/name generation
consumer_test.go - Consumer tests and benchmarks
motion_test.go - Motion detector tests and benchmarks
```
### Dependency Graph
```
pkg/hksv/
-> pkg/core (Consumer, Connection, Media, Codec, Receiver, Sender)
-> pkg/hap (Server, Conn, Accessory, Character)
-> pkg/hap/hds (Conn, Session - encrypted DataStream)
-> pkg/hap/camera (TLV8 structs, services, accessory factories)
-> pkg/hap/tlv8 (marshal/unmarshal)
-> pkg/homekit (ServerHandler, ProxyHandler, HandlerFunc)
-> pkg/mp4 (Muxer - fMP4)
-> pkg/h264 (IsKeyframe, RTPDepay, RepairAVCC)
-> pkg/aac (RTPDepay)
-> pkg/mdns (ServiceEntry for mDNS advertisement)
-> github.com/pion/rtp
-> github.com/rs/zerolog
-> ZERO imports from internal/
```
## Quick Start
### Minimal HKSV Camera
```go
package main
import (
"net/http"
"github.com/AlexxIT/go2rtc/pkg/hap"
"github.com/AlexxIT/go2rtc/pkg/hksv"
"github.com/AlexxIT/go2rtc/pkg/mdns"
"github.com/rs/zerolog"
"github.com/rs/zerolog/log"
)
func main() {
logger := log.Output(zerolog.ConsoleWriter{Out: os.Stderr})
srv, err := hksv.NewServer(hksv.Config{
StreamName: "my-camera",
Pin: "27041991",
HKSV: true,
MotionMode: "detect",
Streams: &myStreamProvider{},
Store: &myPairingStore{},
Snapshots: &mySnapshotProvider{},
Logger: logger,
Port: 8080,
})
if err != nil {
logger.Fatal().Err(err).Msg("failed to create server")
}
// Register HAP endpoints
http.HandleFunc(hap.PathPairSetup, func(w http.ResponseWriter, r *http.Request) {
srv.Handle(w, r)
})
http.HandleFunc(hap.PathPairVerify, func(w http.ResponseWriter, r *http.Request) {
srv.Handle(w, r)
})
// Advertise via mDNS
entry := srv.MDNSEntry()
go mdns.Serve(mdns.ServiceHAP, []*mdns.ServiceEntry{entry})
// Start HTTP server
logger.Info().Msg("HomeKit camera running on :8080")
http.ListenAndServe(":8080", nil)
}
```
### HKSV Camera with Live Streaming
```go
srv, err := hksv.NewServer(hksv.Config{
StreamName: "my-camera",
Pin: "27041991",
HKSV: true,
MotionMode: "detect",
// Required interfaces
Streams: &myStreamProvider{},
Store: &myPairingStore{},
Snapshots: &mySnapshotProvider{},
LiveStream: &myLiveStreamHandler{}, // enables live view in Home app
Logger: logger,
Port: 8080,
})
```
### Basic Camera (no HKSV, live streaming only)
```go
srv, err := hksv.NewServer(hksv.Config{
StreamName: "basic-cam",
Pin: "27041991",
HKSV: false, // no HKSV recording
Streams: &myStreamProvider{},
LiveStream: &myLiveStreamHandler{},
Logger: logger,
Port: 8080,
})
```
### Proxy Mode (transparent proxy for existing HomeKit camera)
```go
srv, err := hksv.NewServer(hksv.Config{
StreamName: "proxied-cam",
Pin: "27041991",
ProxyURL: "homekit://192.168.1.100:51827?device_id=AA:BB:CC:DD:EE:FF&...",
Logger: logger,
Port: 8080,
})
```
### HomeKit Doorbell
```go
srv, err := hksv.NewServer(hksv.Config{
StreamName: "my-doorbell",
Pin: "27041991",
CategoryID: "doorbell", // creates doorbell accessory
HKSV: true,
MotionMode: "detect",
Streams: &myStreamProvider{},
Store: &myPairingStore{},
Snapshots: &mySnapshotProvider{},
Logger: logger,
Port: 8080,
})
// Trigger doorbell press from external event
srv.TriggerDoorbell()
```
## Interfaces
The library uses dependency injection via four interfaces. You implement these to connect `hksv` to your own stream management, storage, and media pipeline.
### StreamProvider (required)
Connects HKSV consumers to your video/audio streams.
```go
type StreamProvider interface {
// AddConsumer connects a consumer to the named stream.
// The consumer implements core.Consumer (AddTrack, WriteTo, Stop).
AddConsumer(streamName string, consumer core.Consumer) error
// RemoveConsumer disconnects a consumer from the named stream.
RemoveConsumer(streamName string, consumer core.Consumer)
}
```
**Example implementation:**
```go
type myStreamProvider struct {
streams map[string]*Stream // your stream registry
}
func (p *myStreamProvider) AddConsumer(name string, cons core.Consumer) error {
stream, ok := p.streams[name]
if !ok {
return fmt.Errorf("stream not found: %s", name)
}
return stream.AddConsumer(cons)
}
func (p *myStreamProvider) RemoveConsumer(name string, cons core.Consumer) {
if stream, ok := p.streams[name]; ok {
stream.RemoveConsumer(cons)
}
}
```
### PairingStore (optional)
Persists HomeKit pairing data across restarts. If `nil`, pairings are lost on restart and the device must be re-paired.
```go
type PairingStore interface {
SavePairings(streamName string, pairings []string) error
}
```
**Example implementation (JSON file):**
```go
type filePairingStore struct {
path string
}
func (s *filePairingStore) SavePairings(name string, pairings []string) error {
data := map[string][]string{name: pairings}
b, err := json.Marshal(data)
if err != nil {
return err
}
return os.WriteFile(s.path, b, 0644)
}
```
### SnapshotProvider (optional)
Generates JPEG snapshots for HomeKit `/resource` requests (shown in the Home app timeline and notifications). If `nil`, snapshots are not available.
```go
type SnapshotProvider interface {
GetSnapshot(streamName string, width, height int) ([]byte, error)
}
```
**Example implementation (ffmpeg):**
```go
type ffmpegSnapshotProvider struct {
streams map[string]*Stream
}
func (p *ffmpegSnapshotProvider) GetSnapshot(name string, w, h int) ([]byte, error) {
stream := p.streams[name]
if stream == nil {
return nil, errors.New("stream not found")
}
// Capture one keyframe from the stream
frame, err := stream.CaptureKeyframe()
if err != nil {
return nil, err
}
// Convert to JPEG using ffmpeg
return ffmpegToJPEG(frame, w, h)
}
```
### LiveStreamHandler (optional)
Handles live-streaming requests from the Home app (RTP/SRTP setup). If `nil`, only HKSV recording is available (no live view).
```go
type LiveStreamHandler interface {
// SetupEndpoints handles a SetupEndpoints request (HAP characteristic 118).
// Creates the RTP/SRTP consumer, returns the response value.
SetupEndpoints(conn net.Conn, offer *camera.SetupEndpointsRequest) (any, error)
// GetEndpointsResponse returns the current endpoints response (for GET requests).
GetEndpointsResponse() any
// StartStream starts RTP streaming with the given configuration.
// The connTracker is used to register/unregister the live stream connection
// on the HKSV server (for connection tracking and MarshalJSON).
StartStream(streamName string, conf *camera.SelectedStreamConfiguration, connTracker ConnTracker) error
// StopStream stops a stream matching the given session ID.
StopStream(sessionID string, connTracker ConnTracker) error
}
type ConnTracker interface {
AddConn(v any)
DelConn(v any)
}
```
**Example implementation (SRTP-based):**
```go
type srtpLiveStreamHandler struct {
mu sync.Mutex
consumer *homekit.Consumer
srtp *srtp.Server
streams map[string]*Stream
}
func (h *srtpLiveStreamHandler) SetupEndpoints(conn net.Conn, offer *camera.SetupEndpointsRequest) (any, error) {
consumer := homekit.NewConsumer(conn, h.srtp)
consumer.SetOffer(offer)
h.mu.Lock()
h.consumer = consumer
h.mu.Unlock()
answer := consumer.GetAnswer()
v, err := tlv8.MarshalBase64(answer)
return v, err
}
func (h *srtpLiveStreamHandler) GetEndpointsResponse() any {
h.mu.Lock()
consumer := h.consumer
h.mu.Unlock()
if consumer == nil {
return nil
}
answer := consumer.GetAnswer()
v, _ := tlv8.MarshalBase64(answer)
return v
}
func (h *srtpLiveStreamHandler) StartStream(streamName string, conf *camera.SelectedStreamConfiguration, ct hksv.ConnTracker) error {
h.mu.Lock()
consumer := h.consumer
h.mu.Unlock()
if consumer == nil {
return errors.New("no consumer")
}
if !consumer.SetConfig(conf) {
return errors.New("wrong config")
}
ct.AddConn(consumer)
stream := h.streams[streamName]
if err := stream.AddConsumer(consumer); err != nil {
return err
}
go func() {
_, _ = consumer.WriteTo(nil) // blocks until stream ends
stream.RemoveConsumer(consumer)
ct.DelConn(consumer)
}()
return nil
}
func (h *srtpLiveStreamHandler) StopStream(sessionID string, ct hksv.ConnTracker) error {
h.mu.Lock()
consumer := h.consumer
h.mu.Unlock()
if consumer != nil && consumer.SessionID() == sessionID {
_ = consumer.Stop()
}
return nil
}
```
## Config Reference
```go
type Config struct {
// Required
StreamName string // stream identifier (used for lookups)
Pin string // HomeKit pairing PIN, e.g. "27041991" (default)
Port uint16 // HAP HTTP port
Logger zerolog.Logger // structured logger
Streams StreamProvider // stream registry (required for HKSV/live/motion)
// Optional - server identity
Name string // mDNS display name (auto-generated from DeviceID if empty)
DeviceID string // MAC-like ID, e.g. "AA:BB:CC:DD:EE:FF" (auto-generated if empty)
DevicePrivate string // ed25519 private key hex (auto-generated if empty)
CategoryID string // "camera" (default), "doorbell", "bridge", or numeric
Pairings []string // pre-existing pairings from storage
// Optional - mode
ProxyURL string // if set, acts as transparent proxy (no local accessory)
HKSV bool // enable HKSV recording support
// Optional - motion detection
MotionMode string // "api" (external trigger), "continuous" (always on), "detect" (P-frame analysis)
MotionThreshold float64 // ratio threshold for "detect" mode (default 2.0, lower = more sensitive)
// Optional - hardware
Speaker *bool // include Speaker service for 2-way audio (default false)
// Optional - metadata
UserAgent string // for mDNS TXTModel field
Version string // for accessory firmware version
// Optional - persistence and features
Store PairingStore // nil = pairings not persisted
Snapshots SnapshotProvider // nil = no snapshot support
LiveStream LiveStreamHandler // nil = no live streaming (HKSV recording only)
}
```
## Motion Detection
The library includes a built-in P-frame based motion detector that works without any external motion detection system.
### How It Works
1. During a **warmup phase** (30 P-frames), the detector establishes a baseline average frame size using fast EMA (alpha=0.1).
2. After warmup, each P-frame size is compared against the baseline multiplied by the threshold.
3. If `frame_size > baseline * threshold`, motion is detected.
4. Motion stays active for a **hold period** (30 seconds) after the last trigger frame.
5. After motion ends, there is a **cooldown period** (5 seconds) before new motion can be detected.
6. The baseline is updated continuously with slow EMA (alpha=0.02) during idle periods.
7. FPS is recalibrated every 150 frames for accurate hold/cooldown timing.
### Motion Modes
| Mode | Description |
|------|-------------|
| `"api"` | Motion is triggered externally via `srv.SetMotionDetected(true/false)` |
| `"detect"` | Automatic P-frame analysis (starts on first Home Hub connection) |
| `"continuous"` | Always reports motion every 30 seconds (for testing/always-record) |
### Using the MotionDetector Standalone
The `MotionDetector` can be used independently as a `core.Consumer`:
```go
onMotion := func(detected bool) {
if detected {
log.Println("Motion started!")
// start recording, send notification, etc.
} else {
log.Println("Motion ended")
}
}
detector := hksv.NewMotionDetector(2.0, onMotion, logger)
// Attach to a stream (detector implements core.Consumer)
err := stream.AddConsumer(detector)
// Blocks until Stop() is called
go func() {
detector.WriteTo(nil)
}()
// Later, stop the detector
detector.Stop()
```
## Server API
### Motion Control
```go
// Trigger motion detected (for "api" mode or external sensors)
srv.SetMotionDetected(true)
// Clear motion
srv.SetMotionDetected(false)
// Trigger doorbell press event
srv.TriggerDoorbell()
```
### Connection Tracking
```go
// Register a connection (for monitoring/JSON output)
srv.AddConn(conn)
// Unregister a connection
srv.DelConn(conn)
```
### Pairing Management
```go
// Add a new pairing (called automatically during HAP pair-setup)
srv.AddPair(clientID, publicKey, hap.PermissionAdmin)
// Remove a pairing
srv.DelPair(clientID)
// Get client's public key (used by HAP pair-verify)
pubKey := srv.GetPair(clientID)
```
### JSON Serialization
The server implements `json.Marshaler` for status reporting:
```go
b, _ := json.Marshal(srv)
// {"name":"go2rtc-A1B2","device_id":"AA:BB:CC:DD:EE:FF","paired":1,"category_id":"17","connections":[...]}
// If not paired, includes setup_code and setup_id for QR code generation
// {"name":"go2rtc-A1B2","device_id":"AA:BB:CC:DD:EE:FF","setup_code":"195-50-224","setup_id":"A1B2"}
```
### mDNS Advertisement
```go
entry := srv.MDNSEntry()
// Start mDNS advertisement
go mdns.Serve(mdns.ServiceHAP, []*mdns.ServiceEntry{entry})
```
## Helper Functions
For deterministic ID generation from stream names:
```go
// Generate a display name from a seed
name := hksv.CalcName("", "my-camera")
// => "go2rtc-A1B2" (deterministic from seed)
name = hksv.CalcName("My Camera", "")
// => "My Camera" (uses provided name)
// Generate a MAC-like device ID
deviceID := hksv.CalcDeviceID("", "my-camera")
// => "AA:BB:CC:DD:EE:FF" (deterministic from seed)
// Generate an ed25519 private key
privateKey := hksv.CalcDevicePrivate("", "my-camera")
// => []byte{...} (deterministic 64-byte ed25519 key)
// Generate a setup ID for QR codes
setupID := hksv.CalcSetupID("my-camera")
// => "A1B2"
// Convert category string to HAP constant
catID := hksv.CalcCategoryID("doorbell")
// => "18" (hap.CategoryDoorbell)
```
## Multiple Cameras
You can run multiple HKSV cameras on a single port. Each camera gets its own mDNS entry and is resolved by hostname:
```go
cameras := []string{"front-door", "backyard", "garage"}
var entries []*mdns.ServiceEntry
for _, name := range cameras {
srv, _ := hksv.NewServer(hksv.Config{
StreamName: name,
Pin: "27041991",
HKSV: true,
MotionMode: "detect",
Streams: provider,
Logger: logger,
Port: 8080,
})
entry := srv.MDNSEntry()
entries = append(entries, entry)
// Map hostname -> server for HTTP routing
host := entry.Host(mdns.ServiceHAP)
handlers[host] = srv
}
// Single HTTP server handles all cameras
http.HandleFunc(hap.PathPairSetup, func(w http.ResponseWriter, r *http.Request) {
if srv := handlers[r.Host]; srv != nil {
srv.Handle(w, r)
}
})
http.HandleFunc(hap.PathPairVerify, func(w http.ResponseWriter, r *http.Request) {
if srv := handlers[r.Host]; srv != nil {
srv.Handle(w, r)
}
})
go mdns.Serve(mdns.ServiceHAP, entries)
http.ListenAndServe(":8080", nil)
```
## HKSV Recording Flow
Understanding the recording flow helps with debugging:
```
1. Home Hub discovers camera via mDNS
2. Home Hub connects -> PairSetup (first time) or PairVerify (subsequent)
3. On PairVerify success:
- If motion="detect": MotionDetector starts consuming the video stream
- If motion="continuous": prepareHKSVConsumer() + startContinuousMotion()
4. Motion detected -> SetMotionDetected(true) -> HAP event notification
5. Home Hub receives motion event -> sets up HDS DataStream:
- SetCharacteristic(TypeSetupDataStreamTransport) -> TCP listener created
- Home Hub connects to TCP port -> encrypted HDS connection established
- hksvSession created
6. Home Hub opens dataSend stream:
- handleOpen() -> takes prepared consumer (or creates new one)
- consumer.Activate() -> sends fMP4 init segment over HDS
- H264 keyframes trigger GOP flush -> mediaFragment sent over HDS
7. Home Hub closes dataSend -> handleClose() -> consumer stopped
8. Motion timeout -> SetMotionDetected(false)
```
## Testing
```bash
# Run all tests
go test ./pkg/hksv/...
# Run with verbose output
go test -v ./pkg/hksv/...
# Run benchmarks
go test -bench=. ./pkg/hksv/...
# Run specific test
go test -v -run TestMotionDetector_BasicTrigger ./pkg/hksv/...
```
## Requirements
- Go 1.22+
- Dependencies: `github.com/pion/rtp`, `github.com/rs/zerolog` (plus go2rtc `pkg/` packages)
pkg/hksv/consumer.go
+257
View File
@@ -0,0 +1,257 @@
// Author: Sergei "svk" Krashevich <svk@svk.su>
package hksv
import (
"errors"
"io"
"sync"
"time"
"github.com/AlexxIT/go2rtc/pkg/aac"
"github.com/AlexxIT/go2rtc/pkg/core"
"github.com/AlexxIT/go2rtc/pkg/h264"
"github.com/AlexxIT/go2rtc/pkg/hap/hds"
"github.com/AlexxIT/go2rtc/pkg/mp4"
"github.com/pion/rtp"
"github.com/rs/zerolog"
)
// HKSVConsumer implements core.Consumer, generates fMP4 and sends over HDS.
// It can be pre-started without an HDS session, buffering init data until activated.
type HKSVConsumer struct {
core.Connection
muxer *mp4.Muxer
mu sync.Mutex
done chan struct{}
log zerolog.Logger
// Set by Activate() when HDS session is available
session *hds.Session
streamID int
seqNum int
active bool
start bool // waiting for first keyframe
// GOP buffer - accumulate moof+mdat pairs, flush on next keyframe
fragBuf []byte
// Pre-built init segment (built when tracks connect)
initData []byte
initErr error
initDone chan struct{} // closed when init is ready
}
// NewHKSVConsumer creates a new HKSV consumer that muxes H264+AAC into fMP4
// and sends fragments over an HDS DataStream session.
func NewHKSVConsumer(log zerolog.Logger) *HKSVConsumer {
medias := []*core.Media{
{
Kind: core.KindVideo,
Direction: core.DirectionSendonly,
Codecs: []*core.Codec{
{Name: core.CodecH264},
},
},
{
Kind: core.KindAudio,
Direction: core.DirectionSendonly,
Codecs: []*core.Codec{
{Name: core.CodecAAC},
},
},
}
return &HKSVConsumer{
Connection: core.Connection{
ID: core.NewID(),
FormatName: "hksv",
Protocol: "hds",
Medias: medias,
},
muxer: &mp4.Muxer{},
done: make(chan struct{}),
initDone: make(chan struct{}),
log: log,
}
}
func (c *HKSVConsumer) AddTrack(media *core.Media, _ *core.Codec, track *core.Receiver) error {
// Reject late tracks after init segment is built (can't modify fMP4 header)
select {
case <-c.initDone:
c.log.Debug().Str("codec", track.Codec.Name).Msg("[hksv] ignoring late track (init already built)")
return nil
default:
}
trackID := byte(len(c.Senders))
c.log.Debug().Str("codec", track.Codec.Name).Uint8("trackID", trackID).Msg("[hksv] AddTrack")
codec := track.Codec.Clone()
handler := core.NewSender(media, codec)
switch track.Codec.Name {
case core.CodecH264:
handler.Handler = func(packet *rtp.Packet) {
c.mu.Lock()
if !c.active {
c.mu.Unlock()
return
}
if !c.start {
if !h264.IsKeyframe(packet.Payload) {
c.mu.Unlock()
return
}
c.start = true
c.log.Debug().Int("payloadLen", len(packet.Payload)).Msg("[hksv] first keyframe")
} else if h264.IsKeyframe(packet.Payload) && len(c.fragBuf) > 0 {
// New keyframe = flush previous GOP as one mediaFragment
c.flushFragment()
}
b := c.muxer.GetPayload(trackID, packet)
c.fragBuf = append(c.fragBuf, b...)
c.mu.Unlock()
}
if track.Codec.IsRTP() {
handler.Handler = h264.RTPDepay(track.Codec, handler.Handler)
} else {
handler.Handler = h264.RepairAVCC(track.Codec, handler.Handler)
}
case core.CodecAAC:
handler.Handler = func(packet *rtp.Packet) {
c.mu.Lock()
if !c.active || !c.start {
c.mu.Unlock()
return
}
b := c.muxer.GetPayload(trackID, packet)
c.fragBuf = append(c.fragBuf, b...)
c.mu.Unlock()
}
if track.Codec.IsRTP() {
handler.Handler = aac.RTPDepay(handler.Handler)
}
default:
return nil // skip unsupported codecs
}
c.muxer.AddTrack(codec)
handler.HandleRTP(track)
c.Senders = append(c.Senders, handler)
// Build init segment when all expected tracks are ready
select {
case <-c.initDone:
// already built — ignore late tracks (init is immutable)
default:
if len(c.Senders) >= len(c.Medias) {
c.buildInit()
}
}
return nil
}
// buildInit creates the init segment from currently connected tracks.
// Must only be called once (closes initDone).
func (c *HKSVConsumer) buildInit() {
initData, err := c.muxer.GetInit()
c.initData = initData
c.initErr = err
close(c.initDone)
if err != nil {
c.log.Error().Err(err).Msg("[hksv] GetInit failed")
} else {
c.log.Debug().Int("initSize", len(initData)).Int("tracks", len(c.Senders)).Msg("[hksv] init segment ready")
}
}
// Activate is called when the HDS session is ready (dataSend.open).
// It sends the pre-built init segment and starts streaming.
func (c *HKSVConsumer) Activate(session *hds.Session, streamID int) error {
// Wait for init to be ready (should already be done if consumer was pre-started)
select {
case <-c.initDone:
case <-time.After(5 * time.Second):
// Build init with whatever tracks we have (audio may be missing)
select {
case <-c.initDone:
default:
if len(c.Senders) > 0 {
c.log.Warn().Int("tracks", len(c.Senders)).Msg("[hksv] init timeout, building with available tracks")
c.buildInit()
} else {
return errors.New("hksv: no tracks connected after timeout")
}
}
}
if c.initErr != nil {
return c.initErr
}
c.log.Debug().Int("initSize", len(c.initData)).Msg("[hksv] sending init segment")
if err := session.SendMediaInit(streamID, c.initData); err != nil {
return err
}
c.log.Debug().Msg("[hksv] init segment sent OK")
// Enable live streaming (seqNum=2 because init used seqNum=1)
c.mu.Lock()
c.session = session
c.streamID = streamID
c.seqNum = 2
c.active = true
c.mu.Unlock()
return nil
}
// flushFragment sends the accumulated GOP buffer as a single mediaFragment.
// Must be called while holding c.mu.
func (c *HKSVConsumer) flushFragment() {
fragment := c.fragBuf
c.fragBuf = make([]byte, 0, len(fragment))
c.log.Debug().Int("fragSize", len(fragment)).Int("seq", c.seqNum).Msg("[hksv] flush fragment")
if err := c.session.SendMediaFragment(c.streamID, fragment, c.seqNum); err == nil {
c.Send += len(fragment)
}
c.seqNum++
}
func (c *HKSVConsumer) WriteTo(io.Writer) (int64, error) {
<-c.done
return 0, nil
}
func (c *HKSVConsumer) Stop() error {
select {
case <-c.done:
default:
close(c.done)
}
c.mu.Lock()
c.active = false
c.mu.Unlock()
return c.Connection.Stop()
}
// Done returns a channel that is closed when the consumer is stopped.
func (c *HKSVConsumer) Done() <-chan struct{} {
return c.done
}
func (c *HKSVConsumer) String() string {
return "hksv consumer"
}
pkg/hksv/consumer_test.go
+460
View File
@@ -0,0 +1,460 @@
// Author: Sergei "svk" Krashevich <svk@svk.su>
package hksv
import (
"net"
"sync"
"testing"
"time"
"github.com/AlexxIT/go2rtc/pkg/core"
"github.com/AlexxIT/go2rtc/pkg/hap/hds"
"github.com/rs/zerolog"
"github.com/stretchr/testify/require"
)
var testLog = zerolog.Nop()
// newTestSessionPair creates connected HDS sessions for testing.
func newTestSessionPair(t *testing.T) (accessory *hds.Session, controller *hds.Session) {
t.Helper()
key := []byte(core.RandString(16, 0))
salt := core.RandString(32, 0)
c1, c2 := net.Pipe()
t.Cleanup(func() { c1.Close(); c2.Close() })
accConn, err := hds.NewConn(c1, key, salt, false)
require.NoError(t, err)
ctrlConn, err := hds.NewConn(c2, key, salt, true)
require.NoError(t, err)
return hds.NewSession(accConn), hds.NewSession(ctrlConn)
}
func TestHKSVConsumer_Creation(t *testing.T) {
c := NewHKSVConsumer(testLog)
require.Equal(t, "hksv", c.FormatName)
require.Equal(t, "hds", c.Protocol)
require.Len(t, c.Medias, 2)
require.Equal(t, core.KindVideo, c.Medias[0].Kind)
require.Equal(t, core.KindAudio, c.Medias[1].Kind)
require.Equal(t, core.CodecH264, c.Medias[0].Codecs[0].Name)
require.Equal(t, core.CodecAAC, c.Medias[1].Codecs[0].Name)
require.NotNil(t, c.muxer)
require.NotNil(t, c.done)
require.NotNil(t, c.initDone)
require.False(t, c.active)
require.False(t, c.start)
require.Equal(t, 0, c.seqNum)
require.Nil(t, c.fragBuf)
require.Nil(t, c.initData)
}
func TestHKSVConsumer_FlushFragment_SendsAndIncrements(t *testing.T) {
acc, ctrl := newTestSessionPair(t)
c := NewHKSVConsumer(testLog)
// Manually set up the consumer as if Activate() was called
c.session = acc
c.streamID = 1
c.seqNum = 2
c.active = true
c.fragBuf = []byte("fake-fragment-data-here")
done := make(chan struct{})
go func() {
defer close(done)
msg, err := ctrl.ReadMessage()
require.NoError(t, err)
require.Equal(t, "dataSend", msg.Protocol)
require.Equal(t, "data", msg.Topic)
require.True(t, msg.IsEvent)
packets, ok := msg.Body["packets"].([]any)
require.True(t, ok)
pkt := packets[0].(map[string]any)
meta := pkt["metadata"].(map[string]any)
require.Equal(t, "mediaFragment", meta["dataType"])
require.Equal(t, int64(2), meta["dataSequenceNumber"].(int64))
require.Equal(t, true, meta["isLastDataChunk"])
}()
c.mu.Lock()
c.flushFragment()
c.mu.Unlock()
<-done
require.Equal(t, 3, c.seqNum, "seqNum should increment after flush")
require.Empty(t, c.fragBuf, "fragBuf should be empty after flush")
}
func TestHKSVConsumer_FlushFragment_MultipleFlushes(t *testing.T) {
acc, ctrl := newTestSessionPair(t)
c := NewHKSVConsumer(testLog)
c.session = acc
c.streamID = 1
c.seqNum = 2
c.active = true
var received []int64
var mu sync.Mutex
done := make(chan struct{})
go func() {
defer close(done)
for i := 0; i < 3; i++ {
msg, err := ctrl.ReadMessage()
if err != nil {
return
}
packets := msg.Body["packets"].([]any)
pkt := packets[0].(map[string]any)
meta := pkt["metadata"].(map[string]any)
mu.Lock()
received = append(received, meta["dataSequenceNumber"].(int64))
mu.Unlock()
}
}()
for i := 0; i < 3; i++ {
c.mu.Lock()
c.fragBuf = []byte("data")
c.flushFragment()
c.mu.Unlock()
}
<-done
mu.Lock()
defer mu.Unlock()
require.Equal(t, []int64{2, 3, 4}, received)
require.Equal(t, 5, c.seqNum)
}
func TestHKSVConsumer_FlushFragment_EmptyBuffer(t *testing.T) {
c := NewHKSVConsumer(testLog)
c.seqNum = 2
// flushFragment with empty/nil buffer should still increment seqNum
// but send empty data (protocol layer handles it)
// In practice, flushFragment is only called when fragBuf has data
c.mu.Lock()
c.fragBuf = nil
initialSeq := c.seqNum
c.mu.Unlock()
// No crash = pass (no session to write to, would panic on nil session)
require.Equal(t, initialSeq, c.seqNum)
}
func TestHKSVConsumer_BufferAccumulation(t *testing.T) {
c := NewHKSVConsumer(testLog)
c.active = true
data1 := []byte("chunk-1")
data2 := []byte("chunk-2")
data3 := []byte("chunk-3")
c.fragBuf = append(c.fragBuf, data1...)
c.fragBuf = append(c.fragBuf, data2...)
c.fragBuf = append(c.fragBuf, data3...)
require.Equal(t, len(data1)+len(data2)+len(data3), len(c.fragBuf))
require.Equal(t, "chunk-1chunk-2chunk-3", string(c.fragBuf))
}
func TestHKSVConsumer_ActivateSeqNum(t *testing.T) {
acc, ctrl := newTestSessionPair(t)
c := NewHKSVConsumer(testLog)
// Simulate init ready
c.initData = []byte("fake-init")
close(c.initDone)
done := make(chan struct{})
go func() {
defer close(done)
// Read the init message
msg, err := ctrl.ReadMessage()
require.NoError(t, err)
require.True(t, msg.IsEvent)
packets := msg.Body["packets"].([]any)
pkt := packets[0].(map[string]any)
meta := pkt["metadata"].(map[string]any)
require.Equal(t, "mediaInitialization", meta["dataType"])
require.Equal(t, int64(1), meta["dataSequenceNumber"].(int64))
}()
err := c.Activate(acc, 5)
require.NoError(t, err)
<-done
require.Equal(t, 2, c.seqNum, "seqNum should be 2 after activate (init uses 1)")
require.True(t, c.active)
require.Equal(t, 5, c.streamID)
require.Equal(t, acc, c.session)
}
func TestHKSVConsumer_ActivateTimeout(t *testing.T) {
acc, _ := newTestSessionPair(t)
c := NewHKSVConsumer(testLog)
// Don't close initDone — simulate init never becoming ready
// Override the timeout for faster test
err := func() error {
select {
case <-c.initDone:
case <-time.After(50 * time.Millisecond):
return errActivateTimeout
}
return nil
}()
require.Error(t, err)
_ = acc // prevent unused
}
var errActivateTimeout = func() error {
return &timeoutError{}
}()
type timeoutError struct{}
func (e *timeoutError) Error() string { return "activate timeout" }
func TestHKSVConsumer_ActivateWithError(t *testing.T) {
c := NewHKSVConsumer(testLog)
c.initErr = &timeoutError{}
close(c.initDone)
acc, _ := newTestSessionPair(t)
err := c.Activate(acc, 1)
require.Error(t, err)
require.False(t, c.active)
}
func TestHKSVConsumer_StopSafety(t *testing.T) {
c := NewHKSVConsumer(testLog)
c.active = true
// First stop
err := c.Stop()
require.NoError(t, err)
require.False(t, c.active)
// Second stop — should not panic
err = c.Stop()
require.NoError(t, err)
}
func TestHKSVConsumer_StopDeactivates(t *testing.T) {
c := NewHKSVConsumer(testLog)
c.active = true
c.start = true
_ = c.Stop()
require.False(t, c.active)
}
func TestHKSVConsumer_WriteToDone(t *testing.T) {
c := NewHKSVConsumer(testLog)
done := make(chan struct{})
go func() {
n, err := c.WriteTo(nil)
require.NoError(t, err)
require.Equal(t, int64(0), n)
close(done)
}()
// WriteTo should block until done channel is closed
select {
case <-done:
t.Fatal("WriteTo returned before Stop")
case <-time.After(50 * time.Millisecond):
}
_ = c.Stop()
select {
case <-done:
case <-time.After(time.Second):
t.Fatal("WriteTo did not return after Stop")
}
}
func TestHKSVConsumer_GOPFlushIntegration(t *testing.T) {
acc, ctrl := newTestSessionPair(t)
c := NewHKSVConsumer(testLog)
c.session = acc
c.streamID = 1
c.seqNum = 2
c.active = true
c.start = true // already started
// Simulate a sequence: buffer data, then flush
frag1 := []byte("keyframe-1-data-plus-p-frames")
frag2 := []byte("keyframe-2-data")
var received [][]byte
done := make(chan struct{})
go func() {
defer close(done)
for i := 0; i < 2; i++ {
msg, err := ctrl.ReadMessage()
if err != nil {
return
}
packets := msg.Body["packets"].([]any)
pkt := packets[0].(map[string]any)
data := pkt["data"].([]byte)
received = append(received, data)
}
}()
// First GOP
c.mu.Lock()
c.fragBuf = append(c.fragBuf, frag1...)
c.flushFragment()
c.mu.Unlock()
// Second GOP
c.mu.Lock()
c.fragBuf = append(c.fragBuf, frag2...)
c.flushFragment()
c.mu.Unlock()
<-done
require.Len(t, received, 2)
require.Equal(t, frag1, received[0])
require.Equal(t, frag2, received[1])
require.Equal(t, 4, c.seqNum) // 2 + 2 flushes
}
func TestHKSVConsumer_FlushClearsBuffer(t *testing.T) {
acc, ctrl := newTestSessionPair(t)
c := NewHKSVConsumer(testLog)
c.session = acc
c.streamID = 1
c.seqNum = 2
c.active = true
done := make(chan struct{})
go func() {
defer close(done)
// drain messages
for i := 0; i < 3; i++ {
ctrl.ReadMessage()
}
}()
for i := 0; i < 3; i++ {
c.mu.Lock()
c.fragBuf = append(c.fragBuf, []byte("frame-data")...)
prevLen := len(c.fragBuf)
c.flushFragment()
require.Empty(t, c.fragBuf, "fragBuf should be empty after flush")
require.Greater(t, prevLen, 0, "had data before flush")
c.mu.Unlock()
}
<-done
require.Equal(t, 5, c.seqNum, "3 flushes from seqNum=2 → 5")
}
func TestHKSVConsumer_SendTracking(t *testing.T) {
acc, ctrl := newTestSessionPair(t)
c := NewHKSVConsumer(testLog)
c.session = acc
c.streamID = 1
c.seqNum = 2
c.active = true
data := []byte("12345678") // 8 bytes
done := make(chan struct{})
go func() {
defer close(done)
ctrl.ReadMessage()
}()
c.mu.Lock()
c.fragBuf = append(c.fragBuf, data...)
c.flushFragment()
c.mu.Unlock()
<-done
require.Equal(t, 8, c.Send, "Send counter should track bytes sent")
}
// --- Benchmarks ---
func BenchmarkHKSVConsumer_FlushFragment(b *testing.B) {
key := []byte(core.RandString(16, 0))
salt := core.RandString(32, 0)
c1, c2 := net.Pipe()
defer c1.Close()
defer c2.Close()
accConn, _ := hds.NewConn(c1, key, salt, false)
ctrlConn, _ := hds.NewConn(c2, key, salt, true)
acc := hds.NewSession(accConn)
go func() {
buf := make([]byte, 512*1024) // must be > 256KB chunk size
for {
if _, err := ctrlConn.Read(buf); err != nil {
return
}
}
}()
c := NewHKSVConsumer(testLog)
c.session = acc
c.streamID = 1
c.seqNum = 2
c.active = true
gopData := make([]byte, 4*1024*1024) // 4MB GOP
b.SetBytes(int64(len(gopData)))
b.ResetTimer()
for i := 0; i < b.N; i++ {
c.mu.Lock()
c.fragBuf = append(c.fragBuf[:0], gopData...)
c.flushFragment()
c.mu.Unlock()
}
}
func BenchmarkHKSVConsumer_BufferAppend(b *testing.B) {
c := NewHKSVConsumer(testLog)
frame := make([]byte, 1500) // typical frame fragment
b.SetBytes(int64(len(frame)))
b.ResetTimer()
for i := 0; i < b.N; i++ {
c.fragBuf = append(c.fragBuf, frame...)
if len(c.fragBuf) > 5*1024*1024 {
c.fragBuf = c.fragBuf[:0]
}
}
}
func BenchmarkHKSVConsumer_CreateAndStop(b *testing.B) {
b.ResetTimer()
for i := 0; i < b.N; i++ {
c := NewHKSVConsumer(testLog)
_ = c.Stop()
}
}
pkg/hksv/helpers.go
+65
View File
@@ -0,0 +1,65 @@
// Author: Sergei "svk" Krashevich <svk@svk.su>
package hksv
import (
"crypto/ed25519"
"crypto/sha512"
"encoding/hex"
"fmt"
"github.com/AlexxIT/go2rtc/pkg/core"
"github.com/AlexxIT/go2rtc/pkg/hap"
)
// CalcName generates a HomeKit display name from a seed if name is empty.
func CalcName(name, seed string) string {
if name != "" {
return name
}
b := sha512.Sum512([]byte(seed))
return fmt.Sprintf("go2rtc-%02X%02X", b[0], b[2])
}
// CalcDeviceID generates a MAC-like device ID from a seed if deviceID is empty.
func CalcDeviceID(deviceID, seed string) string {
if deviceID != "" {
if len(deviceID) >= 17 {
return deviceID
}
seed = deviceID
}
b := sha512.Sum512([]byte(seed))
return fmt.Sprintf("%02X:%02X:%02X:%02X:%02X:%02X", b[32], b[34], b[36], b[38], b[40], b[42])
}
// CalcDevicePrivate generates an ed25519 private key from a seed if private is empty.
func CalcDevicePrivate(private, seed string) []byte {
if private != "" {
if b, _ := hex.DecodeString(private); len(b) == ed25519.PrivateKeySize {
return b
}
seed = private
}
b := sha512.Sum512([]byte(seed))
return ed25519.NewKeyFromSeed(b[:ed25519.SeedSize])
}
// CalcSetupID generates a setup ID from a seed.
func CalcSetupID(seed string) string {
b := sha512.Sum512([]byte(seed))
return fmt.Sprintf("%02X%02X", b[44], b[46])
}
// CalcCategoryID converts a category string to a HAP category constant.
func CalcCategoryID(categoryID string) string {
switch categoryID {
case "bridge":
return hap.CategoryBridge
case "doorbell":
return hap.CategoryDoorbell
}
if core.Atoi(categoryID) > 0 {
return categoryID
}
return hap.CategoryCamera
}
pkg/hksv/hksv.go
+800
View File
@@ -0,0 +1,800 @@
// Package hksv provides a reusable HomeKit Secure Video server library.
//
// It implements HKSV recording (fMP4 over HDS DataStream), motion detection,
// and integrates with the HAP protocol for HomeKit pairing and communication.
//
// Usage:
//
// srv, err := hksv.NewServer(hksv.Config{
// StreamName: "camera1",
// Pin: "27041991",
// HKSV: true,
// MotionMode: "detect",
// Streams: myStreamProvider,
// Logger: logger,
// Port: 8080,
// })
// // Register srv.Handle as HTTP handler for HAP paths
// // Advertise srv.MDNSEntry() via mDNS
//
// Author: Sergei "svk" Krashevich <svk@svk.su>
package hksv
import (
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"io"
"net"
"net/http"
"net/url"
"slices"
"strings"
"sync"
"time"
"github.com/AlexxIT/go2rtc/pkg/core"
"github.com/AlexxIT/go2rtc/pkg/hap"
"github.com/AlexxIT/go2rtc/pkg/hap/camera"
"github.com/AlexxIT/go2rtc/pkg/hap/hds"
"github.com/AlexxIT/go2rtc/pkg/hap/tlv8"
"github.com/AlexxIT/go2rtc/pkg/homekit"
"github.com/AlexxIT/go2rtc/pkg/mdns"
"github.com/rs/zerolog"
)
// StreamProvider provides access to media streams.
// The host application implements this to connect the HKSV library
// to its own stream management system.
type StreamProvider interface {
// AddConsumer connects a consumer to the named stream.
AddConsumer(streamName string, consumer core.Consumer) error
// RemoveConsumer disconnects a consumer from the named stream.
RemoveConsumer(streamName string, consumer core.Consumer)
}
// PairingStore persists HAP pairing data.
type PairingStore interface {
SavePairings(streamName string, pairings []string) error
}
// SnapshotProvider generates JPEG snapshots for HomeKit /resource requests.
type SnapshotProvider interface {
GetSnapshot(streamName string, width, height int) ([]byte, error)
}
// LiveStreamHandler handles live-streaming requests (SetupEndpoints, SelectedStreamConfiguration).
// Implementation is external because it depends on SRTP.
type LiveStreamHandler interface {
// SetupEndpoints handles a SetupEndpoints request (ch118).
// Returns the response to store as characteristic value.
SetupEndpoints(conn net.Conn, offer *camera.SetupEndpointsRequest) (any, error)
// GetEndpointsResponse returns the current endpoints response (for GET requests).
GetEndpointsResponse() any
// StartStream starts RTP streaming with the given configuration (ch117 command=start).
// The connTracker is used to register/unregister the live stream connection.
StartStream(streamName string, conf *camera.SelectedStreamConfiguration, connTracker ConnTracker) error
// StopStream stops a stream matching the given session ID.
StopStream(sessionID string, connTracker ConnTracker) error
}
// ConnTracker allows the live stream handler to track connections on the server.
type ConnTracker interface {
AddConn(v any)
DelConn(v any)
}
// Config for creating an HKSV server.
type Config struct {
StreamName string
Pin string // HomeKit pairing PIN (e.g., "27041991")
Name string // mDNS display name (auto-generated if empty)
DeviceID string // MAC-like device ID (auto-generated if empty)
DevicePrivate string // ed25519 private key hex (auto-generated if empty)
CategoryID string // "camera" or "doorbell"
Pairings []string // pre-existing pairings
ProxyURL string // if set, acts as transparent proxy (no local accessory)
HKSV bool
MotionMode string // "api", "continuous", "detect"
MotionThreshold float64 // ratio threshold for "detect" mode (default 2.0)
Speaker *bool // include Speaker service for 2-way audio (default false)
UserAgent string // for mDNS TXTModel field
Version string // for accessory firmware version
// Dependencies (injected by host)
Streams StreamProvider
Store PairingStore // optional, nil = no persistence
Snapshots SnapshotProvider // optional, nil = no snapshots
LiveStream LiveStreamHandler // optional, nil = HKSV only (no live streaming)
Logger zerolog.Logger
// Network
Port uint16 // HAP HTTP port
}
// Server is a complete HKSV camera server.
type Server struct {
hap *hap.Server
mdns *mdns.ServiceEntry
log zerolog.Logger
pairings []string
conns []any
mu sync.Mutex
accessory *hap.Accessory
setupID string
stream string // stream name
proxyURL string // transparent proxy URL
// Injected dependencies
streams StreamProvider
store PairingStore
snapshots SnapshotProvider
liveStream LiveStreamHandler
// HKSV fields
motionMode string
motionThreshold float64
motionDetector *MotionDetector
hksvSession *hksvSession
continuousMotion bool
preparedConsumer *HKSVConsumer
}
// NewServer creates a new HKSV server with the given configuration.
func NewServer(cfg Config) (*Server, error) {
if cfg.Pin == "" {
cfg.Pin = "27041991"
}
pin, err := hap.SanitizePin(cfg.Pin)
if err != nil {
return nil, fmt.Errorf("hksv: invalid pin: %w", err)
}
deviceID := CalcDeviceID(cfg.DeviceID, cfg.StreamName)
name := CalcName(cfg.Name, deviceID)
setupID := CalcSetupID(cfg.StreamName)
srv := &Server{
stream: cfg.StreamName,
pairings: cfg.Pairings,
setupID: setupID,
log: cfg.Logger,
streams: cfg.Streams,
store: cfg.Store,
snapshots: cfg.Snapshots,
liveStream: cfg.LiveStream,
motionMode: cfg.MotionMode,
motionThreshold: cfg.MotionThreshold,
}
srv.hap = &hap.Server{
Pin: pin,
DeviceID: deviceID,
DevicePrivate: CalcDevicePrivate(cfg.DevicePrivate, cfg.StreamName),
GetClientPublic: srv.GetPair,
}
categoryID := CalcCategoryID(cfg.CategoryID)
srv.mdns = &mdns.ServiceEntry{
Name: name,
Port: cfg.Port,
Info: map[string]string{
hap.TXTConfigNumber: "1",
hap.TXTFeatureFlags: "0",
hap.TXTDeviceID: deviceID,
hap.TXTModel: cfg.UserAgent,
hap.TXTProtoVersion: "1.1",
hap.TXTStateNumber: "1",
hap.TXTStatusFlags: hap.StatusNotPaired,
hap.TXTCategory: categoryID,
hap.TXTSetupHash: hap.SetupHash(setupID, deviceID),
},
}
srv.UpdateStatus()
if cfg.ProxyURL != "" {
// Proxy mode: no local accessory
srv.proxyURL = cfg.ProxyURL
} else if cfg.HKSV {
if srv.motionThreshold <= 0 {
srv.motionThreshold = defaultThreshold
}
srv.log.Debug().Str("stream", cfg.StreamName).Str("motion", cfg.MotionMode).
Float64("threshold", srv.motionThreshold).Msg("[hksv] HKSV mode")
if cfg.CategoryID == "doorbell" {
srv.accessory = camera.NewHKSVDoorbellAccessory("AlexxIT", "go2rtc", name, "-", cfg.Version)
} else {
srv.accessory = camera.NewHKSVAccessory("AlexxIT", "go2rtc", name, "-", cfg.Version)
}
} else {
srv.accessory = camera.NewAccessory("AlexxIT", "go2rtc", name, "-", cfg.Version)
}
// Remove Speaker service unless explicitly enabled (default: disabled)
if (cfg.Speaker == nil || !*cfg.Speaker) && srv.accessory != nil {
filtered := srv.accessory.Services[:0]
for _, svc := range srv.accessory.Services {
if svc.Type != "113" { // 113 = Speaker
filtered = append(filtered, svc)
}
}
srv.accessory.Services = filtered
srv.accessory.InitIID() // recalculate IIDs
}
return srv, nil
}
// MDNSEntry returns the mDNS service entry for advertisement.
func (s *Server) MDNSEntry() *mdns.ServiceEntry {
return s.mdns
}
// Accessory returns the HAP accessory.
func (s *Server) Accessory() *hap.Accessory {
return s.accessory
}
// StreamName returns the configured stream name.
func (s *Server) StreamName() string {
return s.stream
}
func (s *Server) MarshalJSON() ([]byte, error) {
v := struct {
Name string `json:"name"`
DeviceID string `json:"device_id"`
Paired int `json:"paired,omitempty"`
CategoryID string `json:"category_id,omitempty"`
SetupCode string `json:"setup_code,omitempty"`
SetupID string `json:"setup_id,omitempty"`
Conns []any `json:"connections,omitempty"`
}{
Name: s.mdns.Name,
DeviceID: s.mdns.Info[hap.TXTDeviceID],
CategoryID: s.mdns.Info[hap.TXTCategory],
Paired: len(s.pairings),
Conns: s.conns,
}
if v.Paired == 0 {
v.SetupCode = s.hap.Pin
v.SetupID = s.setupID
}
return json.Marshal(v)
}
// Handle processes an incoming HAP connection (called from your HTTP server).
func (s *Server) Handle(w http.ResponseWriter, r *http.Request) {
conn, rw, err := w.(http.Hijacker).Hijack()
if err != nil {
return
}
defer conn.Close()
// Fix reading from Body after Hijack.
r.Body = io.NopCloser(rw)
switch r.RequestURI {
case hap.PathPairSetup:
id, key, err := s.hap.PairSetup(r, rw)
if err != nil {
s.log.Error().Err(err).Caller().Send()
return
}
s.AddPair(id, key, hap.PermissionAdmin)
case hap.PathPairVerify:
id, key, err := s.hap.PairVerify(r, rw)
if err != nil {
s.log.Debug().Err(err).Caller().Send()
return
}
s.log.Debug().Str("stream", s.stream).Str("client_id", id).Msgf("[hksv] %s: new conn", conn.RemoteAddr())
controller, err := hap.NewConn(conn, rw, key, false)
if err != nil {
s.log.Error().Err(err).Caller().Send()
return
}
s.AddConn(controller)
defer s.DelConn(controller)
// start motion on first Home Hub connection
switch s.motionMode {
case "detect":
go s.startMotionDetector()
case "continuous":
go s.prepareHKSVConsumer()
go s.startContinuousMotion()
}
var handler homekit.HandlerFunc
switch {
case s.accessory != nil:
handler = homekit.ServerHandler(s)
case s.proxyURL != "":
client, err := hap.Dial(s.proxyURL)
if err != nil {
s.log.Error().Err(err).Caller().Send()
return
}
handler = homekit.ProxyHandler(s, client.Conn)
}
s.log.Debug().Str("stream", s.stream).Msgf("[hksv] handler started for %s", conn.RemoteAddr())
if err = handler(controller); err != nil {
if errors.Is(err, io.EOF) || isClosedConnErr(err) {
s.log.Debug().Str("stream", s.stream).Msgf("[hksv] %s: connection closed", conn.RemoteAddr())
} else {
s.log.Error().Err(err).Str("stream", s.stream).Caller().Send()
}
return
}
}
}
// AddConn registers a connection for tracking.
func (s *Server) AddConn(v any) {
s.log.Trace().Str("stream", s.stream).Msgf("[hksv] add conn %s", connLabel(v))
s.mu.Lock()
s.conns = append(s.conns, v)
s.mu.Unlock()
}
// DelConn unregisters a connection.
func (s *Server) DelConn(v any) {
s.log.Trace().Str("stream", s.stream).Msgf("[hksv] del conn %s", connLabel(v))
s.mu.Lock()
if i := slices.Index(s.conns, v); i >= 0 {
s.conns = slices.Delete(s.conns, i, i+1)
}
s.mu.Unlock()
}
// connLabel returns a short human-readable label for a connection.
func connLabel(v any) string {
switch v := v.(type) {
case *hap.Conn:
return "hap " + v.RemoteAddr().String()
case *hds.Conn:
return "hds " + v.RemoteAddr().String()
}
if s, ok := v.(fmt.Stringer); ok {
return s.String()
}
return fmt.Sprintf("%T", v)
}
func (s *Server) UpdateStatus() {
if len(s.pairings) == 0 {
s.mdns.Info[hap.TXTStatusFlags] = hap.StatusNotPaired
} else {
s.mdns.Info[hap.TXTStatusFlags] = hap.StatusPaired
}
}
func (s *Server) pairIndex(id string) int {
id = "client_id=" + id
for i, pairing := range s.pairings {
if strings.HasPrefix(pairing, id) {
return i
}
}
return -1
}
func (s *Server) GetPair(id string) []byte {
s.mu.Lock()
defer s.mu.Unlock()
if i := s.pairIndex(id); i >= 0 {
query, _ := url.ParseQuery(s.pairings[i])
b, _ := hex.DecodeString(query.Get("client_public"))
return b
}
return nil
}
func (s *Server) AddPair(id string, public []byte, permissions byte) {
s.log.Debug().Str("stream", s.stream).Msgf("[hksv] add pair id=%s public=%x perm=%d", id, public, permissions)
s.mu.Lock()
if s.pairIndex(id) < 0 {
s.pairings = append(s.pairings, fmt.Sprintf(
"client_id=%s&client_public=%x&permissions=%d", id, public, permissions,
))
s.UpdateStatus()
s.savePairings()
}
s.mu.Unlock()
}
func (s *Server) DelPair(id string) {
s.log.Debug().Str("stream", s.stream).Msgf("[hksv] del pair id=%s", id)
s.mu.Lock()
if i := s.pairIndex(id); i >= 0 {
s.pairings = append(s.pairings[:i], s.pairings[i+1:]...)
s.UpdateStatus()
s.savePairings()
}
s.mu.Unlock()
}
func (s *Server) savePairings() {
if s.store != nil {
if err := s.store.SavePairings(s.stream, s.pairings); err != nil {
s.log.Error().Err(err).Msgf("[hksv] can't save %s pairings=%v", s.stream, s.pairings)
}
}
}
func (s *Server) GetAccessories(_ net.Conn) []*hap.Accessory {
s.log.Trace().Str("stream", s.stream).Msg("[hksv] GET /accessories")
if s.log.Trace().Enabled() {
if b, err := json.Marshal(s.accessory); err == nil {
s.log.Trace().Str("stream", s.stream).Str("accessory", string(b)).Msg("[hksv] accessory JSON")
}
}
return []*hap.Accessory{s.accessory}
}
func (s *Server) GetCharacteristic(conn net.Conn, aid uint8, iid uint64) any {
s.log.Trace().Str("stream", s.stream).Msgf("[hksv] get char aid=%d iid=0x%x", aid, iid)
char := s.accessory.GetCharacterByID(iid)
if char == nil {
s.log.Warn().Msgf("[hksv] get unknown characteristic: %d", iid)
return nil
}
switch char.Type {
case camera.TypeSetupEndpoints:
if s.liveStream != nil {
return s.liveStream.GetEndpointsResponse()
}
return nil
}
return char.Value
}
func (s *Server) SetCharacteristic(conn net.Conn, aid uint8, iid uint64, value any) {
s.log.Trace().Str("stream", s.stream).Msgf("[hksv] set char aid=%d iid=0x%x value=%v", aid, iid, value)
char := s.accessory.GetCharacterByID(iid)
if char == nil {
s.log.Warn().Msgf("[hksv] set unknown characteristic: %d", iid)
return
}
switch char.Type {
case camera.TypeSetupEndpoints:
if s.liveStream == nil {
return
}
var offer camera.SetupEndpointsRequest
if err := tlv8.UnmarshalBase64(value, &offer); err != nil {
return
}
resp, err := s.liveStream.SetupEndpoints(conn, &offer)
if err != nil {
s.log.Error().Err(err).Msg("[hksv] setup endpoints failed")
}
_ = resp // stored by the handler
case camera.TypeSelectedStreamConfiguration:
if s.liveStream == nil {
return
}
var conf camera.SelectedStreamConfiguration
if err := tlv8.UnmarshalBase64(value, &conf); err != nil {
return
}
s.log.Trace().Str("stream", s.stream).Msgf("[hksv] stream id=%x cmd=%d", conf.Control.SessionID, conf.Control.Command)
switch conf.Control.Command {
case camera.SessionCommandEnd:
_ = s.liveStream.StopStream(conf.Control.SessionID, s)
case camera.SessionCommandStart:
_ = s.liveStream.StartStream(s.stream, &conf, s)
}
case camera.TypeSetupDataStreamTransport:
var req camera.SetupDataStreamTransportRequest
if err := tlv8.UnmarshalBase64(value, &req); err != nil {
s.log.Error().Err(err).Str("stream", s.stream).Msg("[hksv] parse ch131 failed")
return
}
s.log.Debug().Str("stream", s.stream).Uint8("cmd", req.SessionCommandType).
Uint8("transport", req.TransportType).Msg("[hksv] DataStream setup")
if req.SessionCommandType != 0 {
s.log.Debug().Str("stream", s.stream).Msg("[hksv] DataStream close request")
if s.hksvSession != nil {
s.hksvSession.Close()
}
return
}
accessoryKeySalt := core.RandString(32, 0)
combinedSalt := req.ControllerKeySalt + accessoryKeySalt
ln, err := net.ListenTCP("tcp", nil)
if err != nil {
s.log.Error().Err(err).Str("stream", s.stream).Msg("[hksv] listen failed")
return
}
port := ln.Addr().(*net.TCPAddr).Port
resp := camera.SetupDataStreamTransportResponse{
Status: 0,
AccessoryKeySalt: accessoryKeySalt,
}
resp.TransportTypeSessionParameters.TCPListeningPort = uint16(port)
v, err := tlv8.MarshalBase64(resp)
if err != nil {
ln.Close()
return
}
char.Value = v
s.log.Debug().Str("stream", s.stream).Int("port", port).Msg("[hksv] listening for HDS")
hapConn := conn.(*hap.Conn)
go s.acceptHDS(hapConn, ln, combinedSalt)
case camera.TypeSelectedCameraRecordingConfiguration:
s.log.Debug().Str("stream", s.stream).Str("motion", s.motionMode).Msg("[hksv] selected recording config")
char.Value = value
switch s.motionMode {
case "continuous":
go s.startContinuousMotion()
case "detect":
go s.startMotionDetector()
}
default:
char.Value = value
}
}
func (s *Server) GetImage(conn net.Conn, width, height int) []byte {
s.log.Trace().Str("stream", s.stream).Msgf("[hksv] get image width=%d height=%d", width, height)
if s.snapshots == nil {
return nil
}
b, err := s.snapshots.GetSnapshot(s.stream, width, height)
if err != nil {
s.log.Error().Err(err).Msg("[hksv] snapshot failed")
return nil
}
return b
}
// SetMotionDetected triggers or clears the motion detected characteristic.
func (s *Server) SetMotionDetected(detected bool) {
if s.accessory == nil {
return
}
char := s.accessory.GetCharacter("22") // MotionDetected
if char == nil {
return
}
char.Value = detected
_ = char.NotifyListeners(nil)
s.log.Debug().Str("stream", s.stream).Bool("motion", detected).Msg("[hksv] motion")
}
// TriggerDoorbell triggers a doorbell press event.
func (s *Server) TriggerDoorbell() {
if s.accessory == nil {
return
}
char := s.accessory.GetCharacter("73") // ProgrammableSwitchEvent
if char == nil {
return
}
char.Value = 0 // SINGLE_PRESS
_ = char.NotifyListeners(nil)
s.log.Debug().Str("stream", s.stream).Msg("[hksv] doorbell")
}
// acceptHDS opens a TCP listener for the HDS DataStream connection from the Home Hub
func (s *Server) acceptHDS(hapConn *hap.Conn, ln net.Listener, salt string) {
defer ln.Close()
if tcpLn, ok := ln.(*net.TCPListener); ok {
_ = tcpLn.SetDeadline(time.Now().Add(30 * time.Second))
}
rawConn, err := ln.Accept()
if err != nil {
s.log.Error().Err(err).Str("stream", s.stream).Msg("[hksv] accept failed")
return
}
defer rawConn.Close()
hdsConn, err := hds.NewConn(rawConn, hapConn.SharedKey, salt, false)
if err != nil {
s.log.Error().Err(err).Str("stream", s.stream).Msg("[hksv] hds conn failed")
return
}
s.AddConn(hdsConn)
defer s.DelConn(hdsConn)
session := newHKSVSession(s, hapConn, hdsConn)
s.mu.Lock()
s.hksvSession = session
s.mu.Unlock()
defer func() {
s.mu.Lock()
if s.hksvSession == session {
s.hksvSession = nil
}
s.mu.Unlock()
session.Close()
}()
s.log.Debug().Str("stream", s.stream).Msg("[hksv] session started")
if err := session.Run(); err != nil {
s.log.Debug().Err(err).Str("stream", s.stream).Msg("[hksv] session ended")
}
}
// prepareHKSVConsumer pre-starts a consumer and adds it to the stream.
func (s *Server) prepareHKSVConsumer() {
consumer := NewHKSVConsumer(s.log)
if err := s.streams.AddConsumer(s.stream, consumer); err != nil {
s.log.Debug().Err(err).Str("stream", s.stream).Msg("[hksv] prepare consumer failed")
return
}
s.log.Debug().Str("stream", s.stream).Msg("[hksv] consumer prepared")
s.mu.Lock()
if s.preparedConsumer != nil {
old := s.preparedConsumer
s.preparedConsumer = nil
s.mu.Unlock()
s.streams.RemoveConsumer(s.stream, old)
_ = old.Stop()
s.mu.Lock()
}
s.preparedConsumer = consumer
s.mu.Unlock()
// Keep alive until used or timeout (60 seconds)
select {
case <-consumer.Done():
// consumer was stopped (used or server closed)
case <-time.After(60 * time.Second):
s.mu.Lock()
if s.preparedConsumer == consumer {
s.preparedConsumer = nil
s.mu.Unlock()
s.streams.RemoveConsumer(s.stream, consumer)
_ = consumer.Stop()
s.log.Debug().Str("stream", s.stream).Msg("[hksv] prepared consumer expired")
} else {
s.mu.Unlock()
}
}
}
func (s *Server) takePreparedConsumer() *HKSVConsumer {
s.mu.Lock()
defer s.mu.Unlock()
consumer := s.preparedConsumer
s.preparedConsumer = nil
return consumer
}
func (s *Server) startMotionDetector() {
s.mu.Lock()
if s.motionDetector != nil {
s.mu.Unlock()
return
}
det := NewMotionDetector(s.motionThreshold, s.SetMotionDetected, s.log)
s.motionDetector = det
s.mu.Unlock()
s.AddConn(det)
if err := s.streams.AddConsumer(s.stream, det); err != nil {
s.log.Error().Err(err).Str("stream", s.stream).Msg("[hksv] motion detector add consumer failed")
s.DelConn(det)
s.mu.Lock()
s.motionDetector = nil
s.mu.Unlock()
return
}
s.log.Debug().Str("stream", s.stream).Msg("[hksv] motion detector started")
_, _ = det.WriteTo(nil) // blocks until Stop()
s.streams.RemoveConsumer(s.stream, det)
s.DelConn(det)
s.mu.Lock()
if s.motionDetector == det {
s.motionDetector = nil
}
s.mu.Unlock()
s.log.Debug().Str("stream", s.stream).Msg("[hksv] motion detector stopped")
}
func (s *Server) stopMotionDetector() {
s.mu.Lock()
det := s.motionDetector
s.mu.Unlock()
if det != nil {
_ = det.Stop()
}
}
func (s *Server) startContinuousMotion() {
s.mu.Lock()
if s.continuousMotion {
s.mu.Unlock()
return
}
s.continuousMotion = true
s.mu.Unlock()
s.log.Debug().Str("stream", s.stream).Msg("[hksv] continuous motion started")
// delay to allow Home Hub to subscribe to events
time.Sleep(5 * time.Second)
s.SetMotionDetected(true)
ticker := time.NewTicker(30 * time.Second)
defer ticker.Stop()
for range ticker.C {
if s.accessory == nil {
return
}
s.SetMotionDetected(true)
}
}
// isClosedConnErr checks if the error is a "use of closed network connection" error.
// This happens when the remote side (e.g., iPhone) closes the TCP connection.
func isClosedConnErr(err error) bool {
if err == nil {
return false
}
return strings.Contains(err.Error(), "use of closed network connection")
}
pkg/hksv/motion.go
+243
View File
@@ -0,0 +1,243 @@
// Author: Sergei "svk" Krashevich <svk@svk.su>
package hksv
import (
"io"
"time"
"github.com/AlexxIT/go2rtc/pkg/core"
"github.com/AlexxIT/go2rtc/pkg/h264"
"github.com/pion/rtp"
"github.com/rs/zerolog"
)
const (
motionWarmupFrames = 30
defaultThreshold = 2.0
motionAlphaFast = 0.1
motionAlphaSlow = 0.02
motionHoldTime = 30 * time.Second
motionCooldown = 5 * time.Second
motionDefaultFPS = 30.0
// recalibrate FPS and emit trace log every N frames (~5s at 30fps)
motionTraceFrames = 150
)
// MotionDetector implements core.Consumer for P-frame based motion detection.
// It analyzes H.264 P-frame sizes using an EMA baseline and triggers a callback
// when the frame size exceeds the baseline by the configured threshold.
type MotionDetector struct {
core.Connection
done chan struct{}
log zerolog.Logger
// algorithm state (accessed only from Sender goroutine — no mutex needed)
threshold float64
triggerLevel int // pre-computed: int(baseline * threshold)
baseline float64
initialized bool
frameCount int
// frame-based timing (calibrated periodically, no time.Now() in per-frame hot path)
holdBudget int // motionHoldTime converted to frames
cooldownBudget int // motionCooldown converted to frames
remainingHold int // frames left until hold expires (active motion)
remainingCooldown int // frames left until cooldown expires (after OFF)
// motion state
motionActive bool
// periodic FPS recalibration
lastFPSCheck time.Time
lastFPSFrame int
// for testing: injectable time and callback
now func() time.Time
OnMotion func(bool) `json:"-"` // callback when motion state changes
}
// NewMotionDetector creates a new motion detector with the given threshold and callback.
// If threshold <= 0, the default of 2.0 is used.
// onMotion is called when motion state changes (true=detected, false=ended).
func NewMotionDetector(threshold float64, onMotion func(bool), log zerolog.Logger) *MotionDetector {
if threshold <= 0 {
threshold = defaultThreshold
}
medias := []*core.Media{
{
Kind: core.KindVideo,
Direction: core.DirectionSendonly,
Codecs: []*core.Codec{
{Name: core.CodecH264},
},
},
}
return &MotionDetector{
Connection: core.Connection{
ID: core.NewID(),
FormatName: "motion",
Protocol: "detect",
Medias: medias,
},
threshold: threshold,
done: make(chan struct{}),
now: time.Now,
OnMotion: onMotion,
log: log,
}
}
func (m *MotionDetector) AddTrack(media *core.Media, _ *core.Codec, track *core.Receiver) error {
m.log.Debug().Str("codec", track.Codec.Name).Msg("[hksv] motion: add track")
codec := track.Codec.Clone()
sender := core.NewSender(media, codec)
sender.Handler = func(packet *rtp.Packet) {
m.handlePacket(packet)
}
if track.Codec.IsRTP() {
sender.Handler = h264.RTPDepay(track.Codec, sender.Handler)
} else {
sender.Handler = h264.RepairAVCC(track.Codec, sender.Handler)
}
sender.HandleRTP(track)
m.Senders = append(m.Senders, sender)
return nil
}
func (m *MotionDetector) calibrate() {
m.holdBudget = int(motionHoldTime.Seconds() * motionDefaultFPS)
m.cooldownBudget = int(motionCooldown.Seconds() * motionDefaultFPS)
m.triggerLevel = int(m.baseline * m.threshold)
m.lastFPSCheck = m.now()
m.lastFPSFrame = m.frameCount
m.log.Debug().
Float64("baseline", m.baseline).
Int("holdFrames", m.holdBudget).Int("cooldownFrames", m.cooldownBudget).
Msg("[hksv] motion: warmup complete")
}
func (m *MotionDetector) handlePacket(packet *rtp.Packet) {
payload := packet.Payload
if len(payload) < 5 {
return
}
// skip keyframes — always large, not informative for motion
if h264.IsKeyframe(payload) {
return
}
size := len(payload)
m.frameCount++
if m.frameCount <= motionWarmupFrames {
fsize := float64(size)
if !m.initialized {
m.baseline = fsize
m.initialized = true
} else {
m.baseline += motionAlphaFast * (fsize - m.baseline)
}
if m.frameCount == motionWarmupFrames {
m.calibrate()
}
return
}
if m.triggerLevel <= 0 {
return
}
// integer comparison — no float division needed
triggered := size > m.triggerLevel
if !m.motionActive {
// idle path: decrement cooldown, check for trigger, update baseline
if m.remainingCooldown > 0 {
m.remainingCooldown--
}
if triggered && m.remainingCooldown <= 0 {
m.motionActive = true
m.remainingHold = m.holdBudget
m.log.Debug().
Float64("ratio", float64(size)/m.baseline).
Msg("[hksv] motion: ON")
m.setMotion(true)
}
// update baseline only if still idle (trigger frame doesn't pollute baseline)
if !m.motionActive {
fsize := float64(size)
m.baseline += motionAlphaSlow * (fsize - m.baseline)
m.triggerLevel = int(m.baseline * m.threshold)
}
} else {
// active motion path: pure integer arithmetic, zero time.Now() calls
if triggered {
m.remainingHold = m.holdBudget
} else {
m.remainingHold--
if m.remainingHold <= 0 {
m.motionActive = false
m.remainingCooldown = m.cooldownBudget
m.log.Debug().Msg("[hksv] motion: OFF (hold expired)")
m.setMotion(false)
}
}
}
// periodic: recalibrate FPS and emit trace log
if m.frameCount%motionTraceFrames == 0 {
now := m.now()
frames := m.frameCount - m.lastFPSFrame
if frames > 0 {
if elapsed := now.Sub(m.lastFPSCheck); elapsed > time.Millisecond {
fps := float64(frames) / elapsed.Seconds()
m.holdBudget = int(motionHoldTime.Seconds() * fps)
m.cooldownBudget = int(motionCooldown.Seconds() * fps)
}
}
m.lastFPSCheck = now
m.lastFPSFrame = m.frameCount
m.log.Trace().
Float64("baseline", m.baseline).Float64("ratio", float64(size)/m.baseline).
Bool("active", m.motionActive).Msg("[hksv] motion: status")
}
}
func (m *MotionDetector) setMotion(detected bool) {
if m.OnMotion != nil {
m.OnMotion(detected)
}
}
func (m *MotionDetector) String() string {
return "motion detector"
}
func (m *MotionDetector) WriteTo(io.Writer) (int64, error) {
<-m.done
return 0, nil
}
func (m *MotionDetector) Stop() error {
select {
case <-m.done:
default:
if m.motionActive {
m.motionActive = false
m.log.Debug().Msg("[hksv] motion: OFF (stop)")
m.setMotion(false)
}
close(m.done)
}
return m.Connection.Stop()
}
pkg/hksv/motion_test.go
+510
View File
@@ -0,0 +1,510 @@
// Author: Sergei "svk" Krashevich <svk@svk.su>
package hksv
import (
"encoding/binary"
"testing"
"time"
"github.com/AlexxIT/go2rtc/pkg/h264"
"github.com/pion/rtp"
"github.com/rs/zerolog"
)
// makeAVCC creates a fake AVCC packet with the given NAL type and total size.
func makeAVCC(nalType byte, totalSize int) []byte {
if totalSize < 5 {
totalSize = 5
}
b := make([]byte, totalSize)
binary.BigEndian.PutUint32(b[:4], uint32(totalSize-4))
b[4] = nalType
return b
}
func makePFrame(size int) *rtp.Packet {
return &rtp.Packet{Payload: makeAVCC(h264.NALUTypePFrame, size)}
}
func makeIFrame(size int) *rtp.Packet {
return &rtp.Packet{Payload: makeAVCC(h264.NALUTypeIFrame, size)}
}
type mockClock struct {
t time.Time
}
func (c *mockClock) now() time.Time { return c.t }
func (c *mockClock) advance(d time.Duration) { c.t = c.t.Add(d) }
type motionRecorder struct {
calls []bool
}
func (r *motionRecorder) onMotion(detected bool) {
r.calls = append(r.calls, detected)
}
func (r *motionRecorder) lastCall() (bool, bool) {
if len(r.calls) == 0 {
return false, false
}
return r.calls[len(r.calls)-1], true
}
func newTestDetector() (*MotionDetector, *mockClock, *motionRecorder) {
rec := &motionRecorder{}
det := NewMotionDetector(0, rec.onMotion, zerolog.Nop())
clock := &mockClock{t: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)}
det.now = clock.now
return det, clock, rec
}
// warmup feeds the detector with small P-frames to build baseline.
func warmup(det *MotionDetector, clock *mockClock, size int) {
for i := 0; i < motionWarmupFrames; i++ {
det.handlePacket(makePFrame(size))
clock.advance(33 * time.Millisecond) // ~30fps
}
}
// warmupWithBudgets performs warmup then sets test-friendly hold/cooldown budgets.
func warmupWithBudgets(det *MotionDetector, clock *mockClock, size, hold, cooldown int) {
warmup(det, clock, size)
det.holdBudget = hold
det.cooldownBudget = cooldown
}
func TestMotionDetector_NoMotion(t *testing.T) {
det, clock, rec := newTestDetector()
warmup(det, clock, 500)
// feed same-size P-frames — no motion
for i := 0; i < 100; i++ {
det.handlePacket(makePFrame(500))
}
if len(rec.calls) != 0 {
t.Fatalf("expected no motion calls, got %d: %v", len(rec.calls), rec.calls)
}
}
func TestMotionDetector_MotionDetected(t *testing.T) {
det, clock, rec := newTestDetector()
warmup(det, clock, 500)
// large P-frame triggers motion
det.handlePacket(makePFrame(5000))
last, ok := rec.lastCall()
if !ok || !last {
t.Fatal("expected motion detected")
}
}
func TestMotionDetector_HoldTime(t *testing.T) {
det, clock, rec := newTestDetector()
warmupWithBudgets(det, clock, 500, 30, 5)
// trigger motion
det.handlePacket(makePFrame(5000))
if len(rec.calls) != 1 || !rec.calls[0] {
t.Fatal("expected motion ON")
}
// send 20 non-triggered frames — still active (< holdBudget=30)
for i := 0; i < 20; i++ {
det.handlePacket(makePFrame(500))
}
if len(rec.calls) != 1 {
t.Fatalf("expected only ON call during hold, got %v", rec.calls)
}
// send 15 more (total 35 > holdBudget=30) — should turn OFF
for i := 0; i < 15; i++ {
det.handlePacket(makePFrame(500))
}
last, _ := rec.lastCall()
if last {
t.Fatal("expected motion OFF after hold budget exhausted")
}
}
func TestMotionDetector_Cooldown(t *testing.T) {
det, clock, rec := newTestDetector()
warmupWithBudgets(det, clock, 500, 30, 5)
// trigger and expire motion
det.handlePacket(makePFrame(5000))
for i := 0; i < 30; i++ {
det.handlePacket(makePFrame(500))
}
if len(rec.calls) != 2 || rec.calls[1] != false {
t.Fatalf("expected ON then OFF, got %v", rec.calls)
}
// try to trigger again immediately — should be blocked by cooldown
det.handlePacket(makePFrame(5000))
if len(rec.calls) != 2 {
t.Fatalf("expected cooldown to block re-trigger, got %v", rec.calls)
}
// send frames to expire cooldown (blocked trigger consumed 1 decrement)
for i := 0; i < 5; i++ {
det.handlePacket(makePFrame(500))
}
// now re-trigger should work
det.handlePacket(makePFrame(5000))
if len(rec.calls) != 3 || !rec.calls[2] {
t.Fatalf("expected motion ON after cooldown, got %v", rec.calls)
}
}
func TestMotionDetector_SkipsKeyframes(t *testing.T) {
det, clock, rec := newTestDetector()
warmup(det, clock, 500)
// huge keyframe should not trigger motion
det.handlePacket(makeIFrame(50000))
if len(rec.calls) != 0 {
t.Fatal("keyframes should not trigger motion")
}
// verify baseline didn't change
det.handlePacket(makePFrame(500))
if len(rec.calls) != 0 {
t.Fatal("baseline should be unaffected by keyframes")
}
}
func TestMotionDetector_Warmup(t *testing.T) {
det, clock, rec := newTestDetector()
// during warmup, even large frames should not trigger
for i := 0; i < motionWarmupFrames; i++ {
det.handlePacket(makePFrame(5000))
clock.advance(33 * time.Millisecond)
}
if len(rec.calls) != 0 {
t.Fatal("warmup should not trigger motion")
}
}
func TestMotionDetector_BaselineFreeze(t *testing.T) {
det, clock, rec := newTestDetector()
warmup(det, clock, 500)
baselineBefore := det.baseline
// trigger motion
det.handlePacket(makePFrame(5000))
if len(rec.calls) != 1 || !rec.calls[0] {
t.Fatal("expected motion ON")
}
// feed large frames during motion — baseline should not change
for i := 0; i < 50; i++ {
det.handlePacket(makePFrame(5000))
}
if det.baseline != baselineBefore {
t.Fatalf("baseline changed during motion: %f -> %f", baselineBefore, det.baseline)
}
}
func TestMotionDetector_CustomThreshold(t *testing.T) {
det, clock, rec := newTestDetector()
det.threshold = 1.5
warmup(det, clock, 500)
// 1.6x — below default 2.0 but above custom 1.5
det.handlePacket(makePFrame(800))
if len(rec.calls) != 1 || !rec.calls[0] {
t.Fatalf("expected motion ON with custom threshold 1.5, got %v", rec.calls)
}
}
func TestMotionDetector_CustomThresholdNoFalsePositive(t *testing.T) {
det, clock, rec := newTestDetector()
det.threshold = 3.0
warmup(det, clock, 500)
// 2.5x — above default 2.0 but below custom 3.0
det.handlePacket(makePFrame(1250))
if len(rec.calls) != 0 {
t.Fatalf("expected no motion with high threshold 3.0, got %v", rec.calls)
}
}
func TestMotionDetector_HoldTimeExtended(t *testing.T) {
det, clock, rec := newTestDetector()
warmupWithBudgets(det, clock, 500, 30, 5)
// trigger motion
det.handlePacket(makePFrame(5000))
if len(rec.calls) != 1 || !rec.calls[0] {
t.Fatal("expected motion ON")
}
// send 25 non-triggered frames (remainingHold 30→5)
for i := 0; i < 25; i++ {
det.handlePacket(makePFrame(500))
}
// re-trigger — remainingHold resets to 30
det.handlePacket(makePFrame(5000))
// send 25 more non-triggered (remainingHold 30→5)
for i := 0; i < 25; i++ {
det.handlePacket(makePFrame(500))
}
// should still be ON
if len(rec.calls) != 1 {
t.Fatalf("expected hold time to be extended by re-trigger, got %v", rec.calls)
}
// send 10 more to exhaust hold
for i := 0; i < 10; i++ {
det.handlePacket(makePFrame(500))
}
last, _ := rec.lastCall()
if last {
t.Fatal("expected motion OFF after extended hold expired")
}
}
func TestMotionDetector_SmallPayloadIgnored(t *testing.T) {
det, clock, rec := newTestDetector()
warmup(det, clock, 500)
det.handlePacket(&rtp.Packet{Payload: []byte{1, 2, 3, 4}})
det.handlePacket(&rtp.Packet{Payload: nil})
det.handlePacket(&rtp.Packet{Payload: []byte{}})
if len(rec.calls) != 0 {
t.Fatalf("small payloads should be ignored, got %v", rec.calls)
}
}
func TestMotionDetector_BaselineAdapts(t *testing.T) {
det, clock, _ := newTestDetector()
warmup(det, clock, 500)
baselineAfterWarmup := det.baseline
// feed gradually larger frames — baseline should drift up
for i := 0; i < 200; i++ {
det.handlePacket(makePFrame(700))
}
if det.baseline <= baselineAfterWarmup {
t.Fatalf("baseline should adapt upward: before=%f after=%f", baselineAfterWarmup, det.baseline)
}
}
func TestMotionDetector_DoubleStopSafe(t *testing.T) {
det, clock, rec := newTestDetector()
warmup(det, clock, 500)
det.handlePacket(makePFrame(5000))
_ = det.Stop()
_ = det.Stop() // second stop should not panic
if len(rec.calls) != 2 {
t.Fatalf("expected ON+OFF, got %v", rec.calls)
}
}
func TestMotionDetector_StopWithoutMotion(t *testing.T) {
det, clock, _ := newTestDetector()
warmup(det, clock, 500)
rec := &motionRecorder{}
det.OnMotion = rec.onMotion
_ = det.Stop()
if len(rec.calls) != 0 {
t.Fatalf("stop without motion should not call onMotion, got %v", rec.calls)
}
}
func TestMotionDetector_StopClearsMotion(t *testing.T) {
det, clock, rec := newTestDetector()
warmup(det, clock, 500)
det.handlePacket(makePFrame(5000))
if len(rec.calls) != 1 || !rec.calls[0] {
t.Fatal("expected motion ON")
}
_ = det.Stop()
if len(rec.calls) != 2 || rec.calls[1] != false {
t.Fatalf("expected Stop to clear motion, got %v", rec.calls)
}
}
func TestMotionDetector_WarmupBaseline(t *testing.T) {
det, clock, _ := newTestDetector()
for i := 0; i < motionWarmupFrames; i++ {
size := 400 + (i%5)*50
det.handlePacket(makePFrame(size))
clock.advance(33 * time.Millisecond)
}
if det.baseline < 400 || det.baseline > 600 {
t.Fatalf("baseline should be in 400-600 range, got %f", det.baseline)
}
}
func TestMotionDetector_MultipleCycles(t *testing.T) {
det, clock, rec := newTestDetector()
warmupWithBudgets(det, clock, 500, 30, 5)
for cycle := 0; cycle < 3; cycle++ {
det.handlePacket(makePFrame(5000)) // trigger ON
for i := 0; i < 30; i++ { // expire hold
det.handlePacket(makePFrame(500))
}
for i := 0; i < 6; i++ { // expire cooldown
det.handlePacket(makePFrame(500))
}
}
if len(rec.calls) != 6 {
t.Fatalf("expected 6 calls (3 cycles), got %d: %v", len(rec.calls), rec.calls)
}
for i, v := range rec.calls {
expected := i%2 == 0
if v != expected {
t.Fatalf("call[%d] = %v, expected %v", i, v, expected)
}
}
}
func TestMotionDetector_TriggerLevel(t *testing.T) {
det, clock, _ := newTestDetector()
warmup(det, clock, 500)
expected := int(det.baseline * det.threshold)
if det.triggerLevel != expected {
t.Fatalf("triggerLevel = %d, expected %d", det.triggerLevel, expected)
}
}
func TestMotionDetector_DefaultFPSCalibration(t *testing.T) {
det, clock, _ := newTestDetector()
warmup(det, clock, 500)
// calibrate uses default 30fps
expectedHold := int(motionHoldTime.Seconds() * motionDefaultFPS)
expectedCooldown := int(motionCooldown.Seconds() * motionDefaultFPS)
if det.holdBudget != expectedHold {
t.Fatalf("holdBudget = %d, expected %d", det.holdBudget, expectedHold)
}
if det.cooldownBudget != expectedCooldown {
t.Fatalf("cooldownBudget = %d, expected %d", det.cooldownBudget, expectedCooldown)
}
}
func TestMotionDetector_FPSRecalibration(t *testing.T) {
det, clock, _ := newTestDetector()
warmup(det, clock, 500)
// initial budgets use default 30fps
initialHold := det.holdBudget
// send motionTraceFrames frames with 100ms intervals → FPS=10
for i := 0; i < motionTraceFrames; i++ {
clock.advance(100 * time.Millisecond)
det.handlePacket(makePFrame(500))
}
// after recalibration, holdBudget should reflect ~10fps (±5% due to warmup tail)
expectedHold := int(motionHoldTime.Seconds() * 10.0) // ~300
if det.holdBudget < expectedHold-20 || det.holdBudget > expectedHold+20 {
t.Fatalf("holdBudget after recalibration = %d, expected ~%d (was %d)", det.holdBudget, expectedHold, initialHold)
}
}
func BenchmarkMotionDetector_HandlePacket(b *testing.B) {
det, clock, _ := newTestDetector()
warmup(det, clock, 500)
pkt := makePFrame(600)
b.ResetTimer()
for i := 0; i < b.N; i++ {
det.handlePacket(pkt)
}
}
func BenchmarkMotionDetector_WithKeyframes(b *testing.B) {
det, clock, _ := newTestDetector()
warmup(det, clock, 500)
pFrame := makePFrame(600)
iFrame := makeIFrame(10000)
b.ResetTimer()
for i := 0; i < b.N; i++ {
if i%30 == 0 {
det.handlePacket(iFrame)
} else {
det.handlePacket(pFrame)
}
}
}
func BenchmarkMotionDetector_MotionActive(b *testing.B) {
det, clock, _ := newTestDetector()
warmup(det, clock, 500)
// trigger motion and keep it active
det.handlePacket(makePFrame(5000))
pkt := makePFrame(5000)
b.ResetTimer()
for i := 0; i < b.N; i++ {
det.handlePacket(pkt)
}
}
func BenchmarkMotionDetector_Warmup(b *testing.B) {
pkt := makePFrame(500)
b.ResetTimer()
for i := 0; i < b.N; i++ {
det := NewMotionDetector(0, func(bool) {}, zerolog.Nop())
det.now = time.Now
for j := 0; j < motionWarmupFrames; j++ {
det.handlePacket(pkt)
}
}
}
pkg/hksv/session.go
+117
View File
@@ -0,0 +1,117 @@
// Author: Sergei "svk" Krashevich <svk@svk.su>
package hksv
import (
"sync"
"github.com/AlexxIT/go2rtc/pkg/hap"
"github.com/AlexxIT/go2rtc/pkg/hap/hds"
"github.com/rs/zerolog"
)
// hksvSession manages the HDS DataStream connection for HKSV recording
type hksvSession struct {
server *Server
hapConn *hap.Conn
hdsConn *hds.Conn
session *hds.Session
log zerolog.Logger
mu sync.Mutex
consumer *HKSVConsumer
}
func newHKSVSession(srv *Server, hapConn *hap.Conn, hdsConn *hds.Conn) *hksvSession {
session := hds.NewSession(hdsConn)
hs := &hksvSession{
server: srv,
hapConn: hapConn,
hdsConn: hdsConn,
session: session,
log: srv.log,
}
session.OnDataSendOpen = hs.handleOpen
session.OnDataSendClose = hs.handleClose
return hs
}
func (hs *hksvSession) Run() error {
return hs.session.Run()
}
func (hs *hksvSession) Close() {
hs.mu.Lock()
defer hs.mu.Unlock()
if hs.consumer != nil {
hs.stopRecording()
}
_ = hs.session.Close()
}
func (hs *hksvSession) handleOpen(streamID int) error {
hs.mu.Lock()
defer hs.mu.Unlock()
hs.log.Debug().Str("stream", hs.server.stream).Int("streamID", streamID).Msg("[hksv] dataSend open")
if hs.consumer != nil {
hs.stopRecording()
}
// Try to use the pre-started consumer from pair-verify
consumer := hs.server.takePreparedConsumer()
if consumer != nil {
hs.log.Debug().Str("stream", hs.server.stream).Msg("[hksv] using prepared consumer")
hs.consumer = consumer
hs.server.AddConn(consumer)
// Activate: set the HDS session and send init + start streaming
if err := consumer.Activate(hs.session, streamID); err != nil {
hs.log.Error().Err(err).Str("stream", hs.server.stream).Msg("[hksv] activate failed")
hs.stopRecording()
return nil
}
return nil
}
// Fallback: create new consumer (will be slow ~3s)
hs.log.Debug().Str("stream", hs.server.stream).Msg("[hksv] no prepared consumer, creating new")
consumer = NewHKSVConsumer(hs.log)
if err := hs.server.streams.AddConsumer(hs.server.stream, consumer); err != nil {
hs.log.Error().Err(err).Str("stream", hs.server.stream).Msg("[hksv] add consumer failed")
return nil
}
hs.consumer = consumer
hs.server.AddConn(consumer)
go func() {
if err := consumer.Activate(hs.session, streamID); err != nil {
hs.log.Error().Err(err).Str("stream", hs.server.stream).Msg("[hksv] activate failed")
}
}()
return nil
}
func (hs *hksvSession) handleClose(streamID int) error {
hs.mu.Lock()
defer hs.mu.Unlock()
hs.log.Debug().Str("stream", hs.server.stream).Int("streamID", streamID).Msg("[hksv] dataSend close")
if hs.consumer != nil {
hs.stopRecording()
}
return nil
}
func (hs *hksvSession) stopRecording() {
consumer := hs.consumer
hs.consumer = nil
hs.server.streams.RemoveConsumer(hs.server.stream, consumer)
_ = consumer.Stop()
hs.server.DelConn(consumer)
}