Compare commits

..

63 Commits

Author SHA1 Message Date
Alex X cd777ba2b4 Update version to 1.9.3 2024-06-06 16:01:01 +03:00
Alex X e3188a0a6d Update docs about config 2024-06-06 15:21:32 +03:00
Alex X 2bab0a014d Update dependencies 2024-06-06 14:34:16 +03:00
Alex X a01da18018 Merge pull request #1150 from skrashevich/go122
update Go version to 1.22
2024-06-06 14:25:27 +03:00
Alex X 9d5a5c1e45 Merge remote-tracking branch 'origin/master' 2024-06-06 14:15:20 +03:00
Alex X 8377ad1d05 Update codec section in stream info 2024-06-06 13:16:12 +03:00
Alex X ec33796bd3 Add goweight to useful commands 2024-06-05 20:02:10 +03:00
Alex X 31e4ba2722 Rewrite Receiver/Sender classes 2024-06-05 20:01:47 +03:00
Alex X e0b1a50356 Add rtsp_client for testing ghost exec process 2024-06-05 20:00:41 +03:00
Alex X 9bb36ebb6c Fix ghost exec/ffmpeg process 2024-06-05 19:59:22 +03:00
Alex X 756be9801e Code refactoring for app module 2024-06-02 07:00:29 +03:00
Alex X bd73b07ed8 Merge pull request #1147 from skrashevich/docker-ghcr-repo
ci(workflow): add GitHub Container Registry
2024-05-31 07:22:47 +03:00
Sergey Krashevich df1d44d24e chore(deps): update Go version to 1.22 across project files 2024-05-30 17:12:56 +03:00
Sergey Krashevich 79245eeff4 fix(ci): skip GitHub Container Registry login on pull requests 2024-05-30 11:48:15 +03:00
Sergey Krashevich aa86c1ec25 ci(workflow): add GitHub Container Registry login and update image paths 2024-05-30 11:28:06 +03:00
Alex X 2ab1d9d774 Add handling if mp4 client drops connection 2024-05-29 17:32:18 +03:00
Alex X a9e7a73cc8 Add video bitrate setting for HomeKit source 2024-05-28 22:57:43 +03:00
Alex X ea17b420d6 Fix two-way audio for webrtc client 2024-05-28 21:36:12 +03:00
Alex X 660979dfda Merge pull request #1141 from skrashevich/feat-log-terminal-check
feat(logging): add interactive shell detection for console output
2024-05-28 13:26:14 +03:00
Alex X a6b9b4993f Code refactoring after #1141 2024-05-28 13:21:33 +03:00
Sergey Krashevich cc74504ed8 feat(shell): add Windows support for TTY detection 2024-05-28 10:19:51 +03:00
Sergey Krashevich 791239be12 Merge branch 'master' into feat-log-terminal-check 2024-05-28 09:15:01 +03:00
Sergey Krashevich a79061c7c2 feat(logging): add interactive shell detection for console output 2024-05-28 09:10:51 +03:00
Alex X 50ad3b20c4 Add config schema.json 2024-05-28 09:08:57 +03:00
Alex X 649de0131c Change logs timestamp format in WebUI 2024-05-27 20:25:09 +03:00
Alex X 8cb513cb89 Add log level for ffmpeg module 2024-05-27 20:24:24 +03:00
Alex X 3932dbaa84 Add print exec stderr to logs for debug level 2024-05-27 20:23:55 +03:00
Alex X 4534b4d8ca Add more log customization options 2024-05-26 21:28:34 +03:00
Alex X 8e571a66e3 Code refactoring for debug packet logger 2024-05-26 00:19:26 +03:00
Alex X 0ccfcb0ec0 Fix timestamps for RTMP client 2024-05-26 00:18:56 +03:00
Alex X 8bae4631d2 Fix support some RTSP servers 2024-05-26 00:18:36 +03:00
Alex X 268629f551 Fix pix_fmt for publishing to RTMP servers 2024-05-25 19:45:29 +03:00
Alex X 0bd2fcde54 Update color index func for ascii stream 2024-05-25 13:52:55 +03:00
Alex X 6f34cf0c95 Add streaming to rawvideo format 2024-05-25 11:55:28 +03:00
Alex X f8bc25d0ae Add support rawvideo format 2024-05-25 08:22:38 +03:00
Alex X 8749562c96 Fix detection webrtc without audio #1106 2024-05-24 20:41:46 +03:00
Alex X d9d2bdff44 Add timeout query param to RTSP incoming source #1118 2024-05-24 16:26:06 +03:00
Alex X b3e9ed23ac Add /api/ffmpeg for playing files and tts on cameras with two-way audio 2024-05-24 15:57:18 +03:00
Alex X bf3f81ccac Update ffmpeg pkg for reading files and parsing ffmpeg version 2024-05-24 11:06:51 +03:00
Alex X ff39e2e496 Change import log for hass module from debug to trace 2024-05-24 11:04:26 +03:00
Alex X d2346a2aed Fix FFmpeg producer codecs 2024-05-24 07:48:44 +03:00
Alex X 8f57b1acb6 Fix TTS template 2024-05-24 07:48:17 +03:00
Alex X 6fafd10482 Add stream source validation for dynamic streams 2024-05-23 17:40:27 +03:00
Alex X c726651b8b Add ffmpeg version checker 2024-05-23 17:31:02 +03:00
Alex X 02af2e2849 Code refactoring for FFmpeg producer 2024-05-23 12:40:29 +03:00
Alex X 6d9c7012b0 Add output/aac for ffmpeg source 2024-05-23 12:24:41 +03:00
Alex X 8a7712a4c8 Add ffmpeg auto codec selection logic 2024-05-22 18:49:43 +03:00
Alex X 82fa803a37 Add ffmpeg virtual tests 2024-05-22 18:48:40 +03:00
Alex X 78a74da8d6 Fix aac.DecodeConfig sampleRate parsing 2024-05-22 18:46:30 +03:00
Alex X 53242ea02f Add ffmpeg tts source 2024-05-22 13:00:39 +03:00
Alex X af05083a1f Code refactoring for ffmpeg device and virtual 2024-05-22 12:58:21 +03:00
Alex X c41bddbbea Add using wav format for ffmpeg transcoding to PCMA/PCMU 2024-05-21 17:50:15 +03:00
Alex X 54c8ca0112 Add wav format to magic producer 2024-05-21 17:48:31 +03:00
Alex X a518488289 Add debug logs for run RTSP pipe 2024-05-21 17:46:43 +03:00
Alex X 99cc21aacb Code refactoring for magic producer 2024-05-20 14:24:04 +03:00
Alex X bc8295baee Improve play audio on RTSP backchannel 2024-05-19 11:56:33 +03:00
Alex X 50f9913c41 Add hls.html 2024-05-19 10:33:11 +03:00
Alex X 4c135b5a46 Add binaries to gitignore 2024-05-19 07:25:50 +03:00
Alex X 686fb374e9 Remove PCMU for two way for DVRIP source #1111 2024-05-18 17:14:55 +03:00
Alex X 2b3e6a2730 Merge pull request #1122 from isegals/master
Update client.go
2024-05-18 17:09:02 +03:00
Alex X 9143729042 Merge pull request #1123 from skrashevich/fix-vcs-tags-in-docker-builds
add git to build stage
2024-05-18 16:20:42 +03:00
Sergey Krashevich 3952f0ba0f add git to build stage 2024-05-18 13:47:02 +03:00
isegals 7a131822db Update client.go
Add  "AudioFormat":{"EncodeType":"G711_ALAW"} to suppoet new firmware
2024-05-18 11:14:16 +03:00
84 changed files with 2691 additions and 671 deletions
+23 -3
View File
@@ -19,7 +19,7 @@ jobs:
- name: Setup Go
uses: actions/setup-go@v5
with: { go-version: '1.21' }
with: { go-version: '1.22' }
- name: Build go2rtc_win64
env: { GOOS: windows, GOARCH: amd64 }
@@ -123,7 +123,9 @@ jobs:
id: meta
uses: docker/metadata-action@v5
with:
images: ${{ github.repository }}
images: |
${{ github.repository }}
ghcr.io/${{ github.repository }}
tags: |
type=ref,event=branch
type=semver,pattern={{version}},enable=false
@@ -142,6 +144,14 @@ jobs:
username: ${{ secrets.DOCKERHUB_USERNAME }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
- name: Login to GitHub Container Registry
if: github.event_name != 'pull_request'
uses: docker/login-action@v3
with:
registry: ghcr.io
username: ${{ github.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
- name: Build and push
uses: docker/build-push-action@v5
with:
@@ -168,7 +178,9 @@ jobs:
id: meta-hw
uses: docker/metadata-action@v5
with:
images: ${{ github.repository }}
images: |
${{ github.repository }}
ghcr.io/${{ github.repository }}
flavor: |
suffix=-hardware,onlatest=true
latest=auto
@@ -189,6 +201,14 @@ jobs:
with:
username: ${{ secrets.DOCKERHUB_USERNAME }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
- name: Login to GitHub Container Registry
if: github.event_name != 'pull_request'
uses: docker/login-action@v3
with:
registry: ghcr.io
username: ${{ github.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
- name: Build and push
uses: docker/build-push-action@v5
+1 -1
View File
@@ -26,7 +26,7 @@ jobs:
- name: Setup Go
uses: actions/setup-go@v5
with:
go-version: '1.21'
go-version: '1.22'
- name: Build Go binary
run: go build -ldflags "-s -w" -trimpath -o ./go2rtc
+4
View File
@@ -4,6 +4,10 @@
go2rtc.yaml
go2rtc.json
go2rtc_linux*
go2rtc_mac*
go2rtc_win*
0_test.go
.DS_Store
+3 -1
View File
@@ -2,7 +2,7 @@
# 0. Prepare images
ARG PYTHON_VERSION="3.11"
ARG GO_VERSION="1.21"
ARG GO_VERSION="1.22"
ARG NGROK_VERSION="3"
FROM python:${PYTHON_VERSION}-alpine AS base
@@ -20,6 +20,8 @@ ENV GOARCH=${TARGETARCH}
WORKDIR /build
RUN apk add git
# Cache dependencies
COPY go.mod go.sum ./
RUN --mount=type=cache,target=/root/.cache/go-build go mod download
+11 -8
View File
@@ -779,7 +779,7 @@ POST http://localhost:1984/api/streams?dst=camera1&src=ffmpeg:http://example.com
You can publish any stream to streaming services (YouTube, Telegram, etc.) via RTMP/RTMPS. Important:
- Supported codecs: H264 for video and AAC for audio
- Pixel format should be `yuv420p`, for cameras with `yuvj420p` format you SHOULD use [transcoding](#source-ffmpeg)
- AAC audio is required for YouTube, videos without audio will not work
- You don't need to enable [RTMP module](#module-rtmp) listening for this task
You can use API:
@@ -792,16 +792,19 @@ Or config file:
```yaml
publish:
# publish stream "tplink_tapo" to Telegram
tplink_tapo: rtmps://xxx-x.rtmp.t.me/s/xxxxxxxxxx:xxxxxxxxxxxxxxxxxxxxxx
# publish stream "other_camera" to Telegram and YouTube
other_camera:
# publish stream "video_audio_transcode" to Telegram
video_audio_transcode:
- rtmps://xxx-x.rtmp.t.me/s/xxxxxxxxxx:xxxxxxxxxxxxxxxxxxxxxx
- rtmps://xxx.rtmp.youtube.com/live2/xxxx-xxxx-xxxx-xxxx-xxxx
# publish stream "audio_transcode" to Telegram and YouTube
audio_transcode:
- rtmps://xxx-x.rtmp.t.me/s/xxxxxxxxxx:xxxxxxxxxxxxxxxxxxxxxx
- rtmp://xxx.rtmp.youtube.com/live2/xxxx-xxxx-xxxx-xxxx-xxxx
streams:
# for TP-Link cameras it's important to use transcoding because of wrong pixel format
tplink_tapo: ffmpeg:rtsp://user:pass@192.168.1.123/stream1#video=h264#hardware#audio=aac
video_audio_transcode:
- ffmpeg:rtsp://user:pass@192.168.1.123/stream1#video=h264#hardware#audio=aac
audio_transcode:
- ffmpeg:rtsp://user:pass@192.168.1.123/stream1#video=copy#audio=aac
```
- **Telegram Desktop App** > Any public or private channel or group (where you admin) > Live stream > Start with... > Start streaming.
+39
View File
@@ -0,0 +1,39 @@
package main
import (
"log"
"os"
"github.com/AlexxIT/go2rtc/pkg/core"
"github.com/AlexxIT/go2rtc/pkg/rtsp"
"github.com/AlexxIT/go2rtc/pkg/shell"
)
func main() {
client := rtsp.NewClient(os.Args[1])
if err := client.Dial(); err != nil {
log.Panic(err)
}
client.Medias = []*core.Media{
{
Kind: core.KindAudio,
Direction: core.DirectionRecvonly,
Codecs: []*core.Codec{
{Name: core.CodecPCMU, ClockRate: 8000},
},
ID: "streamid=0",
},
}
if err := client.Announce(); err != nil {
log.Panic(err)
}
if _, err := client.SetupMedia(client.Medias[0]); err != nil {
log.Panic(err)
}
if err := client.Record(); err != nil {
log.Panic(err)
}
shell.RunUntilSignal()
}
+9 -9
View File
@@ -1,11 +1,12 @@
module github.com/AlexxIT/go2rtc
go 1.21
go 1.22
require (
github.com/asticode/go-astits v1.13.0
github.com/expr-lang/expr v1.16.5
github.com/expr-lang/expr v1.16.9
github.com/gorilla/websocket v1.5.1
github.com/mattn/go-isatty v0.0.20
github.com/miekg/dns v1.1.59
github.com/pion/ice/v2 v2.3.24
github.com/pion/interceptor v0.1.29
@@ -15,12 +16,12 @@ require (
github.com/pion/srtp/v2 v2.0.18
github.com/pion/stun v0.6.1
github.com/pion/webrtc/v3 v3.2.40
github.com/rs/zerolog v1.32.0
github.com/rs/zerolog v1.33.0
github.com/sigurn/crc16 v0.0.0-20240131213347-83fcde1e29d1
github.com/sigurn/crc8 v0.0.0-20220107193325-2243fe600f9f
github.com/stretchr/testify v1.9.0
github.com/tadglines/go-pkgs v0.0.0-20210623144937-b983b20f54f9
golang.org/x/crypto v0.23.0
golang.org/x/crypto v0.24.0
gopkg.in/yaml.v3 v3.0.1
)
@@ -30,7 +31,6 @@ require (
github.com/google/uuid v1.6.0 // indirect
github.com/kr/pretty v0.2.1 // indirect
github.com/mattn/go-colorable v0.1.13 // indirect
github.com/mattn/go-isatty v0.0.20 // indirect
github.com/pion/datachannel v1.5.6 // indirect
github.com/pion/dtls/v2 v2.2.11 // indirect
github.com/pion/logging v0.2.2 // indirect
@@ -40,9 +40,9 @@ require (
github.com/pion/transport/v2 v2.2.5 // indirect
github.com/pion/turn/v2 v2.1.6 // indirect
github.com/pmezard/go-difflib v1.0.0 // indirect
golang.org/x/mod v0.17.0 // indirect
golang.org/x/net v0.25.0 // indirect
golang.org/x/mod v0.18.0 // indirect
golang.org/x/net v0.26.0 // indirect
golang.org/x/sync v0.7.0 // indirect
golang.org/x/sys v0.20.0 // indirect
golang.org/x/tools v0.20.0 // indirect
golang.org/x/sys v0.21.0 // indirect
golang.org/x/tools v0.22.0 // indirect
)
+14
View File
@@ -8,6 +8,8 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/expr-lang/expr v1.16.5 h1:m2hvtguFeVaVNTHj8L7BoAyt7O0PAIBaSVbjdHgRXMs=
github.com/expr-lang/expr v1.16.5/go.mod h1:uCkhfG+x7fcZ5A5sXHKuQ07jGZRl6J0FCAaf2k4PtVQ=
github.com/expr-lang/expr v1.16.9 h1:WUAzmR0JNI9JCiF0/ewwHB1gmcGw5wW7nWt8gc6PpCI=
github.com/expr-lang/expr v1.16.9/go.mod h1:8/vRC7+7HBzESEqt5kKpYXxrxkr31SaO8r40VO/1IT4=
github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
github.com/google/uuid v1.3.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
@@ -85,6 +87,8 @@ github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZN
github.com/rs/xid v1.5.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
github.com/rs/zerolog v1.32.0 h1:keLypqrlIjaFsbmJOBdB/qvyF8KEtCWHwobLp5l/mQ0=
github.com/rs/zerolog v1.32.0/go.mod h1:/7mN4D5sKwJLZQ2b/znpjC3/GQWY/xaDXUM0kKWRHss=
github.com/rs/zerolog v1.33.0 h1:1cU2KZkvPxNyfgEmhHAz/1A9Bz+llsdYzklWFzgp0r8=
github.com/rs/zerolog v1.33.0/go.mod h1:/7mN4D5sKwJLZQ2b/znpjC3/GQWY/xaDXUM0kKWRHss=
github.com/sigurn/crc16 v0.0.0-20240131213347-83fcde1e29d1 h1:NVK+OqnavpyFmUiKfUMHrpvbCi2VFoWTrcpI7aDaJ2I=
github.com/sigurn/crc16 v0.0.0-20240131213347-83fcde1e29d1/go.mod h1:9/etS5gpQq9BJsJMWg1wpLbfuSnkm8dPF6FdW2JXVhA=
github.com/sigurn/crc8 v0.0.0-20220107193325-2243fe600f9f h1:1R9KdKjCNSd7F8iGTxIpoID9prlYH8nuNYKt0XvweHA=
@@ -115,10 +119,14 @@ golang.org/x/crypto v0.22.0 h1:g1v0xeRhjcugydODzvb3mEM9SQ0HGp9s/nh3COQ/C30=
golang.org/x/crypto v0.22.0/go.mod h1:vr6Su+7cTlO45qkww3VDJlzDn0ctJvRgYbC2NvXHt+M=
golang.org/x/crypto v0.23.0 h1:dIJU/v2J8Mdglj/8rJ6UUOM3Zc9zLZxVZwwxMooUSAI=
golang.org/x/crypto v0.23.0/go.mod h1:CKFgDieR+mRhux2Lsu27y0fO304Db0wZe70UKqHu0v8=
golang.org/x/crypto v0.24.0 h1:mnl8DM0o513X8fdIkmyFE/5hTYxbwYOjDS/+rK6qpRI=
golang.org/x/crypto v0.24.0/go.mod h1:Z1PMYSOR5nyMcyAVAIQSKCDwalqy85Aqn1x3Ws4L5DM=
golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
golang.org/x/mod v0.17.0 h1:zY54UmvipHiNd+pm+m0x9KhZ9hl1/7QNMyxXbc6ICqA=
golang.org/x/mod v0.17.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
golang.org/x/mod v0.18.0 h1:5+9lSbEzPSdWkH32vYPBwEpX8KwDbM52Ud9xBUvNlb0=
golang.org/x/mod v0.18.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
@@ -134,6 +142,8 @@ golang.org/x/net v0.24.0 h1:1PcaxkF854Fu3+lvBIx5SYn9wRlBzzcnHZSiaFFAb0w=
golang.org/x/net v0.24.0/go.mod h1:2Q7sJY5mzlzWjKtYUEXSlBWCdyaioyXzRB2RtU8KVE8=
golang.org/x/net v0.25.0 h1:d/OCCoBEUq33pjydKrGQhw7IlUPI2Oylr+8qLx49kac=
golang.org/x/net v0.25.0/go.mod h1:JkAGAh7GEvH74S6FOH42FLoXpXbE/aqXSrIQjXgsiwM=
golang.org/x/net v0.26.0 h1:soB7SVo0PWrY4vPW/+ay0jKDNScG2X9wFeYlXIvJsOQ=
golang.org/x/net v0.26.0/go.mod h1:5YKkiSynbBIh3p6iOc/vibscux0x38BZDkn8sCUPxHE=
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -160,6 +170,8 @@ golang.org/x/sys v0.19.0 h1:q5f1RH2jigJ1MoAWp2KTp3gm5zAGFUTarQZ5U386+4o=
golang.org/x/sys v0.19.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/sys v0.20.0 h1:Od9JTbYCk261bKm4M/mw7AklTlFYIa0bIp9BgSm1S8Y=
golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/sys v0.21.0 h1:rF+pYz3DAGSQAxAu1CbC7catZg4ebC4UIeIhKxBZvws=
golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
@@ -184,6 +196,8 @@ golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc
golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
golang.org/x/tools v0.20.0 h1:hz/CVckiOxybQvFw6h7b/q80NTr9IUQb4s1IIzW7KNY=
golang.org/x/tools v0.20.0/go.mod h1:WvitBU7JJf6A4jOdg4S1tviW9bhUxkgeCui/0JHctQg=
golang.org/x/tools v0.22.0 h1:gqSGLZqv+AI9lIQzniJ0nZDRG5GBPsSi+DRNHWNz6yA=
golang.org/x/tools v0.22.0/go.mod h1:aCwcsjqvq7Yqt6TNyX7QMU2enbQ/Gt0bo6krSeEri+c=
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
+1 -1
View File
@@ -4,7 +4,7 @@
# only debian 13 (trixie) has latest ffmpeg
# https://packages.debian.org/trixie/ffmpeg
ARG DEBIAN_VERSION="trixie-slim"
ARG GO_VERSION="1.21-bookworm"
ARG GO_VERSION="1.22-bookworm"
ARG NGROK_VERSION="3"
FROM debian:${DEBIAN_VERSION} AS base
+21 -5
View File
@@ -1,6 +1,10 @@
- By default go2rtc will search config file `go2rtc.yaml` in current work directory
- go2rtc support multiple config files
- go2rtc support inline config as `YAML`, `JSON` or `key=value` format from command line
- go2rtc support multiple config files:
- `go2rtc -c config1.yaml -c config2.yaml -c config3.yaml`
- go2rtc support inline config as multiple formats from command line:
- **YAML**: `go2rtc -c '{log: {format: text}}'`
- **JSON**: `go2rtc -c '{"log":{"format":"text"}}'`
- **key=value**: `go2rtc -c log.format=text`
- Every next config will overwrite previous (but only defined params)
```
@@ -21,15 +25,24 @@ Also go2rtc support templates for using environment variables in any part of con
streams:
camera1: rtsp://rtsp:${CAMERA_PASSWORD}@192.168.1.123/av_stream/ch0
${LOGS:} # empty default value
rtsp:
username: ${RTSP_USER:admin} # "admin" if env "RTSP_USER" not set
password: ${RTSP_PASS:secret} # "secret" if env "RTSP_PASS" not set
```
## JSON Schema
Editors like [GoLand](https://www.jetbrains.com/go/) and [VS Code](https://code.visualstudio.com/) supports autocomplete and syntax validation.
```yaml
# yaml-language-server: $schema=https://raw.githubusercontent.com/AlexxIT/go2rtc/master/website/schema.json
```
## Defaults
- Default values may change in updates
- FFmpeg module has many presets, they are not listed here because they may also change in updates
```yaml
api:
listen: ":1984"
@@ -38,7 +51,10 @@ ffmpeg:
bin: "ffmpeg"
log:
format: "color"
level: "info"
output: "stdout"
time: "UNIXMS"
rtsp:
listen: ":8554"
@@ -51,4 +67,4 @@ webrtc:
listen: ":8555/tcp"
ice_servers:
- urls: [ "stun:stun.l.google.com:19302" ]
```
```
+18 -129
View File
@@ -1,28 +1,20 @@
package app
import (
"errors"
"flag"
"fmt"
"os"
"os/exec"
"path/filepath"
"runtime"
"runtime/debug"
"strings"
"github.com/AlexxIT/go2rtc/pkg/shell"
"github.com/AlexxIT/go2rtc/pkg/yaml"
"github.com/rs/zerolog/log"
)
var Version = "1.9.2"
var UserAgent = "go2rtc/" + Version
var ConfigPath string
var Info = map[string]any{
"version": Version,
}
var (
Version string
UserAgent string
ConfigPath string
Info = make(map[string]any)
)
const usage = `Usage of go2rtc:
@@ -32,12 +24,12 @@ const usage = `Usage of go2rtc:
`
func Init() {
var confs Config
var config flagConfig
var daemon bool
var version bool
flag.Var(&confs, "config", "")
flag.Var(&confs, "c", "")
flag.Var(&config, "config", "")
flag.Var(&config, "c", "")
flag.BoolVar(&daemon, "daemon", false, "")
flag.BoolVar(&daemon, "d", false, "")
flag.BoolVar(&version, "version", false, "")
@@ -68,111 +60,30 @@ func Init() {
// Re-run the program in background and exit
cmd := exec.Command(os.Args[0], args...)
if err := cmd.Start(); err != nil {
log.Fatal().Err(err).Send()
fmt.Println(err)
os.Exit(1)
}
fmt.Println("Running in daemon mode with PID:", cmd.Process.Pid)
os.Exit(0)
}
if confs == nil {
confs = []string{"go2rtc.yaml"}
}
for _, conf := range confs {
if len(conf) == 0 {
continue
}
if conf[0] == '{' {
// config as raw YAML or JSON
configs = append(configs, []byte(conf))
} else if data := parseConfString(conf); data != nil {
configs = append(configs, data)
} else {
// config as file
if ConfigPath == "" {
ConfigPath = conf
}
if data, _ = os.ReadFile(conf); data == nil {
continue
}
data = []byte(shell.ReplaceEnvVars(string(data)))
configs = append(configs, data)
}
}
if ConfigPath != "" {
if !filepath.IsAbs(ConfigPath) {
if cwd, err := os.Getwd(); err == nil {
ConfigPath = filepath.Join(cwd, ConfigPath)
}
}
Info["config_path"] = ConfigPath
}
UserAgent = "go2rtc/" + Version
Info["version"] = Version
Info["revision"] = revision
var cfg struct {
Mod map[string]string `yaml:"log"`
}
LoadConfig(&cfg)
log.Logger = NewLogger(cfg.Mod["format"], cfg.Mod["level"])
modules = cfg.Mod
initConfig(config)
initLogger()
platform := fmt.Sprintf("%s/%s", runtime.GOOS, runtime.GOARCH)
log.Info().Str("version", Version).Str("platform", platform).Str("revision", revision).Msg("go2rtc")
log.Debug().Str("version", runtime.Version()).Str("vcs.time", vcsTime).Msg("build")
Logger.Info().Str("version", Version).Str("platform", platform).Str("revision", revision).Msg("go2rtc")
Logger.Debug().Str("version", runtime.Version()).Str("vcs.time", vcsTime).Msg("build")
if ConfigPath != "" {
log.Info().Str("path", ConfigPath).Msg("config")
}
migrateStore()
}
func LoadConfig(v any) {
for _, data := range configs {
if err := yaml.Unmarshal(data, v); err != nil {
log.Warn().Err(err).Msg("[app] read config")
}
Logger.Info().Str("path", ConfigPath).Msg("config")
}
}
func PatchConfig(key string, value any, path ...string) error {
if ConfigPath == "" {
return errors.New("config file disabled")
}
// empty config is OK
b, _ := os.ReadFile(ConfigPath)
b, err := yaml.Patch(b, key, value, path...)
if err != nil {
return err
}
return os.WriteFile(ConfigPath, b, 0644)
}
// internal
type Config []string
func (c *Config) String() string {
return strings.Join(*c, " ")
}
func (c *Config) Set(value string) error {
*c = append(*c, value)
return nil
}
var configs [][]byte
func readRevisionTime() (revision, vcsTime string) {
if info, ok := debug.ReadBuildInfo(); ok {
for _, setting := range info.Settings {
@@ -194,25 +105,3 @@ func readRevisionTime() (revision, vcsTime string) {
}
return
}
func parseConfString(s string) []byte {
i := strings.IndexByte(s, '=')
if i < 0 {
return nil
}
items := strings.Split(s[:i], ".")
if len(items) < 2 {
return nil
}
// `log.level=trace` => `{log: {level: trace}}`
var pre string
var suf = s[i+1:]
for _, item := range items {
pre += "{" + item + ": "
suf += "}"
}
return []byte(pre + suf)
}
+109
View File
@@ -0,0 +1,109 @@
package app
import (
"errors"
"os"
"path/filepath"
"strings"
"github.com/AlexxIT/go2rtc/pkg/shell"
"github.com/AlexxIT/go2rtc/pkg/yaml"
)
func LoadConfig(v any) {
for _, data := range configs {
if err := yaml.Unmarshal(data, v); err != nil {
Logger.Warn().Err(err).Send()
}
}
}
func PatchConfig(key string, value any, path ...string) error {
if ConfigPath == "" {
return errors.New("config file disabled")
}
// empty config is OK
b, _ := os.ReadFile(ConfigPath)
b, err := yaml.Patch(b, key, value, path...)
if err != nil {
return err
}
return os.WriteFile(ConfigPath, b, 0644)
}
type flagConfig []string
func (c *flagConfig) String() string {
return strings.Join(*c, " ")
}
func (c *flagConfig) Set(value string) error {
*c = append(*c, value)
return nil
}
var configs [][]byte
func initConfig(confs flagConfig) {
if confs == nil {
confs = []string{"go2rtc.yaml"}
}
for _, conf := range confs {
if len(conf) == 0 {
continue
}
if conf[0] == '{' {
// config as raw YAML or JSON
configs = append(configs, []byte(conf))
} else if data := parseConfString(conf); data != nil {
configs = append(configs, data)
} else {
// config as file
if ConfigPath == "" {
ConfigPath = conf
}
if data, _ = os.ReadFile(conf); data == nil {
continue
}
data = []byte(shell.ReplaceEnvVars(string(data)))
configs = append(configs, data)
}
}
if ConfigPath != "" {
if !filepath.IsAbs(ConfigPath) {
if cwd, err := os.Getwd(); err == nil {
ConfigPath = filepath.Join(cwd, ConfigPath)
}
}
Info["config_path"] = ConfigPath
}
}
func parseConfString(s string) []byte {
i := strings.IndexByte(s, '=')
if i < 0 {
return nil
}
items := strings.Split(s[:i], ".")
if len(items) < 2 {
return nil
}
// `log.level=trace` => `{log: {level: trace}}`
var pre string
var suf = s[i+1:]
for _, item := range items {
pre += "{" + item + ": "
suf += "}"
}
return []byte(pre + suf)
}
+80 -29
View File
@@ -4,49 +4,100 @@ import (
"io"
"os"
"github.com/mattn/go-isatty"
"github.com/rs/zerolog"
"github.com/rs/zerolog/log"
)
var MemoryLog *circularBuffer
func NewLogger(format string, level string) zerolog.Logger {
var writer io.Writer = os.Stdout
if format != "json" {
writer = zerolog.ConsoleWriter{
Out: writer, TimeFormat: "15:04:05.000", NoColor: format == "text",
}
}
MemoryLog = newBuffer(16)
writer = zerolog.MultiLevelWriter(writer, MemoryLog)
zerolog.TimeFieldFormat = zerolog.TimeFormatUnixMs
lvl, err := zerolog.ParseLevel(level)
if err != nil || lvl == zerolog.NoLevel {
lvl = zerolog.InfoLevel
}
return zerolog.New(writer).With().Timestamp().Logger().Level(lvl)
}
var MemoryLog = newBuffer(16)
func GetLogger(module string) zerolog.Logger {
if s, ok := modules[module]; ok {
lvl, err := zerolog.ParseLevel(s)
if err == nil {
return log.Level(lvl)
return Logger.Level(lvl)
}
log.Warn().Err(err).Caller().Send()
Logger.Warn().Err(err).Caller().Send()
}
return log.Logger
return Logger
}
// initLogger support:
// - output: empty (only to memory), stderr, stdout
// - format: empty (autodetect color support), color, json, text
// - time: empty (disable timestamp), UNIXMS, UNIXMICRO, UNIXNANO
// - level: disabled, trace, debug, info, warn, error...
func initLogger() {
var cfg struct {
Mod map[string]string `yaml:"log"`
}
cfg.Mod = modules // defaults
LoadConfig(&cfg)
var writer io.Writer
switch modules["output"] {
case "stderr":
writer = os.Stderr
case "stdout":
writer = os.Stdout
}
timeFormat := modules["time"]
if writer != nil {
if format := modules["format"]; format != "json" {
console := &zerolog.ConsoleWriter{Out: writer}
switch format {
case "text":
console.NoColor = true
case "color":
console.NoColor = false // useless, but anyway
default:
// autodetection if output support color
// go-isatty - dependency for go-colorable - dependency for ConsoleWriter
console.NoColor = !isatty.IsTerminal(writer.(*os.File).Fd())
}
if timeFormat != "" {
console.TimeFormat = "15:04:05.000"
} else {
console.PartsOrder = []string{
zerolog.LevelFieldName,
zerolog.CallerFieldName,
zerolog.MessageFieldName,
}
}
writer = console
}
writer = zerolog.MultiLevelWriter(writer, MemoryLog)
} else {
writer = MemoryLog
}
lvl, _ := zerolog.ParseLevel(modules["level"])
Logger = zerolog.New(writer).Level(lvl)
if timeFormat != "" {
zerolog.TimeFieldFormat = timeFormat
Logger = Logger.With().Timestamp().Logger()
}
}
var Logger zerolog.Logger
// modules log levels
var modules map[string]string
var modules = map[string]string{
"format": "", // useless, but anyway
"level": "info",
"output": "stdout", // TODO: change to stderr someday
"time": zerolog.TimeFormatUnixMs,
}
const chunkSize = 1 << 16
-35
View File
@@ -1,35 +0,0 @@
package app
import (
"encoding/json"
"os"
"github.com/rs/zerolog/log"
)
func migrateStore() {
const name = "go2rtc.json"
data, _ := os.ReadFile(name)
if data == nil {
return
}
var store struct {
Streams map[string]string `json:"streams"`
}
if err := json.Unmarshal(data, &store); err != nil {
log.Warn().Err(err).Caller().Send()
return
}
for id, url := range store.Streams {
if err := PatchConfig(id, url, "streams"); err != nil {
log.Warn().Err(err).Caller().Send()
return
}
}
_ = os.Remove(name)
}
+1 -5
View File
@@ -12,7 +12,6 @@ import (
"github.com/AlexxIT/go2rtc/internal/streams"
"github.com/AlexxIT/go2rtc/pkg/core"
"github.com/AlexxIT/go2rtc/pkg/dvrip"
"github.com/rs/zerolog/log"
)
func Init() {
@@ -92,10 +91,7 @@ func sendBroadcasts(conn *net.UDPConn) {
for i := 0; i < 3; i++ {
time.Sleep(100 * time.Millisecond)
if _, err = conn.WriteToUDP(data, addr); err != nil {
log.Err(err).Caller().Send()
}
_, _ = conn.WriteToUDP(data, addr)
}
}
+52 -24
View File
@@ -5,7 +5,6 @@ import (
"encoding/hex"
"errors"
"fmt"
"io"
"net/url"
"os"
"os/exec"
@@ -69,8 +68,9 @@ func execHandle(rawURL string) (core.Producer, error) {
args := shell.QuoteSplit(rawURL[5:]) // remove `exec:`
cmd := exec.Command(args[0], args[1:]...)
if log.Debug().Enabled() {
cmd.Stderr = os.Stderr
cmd.Stderr = &logWriter{
buf: make([]byte, 512),
debug: log.Debug().Enabled(),
}
if path == "" {
@@ -90,6 +90,10 @@ func handlePipe(_ string, cmd *exec.Cmd, query url.Values) (core.Producer, error
return nil, err
}
log.Debug().Strs("args", cmd.Args).Msg("[exec] run pipe")
ts := time.Now()
if err = cmd.Start(); err != nil {
return nil, err
}
@@ -99,23 +103,17 @@ func handlePipe(_ string, cmd *exec.Cmd, query url.Values) (core.Producer, error
_ = r.Close()
}
return prod, err
log.Debug().Stringer("launch", time.Since(ts)).Msg("[exec] run pipe")
return prod, fmt.Errorf("exec/pipe: %w\n%s", err, cmd.Stderr)
}
func handleRTSP(url string, cmd *exec.Cmd, path string) (core.Producer, error) {
stderr := limitBuffer{buf: make([]byte, 512)}
if cmd.Stderr != nil {
cmd.Stderr = io.MultiWriter(cmd.Stderr, &stderr)
} else {
cmd.Stderr = &stderr
}
if log.Trace().Enabled() {
cmd.Stdout = os.Stdout
}
waiter := make(chan core.Producer)
waiter := make(chan *pkg.Conn, 1)
waitersMu.Lock()
waiters[path] = waiter
@@ -127,7 +125,7 @@ func handleRTSP(url string, cmd *exec.Cmd, path string) (core.Producer, error) {
waitersMu.Unlock()
}()
log.Debug().Str("url", url).Str("cmd", fmt.Sprintf("%s", strings.Join(cmd.Args, " "))).Msg("[exec] run")
log.Debug().Strs("args", cmd.Args).Msg("[exec] run rtsp")
ts := time.Now()
@@ -145,12 +143,16 @@ func handleRTSP(url string, cmd *exec.Cmd, path string) (core.Producer, error) {
case <-time.After(time.Second * 60):
_ = cmd.Process.Kill()
log.Error().Str("url", url).Msg("[exec] timeout")
return nil, errors.New("timeout")
return nil, errors.New("exec: timeout")
case <-done:
// limit message size
return nil, errors.New("exec: " + stderr.String())
return nil, fmt.Errorf("exec/rtsp\n%s", cmd.Stderr)
case prod := <-waiter:
log.Debug().Stringer("launch", time.Since(ts)).Msg("[exec] run")
log.Debug().Stringer("launch", time.Since(ts)).Msg("[exec] run rtsp")
prod.OnClose = func() error {
log.Debug().Msgf("[exec] kill rtsp")
return errors.Join(cmd.Process.Kill(), cmd.Wait())
}
return prod, nil
}
}
@@ -159,25 +161,51 @@ func handleRTSP(url string, cmd *exec.Cmd, path string) (core.Producer, error) {
var (
log zerolog.Logger
waiters = map[string]chan core.Producer{}
waiters = make(map[string]chan *pkg.Conn)
waitersMu sync.Mutex
)
type limitBuffer struct {
buf []byte
n int
type logWriter struct {
buf []byte
debug bool
n int
}
func (l *limitBuffer) String() string {
func (l *logWriter) String() string {
if l.n == len(l.buf) {
return string(l.buf) + "..."
}
return string(l.buf[:l.n])
}
func (l *limitBuffer) Write(p []byte) (int, error) {
func (l *logWriter) Write(p []byte) (n int, err error) {
if l.n < cap(l.buf) {
l.n += copy(l.buf[l.n:], p)
}
return len(p), nil
n = len(p)
if l.debug {
if p = trimSpace(p); p != nil {
log.Debug().Msgf("[exec] %s", p)
}
}
return
}
func trimSpace(b []byte) []byte {
start := 0
stop := len(b)
for ; start < stop; start++ {
if b[start] >= ' ' {
break // trim all ASCII before 0x20
}
}
for ; ; stop-- {
if stop == start {
return nil // skip empty output
}
if b[stop-1] > ' ' {
break // trim all ASCII before 0x21
}
}
return b[start:stop]
}
+7
View File
@@ -45,6 +45,13 @@
[video4linux2,v4l2 @ 0x7f7de7c58bc0] Compressed: mjpeg : Motion-JPEG : 640x480 160x120 176x144 320x176 320x240 352x288 432x240 544x288 640x360 752x416 800x448 800x600 864x480 960x544 960x720 1024x576 1184x656 1280x720 1280x960
```
## TTS
```yaml
streams:
tts: ffmpeg:#input=-readrate 1 -readrate_initial_burst 0.001 -f lavfi -i "flite=text='1 2 3 4 5 6 7 8 9 0'"#audio=pcma
```
## Useful links
- https://superuser.com/questions/564402/explanation-of-x264-tune
+51
View File
@@ -0,0 +1,51 @@
package ffmpeg
import (
"net/http"
"strings"
"github.com/AlexxIT/go2rtc/internal/streams"
)
func apiFFmpeg(w http.ResponseWriter, r *http.Request) {
if r.Method != "POST" {
http.Error(w, "", http.StatusMethodNotAllowed)
return
}
query := r.URL.Query()
dst := query.Get("dst")
stream := streams.Get(dst)
if stream == nil {
http.Error(w, "", http.StatusNotFound)
return
}
var src string
if s := query.Get("file"); s != "" {
if streams.Validate(s) == nil {
src = "ffmpeg:" + s + "#audio=auto#input=file"
}
} else if s = query.Get("live"); s != "" {
if streams.Validate(s) == nil {
src = "ffmpeg:" + s + "#audio=auto"
}
} else if s = query.Get("text"); s != "" {
if strings.IndexAny(s, `'"&%$`) < 0 {
src = "ffmpeg:tts?text=" + s
if s = query.Get("voice"); s != "" {
src += "&voice=" + s
}
src += "#audio=auto"
}
}
if src == "" {
http.Error(w, "", http.StatusBadRequest)
return
}
if err := stream.Play(src); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
}
}
+4 -15
View File
@@ -1,11 +1,9 @@
package device
import (
"errors"
"net/http"
"net/url"
"strconv"
"strings"
"sync"
"github.com/AlexxIT/go2rtc/internal/api"
@@ -17,24 +15,15 @@ func Init(bin string) {
api.HandleFunc("api/ffmpeg/devices", apiDevices)
}
func GetInput(src string) (string, error) {
i := strings.IndexByte(src, '?')
if i < 0 {
return "", errors.New("empty query: " + src)
}
query, err := url.ParseQuery(src[i+1:])
func GetInput(src string) string {
query, err := url.ParseQuery(src)
if err != nil {
return "", err
return ""
}
runonce.Do(initDevices)
if input := queryToInput(query); input != "" {
return input, nil
}
return "", errors.New("wrong query: " + src)
return queryToInput(query)
}
var Bin string
+69 -20
View File
@@ -2,35 +2,58 @@ package ffmpeg
import (
"net/url"
"slices"
"strings"
"github.com/AlexxIT/go2rtc/internal/api"
"github.com/AlexxIT/go2rtc/internal/app"
"github.com/AlexxIT/go2rtc/internal/ffmpeg/device"
"github.com/AlexxIT/go2rtc/internal/ffmpeg/hardware"
"github.com/AlexxIT/go2rtc/internal/ffmpeg/virtual"
"github.com/AlexxIT/go2rtc/internal/rtsp"
"github.com/AlexxIT/go2rtc/internal/streams"
"github.com/AlexxIT/go2rtc/pkg/core"
"github.com/AlexxIT/go2rtc/pkg/ffmpeg"
"github.com/rs/zerolog"
)
func Init() {
var cfg struct {
Mod map[string]string `yaml:"ffmpeg"`
Log struct {
Level string `yaml:"ffmpeg"`
} `yaml:"log"`
}
cfg.Mod = defaults // will be overriden from yaml
cfg.Log.Level = "error"
app.LoadConfig(&cfg)
if app.GetLogger("exec").GetLevel() >= 0 {
defaults["global"] += " -v error"
log = app.GetLogger("ffmpeg")
// zerolog levels: trace debug info warn error fatal panic disabled
// FFmpeg levels: trace debug verbose info warning error fatal panic quiet
if cfg.Log.Level == "warn" {
cfg.Log.Level = "warning"
}
defaults["global"] += " -v " + cfg.Log.Level
streams.RedirectFunc("ffmpeg", func(url string) (string, error) {
if _, err := Version(); err != nil {
return "", err
}
args := parseArgs(url[7:])
if slices.Contains(args.Codecs, "auto") {
return "", nil // force call streams.HandleFunc("ffmpeg")
}
return "exec:" + args.String(), nil
})
streams.HandleFunc("ffmpeg", NewProducer)
api.HandleFunc("api/ffmpeg", apiFFmpeg)
device.Init(defaults["bin"])
hardware.Init(defaults["bin"])
}
@@ -49,6 +72,9 @@ var defaults = map[string]string{
// output
"output": "-user_agent ffmpeg/go2rtc -rtsp_transport tcp -f rtsp {output}",
"output/mjpeg": "-f mjpeg -",
"output/raw": "-f yuv4mpegpipe -",
"output/aac": "-f adts -",
"output/wav": "-f wav -",
// `-preset superfast` - we can't use ultrafast because it doesn't support `-profile main -level 4.1`
// `-tune zerolatency` - for minimal latency
@@ -59,6 +85,12 @@ var defaults = map[string]string{
"mjpeg": "-c:v mjpeg",
//"mjpeg": "-c:v mjpeg -force_duplicated_matrix:v 1 -huffman:v 0 -pix_fmt:v yuvj420p",
"raw": "-c:v rawvideo",
"raw/gray8": "-c:v rawvideo -pix_fmt:v gray8",
"raw/yuv420p": "-c:v rawvideo -pix_fmt:v yuv420p",
"raw/yuv422p": "-c:v rawvideo -pix_fmt:v yuv422p",
"raw/yuv444p": "-c:v rawvideo -pix_fmt:v yuv444p",
// https://ffmpeg.org/ffmpeg-codecs.html#libopus-1
// https://github.com/pion/webrtc/issues/1514
// https://ffmpeg.org/ffmpeg-resampler.html
@@ -116,6 +148,8 @@ var defaults = map[string]string{
"h265/videotoolbox": "-c:v hevc_videotoolbox -g 50 -bf 0 -profile:v main -level:v 5.1",
}
var log zerolog.Logger
// configTemplate - return template from config (defaults) if exist or return raw template
func configTemplate(template string) string {
if s := defaults[template]; s != "" {
@@ -140,9 +174,10 @@ func inputTemplate(name, s string, query url.Values) string {
func parseArgs(s string) *ffmpeg.Args {
// init FFmpeg arguments
args := &ffmpeg.Args{
Bin: defaults["bin"],
Global: defaults["global"],
Output: defaults["output"],
Bin: defaults["bin"],
Global: defaults["global"],
Output: defaults["output"],
Version: verAV,
}
var query url.Values
@@ -188,16 +223,14 @@ func parseArgs(s string) *ffmpeg.Args {
s += "?video&audio"
}
args.Input = inputTemplate("rtsp", s, query)
} else if strings.HasPrefix(s, "device?") {
var err error
args.Input, err = device.GetInput(s)
if err != nil {
return nil
}
} else if strings.HasPrefix(s, "virtual?") {
var err error
if args.Input, err = virtual.GetInput(s[8:]); err != nil {
return nil
} else if i = strings.Index(s, "?"); i > 0 {
switch s[:i] {
case "device":
args.Input = device.GetInput(s[i+1:])
case "virtual":
args.Input = virtual.GetInput(s[i+1:])
case "tts":
args.Input = virtual.GetInputTTS(s[i+1:])
}
} else {
args.Input = inputTemplate("file", s, query)
@@ -315,11 +348,27 @@ func parseArgs(s string) *ffmpeg.Args {
args.AddCodec("-an")
}
// transcoding to only mjpeg
if (args.Video == 1 && args.Audio == 0 && query.Get("video") == "mjpeg") ||
// no transcoding from mjpeg input
(args.Video == 0 && args.Audio == 0 && strings.Contains(args.Input, " mjpeg ")) {
args.Output = defaults["output/mjpeg"]
// change otput from RTSP to some other pipe format
switch {
case args.Video == 0 && args.Audio == 0:
// no transcoding from mjpeg input (ffmpeg device with support output as raw MJPEG)
if strings.Contains(args.Input, " mjpeg ") {
args.Output = defaults["output/mjpeg"]
}
case args.Video == 1 && args.Audio == 0:
switch core.Before(query.Get("video"), "/") {
case "mjpeg":
args.Output = defaults["output/mjpeg"]
case "raw":
args.Output = defaults["output/raw"]
}
case args.Video == 0 && args.Audio == 1:
switch core.Before(query.Get("audio"), "/") {
case "aac":
args.Output = defaults["output/aac"]
case "pcma", "pcmu", "pcml":
args.Output = defaults["output/wav"]
}
}
return args
+21
View File
@@ -3,6 +3,7 @@ package ffmpeg
import (
"testing"
"github.com/AlexxIT/go2rtc/pkg/ffmpeg"
"github.com/stretchr/testify/require"
)
@@ -292,3 +293,23 @@ func TestDrawText(t *testing.T) {
})
}
}
func TestVersion(t *testing.T) {
verAV = ffmpeg.Version61
tests := []struct {
name string
source string
expect string
}{
{
source: "/media/bbb.mp4",
expect: `ffmpeg -hide_banner -readrate_initial_burst 0.001 -re -i /media/bbb.mp4 -c copy -user_agent ffmpeg/go2rtc -rtsp_transport tcp -f rtsp {output}`,
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
args := parseArgs(test.source)
require.Equal(t, test.expect, args.String())
})
}
}
-3
View File
@@ -7,8 +7,6 @@ import (
"github.com/AlexxIT/go2rtc/internal/api"
"github.com/AlexxIT/go2rtc/pkg/ffmpeg"
"github.com/rs/zerolog/log"
)
const (
@@ -152,7 +150,6 @@ var cache = map[string]string{}
func run(bin string, args string) bool {
err := exec.Command(bin, strings.Split(args, " ")...).Run()
log.Printf("%v %v", args, err)
return err == nil
}
+117
View File
@@ -0,0 +1,117 @@
package ffmpeg
import (
"encoding/json"
"errors"
"net/url"
"strconv"
"strings"
"github.com/AlexxIT/go2rtc/internal/streams"
"github.com/AlexxIT/go2rtc/pkg/aac"
"github.com/AlexxIT/go2rtc/pkg/core"
)
type Producer struct {
core.SuperProducer
url string
query url.Values
ffmpeg core.Producer
}
// NewProducer - FFmpeg producer with auto selection video/audio codec based on client capabilities
func NewProducer(url string) (core.Producer, error) {
p := &Producer{}
i := strings.IndexByte(url, '#')
p.url, p.query = url[:i], streams.ParseQuery(url[i+1:])
// ffmpeg.NewProducer support only one audio
if len(p.query["video"]) != 0 || len(p.query["audio"]) != 1 {
return nil, errors.New("ffmpeg: unsupported params: " + url[i:])
}
p.Type = "FFmpeg producer"
p.Medias = []*core.Media{
{
// we can support only audio, because don't know FmtpLine for H264 and PayloadType for MJPEG
Kind: core.KindAudio,
Direction: core.DirectionRecvonly,
// codecs in order from best to worst
Codecs: []*core.Codec{
// OPUS will always marked as OPUS/48000/2
{Name: core.CodecOpus, ClockRate: 48000, Channels: 2},
{Name: core.CodecPCM, ClockRate: 16000},
{Name: core.CodecPCMA, ClockRate: 16000},
{Name: core.CodecPCMU, ClockRate: 16000},
{Name: core.CodecPCM, ClockRate: 8000},
{Name: core.CodecPCMA, ClockRate: 8000},
{Name: core.CodecPCMU, ClockRate: 8000},
// AAC has unknown problems on Dahua two way
{Name: core.CodecAAC, ClockRate: 16000, FmtpLine: aac.FMTP + "1408"},
},
},
}
return p, nil
}
func (p *Producer) Start() error {
var err error
if p.ffmpeg, err = streams.GetProducer(p.newURL()); err != nil {
return err
}
for i, media := range p.ffmpeg.GetMedias() {
track, err := p.ffmpeg.GetTrack(media, media.Codecs[0])
if err != nil {
return err
}
p.Receivers[i].Replace(track)
}
return p.ffmpeg.Start()
}
func (p *Producer) Stop() error {
if p.ffmpeg == nil {
return nil
}
return p.ffmpeg.Stop()
}
func (p *Producer) MarshalJSON() ([]byte, error) {
if p.ffmpeg == nil {
return json.Marshal(p.SuperProducer)
}
return json.Marshal(p.ffmpeg)
}
func (p *Producer) newURL() string {
s := p.url
// rewrite codecs in url from auto to known presets from defaults
for _, receiver := range p.Receivers {
codec := receiver.Codec
switch codec.Name {
case core.CodecOpus:
s += "#audio=opus"
case core.CodecAAC:
s += "#audio=aac/16000"
case core.CodecPCM:
s += "#audio=pcm/" + strconv.Itoa(int(codec.ClockRate))
case core.CodecPCMA:
s += "#audio=pcma/" + strconv.Itoa(int(codec.ClockRate))
case core.CodecPCMU:
s += "#audio=pcmu/" + strconv.Itoa(int(codec.ClockRate))
}
}
// add other params
for key, values := range p.query {
if key != "audio" {
for _, value := range values {
s += "#" + key + "=" + value
}
}
}
return s
}
+46
View File
@@ -0,0 +1,46 @@
package ffmpeg
import (
"errors"
"os/exec"
"sync"
"github.com/AlexxIT/go2rtc/pkg/ffmpeg"
)
var verMu sync.Mutex
var verErr error
var verFF string
var verAV string
func Version() (string, error) {
verMu.Lock()
defer verMu.Unlock()
if verFF != "" {
return verFF, verErr
}
cmd := exec.Command(defaults["bin"], "-version")
b, err := cmd.Output()
if err != nil {
verFF = "-"
verErr = err
return verFF, verErr
}
verFF, verAV = ffmpeg.ParseVersion(b)
if verFF == "" {
verFF = "?"
}
// better to compare libavformat, because nightly/master builds
if verAV != "" && verAV < ffmpeg.Version50 {
verErr = errors.New("ffmpeg: unsupported version: " + verFF)
}
log.Debug().Str("version", verFF).Str("libavformat", verAV).Msgf("[ffmpeg] bin")
return verFF, verErr
}
+62 -42
View File
@@ -4,56 +4,76 @@ import (
"net/url"
)
func GetInput(src string) (string, error) {
func GetInput(src string) string {
query, err := url.ParseQuery(src)
if err != nil {
return "", err
return ""
}
// set defaults (using Add instead of Set)
query.Add("video", "testsrc")
query.Add("size", "1920x1080")
query.Add("decimals", "2")
input := "-re"
// https://ffmpeg.org/ffmpeg-filters.html
video := query.Get("video")
input := "-re -f lavfi -i " + video
for _, video := range query["video"] {
// https://ffmpeg.org/ffmpeg-filters.html
sep := "=" // first separator
sep := "=" // first separator
for key, values := range query {
value := values[0]
// https://ffmpeg.org/ffmpeg-utils.html#video-size-syntax
switch key {
case "color", "rate", "duration", "sar":
case "size":
switch value {
case "720":
value = "1280x720" // crf=1 -> 12 Mbps
case "1080":
value = "1920x1080" // crf=1 -> 25 Mbps
case "2K":
value = "2560x1440" // crf=1 -> 43 Mbps
case "4K":
value = "3840x2160" // crf=1 -> 103 Mbps
case "8K":
value = "7680x4230" // https://reolink.com/blog/8k-resolution/
}
case "decimals":
if video != "testsrc" {
continue
}
default:
continue
if video == "" {
video = "testsrc=decimals=2" // default video
sep = ":"
}
input += sep + key + "=" + value
sep = ":" // next separator
input += " -f lavfi -i " + video
// set defaults (using Add instead of Set)
query.Add("size", "1920x1080")
for key, values := range query {
value := values[0]
// https://ffmpeg.org/ffmpeg-utils.html#video-size-syntax
switch key {
case "color", "rate", "duration", "sar", "decimals":
case "size":
switch value {
case "720":
value = "1280x720" // crf=1 -> 12 Mbps
case "1080":
value = "1920x1080" // crf=1 -> 25 Mbps
case "2K":
value = "2560x1440" // crf=1 -> 43 Mbps
case "4K":
value = "3840x2160" // crf=1 -> 103 Mbps
case "8K":
value = "7680x4230" // https://reolink.com/blog/8k-resolution/
}
default:
continue
}
input += sep + key + "=" + value
sep = ":" // next separator
}
if s := query.Get("format"); s != "" {
input += ",format=" + s
}
}
if s := query.Get("format"); s != "" {
input += ",format=" + s
}
return input, nil
return input
}
func GetInputTTS(src string) string {
query, err := url.ParseQuery(src)
if err != nil {
return ""
}
input := `-re -f lavfi -i "flite=text='` + query.Get("text") + `'`
// ffmpeg -f lavfi -i flite=list_voices=1
// awb, kal, kal16, rms, slt
if voice := query.Get("voice"); voice != "" {
input += ":voice" + voice
}
return input + `"`
}
+20
View File
@@ -0,0 +1,20 @@
package virtual
import (
"testing"
"github.com/stretchr/testify/require"
)
func TestGetInput(t *testing.T) {
s := GetInput("video")
require.Equal(t, "-re -f lavfi -i testsrc=decimals=2:size=1920x1080", s)
s = GetInput("video=testsrc2&size=4K")
require.Equal(t, "-re -f lavfi -i testsrc2=size=3840x2160", s)
}
func TestGetInputTTS(t *testing.T) {
s := GetInputTTS("text=hello world&voice=slt")
require.Equal(t, `-re -f lavfi -i "flite=text='hello world':voiceslt"`, s)
}
+1 -1
View File
@@ -57,7 +57,7 @@ func Init() {
// load static entries from Hass config
if err := importConfig(conf.Mod.Config); err != nil {
log.Debug().Msgf("[hass] can't import config: %s", err)
log.Trace().Msgf("[hass] can't import config: %s", err)
api.HandleFunc("api/hass", func(w http.ResponseWriter, _ *http.Request) {
http.Error(w, "no hass config", http.StatusNotFound)
+30 -2
View File
@@ -133,12 +133,19 @@ func Init() {
var log zerolog.Logger
var servers map[string]*server
func streamHandler(url string) (core.Producer, error) {
func streamHandler(rawURL string) (core.Producer, error) {
if srtp.Server == nil {
return nil, errors.New("homekit: can't work without SRTP server")
}
return homekit.Dial(url, srtp.Server)
rawURL, rawQuery, _ := strings.Cut(rawURL, "#")
client, err := homekit.Dial(rawURL, srtp.Server)
if client != nil && rawQuery != "" {
query := streams.ParseQuery(rawQuery)
client.Bitrate = parseBitrate(query.Get("bitrate"))
}
return client, err
}
func hapPairSetup(w http.ResponseWriter, r *http.Request) {
@@ -199,3 +206,24 @@ func findHomeKitURL(stream *streams.Stream) string {
return ""
}
func parseBitrate(s string) int {
n := len(s)
if n == 0 {
return 0
}
var k int
switch n--; s[n] {
case 'K':
k = 1024
s = s[:n]
case 'M':
k = 1024 * 1024
s = s[:n]
default:
k = 1
}
return k * core.Atoi(s)
}
+30 -1
View File
@@ -10,6 +10,7 @@ import (
"github.com/AlexxIT/go2rtc/internal/api"
"github.com/AlexxIT/go2rtc/internal/api/ws"
"github.com/AlexxIT/go2rtc/internal/app"
"github.com/AlexxIT/go2rtc/internal/ffmpeg"
"github.com/AlexxIT/go2rtc/internal/streams"
"github.com/AlexxIT/go2rtc/pkg/ascii"
@@ -17,17 +18,23 @@ import (
"github.com/AlexxIT/go2rtc/pkg/magic"
"github.com/AlexxIT/go2rtc/pkg/mjpeg"
"github.com/AlexxIT/go2rtc/pkg/tcp"
"github.com/rs/zerolog/log"
"github.com/AlexxIT/go2rtc/pkg/y4m"
"github.com/rs/zerolog"
)
func Init() {
api.HandleFunc("api/frame.jpeg", handlerKeyframe)
api.HandleFunc("api/stream.mjpeg", handlerStream)
api.HandleFunc("api/stream.ascii", handlerStream)
api.HandleFunc("api/stream.y4m", apiStreamY4M)
ws.HandleFunc("mjpeg", handlerWS)
log = app.GetLogger("mjpeg")
}
var log zerolog.Logger
func handlerKeyframe(w http.ResponseWriter, r *http.Request) {
src := r.URL.Query().Get("src")
stream := streams.Get(src)
@@ -166,3 +173,25 @@ func handlerWS(tr *ws.Transport, _ *ws.Message) error {
return nil
}
func apiStreamY4M(w http.ResponseWriter, r *http.Request) {
src := r.URL.Query().Get("src")
stream := streams.Get(src)
if stream == nil {
http.Error(w, api.StreamNotFound, http.StatusNotFound)
return
}
cons := y4m.NewConsumer()
cons.RemoteAddr = tcp.RemoteAddr(r)
cons.UserAgent = r.UserAgent()
if err := stream.AddConsumer(cons); err != nil {
log.Error().Err(err).Caller().Send()
return
}
_, _ = cons.WriteTo(w)
stream.RemoveConsumer(cons)
}
+14 -13
View File
@@ -1,6 +1,7 @@
package mp4
import (
"context"
"net/http"
"strconv"
"strings"
@@ -127,20 +128,20 @@ func handlerMP4(w http.ResponseWriter, r *http.Request) {
header.Set("Content-Disposition", `attachment; filename="`+filename+`"`)
}
var duration *time.Timer
if s := query.Get("duration"); s != "" {
if i, _ := strconv.Atoi(s); i > 0 {
duration = time.AfterFunc(time.Second*time.Duration(i), func() {
_ = cons.Stop()
})
}
ctx := r.Context() // handle when the client drops the connection
if i := core.Atoi(query.Get("duration")); i > 0 {
timeout := time.Second * time.Duration(i)
var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, timeout)
defer cancel()
}
go func() {
<-ctx.Done()
_ = cons.Stop()
stream.RemoveConsumer(cons)
}()
_, _ = cons.WriteTo(w)
stream.RemoveConsumer(cons)
if duration != nil {
duration.Stop()
}
}
-2
View File
@@ -7,7 +7,6 @@ import (
"github.com/AlexxIT/go2rtc/internal/streams"
"github.com/AlexxIT/go2rtc/pkg/aac"
"github.com/AlexxIT/go2rtc/pkg/tcp"
"github.com/rs/zerolog/log"
)
func apiStreamAAC(w http.ResponseWriter, r *http.Request) {
@@ -23,7 +22,6 @@ func apiStreamAAC(w http.ResponseWriter, r *http.Request) {
cons.UserAgent = r.UserAgent()
if err := stream.AddConsumer(cons); err != nil {
log.Error().Err(err).Caller().Send()
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
-2
View File
@@ -7,7 +7,6 @@ import (
"github.com/AlexxIT/go2rtc/internal/streams"
"github.com/AlexxIT/go2rtc/pkg/mpegts"
"github.com/AlexxIT/go2rtc/pkg/tcp"
"github.com/rs/zerolog/log"
)
func Init() {
@@ -36,7 +35,6 @@ func outputMpegTS(w http.ResponseWriter, r *http.Request) {
cons.UserAgent = r.UserAgent()
if err := stream.AddConsumer(cons); err != nil {
log.Error().Err(err).Caller().Send()
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
+5
View File
@@ -210,6 +210,11 @@ func tcpHandler(conn *rtsp.Conn) {
return
}
query := conn.URL.Query()
if s := query.Get("timeout"); s != "" {
conn.Timeout = core.Atoi(s)
}
log.Debug().Str("stream", name).Msg("[rtsp] new producer")
stream.AddProducer(conn)
+8
View File
@@ -0,0 +1,8 @@
## Testing notes
```yaml
streams:
test1-basic: ffmpeg:virtual?video#video=h264
test2-reconnect: ffmpeg:virtual?video&duration=10#video=h264
test3-execkill: exec:./examples/rtsp_client/rtsp_client/rtsp_client {output}
```
+10 -6
View File
@@ -2,6 +2,8 @@ package streams
import (
"errors"
"time"
"github.com/AlexxIT/go2rtc/pkg/core"
)
@@ -80,18 +82,20 @@ func (s *Stream) Play(source string) error {
s.AddInternalProducer(src)
s.AddInternalConsumer(cons)
go func() {
_ = src.Start()
_ = dst.Stop()
s.RemoveProducer(src)
}()
go func() {
_ = dst.Start()
_ = src.Stop()
s.RemoveInternalConsumer(cons)
}()
go func() {
_ = src.Start()
// little timeout before stop dst, so the buffer can be transferred
time.Sleep(time.Second)
_ = dst.Stop()
s.RemoveProducer(src)
}()
return nil
}
+5 -1
View File
@@ -207,7 +207,7 @@ func (p *Producer) reconnect(workerID, retry int) {
for _, media := range conn.GetMedias() {
switch media.Direction {
case core.DirectionRecvonly:
for _, receiver := range p.receivers {
for i, receiver := range p.receivers {
codec := media.MatchCodec(receiver.Codec)
if codec == nil {
continue
@@ -219,6 +219,7 @@ func (p *Producer) reconnect(workerID, retry int) {
}
receiver.Replace(track)
p.receivers[i] = track
break
}
@@ -234,6 +235,9 @@ func (p *Producer) reconnect(workerID, retry int) {
}
}
// stop previous connection after moving tracks (fix ghost exec/ffmpeg)
_ = p.conn.Stop()
// swap connections
p.conn = conn
go p.worker(conn, workerID)
+16 -4
View File
@@ -1,6 +1,7 @@
package streams
import (
"errors"
"net/http"
"net/url"
"regexp"
@@ -49,9 +50,16 @@ func Get(name string) *Stream {
var sanitize = regexp.MustCompile(`\s`)
func New(name string, source string) *Stream {
// not allow creating dynamic streams with spaces in the source
// Validate - not allow creating dynamic streams with spaces in the source
func Validate(source string) error {
if sanitize.MatchString(source) {
return errors.New("streams: invalid dynamic source")
}
return nil
}
func New(name string, source string) *Stream {
if Validate(source) != nil {
return nil
}
@@ -203,13 +211,17 @@ func streamsHandler(w http.ResponseWriter, r *http.Request) {
// with dst - redirect source to dst
if dst := query.Get("dst"); dst != "" {
if stream := Get(dst); stream != nil {
if err := stream.Play(src); err != nil {
if err := Validate(src); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
} else if err = stream.Play(src); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
} else {
api.ResponseJSON(w, stream)
}
} else if stream = Get(src); stream != nil {
if err := stream.Publish(dst); err != nil {
if err := Validate(dst); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
} else if err = stream.Publish(dst); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
}
} else {
+2
View File
@@ -36,6 +36,8 @@ import (
)
func main() {
app.Version = "1.9.3"
// 1. Core modules: app, api/ws, streams
app.Init() // init config and logs
+2
View File
@@ -69,6 +69,8 @@ func DecodeConfig(b []byte) (objType, sampleFreqIdx, channels byte, sampleRate u
sampleFreqIdx = rd.ReadBits8(4)
if sampleFreqIdx == 0b1111 {
sampleRate = rd.ReadBits(24)
} else {
sampleRate = sampleRates[sampleFreqIdx]
}
channels = rd.ReadBits8(4)
+9
View File
@@ -41,3 +41,12 @@ func TestADTS(t *testing.T) {
require.Equal(t, src[:len(dst)], dst)
}
func TestEncodeConfig(t *testing.T) {
conf := EncodeConfig(TypeAACLC, 48000, 1, false)
require.Equal(t, "1188", hex.EncodeToString(conf))
conf = EncodeConfig(TypeAACLC, 16000, 1, false)
require.Equal(t, "1408", hex.EncodeToString(conf))
conf = EncodeConfig(TypeAACLC, 8000, 1, false)
require.Equal(t, "1588", hex.EncodeToString(conf))
}
+8 -1
View File
@@ -156,7 +156,7 @@ const x256b = "\x00\x00\x00\x00\x80\x80\x80\xc0\x80\x00\x00\x00\xff\xff\xff\xff\
func xterm256color(r, g, b uint8, n int) (index uint8) {
best := uint16(0xFFFF)
for i := 0; i < n; i++ {
diff := uint16(r-x256r[i]) + uint16(g-x256g[i]) + uint16(b-x256b[i])
diff := sqDiff(r, x256r[i]) + sqDiff(g, x256g[i]) + sqDiff(b, x256b[i])
if diff < best {
best = diff
index = uint8(i)
@@ -164,3 +164,10 @@ func xterm256color(r, g, b uint8, n int) (index uint8) {
}
return
}
// sqDiff - just like from image/color/color.go
func sqDiff(x, y uint8) uint16 {
d := uint16(x - y)
//return d
return (d * d) >> 2
}
+4
View File
@@ -131,3 +131,7 @@ func (r *Reader) ReadSEGolomb() int32 {
func (r *Reader) Left() []byte {
return r.buf[r.pos:]
}
func (r *Reader) Pos() (int, byte) {
return r.pos - 1, r.bits
}
+55 -20
View File
@@ -2,8 +2,8 @@ package core
import (
"encoding/base64"
"encoding/json"
"fmt"
"strconv"
"strings"
"unicode"
@@ -18,34 +18,70 @@ type Codec struct {
PayloadType uint8
}
func (c *Codec) String() string {
s := fmt.Sprintf("%d %s", c.PayloadType, c.Name)
// MarshalJSON - return FFprobe compatible output
func (c *Codec) MarshalJSON() ([]byte, error) {
info := map[string]any{}
if name := FFmpegCodecName(c.Name); name != "" {
info["codec_name"] = name
info["codec_type"] = c.Kind()
}
if c.Name == CodecH264 {
profile, level := DecodeH264(c.FmtpLine)
if profile != "" {
info["profile"] = profile
info["level"] = level
}
}
if c.ClockRate != 0 && c.ClockRate != 90000 {
s = fmt.Sprintf("%s/%d", s, c.ClockRate)
info["sample_rate"] = c.ClockRate
}
if c.Channels > 0 {
s = fmt.Sprintf("%s/%d", s, c.Channels)
info["channels"] = c.Channels
}
return s
return json.Marshal(info)
}
func (c *Codec) Text() string {
switch c.Name {
func FFmpegCodecName(name string) string {
switch name {
case CodecH264:
if profile := DecodeH264(c.FmtpLine); profile != "" {
return "H.264 " + profile
}
return c.Name
return "h264"
case CodecH265:
return "h265"
case CodecJPEG:
return "mjpeg"
case CodecRAW:
return "rawvideo"
case CodecPCMA:
return "pcm_alaw"
case CodecPCMU:
return "pcm_mulaw"
case CodecPCM:
return "pcm_s16be"
case CodecPCML:
return "pcm_s16le"
case CodecAAC:
return "aac"
case CodecOpus:
return "opus"
case CodecVP8:
return "vp8"
case CodecVP9:
return "vp9"
case CodecAV1:
return "av1"
}
return ""
}
s := c.Name
func (c *Codec) String() (s string) {
s = c.Name
if c.ClockRate != 0 && c.ClockRate != 90000 {
s += "/" + strconv.Itoa(int(c.ClockRate))
s += fmt.Sprintf("/%d", c.ClockRate)
}
if c.Channels > 0 {
s += "/" + strconv.Itoa(int(c.Channels))
s += fmt.Sprintf("/%d", c.Channels)
}
return s
return
}
func (c *Codec) IsRTP() bool {
@@ -181,10 +217,9 @@ func UnmarshalCodec(md *sdp.MediaDescription, payloadType string) *Codec {
return c
}
func DecodeH264(fmtp string) string {
func DecodeH264(fmtp string) (profile string, level byte) {
if ps := Between(fmtp, "sprop-parameter-sets=", ","); ps != "" {
if sps, _ := base64.StdEncoding.DecodeString(ps); len(sps) >= 4 {
var profile string
switch sps[1] {
case 0x42:
profile = "Baseline"
@@ -198,8 +233,8 @@ func DecodeH264(fmtp string) string {
profile = fmt.Sprintf("0x%02X", sps[1])
}
return fmt.Sprintf("%s %d.%d", profile, sps[3]/10, sps[3]%10)
level = sps[3]
}
}
return ""
return
}
+1
View File
@@ -18,6 +18,7 @@ const (
CodecVP9 = "VP9"
CodecAV1 = "AV1"
CodecJPEG = "JPEG" // payloadType: 26
CodecRAW = "RAW"
CodecPCMU = "PCMU" // payloadType: 0
CodecPCMA = "PCMA" // payloadType: 8
+120
View File
@@ -0,0 +1,120 @@
package core
import (
"fmt"
"testing"
"github.com/stretchr/testify/require"
)
type producer struct {
Medias []*Media
Receivers []*Receiver
id byte
}
func (p *producer) GetMedias() []*Media {
return p.Medias
}
func (p *producer) GetTrack(_ *Media, codec *Codec) (*Receiver, error) {
for _, receiver := range p.Receivers {
if receiver.Codec == codec {
return receiver, nil
}
}
receiver := NewReceiver(nil, codec)
p.Receivers = append(p.Receivers, receiver)
return receiver, nil
}
func (p *producer) Start() error {
pkt := &Packet{Payload: []byte{p.id}}
p.Receivers[0].Input(pkt)
return nil
}
func (p *producer) Stop() error {
for _, receiver := range p.Receivers {
receiver.Close()
}
return nil
}
type consumer struct {
Medias []*Media
Senders []*Sender
cache chan byte
}
func (c *consumer) GetMedias() []*Media {
return c.Medias
}
func (c *consumer) AddTrack(_ *Media, _ *Codec, track *Receiver) error {
c.cache = make(chan byte, 1)
sender := NewSender(nil, track.Codec)
sender.Output = func(packet *Packet) {
c.cache <- packet.Payload[0]
}
sender.HandleRTP(track)
c.Senders = append(c.Senders, sender)
return nil
}
func (c *consumer) Stop() error {
for _, sender := range c.Senders {
sender.Close()
}
return nil
}
func (c *consumer) read() byte {
return <-c.cache
}
func TestName(t *testing.T) {
GetProducer := func(b byte) Producer {
return &producer{
Medias: []*Media{
{
Kind: KindVideo,
Direction: DirectionRecvonly,
Codecs: []*Codec{
{Name: CodecH264},
},
},
},
id: b,
}
}
// stage1
prod1 := GetProducer(1)
cons2 := &consumer{}
media1 := prod1.GetMedias()[0]
track1, _ := prod1.GetTrack(media1, media1.Codecs[0])
_ = cons2.AddTrack(nil, nil, track1)
_ = prod1.Start()
require.Equal(t, byte(1), cons2.read())
// stage2
prod2 := GetProducer(2)
media2 := prod2.GetMedias()[0]
require.NotEqual(t, fmt.Sprintf("%p", media1), fmt.Sprintf("%p", media2))
track2, _ := prod2.GetTrack(media2, media2.Codecs[0])
track1.Replace(track2)
_ = prod1.Stop()
_ = prod2.Start()
require.Equal(t, byte(2), cons2.read())
// stage3
_ = prod2.Stop()
}
+7
View File
@@ -38,6 +38,13 @@ func RandString(size, base byte) string {
return string(b)
}
func Before(s, sep string) string {
if i := strings.Index(s, sep); i > 0 {
return s[:i]
}
return s
}
func Between(s, sub1, sub2 string) string {
i := strings.Index(s, sub1)
if i < 0 {
+1 -1
View File
@@ -22,7 +22,7 @@ type Media struct {
func (m *Media) String() string {
s := fmt.Sprintf("%s, %s", m.Kind, m.Direction)
for _, codec := range m.Codecs {
name := codec.Text()
name := codec.String()
if strings.Contains(s, name) {
continue
+87
View File
@@ -0,0 +1,87 @@
package core
import (
"sync"
"github.com/pion/rtp"
)
//type Packet struct {
// Payload []byte
// Timestamp uint32 // PTS if DTS == 0 else DTS
// Composition uint32 // CTS = PTS-DTS (for support B-frames)
// Sequence uint16
//}
type Packet = rtp.Packet
// HandlerFunc - process input packets (just like http.HandlerFunc)
type HandlerFunc func(packet *Packet)
// Filter - a decorator for any HandlerFunc
type Filter func(handler HandlerFunc) HandlerFunc
// Node - Receiver or Sender or Filter (transform)
type Node struct {
Codec *Codec `json:"codec"`
Input HandlerFunc `json:"-"`
Output HandlerFunc `json:"-"`
childs []*Node
parent *Node
mu sync.Mutex
}
func (n *Node) WithParent(parent *Node) *Node {
parent.AppendChild(n)
return n
}
func (n *Node) AppendChild(child *Node) {
n.mu.Lock()
n.childs = append(n.childs, child)
n.mu.Unlock()
child.parent = n
}
func (n *Node) RemoveChild(child *Node) {
n.mu.Lock()
for i, ch := range n.childs {
if ch == child {
n.childs = append(n.childs[:i], n.childs[i+1:]...)
break
}
}
n.mu.Unlock()
}
func (n *Node) Close() {
if parent := n.parent; parent != nil {
parent.RemoveChild(n)
if len(parent.childs) == 0 {
parent.Close()
}
} else {
for _, childs := range n.childs {
childs.Close()
}
}
}
func MoveNode(dst, src *Node) {
src.mu.Lock()
childs := src.childs
src.childs = nil
src.mu.Unlock()
dst.mu.Lock()
dst.childs = childs
dst.mu.Unlock()
for _, child := range childs {
child.parent = dst
}
}
+115 -167
View File
@@ -1,225 +1,173 @@
package core
import (
"encoding/json"
"errors"
"fmt"
"strconv"
"sync"
"github.com/pion/rtp"
)
type Packet struct {
PayloadType uint8
Sequence uint16
Timestamp uint32 // PTS if DTS == 0 else DTS
Composition uint32 // CTS = PTS-DTS (for support B-frames)
Payload []byte
}
var ErrCantGetTrack = errors.New("can't get track")
type Receiver struct {
Codec *Codec
Media *Media
Node
ID byte // Channel for RTSP, PayloadType for MPEG-TS
// Deprecated: should be removed
Media *Media `json:"-"`
// Deprecated: should be removed
ID byte `json:"-"` // Channel for RTSP, PayloadType for MPEG-TS
senders map[*Sender]chan *rtp.Packet
mu sync.RWMutex
bytes int
Bytes int `json:"bytes,omitempty"`
Packets int `json:"packets,omitempty"`
}
func NewReceiver(media *Media, codec *Codec) *Receiver {
Assert(codec != nil)
return &Receiver{Codec: codec, Media: media}
}
// WriteRTP - fast and non blocking write to all readers buffers
func (t *Receiver) WriteRTP(packet *rtp.Packet) {
t.mu.Lock()
t.bytes += len(packet.Payload)
for sender, buffer := range t.senders {
select {
case buffer <- packet:
default:
sender.overflow++
r := &Receiver{
Node: Node{Codec: codec},
Media: media,
}
r.Input = func(packet *Packet) {
r.Bytes += len(packet.Payload)
r.Packets++
for _, child := range r.childs {
child.Input(packet)
}
}
t.mu.Unlock()
return r
}
func (t *Receiver) Senders() (senders []*Sender) {
t.mu.RLock()
for sender := range t.senders {
senders = append(senders, sender)
// Deprecated: should be removed
func (r *Receiver) WriteRTP(packet *rtp.Packet) {
r.Input(packet)
}
// Deprecated: should be removed
func (r *Receiver) Senders() []*Sender {
if len(r.childs) > 0 {
return []*Sender{{}}
} else {
return nil
}
t.mu.RUnlock()
return
}
func (t *Receiver) Close() {
t.mu.Lock()
// close all sender channel buffers and erase senders list
for _, buffer := range t.senders {
close(buffer)
}
t.senders = nil
t.mu.Unlock()
// Deprecated: should be removed
func (r *Receiver) Replace(target *Receiver) {
MoveNode(&target.Node, &r.Node)
}
func (t *Receiver) Replace(target *Receiver) {
// move this receiver senders to new receiver
t.mu.Lock()
senders := t.senders
t.mu.Unlock()
target.mu.Lock()
target.senders = senders
target.mu.Unlock()
}
func (t *Receiver) String() string {
s := t.Codec.String() + ", bytes=" + strconv.Itoa(t.bytes)
t.mu.RLock()
s += fmt.Sprintf(", senders=%d", len(t.senders))
t.mu.RUnlock()
return s
}
func (t *Receiver) MarshalJSON() ([]byte, error) {
return json.Marshal(t.String())
func (r *Receiver) Close() {
r.Node.Close()
}
type Sender struct {
Codec *Codec
Media *Media
Node
Handler HandlerFunc
// Deprecated:
Media *Media `json:"-"`
// Deprecated:
Handler HandlerFunc `json:"-"`
receivers []*Receiver
mu sync.RWMutex
bytes int
Bytes int `json:"bytes,omitempty"`
Packets int `json:"packets,omitempty"`
Drops int `json:"drops,omitempty"`
overflow int
buf chan *Packet
done chan struct{}
}
func NewSender(media *Media, codec *Codec) *Sender {
return &Sender{Codec: codec, Media: media}
}
var bufSize uint16
// HandlerFunc like http.HandlerFunc
type HandlerFunc func(packet *rtp.Packet)
func (s *Sender) HandleRTP(track *Receiver) {
s.Bind(track)
go s.worker(track)
}
func (s *Sender) Bind(track *Receiver) {
var bufferSize uint16
if GetKind(track.Codec.Name) == KindVideo {
if track.Codec.IsRTP() {
if GetKind(codec.Name) == KindVideo {
if codec.IsRTP() {
// in my tests 40Mbit/s 4K-video can generate up to 1500 items
// for the h264.RTPDepay => RTPPay queue
bufferSize = 5000
bufSize = 4096
} else {
bufferSize = 50
bufSize = 64
}
} else {
bufferSize = 100
bufSize = 128
}
buffer := make(chan *rtp.Packet, bufferSize)
track.mu.Lock()
if track.senders == nil {
track.senders = map[*Sender]chan *rtp.Packet{}
buf := make(chan *Packet, bufSize)
s := &Sender{
Node: Node{Codec: codec},
Media: media,
buf: buf,
}
track.senders[s] = buffer
track.mu.Unlock()
s.mu.Lock()
s.receivers = append(s.receivers, track)
s.mu.Unlock()
s.Input = func(packet *Packet) {
// writing to nil chan - OK, writing to closed chan - panic
s.mu.Lock()
select {
case s.buf <- packet:
s.Bytes += len(packet.Payload)
s.Packets++
default:
s.Drops++
}
s.mu.Unlock()
}
s.Output = func(packet *Packet) {
s.Handler(packet)
}
return s
}
func (s *Sender) worker(track *Receiver) {
track.mu.Lock()
buffer := track.senders[s]
track.mu.Unlock()
// Deprecated: should be removed
func (s *Sender) HandleRTP(parent *Receiver) {
s.WithParent(parent)
s.Start()
}
// read packets from buffer channel until it will be closed
if buffer != nil {
for packet := range buffer {
s.bytes += len(packet.Payload)
s.Handler(packet)
}
}
// Deprecated: should be removed
func (s *Sender) Bind(parent *Receiver) {
s.WithParent(parent)
}
// remove current receiver from list
// it can only happen when receiver close buffer channel
s.mu.Lock()
for i, receiver := range s.receivers {
if receiver == track {
s.receivers = append(s.receivers[:i], s.receivers[i+1:]...)
break
}
}
s.mu.Unlock()
func (s *Sender) WithParent(parent *Receiver) *Sender {
s.Node.WithParent(&parent.Node)
return s
}
func (s *Sender) Start() {
s.mu.Lock()
for _, track := range s.receivers {
go s.worker(track)
defer s.mu.Unlock()
if s.buf == nil || s.done != nil {
return
}
s.mu.Unlock()
s.done = make(chan struct{})
go func() {
for packet := range s.buf {
s.Output(packet)
}
close(s.done)
}()
}
func (s *Sender) Wait() {
if done := s.done; s.done != nil {
<-done
}
}
func (s *Sender) State() string {
if s.buf == nil {
return "closed"
}
if s.done == nil {
return "new"
}
return "connected"
}
func (s *Sender) Close() {
s.mu.Lock()
// remove this sender from all receivers list
for _, receiver := range s.receivers {
receiver.mu.Lock()
if buffer := receiver.senders[s]; buffer != nil {
// remove channel from list
delete(receiver.senders, s)
// close channel
close(buffer)
}
receiver.mu.Unlock()
// close buffer if exists
if buf := s.buf; buf != nil {
s.buf = nil
defer close(buf)
}
s.receivers = nil
s.mu.Unlock()
}
func (s *Sender) String() string {
info := s.Codec.String() + ", bytes=" + strconv.Itoa(s.bytes)
s.mu.RLock()
info += ", receivers=" + strconv.Itoa(len(s.receivers))
s.mu.RUnlock()
if s.overflow > 0 {
info += ", overflow=" + strconv.Itoa(s.overflow)
}
return info
}
func (s *Sender) MarshalJSON() ([]byte, error) {
return json.Marshal(s.String())
}
// VA - helper, for extract video and audio receivers from list
func VA(receivers []*Receiver) (video, audio *Receiver) {
for _, receiver := range receivers {
switch GetKind(receiver.Codec.Name) {
case KindVideo:
video = receiver
case KindAudio:
audio = receiver
}
}
return
s.Node.Close()
}
+2 -2
View File
@@ -24,7 +24,7 @@ func Logger(include func(packet *rtp.Packet) bool) func(packet *rtp.Packet) {
now := time.Now()
fmt.Printf(
"%s: size:%6d, ts:%10d, type:%2d, ssrc:%d, seq:%5d, mark:%t, dts:%4d, dtime:%3d\n",
"%s: size=%6d ts=%10d type=%2d ssrc=%d seq=%5d mark=%t dts=%4d dtime=%3dms\n",
now.Format("15:04:05.000"),
len(packet.Payload), packet.Timestamp, packet.PayloadType, packet.SSRC, packet.SequenceNumber, packet.Marker,
packet.Timestamp-lastTS, now.Sub(lastTime).Milliseconds(),
@@ -41,7 +41,7 @@ func Logger(include func(packet *rtp.Packet) bool) func(packet *rtp.Packet) {
if dt := now.Sub(secTime); dt > time.Second {
fmt.Printf(
"%s: size:%6d, cnt:%d, dts: %d, dtime:%d\n",
"%s: size=%6d cnt=%d dts=%d dtime=%3dms\n",
now.Format("15:04:05.000"),
secSize, secCnt, lastTS-secTS, dt.Milliseconds(),
)
+1 -1
View File
@@ -114,7 +114,7 @@ func (c *Client) Play() error {
}
func (c *Client) Talk() error {
format := `{"Name":"OPTalk","SessionID":"0x%08X","OPTalk":{"Action":"%s"}}` + "\x0A\x00"
format := `{"Name":"OPTalk","SessionID":"0x%08X","OPTalk":{"Action":"%s","AudioFormat":{"EncodeType":"G711_ALAW"}}}` + "\x0A\x00"
data := fmt.Sprintf(format, c.session, "Claim")
if _, err := c.WriteCmd(OPTalkClaim, []byte(data)); err != nil {
+2 -1
View File
@@ -23,8 +23,9 @@ func Dial(url string) (core.Producer, error) {
Kind: core.KindAudio,
Direction: core.DirectionSendonly,
Codecs: []*core.Codec{
// leave only one codec here for better compatibility with cameras
// https://github.com/AlexxIT/go2rtc/issues/1111
{Name: core.CodecPCMA, ClockRate: 8000, PayloadType: 8},
{Name: core.CodecPCMU, ClockRate: 8000, PayloadType: 0},
},
},
}
+30
View File
@@ -6,6 +6,15 @@ import (
"strings"
)
// correlation of libavformat versions with ffmpeg versions
const (
Version50 = "59. 16"
Version51 = "59. 27"
Version60 = "60. 3"
Version61 = "60. 16"
Version70 = "61. 1"
)
type Args struct {
Bin string // ffmpeg
Global string // -hide_banner -v error
@@ -13,6 +22,7 @@ type Args struct {
Codecs []string // -c:v libx264 -g:v 30 -preset:v ultrafast -tune:v zerolatency
Filters []string // scale=1920:1080
Output string // -f rtsp {output}
Version string // libavformat version, it's more reliable than the ffmpeg version
Video, Audio int // count of Video and Audio params
}
@@ -52,6 +62,11 @@ func (a *Args) String() string {
}
b.WriteByte(' ')
// starting from FFmpeg 6.1 readrate=1 has default initial bust 0.5 sec
// it might make us miss the first couple seconds of the file
if strings.HasPrefix(a.Input, "-re ") && a.Version >= Version61 {
b.WriteString("-readrate_initial_burst 0.001 ")
}
b.WriteString(a.Input)
multimode := a.Video > 1 || a.Audio > 1
@@ -91,3 +106,18 @@ func (a *Args) String() string {
return b.String()
}
func ParseVersion(b []byte) (ffmpeg string, libavformat string) {
if len(b) > 100 {
// ffmpeg version n7.0-30-g8b0fe91754-20240520 Copyright (c) 2000-2024 the FFmpeg developers
if i := bytes.IndexByte(b[15:], ' '); i > 0 {
ffmpeg = string(b[15 : 15+i])
}
// libavformat 60. 16.100 / 60. 16.100
if i := strings.Index(string(b), "libavformat"); i > 0 {
libavformat = string(b[i+15 : i+25])
}
}
return
}
+64
View File
@@ -137,6 +137,70 @@ func TestNewReader(t *testing.T) {
},
},
},
{
name: "mediamtx",
actual: "0200075f726573756c74003ff0000000000000030006666d7356657202000d4c4e5820392c302c3132342c32000c6361706162696c697469657300403f0000000000000000090300056c6576656c0200067374617475730004636f646502001d4e6574436f6e6e656374696f6e2e436f6e6e6563742e53756363657373000b6465736372697074696f6e020015436f6e6e656374696f6e207375636365656465642e000e6f626a656374456e636f64696e67000000000000000000000009",
expect: []any{
"_result", float64(1), map[string]any{
"capabilities": float64(31),
"fmsVer": "LNX 9,0,124,2",
}, map[string]any{
"code": "NetConnection.Connect.Success",
"description": "Connection succeeded.",
"level": "status",
"objectEncoding": float64(0),
},
},
},
{
name: "mediamtx",
actual: "0200075f726573756c7400401000000000000005003ff0000000000000",
expect: []any{"_result", float64(4), any(nil), float64(1)},
},
{
name: "mediamtx",
actual: "0200086f6e537461747573004014000000000000050300056c6576656c0200067374617475730004636f64650200144e657453747265616d2e506c61792e5265736574000b6465736372697074696f6e02000a706c6179207265736574000009",
expect: []any{
"onStatus", float64(5), any(nil), map[string]any{
"code": "NetStream.Play.Reset",
"description": "play reset",
"level": "status",
},
},
},
{
name: "mediamtx",
actual: "0200086f6e537461747573004014000000000000050300056c6576656c0200067374617475730004636f64650200144e657453747265616d2e506c61792e5374617274000b6465736372697074696f6e02000a706c6179207374617274000009",
expect: []any{
"onStatus", float64(5), any(nil), map[string]any{
"code": "NetStream.Play.Start",
"description": "play start",
"level": "status",
},
},
},
{
name: "mediamtx",
actual: "0200086f6e537461747573004014000000000000050300056c6576656c0200067374617475730004636f64650200144e657453747265616d2e446174612e5374617274000b6465736372697074696f6e02000a64617461207374617274000009",
expect: []any{
"onStatus", float64(5), any(nil), map[string]any{
"code": "NetStream.Data.Start",
"description": "data start",
"level": "status",
},
},
},
{
name: "mediamtx",
actual: "0200086f6e537461747573004014000000000000050300056c6576656c0200067374617475730004636f646502001c4e657453747265616d2e506c61792e5075626c6973684e6f74696679000b6465736372697074696f6e02000e7075626c697368206e6f74696679000009",
expect: []any{
"onStatus", float64(5), any(nil), map[string]any{
"code": "NetStream.Play.PublishNotify",
"description": "publish notify",
"level": "status",
},
},
},
{
name: "obs-connect",
actual: "020007636f6e6e656374003ff000000000000003000361707002000c617070312f73747265616d3100047479706502000a6e6f6e70726976617465000e737570706f727473476f4177617901010008666c61736856657202001f464d4c452f332e302028636f6d70617469626c653b20464d53632f312e3029000673776655726c02002272746d703a2f2f3139322e3136382e31302e3130312f617070312f73747265616d310005746355726c02002272746d703a2f2f3139322e3136382e31302e3130312f617070312f73747265616d31000009",
+2
View File
@@ -54,6 +54,8 @@ func (m *Muxer) GetInit() []byte {
sps, pps := h264.GetParameterSet(codec.FmtpLine)
if len(sps) == 0 {
sps = []byte{0x67, 0x42, 0x00, 0x0a, 0xf8, 0x41, 0xa2}
} else {
h264.FixPixFmt(sps)
}
if len(pps) == 0 {
pps = []byte{0x68, 0xce, 0x38, 0x80}
+134 -1
View File
@@ -1,6 +1,10 @@
package h264
import "github.com/AlexxIT/go2rtc/pkg/bits"
import (
"fmt"
"github.com/AlexxIT/go2rtc/pkg/bits"
)
// http://www.itu.int/rec/T-REC-H.264
// https://webrtc.googlesource.com/src/+/refs/heads/main/common_video/h264/sps_parser.cc
@@ -229,3 +233,132 @@ func (s *SPS) scaling_list(r *bits.Reader, sizeOfScalingList int) {
}
}
}
func (s *SPS) Profile() string {
switch s.profile_idc {
case 0x42:
return "Baseline"
case 0x4D:
return "Main"
case 0x58:
return "Extended"
case 0x64:
return "High"
}
return fmt.Sprintf("0x%02X", s.profile_idc)
}
func (s *SPS) PixFmt() string {
if s.bit_depth_luma_minus8 == 0 {
switch s.chroma_format_idc {
case 1:
if s.video_full_range_flag == 1 {
return "yuvj420p"
}
return "yuv420p"
case 2:
return "yuv422p"
case 3:
return "yuv444p"
}
}
return ""
}
func (s *SPS) String() string {
return fmt.Sprintf(
"%s %d.%d, %s, %dx%d",
s.Profile(), s.level_idc/10, s.level_idc%10, s.PixFmt(), s.Width(), s.Height(),
)
}
// FixPixFmt - change yuvj420p to yuv420p in SPS
// same as "-c:v copy -bsf:v h264_metadata=video_full_range_flag=0"
func FixPixFmt(sps []byte) {
r := bits.NewReader(sps)
_ = r.ReadByte()
profile := r.ReadByte()
_ = r.ReadByte()
_ = r.ReadByte()
_ = r.ReadUEGolomb()
switch profile {
case 100, 110, 122, 244, 44, 83, 86, 118, 128, 138, 139, 134, 135:
n := byte(8)
if r.ReadUEGolomb() == 3 {
_ = r.ReadBit()
n = 12
}
_ = r.ReadUEGolomb()
_ = r.ReadUEGolomb()
_ = r.ReadBit()
if r.ReadBit() != 0 {
for i := byte(0); i < n; i++ {
if r.ReadBit() != 0 {
return // skip
}
}
}
}
_ = r.ReadUEGolomb()
switch r.ReadUEGolomb() {
case 0:
_ = r.ReadUEGolomb()
case 1:
_ = r.ReadBit()
_ = r.ReadSEGolomb()
_ = r.ReadSEGolomb()
n := r.ReadUEGolomb()
for i := uint32(0); i < n; i++ {
_ = r.ReadSEGolomb()
}
}
_ = r.ReadUEGolomb()
_ = r.ReadBit()
_ = r.ReadUEGolomb()
_ = r.ReadUEGolomb()
if r.ReadBit() == 0 {
_ = r.ReadBit()
}
_ = r.ReadBit()
if r.ReadBit() != 0 {
_ = r.ReadUEGolomb()
_ = r.ReadUEGolomb()
_ = r.ReadUEGolomb()
_ = r.ReadUEGolomb()
}
if r.ReadBit() != 0 {
if r.ReadBit() != 0 {
if r.ReadByte() == 255 {
_ = r.ReadUint16()
_ = r.ReadUint16()
}
}
if r.ReadBit() != 0 {
_ = r.ReadBit()
}
if r.ReadBit() != 0 {
_ = r.ReadBits8(3)
if r.ReadBit() == 1 {
pos, bit := r.Pos()
sps[pos] &= ^byte(1 << bit)
}
}
}
}
+10 -3
View File
@@ -15,7 +15,8 @@ type Stream struct {
}
func NewStream(
client *hap.Client, videoCodec *VideoCodec, audioCodec *AudioCodec, videoSession, audioSession *srtp.Session,
client *hap.Client, videoCodec *VideoCodec, audioCodec *AudioCodec,
videoSession, audioSession *srtp.Session, bitrate int,
) (*Stream, error) {
stream := &Stream{
id: core.RandString(16, 0),
@@ -30,11 +31,17 @@ func NewStream(
return nil, err
}
if bitrate != 0 {
bitrate /= 1024 // convert bps to kbps
} else {
bitrate = 4096 // default kbps for general FullHD camera
}
videoCodec.RTPParams = []RTPParams{
{
PayloadType: 99,
SSRC: videoSession.Local.SSRC,
MaxBitrate: 299,
MaxBitrate: uint16(bitrate), // iPhone query 299Kbps, iPad/AppleTV query 802Kbps
RTCPInterval: 0.5,
MaxMTU: []uint16{1378},
},
@@ -43,7 +50,7 @@ func NewStream(
{
PayloadType: 110,
SSRC: audioSession.Local.SSRC,
MaxBitrate: 24,
MaxBitrate: 24, // any iDevice query 24Kbps (this is OK for 16KHz and 1 channel)
RTCPInterval: 5,
ComfortNoisePayloadType: []uint8{13},
+3 -1
View File
@@ -28,6 +28,8 @@ type Client struct {
audioSession *srtp.Session
stream *camera.Stream
Bitrate int // in bits/s
}
func Dial(rawURL string, server *srtp.Server) (*Client, error) {
@@ -132,7 +134,7 @@ func (c *Client) Start() error {
c.audioSession = &srtp.Session{Local: c.srtpEndpoint()}
var err error
c.stream, err = camera.NewStream(c.hap, videoCodec, audioCodec, c.videoSession, c.audioSession)
c.stream, err = camera.NewStream(c.hap, videoCodec, audioCodec, c.videoSession, c.audioSession, c.Bitrate)
if err != nil {
return err
}
+1 -1
View File
@@ -3,7 +3,7 @@ package homekit
import (
"fmt"
"io"
"math/rand"
"math/rand/v2"
"net"
"time"
+10 -2
View File
@@ -14,6 +14,8 @@ import (
"github.com/AlexxIT/go2rtc/pkg/magic/mjpeg"
"github.com/AlexxIT/go2rtc/pkg/mpegts"
"github.com/AlexxIT/go2rtc/pkg/multipart"
"github.com/AlexxIT/go2rtc/pkg/wav"
"github.com/AlexxIT/go2rtc/pkg/y4m"
)
func Open(r io.Reader) (core.Producer, error) {
@@ -25,9 +27,15 @@ func Open(r io.Reader) (core.Producer, error) {
}
switch {
case bytes.HasPrefix(b, []byte(annexb.StartCode)):
case string(b) == annexb.StartCode:
return bitstream.Open(rd)
case string(b) == wav.FourCC:
return wav.Open(rd)
case string(b) == y4m.FourCC:
return y4m.Open(rd)
case bytes.HasPrefix(b, []byte{0xFF, 0xD8}):
return mjpeg.Open(rd)
@@ -37,7 +45,7 @@ func Open(r io.Reader) (core.Producer, error) {
case bytes.HasPrefix(b, []byte("--")):
return multipart.Open(rd)
case b[0] == 0xFF && b[1]&0xF7 == 0xF1:
case b[0] == 0xFF && (b[1] == 0xF1 || b[1] == 0xF9):
return aac.Open(rd)
case b[0] == mpegts.SyncByte:
+3
View File
@@ -22,6 +22,7 @@ func NewConsumer() *Consumer {
Direction: core.DirectionSendonly,
Codecs: []*core.Codec{
{Name: core.CodecJPEG},
{Name: core.CodecRAW},
},
},
},
@@ -40,6 +41,8 @@ func (c *Consumer) AddTrack(media *core.Media, _ *core.Codec, track *core.Receiv
if track.Codec.IsRTP() {
sender.Handler = RTPDepay(sender.Handler)
} else if track.Codec.Name == core.CodecRAW {
sender.Handler = Encoder(track.Codec, sender.Handler)
}
sender.HandleRTP(track)
+21
View File
@@ -3,6 +3,10 @@ package mjpeg
import (
"bytes"
"image/jpeg"
"github.com/AlexxIT/go2rtc/pkg/core"
"github.com/AlexxIT/go2rtc/pkg/y4m"
"github.com/pion/rtp"
)
// FixJPEG - reencode JPEG if it has wrong header
@@ -33,3 +37,20 @@ func FixJPEG(b []byte) []byte {
}
return buf.Bytes()
}
func Encoder(codec *core.Codec, handler core.HandlerFunc) core.HandlerFunc {
newImage := y4m.NewImage(codec.FmtpLine)
return func(packet *rtp.Packet) {
img := newImage(packet.Payload)
buf := bytes.NewBuffer(nil)
if err := jpeg.Encode(buf, img, nil); err != nil {
return
}
clone := *packet
clone.Payload = buf.Bytes()
handler(&clone)
}
}
-1
View File
@@ -5,7 +5,6 @@ import (
"github.com/AlexxIT/go2rtc/pkg/core"
"github.com/pion/rtp"
"github.com/rs/zerolog/log"
)
func Log(handler core.HandlerFunc) core.HandlerFunc {
+2 -3
View File
@@ -6,7 +6,6 @@ import (
"encoding/json"
"errors"
"fmt"
"log"
"net/rpc"
"net/url"
"strconv"
@@ -138,7 +137,7 @@ func (c *Client) Connect() error {
}
offer := pc.LocalDescription()
log.Printf("[roborock] offer\n%s", offer.SDP)
//log.Printf("[roborock] offer\n%s", offer.SDP)
if err = c.SendSDPtoRobot(offer); err != nil {
return err
}
@@ -151,7 +150,7 @@ func (c *Client) Connect() error {
time.Sleep(time.Second)
if desc, _ := c.GetDeviceSDP(); desc != nil {
log.Printf("[roborock] answer\n%s", desc.SDP)
//log.Printf("[roborock] answer\n%s", desc.SDP)
if err = c.conn.SetAnswer(desc.SDP); err != nil {
return err
}
+4 -4
View File
@@ -6,12 +6,12 @@ import (
"encoding/hex"
"encoding/json"
"fmt"
"github.com/AlexxIT/go2rtc/pkg/mqtt"
"github.com/rs/zerolog/log"
"net"
"net/rpc"
"net/url"
"time"
"github.com/AlexxIT/go2rtc/pkg/mqtt"
)
type Codec struct {
@@ -56,7 +56,7 @@ func (c *Codec) WriteRequest(r *rpc.Request, v any) error {
return err
}
log.Printf("[roborock] send: %s", payload)
//log.Printf("[roborock] send: %s", payload)
payload = c.Encrypt(payload, ts, ts, ts)
@@ -86,7 +86,7 @@ func (c *Codec) ReadResponseHeader(r *rpc.Response) error {
continue
}
log.Printf("[roborock] recv %s", payload)
//log.Printf("[roborock] recv %s", payload)
// get content from response payload:
// {"t":1676871268,"dps":{"102":"{\"id\":315003,\"result\":[\"ok\"]}"}}
+2 -1
View File
@@ -16,4 +16,5 @@ response []interface {}{"onStatus", 0, interface {}(nil), map[string]interface {
- https://en.wikipedia.org/wiki/Flash_Video
- https://en.wikipedia.org/wiki/Real-Time_Messaging_Protocol
- https://rtmp.veriskope.com/pdf/rtmp_specification_1.0.pdf
- https://rtmp.veriskope.com/pdf/rtmp_specification_1.0.pdf
- https://rtmp.veriskope.com/docs/spec/
+1 -1
View File
@@ -65,7 +65,7 @@ func NewClient(conn net.Conn, u *url.URL) (*Conn, error) {
rd: bufio.NewReaderSize(conn, core.BufferSize),
wr: conn,
chunks: map[uint8]*header{},
chunks: map[uint8]*chunk{},
rdPacketSize: 128,
wrPacketSize: 4096, // OBS - 4096, Reolink - 4096
+76 -62
View File
@@ -29,7 +29,7 @@ type Conn struct {
rdPacketSize uint32
wrPacketSize uint32
chunks map[byte]*header
chunks map[byte]*chunk
streamID byte
url string
@@ -52,24 +52,73 @@ func (c *Conn) readResponse(transID float64) ([]any, error) {
if err != nil {
return nil, err
}
//log.Printf("[rtmp] type=%d data=%s", msgType, b)
switch msgType {
case TypeSetPacketSize:
c.rdPacketSize = binary.BigEndian.Uint32(b)
case TypeCommand:
items, _ := amf.NewReader(b).ReadItems()
if len(items) >= 3 && items[1] == transID {
if len(items) >= 3 && (items[1] == transID || items[1] == float64(0)) {
return items, nil
}
}
}
}
type header struct {
timeMS uint32
type chunk struct {
conn *Conn
rawTime uint32
dataSize uint32
tagType byte
streamID uint32
timeMS uint32
}
func (c *chunk) readHeader(typ byte) error {
switch typ {
case 0: // 12 byte header (full header)
b, err := c.conn.readSize(11)
if err != nil {
return err
}
c.rawTime = Uint24(b)
c.dataSize = Uint24(b[3:])
c.tagType = b[6]
c.streamID = binary.LittleEndian.Uint32(b[7:])
c.timeMS = c.readExtendedTime()
case 1: // 8 bytes - like type b00, not including message ID (4 last bytes)
b, err := c.conn.readSize(7)
if err != nil {
return err
}
c.rawTime = Uint24(b)
c.dataSize = Uint24(b[3:]) // msgdatalen
c.tagType = b[6] // msgtypeid
c.timeMS += c.readExtendedTime()
case 2: // 4 bytes - Basic Header and timestamp (3 bytes) are included
b, err := c.conn.readSize(3)
if err != nil {
return err
}
c.rawTime = Uint24(b) // timestamp
c.timeMS += c.readExtendedTime()
case 3: // 1 byte - only the Basic Header is included
// use here hdr from previous msg with same session ID (sid)
}
return nil
}
func (c *chunk) readExtendedTime() uint32 {
if c.rawTime == 0xFFFFFF {
if b, err := c.conn.readSize(4); err == nil {
return binary.BigEndian.Uint32(b)
}
}
return c.rawTime
}
//var ErrNotImplemented = errors.New("rtmp: not implemented")
@@ -84,93 +133,57 @@ func (c *Conn) readMessage() (byte, uint32, []byte, error) {
chunkID := b[0] & 0b111111
// storing header information for support header type 3
hdr, ok := c.chunks[chunkID]
ch, ok := c.chunks[chunkID]
if !ok {
hdr = &header{}
c.chunks[chunkID] = hdr
ch = &chunk{conn: c}
c.chunks[chunkID] = ch
}
switch hdrType {
case 0: // 12 byte header (full header)
if b, err = c.readSize(11); err != nil {
return 0, 0, nil, err
}
_ = b[7]
hdr.timeMS = Uint24(b)
hdr.dataSize = Uint24(b[3:])
hdr.tagType = b[6]
hdr.streamID = binary.LittleEndian.Uint32(b[7:])
case 1: // 8 bytes - like type b00, not including message ID (4 last bytes)
if b, err = c.readSize(7); err != nil {
return 0, 0, nil, err
}
_ = b[6]
hdr.timeMS = Uint24(b) // timestamp
hdr.dataSize = Uint24(b[3:]) // msgdatalen
hdr.tagType = b[6] // msgtypeid
case 2: // 4 bytes - Basic Header and timestamp (3 bytes) are included
if b, err = c.readSize(3); err != nil {
return 0, 0, nil, err
}
hdr.timeMS = Uint24(b) // timestamp
case 3: // 1 byte - only the Basic Header is included
// use here hdr from previous msg with same session ID (sid)
if err = ch.readHeader(hdrType); err != nil {
return 0, 0, nil, err
}
timeMS := hdr.timeMS
if timeMS == 0xFFFFFF {
if b, err = c.readSize(4); err != nil {
return 0, 0, nil, err
}
timeMS = binary.BigEndian.Uint32(b)
}
//log.Printf("[rtmp] hdr=%d chunkID=%d timeMS=%d size=%d tagType=%d streamID=%d", hdrType, chunkID, hdr.timeMS, hdr.dataSize, hdr.tagType, hdr.streamID)
//log.Printf("[rtmp] hdr=%d chunkID=%d timeMS=%d size=%d tagType=%d streamID=%d", hdrType, chunkID, ch.timeMS, ch.dataSize, ch.tagType, ch.streamID)
// 1. Response zero size
if hdr.dataSize == 0 {
return hdr.tagType, timeMS, nil, nil
if ch.dataSize == 0 {
return ch.tagType, ch.timeMS, nil, nil
}
b = make([]byte, hdr.dataSize)
data := make([]byte, ch.dataSize)
// 2. Response small packet
if hdr.dataSize <= c.rdPacketSize {
if _, err = io.ReadFull(c.rd, b); err != nil {
if ch.dataSize <= c.rdPacketSize {
if _, err = io.ReadFull(c.rd, data); err != nil {
return 0, 0, nil, err
}
return hdr.tagType, timeMS, b, nil
return ch.tagType, ch.timeMS, data, nil
}
// 3. Response big packet
var i0 uint32
for i1 := c.rdPacketSize; i1 < hdr.dataSize; i1 += c.rdPacketSize {
if _, err = io.ReadFull(c.rd, b[i0:i1]); err != nil {
for i1 := c.rdPacketSize; i1 < ch.dataSize; i1 += c.rdPacketSize {
if _, err = io.ReadFull(c.rd, data[i0:i1]); err != nil {
return 0, 0, nil, err
}
// hopefully this will be hdrType=3 with same chunkID
if _, err = c.readSize(1); err != nil {
return 0, 0, nil, err
}
if hdr.timeMS == 0xFFFFFF {
if _, err = c.readSize(4); err != nil {
return 0, 0, nil, err
}
}
_ = ch.readExtendedTime()
i0 = i1
}
if _, err = io.ReadFull(c.rd, b[i0:]); err != nil {
if _, err = io.ReadFull(c.rd, data[i0:]); err != nil {
return 0, 0, nil, err
}
return hdr.tagType, timeMS, b, nil
return ch.tagType, ch.timeMS, data, nil
}
func (c *Conn) writeMessage(chunkID, tagType byte, timeMS uint32, payload []byte) error {
c.mu.Lock()
c.resetBuffer()
@@ -288,7 +301,7 @@ func (c *Conn) writePublish() error {
return err
}
v, err := c.readResponse(0)
v, err := c.readResponse(5)
if err != nil {
return nil
}
@@ -307,7 +320,8 @@ func (c *Conn) writePlay() error {
return err
}
v, err := c.readResponse(0)
// Reolink response with ID=0, other software respose with ID=5
v, err := c.readResponse(5)
if err != nil {
return nil
}
@@ -322,7 +336,7 @@ func (c *Conn) writePlay() error {
func (c *Conn) readSize(n uint32) ([]byte, error) {
b := make([]byte, n)
if _, err := io.ReadAtLeast(c.rd, b, int(n)); err != nil {
if _, err := io.ReadFull(c.rd, b); err != nil {
return nil, err
}
return b, nil
+1 -1
View File
@@ -17,7 +17,7 @@ func NewServer(conn net.Conn) (*Conn, error) {
rd: bufio.NewReaderSize(conn, core.BufferSize),
wr: conn,
chunks: map[uint8]*header{},
chunks: map[uint8]*chunk{},
rdPacketSize: 128,
wrPacketSize: 4096,
+15 -2
View File
@@ -186,10 +186,20 @@ func (c *Conn) Announce() (err error) {
return err
}
res, err := c.Do(req)
_, err = c.Do(req)
return
}
_ = res
func (c *Conn) Record() (err error) {
req := &tcp.Request{
Method: MethodRecord,
URL: c.URL,
Header: map[string][]string{
"Range": {"npt=0.000-"},
},
}
_, err = c.Do(req)
return
}
@@ -304,5 +314,8 @@ func (c *Conn) Close() error {
if c.mode == core.ModeActiveProducer {
_ = c.Teardown()
}
if c.OnClose != nil {
_ = c.OnClose()
}
return c.conn.Close()
}
+6 -1
View File
@@ -24,6 +24,7 @@ type Conn struct {
Backchannel bool
Media string
OnClose func() error
PacketSize uint16
SessionName string
Timeout int
@@ -124,7 +125,11 @@ func (c *Conn) Handle() (err error) {
case core.ModePassiveProducer:
// polling frames from remote RTSP Client (ex FFmpeg)
timeout = time.Second * 15
if c.Timeout == 0 {
timeout = time.Second * 15
} else {
timeout = time.Second * time.Duration(c.Timeout)
}
case core.ModePassiveConsumer:
// pushing frames to remote RTSP Client (ex VLC)
+32 -23
View File
@@ -74,19 +74,38 @@ func (c *Conn) AddTrack(media *core.Media, codec *core.Codec, track *core.Receiv
return nil
}
const (
startVideoBuf = 32 * 1024 // 32KB
startAudioBuf = 2 * 1024 // 2KB
maxBuf = 1024 * 1024 // 1MB
rtpHdr = 12 // basic RTP header size
intHdr = 4 // interleaved header size
)
func (c *Conn) packetWriter(codec *core.Codec, channel, payloadType uint8) core.HandlerFunc {
var buf []byte
var n int
video := codec.IsVideo()
if video {
buf = make([]byte, 32*1024) // 32KB
buf = make([]byte, startVideoBuf)
} else {
buf = make([]byte, 2*1024) // 2KB
buf = make([]byte, startAudioBuf)
}
flushBuf := func() {
if err := c.conn.SetWriteDeadline(time.Now().Add(Timeout)); err != nil {
return
}
//log.Printf("[rtsp] channel:%2d write_size:%6d buffer_size:%6d", channel, n, len(buf))
if _, err := c.conn.Write(buf[:n]); err == nil {
c.send += n
}
n = 0
}
handlerFunc := func(packet *rtp.Packet) {
if c.state == StateNone || !c.playOK {
if c.state == StateNone {
return
}
@@ -106,16 +125,13 @@ func (c *Conn) packetWriter(codec *core.Codec, channel, payloadType uint8) core.
packet.Marker = true // better to have marker on all audio packets
}
size := 12 + len(packet.Payload)
size := rtpHdr + len(packet.Payload)
if n+4+size > len(buf) {
if len(buf) < 1024*1024 {
buf = append(buf, make([]byte, len(buf))...)
if l := len(buf); n+intHdr+size > l {
if l < maxBuf {
buf = append(buf, make([]byte, l)...) // double buffer size
} else {
if _, err := c.conn.Write(buf[:n]); err == nil {
c.send += n
}
n = 0
flushBuf()
}
}
@@ -134,21 +150,14 @@ func (c *Conn) packetWriter(codec *core.Codec, channel, payloadType uint8) core.
n += 4 + size
if !packet.Marker {
return // collect continious video packets to buffer
}
if err := c.conn.SetWriteDeadline(time.Now().Add(Timeout)); err != nil {
if !packet.Marker || !c.playOK {
// collect continious video packets to buffer
// or wait OK for PLAY command for backchannel
//log.Printf("[rtsp] collecting buffer ok=%t", c.playOK)
return
}
//log.Printf("[rtsp] channel:%2d write_size:%6d buffer_size:%6d", channel, n, len(buf))
if _, err := c.conn.Write(buf[:n]); err == nil {
c.send += n
}
n = 0
flushBuf()
}
if !codec.IsRTP() {
+127
View File
@@ -0,0 +1,127 @@
package wav
import (
"bufio"
"encoding/binary"
"errors"
"io"
"github.com/AlexxIT/go2rtc/pkg/core"
"github.com/pion/rtp"
)
const FourCC = "RIFF"
func Open(r io.Reader) (*Producer, error) {
// https://en.wikipedia.org/wiki/WAV
// https://www.mmsp.ece.mcgill.ca/Documents/AudioFormats/WAVE/WAVE.html
rd := bufio.NewReaderSize(r, core.BufferSize)
// skip Master RIFF chunk
if _, err := rd.Discard(12); err != nil {
return nil, err
}
codec := &core.Codec{}
for {
chunkID, data, err := readChunk(rd)
if err != nil {
return nil, err
}
if chunkID == "data" {
break
}
if chunkID == "fmt " {
// https://audiocoding.cc/articles/2008-05-22-wav-file-structure/wav_formats.txt
switch data[0] {
case 1:
codec.Name = core.CodecPCML
case 6:
codec.Name = core.CodecPCMA
case 7:
codec.Name = core.CodecPCMU
}
codec.Channels = uint16(data[2])
codec.ClockRate = binary.LittleEndian.Uint32(data[4:])
}
}
if codec.Name == "" {
return nil, errors.New("waw: unsupported codec")
}
prod := &Producer{rd: rd, cl: r.(io.Closer)}
prod.Type = "WAV producer"
prod.Medias = []*core.Media{
{
Kind: core.KindAudio,
Direction: core.DirectionRecvonly,
Codecs: []*core.Codec{codec},
},
}
return prod, nil
}
type Producer struct {
core.SuperProducer
rd *bufio.Reader
cl io.Closer
}
func (c *Producer) Start() error {
var seq uint16
var ts uint32
const PacketSize = 0.040 * 8000 // 40ms
for {
payload := make([]byte, PacketSize)
if _, err := io.ReadFull(c.rd, payload); err != nil {
return err
}
c.Recv += PacketSize
if len(c.Receivers) == 0 {
continue
}
pkt := &rtp.Packet{
Header: rtp.Header{
Version: 2,
Marker: true,
SequenceNumber: seq,
Timestamp: ts,
},
Payload: payload,
}
c.Receivers[0].WriteRTP(pkt)
seq++
ts += PacketSize
}
}
func (c *Producer) Stop() error {
_ = c.SuperProducer.Close()
return c.cl.Close()
}
func readChunk(r io.Reader) (chunkID string, data []byte, err error) {
b := make([]byte, 8)
if _, err = io.ReadFull(r, b); err != nil {
return
}
if chunkID = string(b[:4]); chunkID != "data" {
size := binary.LittleEndian.Uint32(b[4:])
data = make([]byte, size)
_, err = io.ReadFull(r, data)
}
return
}
+7 -1
View File
@@ -77,7 +77,13 @@ func (c *Conn) AddTrack(media *core.Media, codec *core.Codec, track *core.Receiv
sender.Handler = pcm.RepackG711(false, sender.Handler)
}
sender.Bind(track)
// TODO: rewrite this dirty logic
// maybe not best solution, but ActiveProducer connected before AddTrack
if c.Mode != core.ModeActiveProducer {
sender.Bind(track)
} else {
sender.HandleRTP(track)
}
c.senders = append(c.senders, sender)
return nil
+5
View File
@@ -0,0 +1,5 @@
## Useful links
- https://learn.microsoft.com/en-us/windows/win32/medfound/recommended-8-bit-yuv-formats-for-video-rendering
- https://developer.mozilla.org/en-US/docs/Web/Media/Formats/Video_concepts
- https://fourcc.org/yuv.php#YV12
+67
View File
@@ -0,0 +1,67 @@
package y4m
import (
"fmt"
"io"
"github.com/AlexxIT/go2rtc/pkg/core"
"github.com/pion/rtp"
)
type Consumer struct {
core.SuperConsumer
wr *core.WriteBuffer
}
func NewConsumer() *Consumer {
return &Consumer{
core.SuperConsumer{
Type: "YUV4MPEG2 passive consumer",
Medias: []*core.Media{
{
Kind: core.KindVideo,
Direction: core.DirectionSendonly,
Codecs: []*core.Codec{
{Name: core.CodecRAW},
},
},
},
},
core.NewWriteBuffer(nil),
}
}
func (c *Consumer) AddTrack(media *core.Media, _ *core.Codec, track *core.Receiver) error {
sender := core.NewSender(media, track.Codec)
sender.Handler = func(packet *rtp.Packet) {
if n, err := c.wr.Write([]byte(frameHdr)); err == nil {
c.Send += n
}
if n, err := c.wr.Write(packet.Payload); err == nil {
c.Send += n
}
}
hdr := fmt.Sprintf(
"YUV4MPEG2 W%s H%s C%s\n",
core.Between(track.Codec.FmtpLine, "width=", ";"),
core.Between(track.Codec.FmtpLine, "height=", ";"),
core.Between(track.Codec.FmtpLine, "colorspace=", ";"),
)
if _, err := c.wr.Write([]byte(hdr)); err != nil {
return err
}
sender.HandleRTP(track)
c.Senders = append(c.Senders, sender)
return nil
}
func (c *Consumer) WriteTo(wr io.Writer) (int64, error) {
return c.wr.WriteTo(wr)
}
func (c *Consumer) Stop() error {
_ = c.SuperConsumer.Close()
return c.wr.Close()
}
+110
View File
@@ -0,0 +1,110 @@
package y4m
import (
"bufio"
"bytes"
"errors"
"io"
"github.com/AlexxIT/go2rtc/pkg/core"
"github.com/pion/rtp"
)
func Open(r io.Reader) (*Producer, error) {
rd := bufio.NewReaderSize(r, core.BufferSize)
b, err := rd.ReadBytes('\n')
if err != nil {
return nil, err
}
b = b[:len(b)-1] // remove \n
sdp := string(b)
var fmtp string
for b != nil {
// YUV4MPEG2 W1280 H720 F24:1 Ip A1:1 C420mpeg2 XYSCSS=420MPEG2
// https://manned.org/yuv4mpeg.5
// https://github.com/FFmpeg/FFmpeg/blob/master/libavformat/yuv4mpegenc.c
key := b[0]
var value string
if i := bytes.IndexByte(b, ' '); i > 0 {
value = string(b[1:i])
b = b[i+1:]
} else {
value = string(b[1:])
b = nil
}
switch key {
case 'W':
fmtp = "width=" + value
case 'H':
fmtp += ";height=" + value
case 'C':
fmtp += ";colorspace=" + value
}
}
if GetSize(fmtp) == 0 {
return nil, errors.New("y4m: unsupported format: " + sdp)
}
prod := &Producer{rd: rd, cl: r.(io.Closer)}
prod.Type = "YUV4MPEG2 producer"
prod.SDP = sdp
prod.Medias = []*core.Media{
{
Kind: core.KindVideo,
Direction: core.DirectionRecvonly,
Codecs: []*core.Codec{
{
Name: core.CodecRAW,
ClockRate: 90000,
FmtpLine: fmtp,
PayloadType: core.PayloadTypeRAW,
},
},
},
}
return prod, nil
}
type Producer struct {
core.SuperProducer
rd *bufio.Reader
cl io.Closer
}
func (c *Producer) Start() error {
size := GetSize(c.Medias[0].Codecs[0].FmtpLine)
for {
if _, err := c.rd.Discard(len(frameHdr)); err != nil {
return err
}
frame := make([]byte, size)
if _, err := io.ReadFull(c.rd, frame); err != nil {
return err
}
c.Recv += size
if len(c.Receivers) == 0 {
continue
}
pkt := &rtp.Packet{
Header: rtp.Header{Timestamp: core.Now90000()},
Payload: frame,
}
c.Receivers[0].WriteRTP(pkt)
}
}
func (c *Producer) Stop() error {
_ = c.SuperProducer.Close()
return c.cl.Close()
}
+96
View File
@@ -0,0 +1,96 @@
package y4m
import (
"image"
"github.com/AlexxIT/go2rtc/pkg/core"
)
const FourCC = "YUV4"
const frameHdr = "FRAME\n"
func GetSize(fmtp string) int {
w := core.Atoi(core.Between(fmtp, "width=", ";"))
h := core.Atoi(core.Between(fmtp, "height=", ";"))
switch core.Between(fmtp, "colorspace=", ";") {
case "mono":
return w * h
case "420mpeg2", "420jpeg":
return w * h * 3 / 2
case "422":
return w * h * 2
case "444":
return w * h * 3
}
return 0
}
func NewImage(fmtp string) func(frame []byte) image.Image {
w := core.Atoi(core.Between(fmtp, "width=", ";"))
h := core.Atoi(core.Between(fmtp, "height=", ";"))
rect := image.Rect(0, 0, w, h)
switch core.Between(fmtp, "colorspace=", ";") {
case "mono":
return func(frame []byte) image.Image {
return &image.Gray{
Pix: frame,
Stride: w,
Rect: rect,
}
}
case "420mpeg2", "420jpeg":
i1 := w * h
i2 := i1 + i1/4
i3 := i2 + i1/4
return func(frame []byte) image.Image {
return &image.YCbCr{
Y: frame[:i1],
Cb: frame[i1:i2],
Cr: frame[i2:i3],
YStride: w,
CStride: w / 2,
SubsampleRatio: image.YCbCrSubsampleRatio420,
Rect: rect,
}
}
case "422":
i1 := w * h
i2 := i1 + i1/2
i3 := i2 + i1/2
return func(frame []byte) image.Image {
return &image.YCbCr{
Y: frame[:i1],
Cb: frame[i1:i2],
Cr: frame[i2:i3],
YStride: w,
CStride: w / 2,
SubsampleRatio: image.YCbCrSubsampleRatio422,
Rect: rect,
}
}
case "444":
i1 := w * h
i2 := i1 + i1
i3 := i2 + i1
return func(frame []byte) image.Image {
return &image.YCbCr{
Y: frame[:i1],
Cb: frame[i1:i2],
Cr: frame[i2:i3],
YStride: w,
CStride: w,
SubsampleRatio: image.YCbCrSubsampleRatio444,
Rect: rect,
}
}
}
return nil
}
+1
View File
@@ -14,6 +14,7 @@ go get -u
go mod tidy
go mod why github.com/pion/rtcp
go list -deps .\cmd\go2rtc_rtsp\
./goweight
```
## Dependencies
+486
View File
@@ -0,0 +1,486 @@
{
"$schema": "http://json-schema.org/draft-07/schema#",
"title": "go2rtc",
"type": "object",
"additionalProperties": false,
"definitions": {
"listen": {
"type": "string",
"anyOf": [
{
"type": "string",
"pattern": ":[0-9]{1,5}$"
},
{
"type": "string",
"const": ""
}
]
},
"log_level": {
"type": "string",
"enum": [
"trace",
"debug",
"info",
"warn",
"error"
]
}
},
"properties": {
"api": {
"type": "object",
"properties": {
"listen": {
"default": ":1984",
"examples": [
"127.0.0.1:8080"
],
"$ref": "#/definitions/listen"
},
"username": {
"type": "string",
"examples": [
"admin"
]
},
"password": {
"type": "string"
},
"base_path": {
"type": "string",
"examples": [
"/go2rtc"
]
},
"static_dir": {
"type": "string",
"examples": [
"/var/www"
]
},
"origin": {
"type": "string",
"const": "*"
},
"tls_listen": {
"$ref": "#/definitions/listen"
},
"tls_cert": {
"type": "string",
"examples": [
"-----BEGIN CERTIFICATE-----",
"/ssl/fullchain.pem"
]
},
"tls_key": {
"type": "string",
"examples": [
"-----BEGIN PRIVATE KEY-----",
"/ssl/privkey.pem"
]
},
"unix_listen": {
"type": "string",
"examples": [
"/tmp/go2rtc.sock"
]
}
}
},
"ffmpeg": {
"type": "object",
"properties": {
"bin": {
"type": "string",
"default": "ffmpeg"
}
},
"additionalProperties": {
"description": "FFmpeg template",
"type": "string"
}
},
"hass": {
"type": "object",
"properties": {
"config": {
"description": "Home Assistant config directory path",
"type": "string",
"examples": [
"/config"
]
}
}
},
"homekit": {
"type": "object",
"additionalProperties": {
"type": "object",
"properties": {
"pin": {
"type": "string",
"default": "19550224",
"pattern": "^[0-9]{8}$"
},
"name": {
"type": "string"
},
"device_id": {
"type": "string"
},
"device_private": {
"type": "string"
},
"pairings": {
"type": "array",
"items": {
"type": "string"
}
}
}
}
},
"log": {
"type": "object",
"properties": {
"format": {
"type": "string",
"default": "color",
"enum": [
"color",
"json",
"text"
]
},
"level": {
"description": "Defaul log level",
"default": "info",
"$ref": "#/definitions/log_level"
},
"output": {
"type": "string",
"default": "stdout",
"enum": [
"",
"stdout",
"stderr"
]
},
"time": {
"type": "string",
"default": "UNIXMS",
"anyOf": [
{
"type": "string",
"enum": [
"",
"UNIXMS",
"UNIXMICRO",
"UNIXNANO",
"2006-01-02T15:04:05Z07:00",
"2006-01-02T15:04:05.999999999Z07:00"
]
},
{
"type": "string"
}
]
},
"api": {
"$ref": "#/definitions/log_level"
},
"echo": {
"$ref": "#/definitions/log_level"
},
"exec": {
"description": "Value `exec: debug` will print stderr",
"$ref": "#/definitions/log_level"
},
"expr": {
"$ref": "#/definitions/log_level"
},
"ffmpeg": {
"description": "Will only be displayed with `exec: debug` setting",
"default": "error",
"$ref": "#/definitions/log_level"
},
"hass": {
"$ref": "#/definitions/log_level"
},
"hls": {
"$ref": "#/definitions/log_level"
},
"homekit": {
"$ref": "#/definitions/log_level"
},
"mp4": {
"$ref": "#/definitions/log_level"
},
"ngrok": {
"$ref": "#/definitions/log_level"
},
"onvif": {
"$ref": "#/definitions/log_level"
},
"rtmp": {
"$ref": "#/definitions/log_level"
},
"rtsp": {
"$ref": "#/definitions/log_level"
},
"streams": {
"$ref": "#/definitions/log_level"
},
"webrtc": {
"$ref": "#/definitions/log_level"
},
"webtorrent": {
"$ref": "#/definitions/log_level"
}
}
},
"ngrok": {
"type": "object",
"properties": {
"command": {
"type": "string",
"examples": [
"ngrok tcp 8555 --authtoken xxx",
"ngrok start --all --config ngrok.yaml"
]
}
}
},
"publish": {
"type": "object",
"additionalProperties": {
"anyOf": [
{
"type": "string",
"examples": [
"rtmp://xxx.rtmp.youtube.com/live2/xxxx-xxxx-xxxx-xxxx-xxxx",
"rtmps://xxx-x.rtmp.t.me/s/xxxxxxxxxx:xxxxxxxxxxxxxxxxxxxxxx"
]
},
{
"type": "array",
"items": {
"type": "string"
}
}
]
}
},
"rtmp": {
"type": "object",
"properties": {
"listen": {
"examples": [
":1935"
],
"$ref": "#/definitions/listen"
}
}
},
"rtsp": {
"type": "object",
"properties": {
"listen": {
"default": ":8554",
"$ref": "#/definitions/listen"
},
"username": {
"type": "string",
"examples": [
"admin"
]
},
"password": {
"type": "string"
},
"default_query": {
"type": "string",
"default": "video&audio"
},
"pkt_size": {
"type": "integer"
}
}
},
"srtp": {
"description": "SRTP server for HomeKit",
"type": "object",
"properties": {
"listen": {
"default": ":8443",
"$ref": "#/definitions/listen"
}
}
},
"streams": {
"type": "object",
"additionalProperties": {
"title": "Stream",
"anyOf": [
{
"description": "Source",
"type": "string",
"examples": [
"rtsp://username:password@192.168.1.123/cam/realmonitor?channel=1&subtype=0&unicast=true&proto=Onvif",
"rtsp://username:password@192.168.1.123/stream1",
"rtsp://username:password@192.168.1.123/h264Preview_01_main",
"rtmp://192.168.1.123/bcs/channel0_main.bcs?channel=0&stream=0&user=username&password=password",
"http://192.168.1.123/flv?port=1935&app=bcs&stream=channel0_main.bcs&user=username&password=password",
"http://username:password@192.168.1.123/cgi-bin/snapshot.cgi?channel=1",
"ffmpeg:media.mp4#video=h264#hadware#width=1920#height=1080#rotate=180#audio=copy",
"ffmpeg:virtual?video=testsrc&size=4K#video=h264#hardware#bitrate=50M",
"bubble://username:password@192.168.1.123:34567/bubble/live?ch=0&stream=0",
"dvrip://username:password@192.168.1.123:34567?channel=0&subtype=0",
"exec:ffmpeg -re -i media.mp4 -c copy -rtsp_transport tcp -f rtsp {output}",
"isapi://username:password@192.168.1.123:80/",
"kasa://username:password@192.168.1.123:19443/https/stream/mixed",
"onvif://username:password@192.168.1.123:80?subtype=0",
"tapo://password@192.168.1.123:8800?channel=0&subtype=0",
"webtorrent:?share=xxx&pwd=xxx"
]
},
{
"type": "array",
"items": {
"description": "Source",
"type": "string"
}
}
]
}
},
"webrtc": {
"type": "object",
"properties": {
"listen": {
"default": ":8555/tcp",
"type": "string",
"anyOf": [
{
"type": "string",
"pattern": ":[0-9]{1,5}(/tcp|/udp)?$"
},
{
"type": "string",
"const": ""
}
]
},
"candidates": {
"type": "array",
"items": {
"$ref": "#/definitions/listen/anyOf/0"
},
"examples": [
"216.58.210.174:8555",
"stun:8555",
"home.duckdns.org:8555"
]
},
"ice_servers": {
"type": "array",
"items": {
"type": "object",
"properties": {
"urls": {
"type": "array",
"items": {
"type": "string",
"examples": [
"stun:stun.l.google.com:19302",
"turn:123.123.123.123:3478"
]
}
},
"username": {
"type": "string"
},
"credential": {
"type": "string"
}
}
}
},
"filters": {
"type": "object",
"properties": {
"candidates": {
"description": "Keep only these candidates",
"type": "array",
"items": {
"type": "string"
}
},
"interfaces": {
"description": "Keep only these interfaces",
"type": "array",
"items": {
"type": "string"
}
},
"ips": {
"description": "Keep only these IP-addresses",
"type": "array",
"items": {
"type": "string"
}
},
"networks": {
"description": "Use only these network types",
"type": "array",
"items": {
"enum": [
"tcp4",
"tcp6",
"udp4",
"udp6"
],
"type": "string"
}
},
"udp_ports": {
"description": "Use only these UDP ports range [min, max]",
"type": "array",
"items": {
"type": "integer"
},
"maxItems": 2,
"minItems": 2
}
}
}
}
},
"webtorrent": {
"type": "object",
"properties": {
"trackers": {
"type": "array",
"items": {
"type": "string"
}
},
"shares": {
"additionalProperties": {
"type": "object",
"properties": {
"pwd": {
"type": "string"
},
"src": {
"type": "string"
}
}
}
}
}
}
}
}
+37
View File
@@ -0,0 +1,37 @@
<!doctype html>
<html lang="en">
<head>
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>go2rtc - HLS</title>
<style>
body {
background-color: black;
margin: 0;
padding: 0;
}
html, body, video {
height: 100%;
width: 100%;
}
</style>
</head>
<body>
<script src="https://cdn.jsdelivr.net/npm/hls.js@1"></script>
<video id="video" autoplay controls playsinline muted></video>
<script>
// http://192.168.1.123:1984/hls.html?src=demo&mp4
// const url = new URL('api/stream.m3u8' + location.search, location.href);
const url = 'http://192.168.10.5:1984/api/stream.m3u8?src=rtsp-dahua2&mp4';
const video = document.getElementById('video');
if (Hls.isSupported()) {
const hls = new Hls();
hls.loadSource(url.toString());
hls.attachMedia(video);
} else if (video.canPlayType('application/vnd.apple.mpegurl')) {
video.src = url.toString();
}
</script>
</body>
</html>
+9 -4
View File
@@ -85,16 +85,21 @@
<div>
<h2>Play audio</h2>
<pre>example: ffmpeg:https://example.com/song.mp3#audio=pcma#input=file</pre>
<input id="play-url" type="text" placeholder="url">
<label><input type="radio" name="play" value="file" checked>file - play remote (https://example.com/song.mp3) or local (/media/song.mp3) file</label><br>
<label><input type="radio" name="play" value="live">live - play remote live stream (radio, etc.)</label><br>
<label><input type="radio" name="play" value="text">text - play Text To Speech (if your FFmpeg support this)</label><br>
<br>
<input id="play-url" type="text" placeholder="path / url / text">
<a id="play-send" href="#">send</a> / cameras with two way audio support
</div>
<script>
document.getElementById('play-send').addEventListener('click', ev => {
ev.preventDefault();
const url = new URL('api/streams', location.href);
// action - file / live / text
const action = document.querySelector('input[name="play"]:checked').value;
const url = new URL('api/ffmpeg', location.href);
url.searchParams.set('dst', src);
url.searchParams.set('src', document.getElementById('play-url').value);
url.searchParams.set(action, document.getElementById('play-url').value);
fetch(url, {method: 'POST'});
});
</script>
+8 -3
View File
@@ -56,7 +56,7 @@
<table>
<thead>
<tr>
<th style="width: 130px">Time</th>
<th style="width: 100px">Time</th>
<th style="width: 40px">Level</th>
<th>Message</th>
</tr>
@@ -98,11 +98,16 @@
lines = lines.reverse();
}
return lines.map(line => {
const ts = new Date(line['time']);
const ts = new Date(line['time']).toLocaleString(undefined, {
hour: 'numeric',
minute: 'numeric',
second: 'numeric',
fractionalSecondDigits: 3
});
const msg = Object.keys(line).reduce((msg, key) => {
return KEYS.indexOf(key) < 0 ? `${msg} ${key}=${line[key]}` : msg;
}, line['message']);
return `<tr class="${line['level']}"><td>${ts.toLocaleString()}</td><td>${escapeHTML(line['level'])}</td><td>${escapeHTML(msg)}</td></tr>`;
return `<tr class="${line['level']}"><td>${ts}</td><td>${line['level']}</td><td>${escapeHTML(msg)}</td></tr>`;
}).join('');
}
+3 -1
View File
@@ -492,7 +492,9 @@ export class VideoRTC extends HTMLElement {
pc.addEventListener('connectionstatechange', () => {
if (pc.connectionState === 'connected') {
const tracks = pc.getReceivers().map(receiver => receiver.track);
const tracks = pc.getTransceivers()
.filter(tr => tr.currentDirection === 'recvonly') // skip inactive
.map(tr => tr.receiver.track);
/** @type {HTMLVideoElement} */
const video2 = document.createElement('video');
video2.addEventListener('loadeddata', () => this.onpcvideo(video2), {once: true});