rtsp_reader-writer/reader/internal/unpacker/proc.go

package unpacker

import (
	"bytes"
	"encoding/binary"
	"errors"
	"fmt"
	"io"
	"log"
	"os"
	"strconv"
	"strings"
	"time"

	"git.insit.tech/psa/rtsp_reader-writer/writer/pkg/storage"
	"git.insit.tech/sas/rtsp_proxy/core/gen"
	log2 "git.insit.tech/sas/rtsp_proxy/core/log"
	"github.com/bluenviron/gortsplib/v4/pkg/format"
	"go.uber.org/zap"
	logger "reader/internal/log"
	"reader/internal/processor"
)

// dir

// CreateVideo generate TS files and M3U8 playlists.
func CreateVideo(dir string) error {
	// Check if the data folder in the directory.
	homeDir, err := os.UserHomeDir()
	if err != nil {
		return err
	}
	dirData := fmt.Sprintf("%s/%s/data", homeDir, dir)

	exist := storage.Exists(dirData)
	if !exist {
		return errors.New("directory does not exist")
	}

	// Read directory.
	files, err := storage.ReadDir(dirData)
	if err != nil {
		return err
	}

	for _, file := range files {
		// Create logger.
		cam := log2.CamLogging(
			fmt.Sprintf("%s/%s/reader-cam_%s.log", dirData, file.Name(), time.Now().Format("15-04-05_02-01-2006")))

		res, err := storage.ReadDir(fmt.Sprintf("%s/%s", dirData, file.Name()))
		if err != nil {
			cam.Error("error reading directory", zap.String("dir", dir), zap.Error(err))
			return err
		}
		resolutions := make([]string, 0)
		for _, r := range res {
			if r.IsDir() {
				resolutions = append(resolutions, r.Name())
			}
		}

		logger.Log.Info("start process camera:", zap.String("cam_name", file.Name()))
		log.Println("start process camera: ", file.Name())

		// Establish starting period.
		per := 60
		period := time.Duration(per)

		// Create M3U8 playlist.
		go gen.MediaPlaylistGenerator(dirData, file.Name(), float64(period), resolutions, cam)

		for {
			for _, resolution := range resolutions {
				filenames := gen.StringDirEntryList(
					fmt.Sprintf("%s/%s/%s", dirData, file.Name(), resolution), "insit", logger.Log)
				if len(filenames) == 0 {
					break
				}

				for i := len(filenames) - 2; i < len(filenames)-1; i++ {
					segment := storage.Segment{}

					// Open file for reading.
					f, err := os.Open(fmt.Sprintf("%s/%s/%s/%s", dirData, file.Name(), resolution, filenames[i]))
					if err != nil {
						cam.Error(
							"opening file error for file:", zap.String("filename", filenames[i]), zap.Error(err))
						return err
					}
					defer f.Close()

					// Read StreamID.
					var streamIDLen int32
					if err := binary.Read(f, binary.LittleEndian, &streamIDLen); err != nil {
						cam.Error(
							"reading StreamID length error:", zap.String("filename", filenames[i]), zap.Error(err))
						return err
					}
					streamIDBytes := make([]byte, streamIDLen)
					if _, err := io.ReadFull(f, streamIDBytes); err != nil {
						cam.Error("reading StreamID error:", zap.String("filename", filenames[i]), zap.Error(err))
						return err
					}
					streamID := string(streamIDBytes)
					cam.Info(
						"cam with Stream ID started to convert:",
						zap.String("filename", filenames[i]),
						zap.String("stream_id", streamID))

					// Read header of the file.
					var segLen int32
					if err := binary.Read(f, binary.LittleEndian, &segLen); err != nil {
						cam.Error(
							"reading header length error:", zap.String("filename", filenames[i]), zap.Error(err))
						return err
					}
					segData := make([]byte, segLen)
					if _, err := io.ReadFull(f, segData); err != nil {
						cam.Error("reading header error:", zap.String("filename", filenames[i]), zap.Error(err))
						return err
					}
					headerReader := bytes.NewReader(segData)
					headerSeg, err := readHeaderSegment(headerReader)
					if err != nil {
						cam.Error(
							"func readHeaderSegment error:", zap.String("filename", filenames[i]), zap.Error(err))
						return err
					}

					// Parse duration of a segment.
					per, err = strconv.Atoi(headerSeg.Duration)
					if err != nil {
						cam.Error("parsing duration error:", zap.String("filename", filenames[i]), zap.Error(err))
						return err
					}

					// Setup MPEG-TS muxer.
					var h264Format format.H264
					var aacFormat format.MPEG4Audio

					filename, _ := strings.CutSuffix(filenames[i], ".insit")
					tsFilename := fmt.Sprintf("%s/%s/%s/%s.ts", dirData, file.Name(), resolution, filename)

					currentMpegtsMuxer := processor.MpegtsMuxer{
						FileName:         tsFilename,
						H264Format:       &h264Format,
						Mpeg4AudioFormat: &aacFormat,
					}

					err = currentMpegtsMuxer.Initialize()
					if err != nil {
						cam.Error("init muxer error:", zap.String("filename", filenames[i]), zap.Error(err))
						return err
					}

					// Read segments.
					for {
						// Read segments length.
						var segLen int32
						err := binary.Read(f, binary.LittleEndian, &segLen)
						if err != nil {
							if err == io.EOF {
								break
							}
							cam.Error(
								"read segments length error:", zap.String("filename", filenames[i]), zap.Error(err))
							return err
						}
						// Read segments data.
						segData = make([]byte, segLen)
						if _, err := io.ReadFull(f, segData); err != nil {
							cam.Error(
								"read segments error:", zap.String("filename", filenames[i]), zap.Error(err))
							return err
						}
						packetReader := bytes.NewReader(segData)
						packets, err := readPacketSegment(packetReader)
						if err != nil {
							cam.Error(
								"func readPacketSegment error:", zap.String("filename", filenames[i]), zap.Error(err))
							return err
						}

						segment.Packets = append(segment.Packets, packets...)

						for _, pkt := range packets {
							switch pkt.Type {
							case storage.PacketTypeH264:
								// Encode the access unit into MPEG-TS.
								err = currentMpegtsMuxer.WriteH264(pkt.Pts, pkt.H264AUs)
								if err != nil {
									cam.Warn(
										"write H264 packet error:", zap.String("filename", filenames[i]), zap.Error(err))
								}
							case storage.PacketTypeLPCM:
								// Convert G711 to AAC.
								au, err := processor.ConvertLPCMToAAC(pkt.LPCMSamples)
								if err != nil {
									cam.Warn(
										"converting to AAC frame error:", zap.String("filename", filenames[i]), zap.Error(err))
								}

								// Encode the access unit into MPEG-TS.
								err = currentMpegtsMuxer.WriteAAC([][]byte{au}, pkt.Pts)
								if err != nil {
									cam.Warn(
										"write G711 packet error:", zap.String("filename", filenames[i]), zap.Error(err))
								}
							}
						}
					}
					currentMpegtsMuxer.Close()

					if storage.Exists(tsFilename) {
						time.Sleep(period * time.Second)
					}
				}
			}
		}
	}
	return errors.New("func CreateVideo downed")
}