1package stream23import (4 "bufio"5 "bytes"6 "context"7 "fmt"8 "io"9 "os"10 "path/filepath"11 "strconv"12 "strings"13 "time"1415 "airwaves/internal/cc"16)1718// A stream whose captions come as text (a custom channel's, from its19// cc.Script) gets them as a WebVTT subtitle rendition beside the video:20// subs.m3u8 lists a .vtt segment for each video segment, with the same21// sequence numbers and lengths, so players fetch them in step, rewinding22// included.23//24// Cue times in the script count from the stream's first frame, which is25// the first picture of the first segment: ffmpeg's transcode keeps the26// source's frames one for one. Each WebVTT segment maps its own segment's27// first picture time (X-TIMESTAMP-MAP's MPEGTS, in the 90 kHz clock the28// player aligns video by) to its time in the stream (LOCAL), and gives its29// cues stream times, so they line up wherever a player starts or seeks.3031const (32 subsPlaylist = "subs.m3u8"33 subsPoll = 200 * time.Millisecond34 subsAhead = 10 * time.Second35)3637// subtitler writes a session's WebVTT segments as its video segments38// appear.39type subtitler struct {40 dir string41 script *cc.Script42 first int64 // PTS of the stream's first picture; -1 until known43 segs map[int]subSegment44 audio bool // the script counts from the first sound (byAudio)45 lead time.Duration // and that's this long after the first picture46}4748// subSegment is a video segment's place in time.49type subSegment struct {50 pts int64 // its first picture's PTS, unwrapped51 raw int64 // the same within the 33-bit clock52 start time.Duration53 dur float64 // seconds, as the playlist lists it54}5556func newSubtitler(dir string, script *cc.Script) *subtitler {57 return &subtitler{dir: dir, script: script, first: -1, segs: map[int]subSegment{}}58}5960// byAudio has the script count from the stream's first sound rather than61// its first picture: for a picture copied as it is (a podcast's), which62// starts at the key frame before where the stream was asked to, while the63// sound starts there.64func (s *subtitler) byAudio() *subtitler {65 s.audio = true66 return s67}6869// run keeps the subtitles in step with the video until ctx ends.70func (s *subtitler) run(ctx context.Context) {71 t := time.NewTicker(subsPoll)72 defer t.Stop()73 for {74 _ = s.update()75 select {76 case <-ctx.Done():77 return78 case <-t.C:79 }80 }81}8283// mediaPlaylist is the part of an HLS media playlist the subtitler reads.84type mediaPlaylist struct {85 target string // EXT-X-TARGETDURATION86 sequence int // of the first segment87 segs []playlistSegment88 kind string // EXT-X-PLAYLIST-TYPE, "" for none89 ended bool // EXT-X-ENDLIST90}9192type playlistSegment struct {93 dur float6494 uri string95}9697func parseMediaPlaylist(b []byte) (mediaPlaylist, error) {98 var p mediaPlaylist99 sc := bufio.NewScanner(bytes.NewReader(b))100 dur := -1.0101 for sc.Scan() {102 line := strings.TrimSpace(sc.Text())103 switch {104 case strings.HasPrefix(line, "#EXT-X-TARGETDURATION:"):105 p.target = strings.TrimPrefix(line, "#EXT-X-TARGETDURATION:")106 case strings.HasPrefix(line, "#EXT-X-PLAYLIST-TYPE:"):107 p.kind = strings.TrimPrefix(line, "#EXT-X-PLAYLIST-TYPE:")108 case line == "#EXT-X-ENDLIST":109 p.ended = true110 case strings.HasPrefix(line, "#EXT-X-MEDIA-SEQUENCE:"):111 p.sequence, _ = strconv.Atoi(strings.TrimPrefix(line, "#EXT-X-MEDIA-SEQUENCE:"))112 case strings.HasPrefix(line, "#EXTINF:"):113 v, _, _ := strings.Cut(strings.TrimPrefix(line, "#EXTINF:"), ",")114 dur, _ = strconv.ParseFloat(v, 64)115 case line != "" && !strings.HasPrefix(line, "#") && dur >= 0:116 p.segs = append(p.segs, playlistSegment{dur: dur, uri: line})117 dur = -1118 }119 }120 if p.target == "" {121 return p, fmt.Errorf("not a media playlist")122 }123 return p, sc.Err()124}125126// update writes WebVTT for video segments new since the last call, and127// subs.m3u8 to list them.128func (s *subtitler) update() error {129 raw, err := os.ReadFile(filepath.Join(s.dir, "stream_0.m3u8"))130 if err != nil {131 return err132 }133 p, err := parseMediaPlaylist(raw)134 if err != nil || len(p.segs) == 0 {135 return err136 }137 var list bytes.Buffer138 fmt.Fprintf(&list, "#EXTM3U\n#EXT-X-VERSION:3\n#EXT-X-TARGETDURATION:%s\n#EXT-X-MEDIA-SEQUENCE:%d\n", p.target, p.sequence)139 if p.kind != "" {140 fmt.Fprintf(&list, "#EXT-X-PLAYLIST-TYPE:%s\n", p.kind)141 }142 ended := p.ended // the video's ended, and so have these once all are written143 for i, seg := range p.segs {144 n := p.sequence + i145 if _, done := s.segs[n]; !done {146 if err := s.write(n, seg); err != nil {147 ended = false148 break // list only what's written; try again shortly149 }150 }151 fmt.Fprintf(&list, "#EXTINF:%.6f,\n%s\n", seg.dur, subName(n))152 }153 if ended {154 list.WriteString("#EXT-X-ENDLIST\n")155 }156 if err := writeAtomic(filepath.Join(s.dir, subsPlaylist), list.Bytes()); err != nil {157 return err158 }159 // Drop what has left the playlist, keeping a couple for players160 // mid-fetch.161 for n := range s.segs {162 if n < p.sequence-2 {163 _ = os.Remove(filepath.Join(s.dir, subName(n)))164 delete(s.segs, n)165 }166 }167 if first, ok := s.segs[p.sequence]; ok {168 s.script.Forget(first.start - time.Minute)169 }170 return nil171}172173func subName(n int) string { return fmt.Sprintf("sub_%05d.vtt", n) }174175// write makes the WebVTT segment for video segment n.176func (s *subtitler) write(n int, seg playlistSegment) error {177 pts, err := firstPTS(filepath.Join(s.dir, seg.uri), false)178 if err != nil {179 prev, ok := s.segs[n-1]180 if !ok {181 return err182 }183 // Gone already, or unreadable: carry on from the one before.184 pts = (prev.raw + int64(prev.dur*90000)) & ptsMask185 }186 cur := subSegment{raw: pts, dur: seg.dur}187 if prev, ok := s.segs[n-1]; ok {188 cur.pts = prev.pts + (pts-prev.raw)&ptsMask189 } else if s.first < 0 {190 s.first, cur.pts = pts, pts191 if s.audio {192 // How long the sound starts after the picture, which the193 // script's times count from.194 if a, err := firstPTS(filepath.Join(s.dir, seg.uri), true); err == nil {195 if d := (a - pts) & ptsMask; d < 10*90000 {196 s.lead = time.Duration(d) * time.Second / 90000197 }198 }199 }200 } else {201 // A gap in what's been seen: the nearest way round the clock.202 cur.pts = s.first + (pts-s.first)&ptsMask203 }204 cur.start = time.Duration(cur.pts-s.first) * time.Second / 90000205 end := cur.start + time.Duration(seg.dur*float64(time.Second))206207 b := fmt.Appendf(nil, "WEBVTT\nX-TIMESTAMP-MAP=MPEGTS:%d,LOCAL:%s\n\n", pts, cc.VTTTime(cur.start))208 // Cues starting a little later come too, so players that lay captions209 // out ahead know what's coming; they're repeated in their own segments.210 for _, c := range cc.Shift(s.script.Between(cur.start-s.lead, end+subsAhead-s.lead), s.lead) {211 b = cc.AppendVTT(b, c)212 }213 if err := writeAtomic(filepath.Join(s.dir, subName(n)), b); err != nil {214 return err215 }216 s.segs[n] = cur217 return nil218}219220const ptsMask = 1<<33 - 1221222// firstPTS reads the presentation time of a segment's first picture, or223// with audio its first sound, in 90 kHz ticks.224func firstPTS(path string, audio bool) (int64, error) {225 f, err := os.Open(path)226 if err != nil {227 return 0, err228 }229 defer f.Close()230 r := bufio.NewReaderSize(f, 64<<10)231 pkt := make([]byte, 188)232 for range 4096 {233 if _, err := io.ReadFull(r, pkt); err != nil {234 return 0, fmt.Errorf("%s: no picture time", filepath.Base(path))235 }236 if pkt[0] != 0x47 || pkt[1]&0x40 == 0 || pkt[3]&0x10 == 0 {237 continue238 }239 p := pkt[4:]240 if pkt[3]&0x20 != 0 {241 if int(pkt[4]) >= 183 {242 continue243 }244 p = pkt[5+int(pkt[4]):]245 }246 // A video (or audio) PES with a PTS.247 if len(p) < 14 || p[0] != 0 || p[1] != 0 || p[2] != 1 || p[7]&0x80 == 0 {248 continue249 }250 if !audio && p[3]&0xf0 == 0xe0 || audio && p[3]&0xe0 == 0xc0 {251 return int64(p[9]>>1&7)<<30 | int64(p[10])<<22 | int64(p[11]>>1)<<15 | int64(p[12])<<7 | int64(p[13]>>1), nil252 }253 }254 return 0, fmt.Errorf("%s: no picture time", filepath.Base(path))255}256257// writeAtomic replaces a file so readers never see it half written.258func writeAtomic(path string, b []byte) error {259 tmp := path + ".tmp"260 if err := os.WriteFile(tmp, b, 0o644); err != nil {261 return err262 }263 return os.Rename(tmp, path)264}