main.go 3.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156
  1. package main
  2. import (
  3. "context"
  4. "flag"
  5. "log"
  6. "net/http"
  7. "golang.org/x/crypto/blake2b"
  8. dev "github.com/vladimirvivien/go4vl/device"
  9. "github.com/vladimirvivien/go4vl/v4l2"
  10. "github.com/gorilla/websocket"
  11. )
  12. func sender(queue chan []byte, secret string) {
  13. log.Println("start sender")
  14. h := make(http.Header, 1)
  15. h.Set("X-Secret", secret)
  16. for {
  17. ws, _, err := websocket.DefaultDialer.Dial("wss://yebi.su/ws", h)
  18. if err != nil {
  19. log.Println("dial error")
  20. continue
  21. }
  22. log.Println("connected")
  23. for {
  24. data := <-queue
  25. log.Printf("sending %db\n", len(data))
  26. if err := ws.WriteMessage(websocket.BinaryMessage, data); err != nil {
  27. log.Println("WriteMessage() failed")
  28. break
  29. }
  30. }
  31. ws.Close()
  32. }
  33. }
  34. func chunkBy[T any](items []T, chunkSize int) (chunks [][]T) {
  35. for chunkSize < len(items) {
  36. items, chunks = items[chunkSize:], append(chunks, items[0:chunkSize:chunkSize])
  37. }
  38. return chunks
  39. }
  40. func processor(inQueue chan []byte, outQueue chan []byte) {
  41. log.Println("start processor")
  42. for {
  43. data := <-inQueue
  44. newData := []byte{}
  45. for _, chunk := range chunkBy(data, 2) {
  46. newData = append(newData, chunk[0]^chunk[1])
  47. }
  48. data = newData
  49. newData = []byte{}
  50. for _, chunk := range chunkBy(data, 8) {
  51. var b byte = 0
  52. for j := 0; j < 8; j++ {
  53. if chunk[j]&1 != 0 {
  54. b |= (1 << (7 - j))
  55. }
  56. }
  57. newData = append(newData, b)
  58. }
  59. data = newData
  60. newData = []byte{}
  61. for _, chunk := range chunkBy(data, 32) {
  62. sum := blake2b.Sum256(chunk)
  63. newData = append(newData, sum[:]...)
  64. }
  65. outQueue <- newData
  66. }
  67. }
  68. func main() {
  69. var deviceName string
  70. var width uint
  71. var height uint
  72. var bufferSize uint
  73. var ident string
  74. var secret string
  75. flag.StringVar(&deviceName, "device", "/dev/video0", "v4l2 device to capture from")
  76. flag.UintVar(&width, "width", 1280, "frame width")
  77. flag.UintVar(&height, "height", 720, "frame height")
  78. flag.UintVar(&bufferSize, "bufferSize", 1024, "sending queue max capacity")
  79. flag.StringVar(&ident, "ident", "", "yebi.su ident")
  80. flag.StringVar(&secret, "secret", "", "yebi.su secret")
  81. flag.Parse()
  82. if ident == "" {
  83. log.Panicln("ident not specified")
  84. }
  85. if secret == "" {
  86. log.Panicln("secret not specified")
  87. }
  88. secret = ident + " " + secret
  89. log.Printf("open v4l2 device: %s\n", deviceName)
  90. device, err := dev.Open(
  91. deviceName,
  92. dev.WithPixFormat(v4l2.PixFormat{Width: uint32(width), Height: uint32(height), PixelFormat: v4l2.PixelFmtYUYV, Field: v4l2.FieldNone}),
  93. )
  94. if err != nil {
  95. log.Panicln("failed to open device")
  96. }
  97. defer device.Close()
  98. processorQueue := make(chan []byte, bufferSize)
  99. senderQueue := make(chan []byte, bufferSize)
  100. go sender(senderQueue, secret)
  101. go processor(processorQueue, senderQueue)
  102. log.Println("start recording")
  103. ctx, stop := context.WithCancel(context.TODO())
  104. if err := device.Start(ctx); err != nil {
  105. log.Panicln("failed to start recording on device")
  106. }
  107. defer stop()
  108. log.Println("start loop")
  109. for {
  110. frame := <-device.GetOutput()
  111. processorQueue <- frame
  112. }
  113. }