diff --git a/machinery/src/cloud/cloud.go b/machinery/src/cloud/cloud.go index c476f23..b941849 100644 --- a/machinery/src/cloud/cloud.go +++ b/machinery/src/cloud/cloud.go @@ -874,7 +874,7 @@ func HandleLiveStreamSD(livestreamCursor *packets.QueueCursor, configuration *mo log.Log.Debug("cloud.HandleLiveStreamSD(): finished") } -func HandleLiveStreamHD(livestreamCursor *packets.QueueCursor, configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, rtspClient capture.RTSPClient) { +func HandleLiveStreamHD(configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, rtspClient capture.RTSPClient, rtspSubClient capture.RTSPClient, subStreamEnabled bool) { config := configuration.Config @@ -888,23 +888,51 @@ func HandleLiveStreamHD(livestreamCursor *packets.QueueCursor, configuration *mo // Create per-peer broadcasters instead of shared tracks. // Each viewer gets its own track with independent, non-blocking writes // so a slow/congested peer cannot stall the others. - streams, _ := rtspClient.GetStreams() - videoBroadcaster := webrtc.NewVideoBroadcaster(streams) - audioBroadcaster := webrtc.NewAudioBroadcaster(streams) + // + // Both the main (high-resolution) and sub (low-resolution) streams are + // exposed as separate broadcasters that are always forwarding, so a + // viewer can pick the resolution it needs per peer connection without + // the agent re-negotiating the RTSP source. + mainStreams, _ := rtspClient.GetStreams() + mainVideoBroadcaster := webrtc.NewVideoBroadcaster(mainStreams) + mainAudioBroadcaster := webrtc.NewAudioBroadcaster(mainStreams) - if videoBroadcaster == nil && audioBroadcaster == nil { - log.Log.Error("cloud.HandleLiveStreamHD(): failed to create both video and audio broadcasters") + if mainVideoBroadcaster == nil && mainAudioBroadcaster == nil { + log.Log.Error("cloud.HandleLiveStreamHD(): failed to create both video and audio broadcasters for the main stream") return } - go webrtc.WriteToTrack(livestreamCursor, configuration, communication, mqttClient, videoBroadcaster, audioBroadcaster, rtspClient) + go webrtc.WriteToTrack(communication.Queue.Latest(), configuration, communication, mqttClient, mainVideoBroadcaster, mainAudioBroadcaster, rtspClient) + + // Sub stream broadcasters, only when a distinct sub stream is available. + var subVideoBroadcaster *webrtc.TrackBroadcaster + var subAudioBroadcaster *webrtc.TrackBroadcaster + if subStreamEnabled && rtspSubClient != nil && communication.SubQueue != nil { + subStreams, _ := rtspSubClient.GetStreams() + subVideoBroadcaster = webrtc.NewVideoBroadcaster(subStreams) + subAudioBroadcaster = webrtc.NewAudioBroadcaster(subStreams) + go webrtc.WriteToTrack(communication.SubQueue.Latest(), configuration, communication, mqttClient, subVideoBroadcaster, subAudioBroadcaster, rtspSubClient) + } + subBroadcastersReady := subVideoBroadcaster != nil || subAudioBroadcaster != nil if config.Capture.ForwardWebRTC == "true" { } else { log.Log.Info("cloud.HandleLiveStreamHD(): Waiting for peer connections.") for handshake := range communication.HandleLiveHDHandshake { - log.Log.Info("cloud.HandleLiveStreamHD(): setting up a peer connection.") + // Route each viewer to the main or sub broadcasters based on the + // quality it requested; "auto" prefers the sub stream when one is + // available, matching the historical default. + useSub := models.SelectSubStreamForQuality(config, handshake.Payload.Quality, subStreamEnabled && subBroadcastersReady) + videoBroadcaster := mainVideoBroadcaster + audioBroadcaster := mainAudioBroadcaster + streamLabel := "main" + if useSub { + videoBroadcaster = subVideoBroadcaster + audioBroadcaster = subAudioBroadcaster + streamLabel = "sub" + } + log.Log.Info("cloud.HandleLiveStreamHD(): setting up a peer connection on the " + streamLabel + " stream (quality=" + handshake.Payload.Quality + ").") go webrtc.InitializeWebRTCConnection(configuration, communication, mqttClient, videoBroadcaster, audioBroadcaster, handshake) } } diff --git a/machinery/src/cloud/livehls.go b/machinery/src/cloud/livehls.go index b975211..e3ebb56 100644 --- a/machinery/src/cloud/livehls.go +++ b/machinery/src/cloud/livehls.go @@ -6,7 +6,6 @@ import ( mqtt "github.com/eclipse/paho.mqtt.golang" - "github.com/kerberos-io/agent/machinery/src/capture" "github.com/kerberos-io/agent/machinery/src/cloud/livehls" "github.com/kerberos-io/agent/machinery/src/log" "github.com/kerberos-io/agent/machinery/src/models" @@ -46,7 +45,7 @@ const hlsReadyReannounceSeconds = 2 // trades a little idle CPU for a near-instant "requesting stream", so viewers no // longer wait a full GOP for the first segment to be cut. Set // AGENT_LIVE_HLS_PREWARM=false to fall back to the lazy on-demand path above. -func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, _ capture.RTSPClient) { +func HandleLiveStreamHLS(configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, subStreamEnabled bool) { log.Log.Debug("cloud.HandleLiveStreamHLS(): started") @@ -81,10 +80,16 @@ func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *m DeviceKey: deviceId, }) + // The live session can be served from the main (high-resolution) or sub + // (low-resolution) stream and switched on demand. requestedQuality tracks the + // latest tier asked for over the keepalive; source holds the cursor plus the + // encoded parameter sets/dimensions for the stream currently being muxed. // Encoded dimensions are only needed for the avcC fallback path (an SPS that - // mp4ff's strict parser rejects); the main stream dimensions are a safe value. - width := uint16(config.Capture.IPCamera.Width) - height := uint16(config.Capture.IPCamera.Height) + // mp4ff's strict parser rejects). + requestedQuality := models.StreamQualityAuto + useSub := models.SelectSubStreamForQuality(config, requestedQuality, subStreamEnabled) + source := buildHLSSource(config, communication, useSub) + log.Log.Info("cloud.HandleLiveStreamHLS(): serving live HLS from the " + source.label + " stream") // prewarm keeps a single long-lived session muxing into an in-memory ring // buffer while idle and flushes it the instant a viewer arrives, eliminating @@ -117,12 +122,15 @@ func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *m var pkt packets.Packet for cursorError == nil { - pkt, cursorError = livestreamCursor.ReadPacket() + pkt, cursorError = source.cursor.ReadPacket() now := time.Now().Unix() select { - case <-communication.HandleLiveHLS: + case q := <-communication.HandleLiveHLS: lastViewerRequest = now + if q != "" { + requestedQuality = q + } // A keepalive may come from a viewer that just connected or hard- // refreshed and therefore missed the one-shot readiness announcement // fired when this session's first segment landed. Re-announce (throttled) @@ -138,6 +146,22 @@ func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *m default: } + // Switch the source stream when the requested quality now maps to the other + // stream. Tearing the current session down makes the producer rebuild the + // init segment and announce a fresh session id from the new stream, which the + // viewer re-attaches to. + if wantSub := models.SelectSubStreamForQuality(config, requestedQuality, subStreamEnabled); wantSub != useSub { + useSub = wantSub + if session != nil { + _ = session.Close() + session = nil + } + source = buildHLSSource(config, communication, useSub) + lastReadyAnnounce = 0 + log.Log.Info("cloud.HandleLiveStreamHLS(): switched live HLS to the " + source.label + " stream (quality=" + requestedQuality + ")") + continue + } + viewerActive := now-lastViewerRequest <= hlsViewerTimeoutSeconds if prewarm { @@ -151,11 +175,11 @@ func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *m } session = livehls.NewSession(publisher, livehls.SessionOptions{ Codec: pkt.Codec, - SPSNALUs: config.Capture.IPCamera.SPSNALUs, - PPSNALUs: config.Capture.IPCamera.PPSNALUs, - VPSNALUs: config.Capture.IPCamera.VPSNALUs, - Width: width, - Height: height, + SPSNALUs: source.sps, + PPSNALUs: source.pps, + VPSNALUs: source.vps, + Width: source.width, + Height: source.height, PartTargetMs: partTargetMs, StartBuffering: true, }) @@ -211,11 +235,11 @@ func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *m } session = livehls.NewSession(publisher, livehls.SessionOptions{ Codec: pkt.Codec, - SPSNALUs: config.Capture.IPCamera.SPSNALUs, - PPSNALUs: config.Capture.IPCamera.PPSNALUs, - VPSNALUs: config.Capture.IPCamera.VPSNALUs, - Width: width, - Height: height, + SPSNALUs: source.sps, + PPSNALUs: source.pps, + VPSNALUs: source.vps, + Width: source.width, + Height: source.height, PartTargetMs: partTargetMs, }) session.SetOnReady(func(sessionID string) { @@ -259,3 +283,45 @@ func publishHLSReady(configuration *models.Configuration, mqttClient mqtt.Client log.Log.Error("cloud.HandleLiveStreamHLS(): failed to package receive-hls-ready message: " + err.Error()) } } + +// hlsStreamSource bundles everything the live HLS producer needs to mux one of +// the camera's streams: the packet cursor it reads from plus the encoded +// parameter sets and dimensions used to build that stream's init segment. +type hlsStreamSource struct { + cursor *packets.QueueCursor + sps [][]byte + pps [][]byte + vps [][]byte + width uint16 + height uint16 + label string +} + +// buildHLSSource resolves the packet cursor and encoded parameter sets/dimensions +// for the selected stream. useSub picks the sub (low-resolution) stream when one +// is available; otherwise the main (high-resolution) stream is used. A fresh +// Latest() cursor is created so muxing resumes from the live edge of the chosen +// stream after a switch. +func buildHLSSource(config models.Config, communication *models.Communication, useSub bool) hlsStreamSource { + cam := config.Capture.IPCamera + if useSub && communication.SubQueue != nil { + return hlsStreamSource{ + cursor: communication.SubQueue.Latest(), + sps: cam.SubSPSNALUs, + pps: cam.SubPPSNALUs, + vps: cam.SubVPSNALUs, + width: uint16(cam.SubWidth), + height: uint16(cam.SubHeight), + label: "sub", + } + } + return hlsStreamSource{ + cursor: communication.Queue.Latest(), + sps: cam.SPSNALUs, + pps: cam.PPSNALUs, + vps: cam.VPSNALUs, + width: uint16(cam.Width), + height: uint16(cam.Height), + label: "main", + } +} diff --git a/machinery/src/components/kerberos.go b/machinery/src/components/kerberos.go index 8689257..f3bcde6 100644 --- a/machinery/src/components/kerberos.go +++ b/machinery/src/components/kerberos.go @@ -73,7 +73,7 @@ func Bootstrap(ctx context.Context, configDirectory string, configuration *model communication.HandleLiveSDHTTP = make(chan int64, 1) communication.HandleLiveHDKeepalive = make(chan string, 1) communication.HandleLiveHDPeers = make(chan string, 1) - communication.HandleLiveHLS = make(chan int64, 1) + communication.HandleLiveHLS = make(chan string, 1) communication.IsConfiguring = abool.New() cameraSettings := &models.Camera{} @@ -237,6 +237,13 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu configuration.Config.Capture.IPCamera.SubWidth = width configuration.Config.Capture.IPCamera.SubHeight = height + // Capture the sub stream parameter sets separately from the main stream so + // the live HLS muxer can build a correct init segment when a viewer asks for + // the sub (low-resolution) stream on demand. + configuration.Config.Capture.IPCamera.SubSPSNALUs = [][]byte{videoSubStream.SPS} + configuration.Config.Capture.IPCamera.SubPPSNALUs = [][]byte{videoSubStream.PPS} + configuration.Config.Capture.IPCamera.SubVPSNALUs = [][]byte{videoSubStream.VPS} + // If we have a substream, we need to set the width and height of the substream. (so we will override above information) // Set the liveview width and height, this is used for the liveview and motion regions (drawing on the hub). configuration.Config.Capture.IPCamera.BaseWidth, configuration.Config.Capture.IPCamera.BaseHeight = @@ -287,26 +294,19 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu } // Handle livestream HLS (adaptive segments over HTTP via hub-api -> vault). - // Uses the sub stream when available (lower bitrate, browser-friendly), else - // the main stream. Like SD it is viewer-keepalive gated and produces no - // traffic while nobody is watching. - if subStreamEnabled { - livestreamHLSCursor := subQueue.Latest() - go cloud.HandleLiveStreamHLS(livestreamHLSCursor, configuration, communication, mqttClient, rtspSubClient) - } else { - livestreamHLSCursor := queue.Latest() - go cloud.HandleLiveStreamHLS(livestreamHLSCursor, configuration, communication, mqttClient, rtspClient) - } + // The producer can serve either the main (high-resolution) or sub + // (low-resolution) stream and switches between them on demand based on the + // quality the viewer requests; "auto" prefers the sub stream when available. + // Like SD it is viewer-keepalive gated and produces no traffic while nobody is + // watching. + go cloud.HandleLiveStreamHLS(configuration, communication, mqttClient, subStreamEnabled) - // Handle livestream HD (high resolution over WEBRTC) + // Handle livestream HD (high resolution over WEBRTC). Both the main and sub + // stream are exposed as separate broadcasters so a viewer can request the + // high (main) or low (sub) resolution per peer connection; "auto" prefers the + // sub stream when available. communication.HandleLiveHDHandshake = make(chan models.LiveHDHandshake, 100) - if subStreamEnabled { - livestreamHDCursor := subQueue.Latest() - go cloud.HandleLiveStreamHD(livestreamHDCursor, configuration, communication, mqttClient, rtspSubClient) - } else { - livestreamHDCursor := queue.Latest() - go cloud.HandleLiveStreamHD(livestreamHDCursor, configuration, communication, mqttClient, rtspClient) - } + go cloud.HandleLiveStreamHD(configuration, communication, mqttClient, rtspClient, rtspSubClient, subStreamEnabled) // Handle recording, will write an mp4 to disk. go capture.HandleRecordStream(queue, configDirectory, configuration, communication, rtspClient) diff --git a/machinery/src/models/communication.go b/machinery/src/models/communication.go index 1eafd4c..cfb39cc 100644 --- a/machinery/src/models/communication.go +++ b/machinery/src/models/communication.go @@ -41,7 +41,10 @@ type Communication struct { HandleLiveHDKeepalive chan string HandleLiveHDHandshake chan LiveHDHandshake HandleLiveHDPeers chan string - HandleLiveHLS chan int64 + // HandleLiveHLS is the live HLS viewer keepalive. It carries the requested + // quality tier ("auto"|"high"|"low"; empty => auto) so the producer can switch + // the live session between the main and sub stream on demand. + HandleLiveHLS chan string HandleONVIF chan OnvifAction IsConfiguring *abool.AtomicBool Queue *packets.Queue diff --git a/machinery/src/models/config.go b/machinery/src/models/config.go index 3759120..19d0cfb 100644 --- a/machinery/src/models/config.go +++ b/machinery/src/models/config.go @@ -99,8 +99,14 @@ type IPCamera struct { SPSNALUs [][]byte `json:"sps_nalus,omitempty" bson:"sps_nalus,omitempty"` PPSNALUs [][]byte `json:"pps_nalus,omitempty" bson:"pps_nalus,omitempty"` VPSNALUs [][]byte `json:"vps_nalus,omitempty" bson:"vps_nalus,omitempty"` - SampleRate int `json:"sample_rate,omitempty" bson:"sample_rate,omitempty"` - Channels int `json:"channels,omitempty" bson:"channels,omitempty"` + // Sub stream parameter sets, captured separately from the main stream so the + // live HLS muxer can build a correct init segment when a viewer switches the + // live view to the sub (low-resolution) stream. + SubSPSNALUs [][]byte `json:"sub_sps_nalus,omitempty" bson:"sub_sps_nalus,omitempty"` + SubPPSNALUs [][]byte `json:"sub_pps_nalus,omitempty" bson:"sub_pps_nalus,omitempty"` + SubVPSNALUs [][]byte `json:"sub_vps_nalus,omitempty" bson:"sub_vps_nalus,omitempty"` + SampleRate int `json:"sample_rate,omitempty" bson:"sample_rate,omitempty"` + Channels int `json:"channels,omitempty" bson:"channels,omitempty"` } // USBCamera configuration, such as the device path (/dev/video*) diff --git a/machinery/src/models/mqtt.go b/machinery/src/models/mqtt.go index f5ec05d..9d7465b 100644 --- a/machinery/src/models/mqtt.go +++ b/machinery/src/models/mqtt.go @@ -179,11 +179,26 @@ type RequestSDStreamPayload struct { Transport string `json:"transport,omitempty"` } +// Stream quality tiers a viewer can request for the live (HD) view. The agent +// maps these onto the camera's main (high-resolution) or sub (low-resolution) +// RTSP stream, so a viewer can pick the resolution it needs instead of the agent +// always preferring the sub stream. Empty/unknown values are treated as "auto" +// for backward compatibility: older frontends that never set a quality keep the +// previous behaviour (sub stream when available, otherwise main). +const ( + StreamQualityAuto = "auto" // agent decides based on availability/resolution + StreamQualityHigh = "high" // main stream (highest resolution) + StreamQualityLow = "low" // sub stream (lowest resolution) +) + // We received a live HLS stream request. Like SD it is a simple viewer // keepalive: the agent owns the live HLS session, so the request only needs to -// signal "a viewer is watching" to keep the segment pipeline alive. +// signal "a viewer is watching" to keep the segment pipeline alive. Quality lets +// the viewer ask for the main (high) or sub (low) stream on demand; the agent +// switches the live session's source stream when it changes. type RequestHLSStreamPayload struct { - Timestamp int64 `json:"timestamp"` // timestamp + Timestamp int64 `json:"timestamp"` // timestamp + Quality string `json:"quality,omitempty"` // "auto" | "high" | "low" (empty => auto) } // We received a request HD stream request @@ -192,6 +207,7 @@ type RequestHDStreamPayload struct { HubKey string `json:"hub_key"` // hub key SessionID string `json:"session_id"` // session id SessionDescription string `json:"session_description"` // session description + Quality string `json:"quality,omitempty"` // "auto" | "high" | "low" (empty => auto) } // We received a receive HD candidates request diff --git a/machinery/src/models/stream.go b/machinery/src/models/stream.go new file mode 100644 index 0000000..3bd32e5 --- /dev/null +++ b/machinery/src/models/stream.go @@ -0,0 +1,40 @@ +package models + +// SelectSubStreamForQuality decides whether the live (HD) view should be served +// from the sub (secondary) RTSP stream for the requested quality tier. +// +// It is resolution-aware: "high" picks whichever configured stream has the +// higher resolution and "low" whichever has the lower resolution, regardless of +// which one is wired as main vs sub. "auto" — the default, also used for the +// empty/unknown value sent by older frontends that never set a quality — keeps +// the historical behaviour of preferring the sub stream when one is available +// (lower bitrate, browser friendly), falling back to the main stream otherwise. +// +// When no sub stream is configured the main stream is always used. +func SelectSubStreamForQuality(config Config, quality string, subStreamEnabled bool) bool { + if !subStreamEnabled { + return false + } + + cam := config.Capture.IPCamera + mainPixels := cam.Width * cam.Height + subPixels := cam.SubWidth * cam.SubHeight + + switch quality { + case StreamQualityHigh: + // Highest resolution available. If the sub stream is (unusually) larger, + // use it; otherwise use the main stream. When dimensions are not yet known + // (0), default to the main stream for "high". + return subPixels > mainPixels + case StreamQualityLow: + // Lowest resolution available. If the main stream is (unusually) the + // smaller of the two, use it; otherwise use the sub stream. When the sub + // dimensions are unknown, still prefer the sub stream for "low". + if mainPixels > 0 && subPixels > 0 && mainPixels < subPixels { + return false + } + return true + default: // StreamQualityAuto, empty, or any unknown value + return true + } +} diff --git a/machinery/src/models/stream_test.go b/machinery/src/models/stream_test.go new file mode 100644 index 0000000..3e34750 --- /dev/null +++ b/machinery/src/models/stream_test.go @@ -0,0 +1,53 @@ +package models + +import "testing" + +func cfgWithDims(mainW, mainH, subW, subH int) Config { + c := Config{} + c.Capture.IPCamera.Width = mainW + c.Capture.IPCamera.Height = mainH + c.Capture.IPCamera.SubWidth = subW + c.Capture.IPCamera.SubHeight = subH + return c +} + +func TestSelectSubStreamForQuality(t *testing.T) { + tests := []struct { + name string + config Config + quality string + subStreamEnabled bool + wantSub bool + }{ + // No sub stream configured -> always the main stream. + {"no sub, auto", cfgWithDims(1920, 1080, 0, 0), StreamQualityAuto, false, false}, + {"no sub, high", cfgWithDims(1920, 1080, 0, 0), StreamQualityHigh, false, false}, + {"no sub, low", cfgWithDims(1920, 1080, 0, 0), StreamQualityLow, false, false}, + + // Typical config: main is the bigger stream, sub the smaller one. + {"auto prefers sub", cfgWithDims(1920, 1080, 640, 480), StreamQualityAuto, true, true}, + {"empty prefers sub", cfgWithDims(1920, 1080, 640, 480), "", true, true}, + {"unknown prefers sub", cfgWithDims(1920, 1080, 640, 480), "potato", true, true}, + {"high picks main", cfgWithDims(1920, 1080, 640, 480), StreamQualityHigh, true, false}, + {"low picks sub", cfgWithDims(1920, 1080, 640, 480), StreamQualityLow, true, true}, + + // Dimensions not probed yet (0): high defaults to main, low/auto to sub. + {"unknown dims, high", cfgWithDims(0, 0, 0, 0), StreamQualityHigh, true, false}, + {"unknown dims, low", cfgWithDims(0, 0, 0, 0), StreamQualityLow, true, true}, + {"unknown dims, auto", cfgWithDims(0, 0, 0, 0), StreamQualityAuto, true, true}, + + // Inverted config: sub is (unusually) the higher-resolution stream. + {"inverted high picks sub", cfgWithDims(640, 480, 1920, 1080), StreamQualityHigh, true, true}, + {"inverted low picks main", cfgWithDims(640, 480, 1920, 1080), StreamQualityLow, true, false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := SelectSubStreamForQuality(tt.config, tt.quality, tt.subStreamEnabled) + if got != tt.wantSub { + t.Errorf("SelectSubStreamForQuality(quality=%q, subEnabled=%v) = %v, want %v", + tt.quality, tt.subStreamEnabled, got, tt.wantSub) + } + }) + } +} diff --git a/machinery/src/routers/mqtt/main.go b/machinery/src/routers/mqtt/main.go index 6fd81b1..4ee921e 100644 --- a/machinery/src/routers/mqtt/main.go +++ b/machinery/src/routers/mqtt/main.go @@ -585,8 +585,11 @@ func HandleRequestHLSStream(mqttClient mqtt.Client, hubKey string, payload model if requestHLSStreamPayload.Timestamp != 0 { if communication.CameraConnected { + // Forward the requested quality ("auto"|"high"|"low"; empty => auto) so + // the producer can switch the live session between the main and sub + // stream on demand. The send doubles as the viewer keepalive. select { - case communication.HandleLiveHLS <- time.Now().Unix(): + case communication.HandleLiveHLS <- requestHLSStreamPayload.Quality: default: } log.Log.Info("routers.mqtt.main.HandleRequestHLSStream(): received request to livestream over HLS.")