From bc516bce7ddf45440791daacd60048e5d520c063 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ren=C3=A9=20Simonsen?= Date: Wed, 3 Jan 2024 15:08:21 +0100 Subject: [PATCH 1/3] Adds automatic extention of nest stream before it expires. --- pkg/nest/api.go | 62 ++++++++++++++++++++++++++++++++++++++++------ pkg/nest/client.go | 39 ++++++++++++++++++++++++++--- 2 files changed, 91 insertions(+), 10 deletions(-) diff --git a/pkg/nest/api.go b/pkg/nest/api.go index 9c7f4546..e260799e 100644 --- a/pkg/nest/api.go +++ b/pkg/nest/api.go @@ -121,7 +121,7 @@ func (a *API) GetDevices(projectID string) (map[string]string, error) { return devices, nil } -func (a *API) ExchangeSDP(projectID, deviceID, offer string) (string, error) { +func (a *API) ExchangeSDP(projectID, deviceID, offer string) (string, string, time.Time, error) { var reqv struct { Command string `json:"command"` Params struct { @@ -133,14 +133,14 @@ func (a *API) ExchangeSDP(projectID, deviceID, offer string) (string, error) { b, err := json.Marshal(reqv) if err != nil { - return "", err + return "", "", time.Time{}, err } uri := "https://smartdevicemanagement.googleapis.com/v1/enterprises/" + projectID + "/devices/" + deviceID + ":executeCommand" req, err := http.NewRequest("POST", uri, bytes.NewReader(b)) if err != nil { - return "", err + return "", "", time.Time{}, err } req.Header.Set("Authorization", "Bearer "+a.Token) @@ -148,11 +148,11 @@ func (a *API) ExchangeSDP(projectID, deviceID, offer string) (string, error) { client := &http.Client{Timeout: time.Second * 5000} res, err := client.Do(req) if err != nil { - return "", err + return "", "", time.Time{}, err } if res.StatusCode != 200 { - return "", errors.New("nest: wrong status: " + res.Status) + return "", "", time.Time{}, errors.New("nest: wrong status: " + res.Status) } var resv struct { @@ -164,10 +164,58 @@ func (a *API) ExchangeSDP(projectID, deviceID, offer string) (string, error) { } if err = json.NewDecoder(res.Body).Decode(&resv); err != nil { - return "", err + return "", "", time.Time{}, err } - return resv.Results.Answer, nil + return resv.Results.Answer, resv.Results.MediaSessionId, resv.Results.ExpiresAt, nil +} + +func (a *API) ExtendStream(projectID, deviceID, mediaSessionID string) (string, time.Time, error) { + var reqv struct { + Command string `json:"command"` + Params struct { + MediaSessionID string `json:"mediaSessionId"` + } `json:"params"` + } + reqv.Command = "sdm.devices.commands.CameraLiveStream.ExtendWebRtcStream" + reqv.Params.MediaSessionID = mediaSessionID + + b, err := json.Marshal(reqv) + if err != nil { + return "", time.Time{}, err + } + + uri := "https://smartdevicemanagement.googleapis.com/v1/enterprises/" + + projectID + "/devices/" + deviceID + ":executeCommand" + req, err := http.NewRequest("POST", uri, bytes.NewReader(b)) + if err != nil { + return "", time.Time{}, err + } + + req.Header.Set("Authorization", "Bearer "+a.Token) + + client := &http.Client{Timeout: time.Second * 5000} + res, err := client.Do(req) + if err != nil { + return "", time.Time{}, err + } + + if res.StatusCode != 200 { + return "", time.Time{}, errors.New("nest: wrong status: " + res.Status) + } + + var resv struct { + Results struct { + ExpiresAt time.Time `json:"expiresAt"` + MediaSessionId string `json:"mediaSessionId"` + } `json:"results"` + } + + if err = json.NewDecoder(res.Body).Decode(&resv); err != nil { + return "", time.Time{}, err + } + + return resv.Results.MediaSessionId, resv.Results.ExpiresAt, nil } type Device struct { diff --git a/pkg/nest/client.go b/pkg/nest/client.go index b2b0c964..6beabd81 100644 --- a/pkg/nest/client.go +++ b/pkg/nest/client.go @@ -4,13 +4,21 @@ import ( "errors" "net/url" + "time" + "github.com/AlexxIT/go2rtc/pkg/core" "github.com/AlexxIT/go2rtc/pkg/webrtc" pion "github.com/pion/webrtc/v3" ) type Client struct { - conn *webrtc.Conn + conn *webrtc.Conn + projectId string + deviceId string + mediaSessionId string + streamExpiresAt time.Time + nestApi *API + timer *time.Timer } func NewClient(rawURL string) (*Client, error) { @@ -64,7 +72,7 @@ func NewClient(rawURL string) (*Client, error) { } // 4. Exchange SDP via Hass - answer, err := nestAPI.ExchangeSDP(projectID, deviceID, offer) + answer, mediaSessionId, expiresAt, err := nestAPI.ExchangeSDP(projectID, deviceID, offer) if err != nil { return nil, err } @@ -74,7 +82,7 @@ func NewClient(rawURL string) (*Client, error) { return nil, err } - return &Client{conn: conn}, nil + return &Client{conn: conn, deviceId: deviceID, projectId: projectID, mediaSessionId: mediaSessionId, streamExpiresAt: expiresAt, nestApi: nestAPI}, nil } func (c *Client) GetMedias() []*core.Media { @@ -90,10 +98,35 @@ func (c *Client) AddTrack(media *core.Media, codec *core.Codec, track *core.Rece } func (c *Client) Start() error { + c.StartExtendStreamTimer() + return c.conn.Start() } +func (c *Client) StartExtendStreamTimer() { + ontimer := func() { + c.ExtendStream() + c.StartExtendStreamTimer() + } + // Calculate the duration until 30 seconds before the stream expires + duration := time.Until(c.streamExpiresAt.Add(-30 * time.Second)) + + // Start the timer + c.timer = time.AfterFunc(duration, ontimer) +} + +func (c *Client) ExtendStream() error { + mediaSessionId, expiresAt, err := c.nestApi.ExtendStream(c.projectId, c.deviceId, c.mediaSessionId) + if err != nil { + return err + } + c.mediaSessionId = mediaSessionId + c.streamExpiresAt = expiresAt + return nil +} + func (c *Client) Stop() error { + c.timer.Stop() return c.conn.Stop() } From db190e69ed9a7d6a85313f57ee9c43897b765db5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ren=C3=A9=20Simonsen?= Date: Wed, 3 Jan 2024 15:16:59 +0100 Subject: [PATCH 2/3] Updated README with more accurate information regarding nest integration. --- README.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 2cfa2ed8..367f9048 100644 --- a/README.md +++ b/README.md @@ -579,7 +579,8 @@ streams: Any cameras in WebRTC format are supported. But at the moment Home Assistant only supports some [Nest](https://www.home-assistant.io/integrations/nest/) cameras in this fomat. -The Nest API only allows you to get a link to a stream for 5 minutes. So every 5 minutes the stream will be reconnected. +The Nest API only allows you to get a link to a stream for 5 minutes, but a call to extend the stream is made before it expires. If for some reason the stream expires anyway, any streaming clients will lose connection. +Note: Do not use this with frigate. If the stream expires, Frigate will consume all available ram on your machine within seconds. ```yaml streams: From 439319141bbe887d7090a9d839a3f73d9649382a Mon Sep 17 00:00:00 2001 From: Alex X Date: Sun, 21 Apr 2024 07:46:59 +0300 Subject: [PATCH 3/3] Code refactoring after #855 --- README.md | 5 ++-- pkg/nest/api.go | 68 ++++++++++++++++++++++++++++++++++------------ pkg/nest/client.go | 42 ++++------------------------ 3 files changed, 58 insertions(+), 57 deletions(-) diff --git a/README.md b/README.md index 367f9048..bb8bde8a 100644 --- a/README.md +++ b/README.md @@ -579,8 +579,7 @@ streams: Any cameras in WebRTC format are supported. But at the moment Home Assistant only supports some [Nest](https://www.home-assistant.io/integrations/nest/) cameras in this fomat. -The Nest API only allows you to get a link to a stream for 5 minutes, but a call to extend the stream is made before it expires. If for some reason the stream expires anyway, any streaming clients will lose connection. -Note: Do not use this with frigate. If the stream expires, Frigate will consume all available ram on your machine within seconds. +**Important.** The Nest API only allows you to get a link to a stream for 5 minutes. Do not use this with Frigate! If the stream expires, Frigate will consume all available ram on your machine within seconds. It's recommended to use [Nest source](#source-nest) - it supports extending the stream. ```yaml streams: @@ -611,7 +610,7 @@ streams: *[New in v1.6.0](https://github.com/AlexxIT/go2rtc/releases/tag/v1.6.0)* -Currently only WebRTC cameras are supported. Stream reconnects every 5 minutes. +Currently only WebRTC cameras are supported. For simplicity, it is recommended to connect the Nest/WebRTC camera to the [Home Assistant](#source-hass). But if you can somehow get the below parameters - Nest/WebRTC source will work without Hass. diff --git a/pkg/nest/api.go b/pkg/nest/api.go index e260799e..035f1546 100644 --- a/pkg/nest/api.go +++ b/pkg/nest/api.go @@ -14,6 +14,13 @@ import ( type API struct { Token string ExpiresAt time.Time + + StreamProjectID string + StreamDeviceID string + StreamSessionID string + StreamExpiresAt time.Time + + extendTimer *time.Timer } type Auth struct { @@ -121,7 +128,7 @@ func (a *API) GetDevices(projectID string) (map[string]string, error) { return devices, nil } -func (a *API) ExchangeSDP(projectID, deviceID, offer string) (string, string, time.Time, error) { +func (a *API) ExchangeSDP(projectID, deviceID, offer string) (string, error) { var reqv struct { Command string `json:"command"` Params struct { @@ -133,14 +140,14 @@ func (a *API) ExchangeSDP(projectID, deviceID, offer string) (string, string, ti b, err := json.Marshal(reqv) if err != nil { - return "", "", time.Time{}, err + return "", err } uri := "https://smartdevicemanagement.googleapis.com/v1/enterprises/" + projectID + "/devices/" + deviceID + ":executeCommand" req, err := http.NewRequest("POST", uri, bytes.NewReader(b)) if err != nil { - return "", "", time.Time{}, err + return "", err } req.Header.Set("Authorization", "Bearer "+a.Token) @@ -148,29 +155,34 @@ func (a *API) ExchangeSDP(projectID, deviceID, offer string) (string, string, ti client := &http.Client{Timeout: time.Second * 5000} res, err := client.Do(req) if err != nil { - return "", "", time.Time{}, err + return "", err } if res.StatusCode != 200 { - return "", "", time.Time{}, errors.New("nest: wrong status: " + res.Status) + return "", errors.New("nest: wrong status: " + res.Status) } var resv struct { Results struct { Answer string `json:"answerSdp"` ExpiresAt time.Time `json:"expiresAt"` - MediaSessionId string `json:"mediaSessionId"` + MediaSessionID string `json:"mediaSessionId"` } `json:"results"` } if err = json.NewDecoder(res.Body).Decode(&resv); err != nil { - return "", "", time.Time{}, err + return "", err } - return resv.Results.Answer, resv.Results.MediaSessionId, resv.Results.ExpiresAt, nil + a.StreamProjectID = projectID + a.StreamDeviceID = deviceID + a.StreamSessionID = resv.Results.MediaSessionID + a.StreamExpiresAt = resv.Results.ExpiresAt + + return resv.Results.Answer, nil } -func (a *API) ExtendStream(projectID, deviceID, mediaSessionID string) (string, time.Time, error) { +func (a *API) ExtendStream() error { var reqv struct { Command string `json:"command"` Params struct { @@ -178,18 +190,18 @@ func (a *API) ExtendStream(projectID, deviceID, mediaSessionID string) (string, } `json:"params"` } reqv.Command = "sdm.devices.commands.CameraLiveStream.ExtendWebRtcStream" - reqv.Params.MediaSessionID = mediaSessionID + reqv.Params.MediaSessionID = a.StreamSessionID b, err := json.Marshal(reqv) if err != nil { - return "", time.Time{}, err + return err } uri := "https://smartdevicemanagement.googleapis.com/v1/enterprises/" + - projectID + "/devices/" + deviceID + ":executeCommand" + a.StreamProjectID + "/devices/" + a.StreamDeviceID + ":executeCommand" req, err := http.NewRequest("POST", uri, bytes.NewReader(b)) if err != nil { - return "", time.Time{}, err + return err } req.Header.Set("Authorization", "Bearer "+a.Token) @@ -197,25 +209,28 @@ func (a *API) ExtendStream(projectID, deviceID, mediaSessionID string) (string, client := &http.Client{Timeout: time.Second * 5000} res, err := client.Do(req) if err != nil { - return "", time.Time{}, err + return err } if res.StatusCode != 200 { - return "", time.Time{}, errors.New("nest: wrong status: " + res.Status) + return errors.New("nest: wrong status: " + res.Status) } var resv struct { Results struct { ExpiresAt time.Time `json:"expiresAt"` - MediaSessionId string `json:"mediaSessionId"` + MediaSessionID string `json:"mediaSessionId"` } `json:"results"` } if err = json.NewDecoder(res.Body).Decode(&resv); err != nil { - return "", time.Time{}, err + return err } - return resv.Results.MediaSessionId, resv.Results.ExpiresAt, nil + a.StreamSessionID = resv.Results.MediaSessionID + a.StreamExpiresAt = resv.Results.ExpiresAt + + return nil } type Device struct { @@ -251,3 +266,20 @@ type Device struct { // DisplayName string `json:"displayName"` //} `json:"parentRelations"` } + +func (a *API) StartExtendStreamTimer() { + // Calculate the duration until 30 seconds before the stream expires + duration := time.Until(a.StreamExpiresAt.Add(-30 * time.Second)) + a.extendTimer = time.AfterFunc(duration, func() { + if err := a.ExtendStream(); err != nil { + return + } + duration = time.Until(a.StreamExpiresAt.Add(-30 * time.Second)) + a.extendTimer.Reset(duration) + }) + +} + +func (a *API) StopExtendStreamTimer() { + a.extendTimer.Stop() +} diff --git a/pkg/nest/client.go b/pkg/nest/client.go index 6beabd81..cb73cc98 100644 --- a/pkg/nest/client.go +++ b/pkg/nest/client.go @@ -4,21 +4,14 @@ import ( "errors" "net/url" - "time" - "github.com/AlexxIT/go2rtc/pkg/core" "github.com/AlexxIT/go2rtc/pkg/webrtc" pion "github.com/pion/webrtc/v3" ) type Client struct { - conn *webrtc.Conn - projectId string - deviceId string - mediaSessionId string - streamExpiresAt time.Time - nestApi *API - timer *time.Timer + conn *webrtc.Conn + api *API } func NewClient(rawURL string) (*Client, error) { @@ -72,7 +65,7 @@ func NewClient(rawURL string) (*Client, error) { } // 4. Exchange SDP via Hass - answer, mediaSessionId, expiresAt, err := nestAPI.ExchangeSDP(projectID, deviceID, offer) + answer, err := nestAPI.ExchangeSDP(projectID, deviceID, offer) if err != nil { return nil, err } @@ -82,7 +75,7 @@ func NewClient(rawURL string) (*Client, error) { return nil, err } - return &Client{conn: conn, deviceId: deviceID, projectId: projectID, mediaSessionId: mediaSessionId, streamExpiresAt: expiresAt, nestApi: nestAPI}, nil + return &Client{conn: conn, api: nestAPI}, nil } func (c *Client) GetMedias() []*core.Media { @@ -98,35 +91,12 @@ func (c *Client) AddTrack(media *core.Media, codec *core.Codec, track *core.Rece } func (c *Client) Start() error { - c.StartExtendStreamTimer() - + c.api.StartExtendStreamTimer() return c.conn.Start() } -func (c *Client) StartExtendStreamTimer() { - ontimer := func() { - c.ExtendStream() - c.StartExtendStreamTimer() - } - // Calculate the duration until 30 seconds before the stream expires - duration := time.Until(c.streamExpiresAt.Add(-30 * time.Second)) - - // Start the timer - c.timer = time.AfterFunc(duration, ontimer) -} - -func (c *Client) ExtendStream() error { - mediaSessionId, expiresAt, err := c.nestApi.ExtendStream(c.projectId, c.deviceId, c.mediaSessionId) - if err != nil { - return err - } - c.mediaSessionId = mediaSessionId - c.streamExpiresAt = expiresAt - return nil -} - func (c *Client) Stop() error { - c.timer.Stop() + c.api.StopExtendStreamTimer() return c.conn.Stop() }