diff --git a/internal/staticsources/rpicamera/camera_arm_.go b/internal/staticsources/rpicamera/camera_arm_.go index 780e5477..a4f2c9c4 100644 --- a/internal/staticsources/rpicamera/camera_arm_.go +++ b/internal/staticsources/rpicamera/camera_arm_.go @@ -4,6 +4,7 @@ package rpicamera import ( "debug/elf" + "errors" "fmt" "os" "os/exec" @@ -14,8 +15,6 @@ import ( "sync" "syscall" "time" - - "github.com/bluenviron/mediacommon/v2/pkg/codecs/h264" ) const ( @@ -174,8 +173,8 @@ func freeComponent() { } type camera struct { - params params - onData func(int64, time.Time, [][]byte) + params cameraParams + onData func(int64, time.Time, []byte) onDataSecondary func(int64, time.Time, []byte) cmd *exec.Cmd @@ -309,7 +308,7 @@ outer: switch buf[0] { case 'e': - return fmt.Errorf(string(buf[1:])) + return errors.New(string(buf[1:])) case 'r': break outer @@ -327,7 +326,7 @@ outer: switch buf[0] { case 'e': - return fmt.Errorf(string(buf[1:])) + return errors.New(string(buf[1:])) case 'd': dts := int64(buf[8])<<56 | int64(buf[7])<<48 | int64(buf[6])<<40 | int64(buf[5])<<32 | @@ -335,16 +334,10 @@ outer: ntpUs := int64(buf[16])<<56 | int64(buf[15])<<48 | int64(buf[14])<<40 | int64(buf[13])<<32 | int64(buf[12])<<24 | int64(buf[11])<<16 | int64(buf[10])<<8 | int64(buf[9]) - var nalus h264.AnnexB - err = nalus.Unmarshal(buf[17:]) - if err != nil { - return err - } - c.onData( multiplyAndDivide(dts, 90000, 1e6), time.Unix(ntpUs/1e6, (ntpUs%1e6)*1000), - nalus) + buf[17:]) case 's': dts := int64(buf[8])<<56 | int64(buf[7])<<48 | int64(buf[6])<<40 | int64(buf[5])<<32 | @@ -363,7 +356,7 @@ outer: } } -func (c *camera) reloadParams(params params) { +func (c *camera) reloadParams(params cameraParams) { c.pipeOut.write(append([]byte{'c'}, params.serialize()...)) } diff --git a/internal/staticsources/rpicamera/camera_other.go b/internal/staticsources/rpicamera/camera_other.go deleted file mode 100644 index 20efc19c..00000000 --- a/internal/staticsources/rpicamera/camera_other.go +++ /dev/null @@ -1,28 +0,0 @@ -//go:build !linux || (!arm && !arm64) - -package rpicamera - -import ( - "fmt" - "time" -) - -type camera struct { - params params - onData func(int64, time.Time, [][]byte) - onDataSecondary func(int64, time.Time, []byte) -} - -func (c *camera) initialize() error { - return fmt.Errorf("server was compiled without support for the Raspberry Pi Camera") -} - -func (c *camera) close() { -} - -func (c *camera) reloadParams(_ params) { -} - -func (c *camera) wait() error { - return nil -} diff --git a/internal/staticsources/rpicamera/camera_params_arm_.go b/internal/staticsources/rpicamera/camera_params_arm_.go new file mode 100644 index 00000000..850719bc --- /dev/null +++ b/internal/staticsources/rpicamera/camera_params_arm_.go @@ -0,0 +1,155 @@ +//go:build (linux && arm) || (linux && arm64) + +package rpicamera + +import ( + "encoding/base64" + "reflect" + "strconv" + "strings" + + "github.com/bluenviron/mediamtx/internal/conf" + "github.com/bluenviron/mediamtx/internal/logger" +) + +type cameraParams struct { + LogLevel string + CameraID uint32 + Width uint32 + Height uint32 + HFlip bool + VFlip bool + Brightness float32 + Contrast float32 + Saturation float32 + Sharpness float32 + Exposure string + AWB string + AWBGainRed float32 + AWBGainBlue float32 + Denoise string + Shutter uint32 + Metering string + Gain float32 + EV float32 + ROI string + HDR bool + TuningFile string + Mode string + FPS float32 + AfMode string + AfRange string + AfSpeed string + LensPosition float32 + AfWindow string + FlickerPeriod uint32 + TextOverlayEnable bool + TextOverlay string + Codec string + IDRPeriod uint32 + Bitrate uint32 + HardwareH264Profile string + HardwareH264Level string + SoftwareH264Profile string + SoftwareH264Level string + + SecondaryWidth uint32 + SecondaryHeight uint32 + SecondaryFPS float32 + SecondaryMJPEGQuality uint32 +} + +func (p *cameraParams) fromConf(logLevel conf.LogLevel, cnf *conf.Path) { + p.LogLevel = func() string { + switch logLevel { + case conf.LogLevel(logger.Debug): + return "debug" + case conf.LogLevel(logger.Info): + return "info" + case conf.LogLevel(logger.Warn): + return "warn" + } + return "error" + }() + p.CameraID = uint32(cnf.RPICameraCamID) + p.Width = uint32(cnf.RPICameraWidth) + p.Height = uint32(cnf.RPICameraHeight) + p.HFlip = cnf.RPICameraHFlip + p.VFlip = cnf.RPICameraVFlip + p.Brightness = float32(cnf.RPICameraBrightness) + p.Contrast = float32(cnf.RPICameraContrast) + p.Saturation = float32(cnf.RPICameraSaturation) + p.Sharpness = float32(cnf.RPICameraSharpness) + p.Exposure = cnf.RPICameraExposure + p.AWB = cnf.RPICameraAWB + p.AWBGainRed = float32(cnf.RPICameraAWBGains[0]) + p.AWBGainBlue = float32(cnf.RPICameraAWBGains[1]) + p.Denoise = cnf.RPICameraDenoise + p.Shutter = uint32(cnf.RPICameraShutter) + p.Metering = cnf.RPICameraMetering + p.Gain = float32(cnf.RPICameraGain) + p.EV = float32(cnf.RPICameraEV) + p.ROI = cnf.RPICameraROI + p.HDR = cnf.RPICameraHDR + p.TuningFile = cnf.RPICameraTuningFile + p.Mode = cnf.RPICameraMode + p.FPS = float32(cnf.RPICameraFPS) + p.AfMode = cnf.RPICameraAfMode + p.AfRange = cnf.RPICameraAfRange + p.AfSpeed = cnf.RPICameraAfSpeed + p.LensPosition = float32(cnf.RPICameraLensPosition) + p.AfWindow = cnf.RPICameraAfWindow + p.FlickerPeriod = uint32(cnf.RPICameraFlickerPeriod) + p.TextOverlayEnable = cnf.RPICameraTextOverlayEnable + p.TextOverlay = cnf.RPICameraTextOverlay + p.Codec = cnf.RPICameraCodec + p.IDRPeriod = uint32(cnf.RPICameraIDRPeriod) + p.Bitrate = uint32(cnf.RPICameraBitrate) + p.HardwareH264Profile = cnf.RPICameraHardwareH264Profile + p.HardwareH264Level = cnf.RPICameraHardwareH264Level + p.SoftwareH264Profile = cnf.RPICameraSoftwareH264Profile + p.SoftwareH264Level = cnf.RPICameraSoftwareH264Level + + p.SecondaryWidth = uint32(cnf.RPICameraSecondaryWidth) + p.SecondaryHeight = uint32(cnf.RPICameraSecondaryHeight) + p.SecondaryFPS = float32(cnf.RPICameraSecondaryFPS) + p.SecondaryMJPEGQuality = uint32(cnf.RPICameraSecondaryMJPEGQuality) +} + +func (p cameraParams) serialize() []byte { + rv := reflect.ValueOf(p) + rt := rv.Type() + nf := rv.NumField() + ret := make([]string, nf) + + for i := range nf { + entry := rt.Field(i).Name + ":" + f := rv.Field(i) + v := f.Interface() + + switch v := v.(type) { + case uint32: + entry += strconv.FormatUint(uint64(v), 10) + + case float32: + entry += strconv.FormatFloat(float64(v), 'f', -1, 64) + + case string: + entry += base64.StdEncoding.EncodeToString([]byte(v)) + + case bool: + if f.Bool() { + entry += "1" + } else { + entry += "0" + } + + default: + panic("unhandled type") + } + + ret[i] = entry + } + + return []byte(strings.Join(ret, " ")) +} diff --git a/internal/staticsources/rpicamera/params.go b/internal/staticsources/rpicamera/params.go deleted file mode 100644 index eed96b54..00000000 --- a/internal/staticsources/rpicamera/params.go +++ /dev/null @@ -1,47 +0,0 @@ -package rpicamera - -type params struct { - LogLevel string - CameraID uint32 - Width uint32 - Height uint32 - HFlip bool - VFlip bool - Brightness float32 - Contrast float32 - Saturation float32 - Sharpness float32 - Exposure string - AWB string - AWBGainRed float32 - AWBGainBlue float32 - Denoise string - Shutter uint32 - Metering string - Gain float32 - EV float32 - ROI string - HDR bool - TuningFile string - Mode string - FPS float32 - AfMode string - AfRange string - AfSpeed string - LensPosition float32 - AfWindow string - FlickerPeriod uint32 - TextOverlayEnable bool - TextOverlay string - Codec string - IDRPeriod uint32 - Bitrate uint32 - HardwareH264Profile string - HardwareH264Level string - SoftwareH264Profile string - SoftwareH264Level string - SecondaryWidth uint32 - SecondaryHeight uint32 - SecondaryFPS float32 - SecondaryMJPEGQuality uint32 -} diff --git a/internal/staticsources/rpicamera/params_serialize.go b/internal/staticsources/rpicamera/params_serialize.go deleted file mode 100644 index 071f0f2f..00000000 --- a/internal/staticsources/rpicamera/params_serialize.go +++ /dev/null @@ -1,48 +0,0 @@ -//go:build (linux && arm) || (linux && arm64) - -package rpicamera - -import ( - "encoding/base64" - "reflect" - "strconv" - "strings" -) - -func (p params) serialize() []byte { - rv := reflect.ValueOf(p) - rt := rv.Type() - nf := rv.NumField() - ret := make([]string, nf) - - for i := range nf { - entry := rt.Field(i).Name + ":" - f := rv.Field(i) - v := f.Interface() - - switch v := v.(type) { - case uint32: - entry += strconv.FormatUint(uint64(v), 10) - - case float32: - entry += strconv.FormatFloat(float64(v), 'f', -1, 64) - - case string: - entry += base64.StdEncoding.EncodeToString([]byte(v)) - - case bool: - if f.Bool() { - entry += "1" - } else { - entry += "0" - } - - default: - panic("unhandled type") - } - - ret[i] = entry - } - - return []byte(strings.Join(ret, " ")) -} diff --git a/internal/staticsources/rpicamera/pipe.go b/internal/staticsources/rpicamera/pipe_arm_.go similarity index 100% rename from internal/staticsources/rpicamera/pipe.go rename to internal/staticsources/rpicamera/pipe_arm_.go diff --git a/internal/staticsources/rpicamera/source.go b/internal/staticsources/rpicamera/source.go index 1f73e840..3ee14932 100644 --- a/internal/staticsources/rpicamera/source.go +++ b/internal/staticsources/rpicamera/source.go @@ -2,104 +2,11 @@ package rpicamera import ( - "context" - "errors" - "fmt" - "time" - - "github.com/bluenviron/gortsplib/v5/pkg/description" - "github.com/bluenviron/gortsplib/v5/pkg/format" - "github.com/bluenviron/gortsplib/v5/pkg/format/rtph264" - "github.com/bluenviron/gortsplib/v5/pkg/format/rtpmjpeg" - "github.com/pion/rtp" - "github.com/bluenviron/mediamtx/internal/conf" "github.com/bluenviron/mediamtx/internal/defs" "github.com/bluenviron/mediamtx/internal/logger" - "github.com/bluenviron/mediamtx/internal/stream" - "github.com/bluenviron/mediamtx/internal/unit" ) -const ( - pauseBetweenErrors = 1 * time.Second -) - -func paramsFromConf(logLevel conf.LogLevel, cnf *conf.Path) params { - return params{ - LogLevel: func() string { - switch logLevel { - case conf.LogLevel(logger.Debug): - return "debug" - case conf.LogLevel(logger.Info): - return "info" - case conf.LogLevel(logger.Warn): - return "warn" - } - return "error" - }(), - CameraID: uint32(cnf.RPICameraCamID), - Width: uint32(cnf.RPICameraWidth), - Height: uint32(cnf.RPICameraHeight), - HFlip: cnf.RPICameraHFlip, - VFlip: cnf.RPICameraVFlip, - Brightness: float32(cnf.RPICameraBrightness), - Contrast: float32(cnf.RPICameraContrast), - Saturation: float32(cnf.RPICameraSaturation), - Sharpness: float32(cnf.RPICameraSharpness), - Exposure: cnf.RPICameraExposure, - AWB: cnf.RPICameraAWB, - AWBGainRed: float32(cnf.RPICameraAWBGains[0]), - AWBGainBlue: float32(cnf.RPICameraAWBGains[1]), - Denoise: cnf.RPICameraDenoise, - Shutter: uint32(cnf.RPICameraShutter), - Metering: cnf.RPICameraMetering, - Gain: float32(cnf.RPICameraGain), - EV: float32(cnf.RPICameraEV), - ROI: cnf.RPICameraROI, - HDR: cnf.RPICameraHDR, - TuningFile: cnf.RPICameraTuningFile, - Mode: cnf.RPICameraMode, - FPS: float32(cnf.RPICameraFPS), - AfMode: cnf.RPICameraAfMode, - AfRange: cnf.RPICameraAfRange, - AfSpeed: cnf.RPICameraAfSpeed, - LensPosition: float32(cnf.RPICameraLensPosition), - AfWindow: cnf.RPICameraAfWindow, - FlickerPeriod: uint32(cnf.RPICameraFlickerPeriod), - TextOverlayEnable: cnf.RPICameraTextOverlayEnable, - TextOverlay: cnf.RPICameraTextOverlay, - Codec: cnf.RPICameraCodec, - IDRPeriod: uint32(cnf.RPICameraIDRPeriod), - Bitrate: uint32(cnf.RPICameraBitrate), - HardwareH264Profile: cnf.RPICameraHardwareH264Profile, - HardwareH264Level: cnf.RPICameraHardwareH264Level, - SoftwareH264Profile: cnf.RPICameraSoftwareH264Profile, - SoftwareH264Level: cnf.RPICameraSoftwareH264Level, - SecondaryWidth: uint32(cnf.RPICameraSecondaryWidth), - SecondaryHeight: uint32(cnf.RPICameraSecondaryHeight), - SecondaryFPS: float32(cnf.RPICameraSecondaryFPS), - SecondaryMJPEGQuality: uint32(cnf.RPICameraSecondaryMJPEGQuality), - } -} - -type secondaryReader struct { - ctx context.Context - ctxCancel func() -} - -// Close implements reader. -func (r *secondaryReader) Close() { - r.ctxCancel() -} - -// APIReaderDescribe implements reader. -func (*secondaryReader) APIReaderDescribe() *defs.APIPathReader { - return &defs.APIPathReader{ - Type: defs.APIPathReaderTypeHidden, - ID: "", - } -} - type parent interface { logger.Writer SetReady(req defs.PathSourceStaticSetReadyReq) defs.PathSourceStaticSetReadyRes @@ -119,245 +26,6 @@ func (s *Source) Log(level logger.Level, format string, args ...any) { s.Parent.Log(level, "[RPI Camera source] "+format, args...) } -// Run implements StaticSource. -func (s *Source) Run(params defs.StaticSourceRunParams) error { - if !params.Conf.RPICameraSecondary { - return s.runPrimary(params) - } - return s.runSecondary(params) -} - -func (s *Source) runPrimary(params defs.StaticSourceRunParams) error { - var medias []*description.Media - - medi := &description.Media{ - Type: description.MediaTypeVideo, - Formats: []format.Format{&format.H264{ - PayloadTyp: 96, - PacketizationMode: 1, - }}, - } - medias = append(medias, medi) - - var mediaSecondary *description.Media - - if params.Conf.RPICameraSecondaryWidth != 0 { - mediaSecondary = &description.Media{ - Type: description.MediaTypeApplication, - Formats: []format.Format{&format.Generic{ - PayloadTyp: 96, - RTPMa: "rpicamera_secondary/90000", - ClockRat: 90000, - }}, - } - medias = append(medias, mediaSecondary) - } - - var subStream *stream.SubStream - - initializeStream := func() { - if subStream == nil { - res := s.Parent.SetReady(defs.PathSourceStaticSetReadyReq{ - Desc: &description.Session{Medias: medias}, - UseRTPPackets: true, - ReplaceNTP: false, - }) - if res.Err != nil { - panic("should not happen") - } - - subStream = res.SubStream - } - } - - encH264 := &rtph264.Encoder{ - PayloadType: 96, - PayloadMaxSize: s.RTPMaxPayloadSize, - PacketizationMode: 1, - } - err := encH264.Init() - if err != nil { - return err - } - - onData := func(pts int64, ntp time.Time, au [][]byte) { - initializeStream() - - pkts, err2 := encH264.Encode(au) - if err2 != nil { - s.Log(logger.Error, err2.Error()) - return - } - - for _, pkt := range pkts { - pkt.Timestamp = uint32(pts) - subStream.WriteUnit(medi, medi.Formats[0], &unit.Unit{ - PTS: pts, - NTP: ntp, - RTPPackets: []*rtp.Packet{pkt}, - }) - } - } - - var onDataSecondary func(pts int64, ntp time.Time, au []byte) - - if params.Conf.RPICameraSecondaryWidth != 0 { - encJpeg := &rtpmjpeg.Encoder{ - PayloadMaxSize: s.RTPMaxPayloadSize, - } - err = encJpeg.Init() - if err != nil { - panic(err) - } - - onDataSecondary = func(pts int64, ntp time.Time, au []byte) { - initializeStream() - - pkts, err2 := encJpeg.Encode(au) - if err2 != nil { - s.Log(logger.Error, err2.Error()) - return - } - - for _, pkt := range pkts { - pkt.Timestamp = uint32(pts) - pkt.PayloadType = 96 - subStream.WriteUnit(mediaSecondary, mediaSecondary.Formats[0], &unit.Unit{ - PTS: pts, - NTP: ntp, - RTPPackets: []*rtp.Packet{pkt}, - }) - } - } - } - - defer func() { - if subStream != nil { - s.Parent.SetNotReady(defs.PathSourceStaticSetNotReadyReq{}) - } - }() - - cam := &camera{ - params: paramsFromConf(s.LogLevel, params.Conf), - onData: onData, - onDataSecondary: onDataSecondary, - } - err = cam.initialize() //nolint:staticcheck - if err != nil { //nolint:staticcheck - return err - } - defer cam.close() - - cameraErr := make(chan error) - go func() { - cameraErr <- cam.wait() - }() - - for { - select { - case err = <-cameraErr: - return err - - case cnf := <-params.ReloadConf: - cam.reloadParams(paramsFromConf(s.LogLevel, cnf)) - - case <-params.Context.Done(): - return nil - } - } -} - -func (s *Source) runSecondary(params defs.StaticSourceRunParams) error { - r := &secondaryReader{} - r.ctx, r.ctxCancel = context.WithCancel(context.Background()) - defer r.ctxCancel() - - path, primaryStream, err := s.waitForPrimary(r, params) - if err != nil { - return err - } - - defer path.RemoveReader(defs.PathRemoveReaderReq{Author: r}) - - media := &description.Media{ - Type: description.MediaTypeVideo, - Formats: []format.Format{&format.MJPEG{}}, - } - - res := s.Parent.SetReady(defs.PathSourceStaticSetReadyReq{ - Desc: &description.Session{Medias: []*description.Media{media}}, - UseRTPPackets: true, - }) - if res.Err != nil { - return res.Err - } - - rdr := &stream.Reader{Parent: s} - - rdr.OnData( - primaryStream.OrigDesc.Medias[1], - primaryStream.OrigDesc.Medias[1].Formats[0], - func(u *unit.Unit) error { - pkt := u.RTPPackets[0] - - newPkt := &rtp.Packet{ - Header: pkt.Header, - Payload: pkt.Payload, - } - newPkt.PayloadType = 26 - - res.SubStream.WriteUnit(media, media.Formats[0], &unit.Unit{ - PTS: u.PTS, - NTP: u.NTP, - RTPPackets: []*rtp.Packet{newPkt}, - }) - return nil - }) - - primaryStream.AddReader(rdr) - defer primaryStream.RemoveReader(rdr) - - select { - case err = <-rdr.Error(): - return err - - case <-r.ctx.Done(): - return fmt.Errorf("primary stream closed") - - case <-params.Context.Done(): - return fmt.Errorf("terminated") - } -} - -func (s *Source) waitForPrimary( - r *secondaryReader, - params defs.StaticSourceRunParams, -) (defs.Path, *stream.Stream, error) { - for { - res, err := s.Parent.AddReader(defs.PathAddReaderReq{ - Author: r, - AccessRequest: defs.PathAccessRequest{ - Name: params.Conf.RPICameraPrimaryName, - SkipAuth: true, - }, - }) - if err != nil { - if _, ok := errors.AsType[*defs.PathNoStreamAvailableError](err); ok { - select { - case <-time.After(pauseBetweenErrors): - case <-params.Context.Done(): - return nil, nil, fmt.Errorf("terminated") - } - continue - } - - return nil, nil, err - } - - return res.Path, res.Stream, nil - } -} - // APISourceDescribe implements StaticSource. func (*Source) APISourceDescribe() *defs.APIPathSource { return &defs.APIPathSource{ diff --git a/internal/staticsources/rpicamera/source_arm_.go b/internal/staticsources/rpicamera/source_arm_.go new file mode 100644 index 00000000..d05a0793 --- /dev/null +++ b/internal/staticsources/rpicamera/source_arm_.go @@ -0,0 +1,299 @@ +//go:build (linux && arm) || (linux && arm64) + +package rpicamera + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/bluenviron/gortsplib/v5/pkg/description" + "github.com/bluenviron/gortsplib/v5/pkg/format" + "github.com/bluenviron/gortsplib/v5/pkg/format/rtph264" + "github.com/bluenviron/gortsplib/v5/pkg/format/rtpmjpeg" + "github.com/bluenviron/mediacommon/v2/pkg/codecs/h264" + "github.com/bluenviron/mediamtx/internal/defs" + "github.com/bluenviron/mediamtx/internal/logger" + "github.com/bluenviron/mediamtx/internal/stream" + "github.com/bluenviron/mediamtx/internal/unit" + "github.com/pion/rtp" +) + +const ( + pauseBetweenErrors = 1 * time.Second +) + +type secondaryReader struct { + ctx context.Context + ctxCancel func() +} + +// Close implements reader. +func (r *secondaryReader) Close() { + r.ctxCancel() +} + +// APIReaderDescribe implements reader. +func (*secondaryReader) APIReaderDescribe() *defs.APIPathReader { + return &defs.APIPathReader{ + Type: defs.APIPathReaderTypeHidden, + ID: "", + } +} + +// Run implements StaticSource. +func (s *Source) Run(params defs.StaticSourceRunParams) error { + if !params.Conf.RPICameraSecondary { + return s.runPrimary(params) + } + return s.runSecondary(params) +} + +func (s *Source) runPrimary(params defs.StaticSourceRunParams) error { + var medias []*description.Media + + forma := &format.H264{ + PayloadTyp: 96, + PacketizationMode: 1, + } + + media := &description.Media{ + Type: description.MediaTypeVideo, + Formats: []format.Format{forma}, + } + medias = append(medias, media) + + var mediaSecondary *description.Media + + if params.Conf.RPICameraSecondaryWidth != 0 { + mediaSecondary = &description.Media{ + Type: description.MediaTypeApplication, + Formats: []format.Format{&format.Generic{ + PayloadTyp: 96, + RTPMa: "rpicamera_secondary/90000", + ClockRat: 90000, + }}, + } + medias = append(medias, mediaSecondary) + } + + encH264 := &rtph264.Encoder{ + PayloadType: 96, + PayloadMaxSize: s.RTPMaxPayloadSize, + PacketizationMode: 1, + } + err := encH264.Init() + if err != nil { + return err + } + + encode := func(au []byte) ([]*rtp.Packet, error) { + var nalus h264.AnnexB + err = nalus.Unmarshal(au) + if err != nil { + return nil, err + } + + return encH264.Encode(nalus) + } + + var subStream *stream.SubStream + + initializeSubStream := func() { + if subStream == nil { + res := s.Parent.SetReady(defs.PathSourceStaticSetReadyReq{ + Desc: &description.Session{Medias: medias}, + UseRTPPackets: true, + ReplaceNTP: false, + }) + if res.Err != nil { + panic("should not happen") + } + + subStream = res.SubStream + } + } + + onData := func(pts int64, ntp time.Time, au []byte) { + initializeSubStream() + + pkts, err2 := encode(au) + if err2 != nil { + s.Log(logger.Error, err2.Error()) + return + } + + for _, pkt := range pkts { + pkt.Timestamp = uint32(pts) + subStream.WriteUnit(media, media.Formats[0], &unit.Unit{ + PTS: pts, + NTP: ntp, + RTPPackets: []*rtp.Packet{pkt}, + }) + } + } + + var onDataSecondary func(pts int64, ntp time.Time, au []byte) + + if params.Conf.RPICameraSecondaryWidth != 0 { + encJpeg := &rtpmjpeg.Encoder{ + PayloadMaxSize: s.RTPMaxPayloadSize, + } + err := encJpeg.Init() + if err != nil { + panic(err) + } + + onDataSecondary = func(pts int64, ntp time.Time, au []byte) { + initializeSubStream() + + pkts, err2 := encJpeg.Encode(au) + if err2 != nil { + s.Log(logger.Error, err2.Error()) + return + } + + for _, pkt := range pkts { + pkt.Timestamp = uint32(pts) + pkt.PayloadType = 96 + subStream.WriteUnit(mediaSecondary, mediaSecondary.Formats[0], &unit.Unit{ + PTS: pts, + NTP: ntp, + RTPPackets: []*rtp.Packet{pkt}, + }) + } + } + } + + defer func() { + if subStream != nil { + s.Parent.SetNotReady(defs.PathSourceStaticSetNotReadyReq{}) + } + }() + + var p cameraParams + p.fromConf(s.LogLevel, params.Conf) + + cam := &camera{ + params: p, + onData: onData, + onDataSecondary: onDataSecondary, + } + err = cam.initialize() //nolint:staticcheck + if err != nil { //nolint:staticcheck + return err + } + defer cam.close() + + cameraErr := make(chan error) + go func() { + cameraErr <- cam.wait() + }() + + for { + select { + case err = <-cameraErr: + return err + + case cnf := <-params.ReloadConf: + var p cameraParams + p.fromConf(s.LogLevel, cnf) + cam.reloadParams(p) + + case <-params.Context.Done(): + return nil + } + } +} + +func (s *Source) runSecondary(params defs.StaticSourceRunParams) error { + r := &secondaryReader{} + r.ctx, r.ctxCancel = context.WithCancel(context.Background()) + defer r.ctxCancel() + + path, primaryStream, err := s.waitForPrimary(r, params) + if err != nil { + return err + } + + defer path.RemoveReader(defs.PathRemoveReaderReq{Author: r}) + + media := &description.Media{ + Type: description.MediaTypeVideo, + Formats: []format.Format{&format.MJPEG{}}, + } + + res := s.Parent.SetReady(defs.PathSourceStaticSetReadyReq{ + Desc: &description.Session{Medias: []*description.Media{media}}, + UseRTPPackets: true, + }) + if res.Err != nil { + return res.Err + } + + rdr := &stream.Reader{Parent: s} + + rdr.OnData( + primaryStream.OrigDesc.Medias[1], + primaryStream.OrigDesc.Medias[1].Formats[0], + func(u *unit.Unit) error { + pkt := u.RTPPackets[0] + + newPkt := &rtp.Packet{ + Header: pkt.Header, + Payload: pkt.Payload, + } + newPkt.PayloadType = 26 + + res.SubStream.WriteUnit(media, media.Formats[0], &unit.Unit{ + PTS: u.PTS, + NTP: u.NTP, + RTPPackets: []*rtp.Packet{newPkt}, + }) + return nil + }) + + primaryStream.AddReader(rdr) + defer primaryStream.RemoveReader(rdr) + + select { + case err = <-rdr.Error(): + return err + + case <-r.ctx.Done(): + return fmt.Errorf("primary stream closed") + + case <-params.Context.Done(): + return fmt.Errorf("terminated") + } +} + +func (s *Source) waitForPrimary( + r *secondaryReader, + params defs.StaticSourceRunParams, +) (defs.Path, *stream.Stream, error) { + for { + res, err := s.Parent.AddReader(defs.PathAddReaderReq{ + Author: r, + AccessRequest: defs.PathAccessRequest{ + Name: params.Conf.RPICameraPrimaryName, + SkipAuth: true, + }, + }) + if err != nil { + if _, ok := errors.AsType[*defs.PathNoStreamAvailableError](err); ok { + select { + case <-time.After(pauseBetweenErrors): + case <-params.Context.Done(): + return nil, nil, fmt.Errorf("terminated") + } + continue + } + + return nil, nil, err + } + + return res.Path, res.Stream, nil + } +} diff --git a/internal/staticsources/rpicamera/source_other.go b/internal/staticsources/rpicamera/source_other.go new file mode 100644 index 00000000..621828d2 --- /dev/null +++ b/internal/staticsources/rpicamera/source_other.go @@ -0,0 +1,14 @@ +//go:build !linux || (!arm && !arm64) + +package rpicamera + +import ( + "fmt" + + "github.com/bluenviron/mediamtx/internal/defs" +) + +// Run implements StaticSource. +func (s *Source) Run(_ defs.StaticSourceRunParams) error { + return fmt.Errorf("server was compiled without support for the Raspberry Pi Camera") +}