1package vchan23import (4 "bytes"5 "cmp"6 "context"7 "fmt"8 "io"9 "log"10 "os"11 "os/exec"12 "regexp"13 "strconv"14 "strings"15 "sync"16 "sync/atomic"17 "time"18 "unicode"19 "unicode/utf8"2021 "airwaves/internal/cc"22 "airwaves/internal/phase"23)2425// item is one video in a looping channel.26type item struct {27 Path string // file or URL28 Frames int64 // length in output frames29 Audio bool30 Title string31 Subtitle string32 Description string33 Category string34 Season string35 Episode string36 Image string37 // Aired, Cast, Rating and Genres describe it further, as Program's.38 Aired string39 Year int40 Cast []string41 Rating string42 Genres []string43 // Input, when set, returns ffmpeg input options (ending "-i <url>") that44 // play the video from offset in, for sources that seek on their side.45 // Otherwise Path is read with a local seek. The stream's pace is set46 // as it goes to the encoder (see pacer), not by the input.47 Input func(offset time.Duration) []string48 // AudioStream is the sound's stream specifier in Path's or Input's49 // streams; "0:a:0" when empty.50 AudioStream string51 // AudioLang is the sound's language as tagged ("eng", "jpn"), "" when52 // unknown, which is taken for English.53 AudioLang string54 // Captions are the video's English captions, nil when it has none.55 Captions *captions56 // Level, when set, evens the item's sound out as it streams (see57 // loudness.go).58 Level *itemLevel59 // Open, when set, takes precedence over Input. It is called right before60 // the video plays, for sources whose addresses must be looked up then61 // and may fail. When it fails, a slate fills the video's time so the62 // stream keeps to the schedule.63 Open func(ctx context.Context, offset time.Duration) (opened, error)64 // Slate makes the item a quiet dark picture, captioned with Title when65 // ffmpeg can draw text.66 Slate bool67}6869// opened is how to play an item found when it opens.70type opened struct {71 // Args are ffmpeg input options for one or more inputs (picture first,72 // each ending "-i <url>") that play from the offset asked for.73 Args []string74 // Audio is the sound's stream specifier ("1:a:0"), empty for silence.75 Audio string76 // AudioLang and Captions take the place of the item's.77 AudioLang string78 Captions *captions79}8081const (82 loopWidth = 128083 loopHeight = 72084 loopFPS = 3085 sampleRate = 4800086 // lead is how far ahead of real time a stream starts: its first87 // seconds are decoded and encoded as fast as they can be, so a viewer88 // tuning in waits for none of them, and after that the stream keeps89 // that far ahead of the clock (see pacer).90 lead = 6 * time.Second91 // maxPrograms bounds a guide request on a channel of very short videos.92 maxPrograms = 200093)9495// scheduleEpoch anchors every looping channel. A schedule only changes when96// the channel's videos do.97var scheduleEpoch = time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)9899// framesSince is how many output frames have passed between the epoch and t.100func framesSince(t time.Time) int64 { return framesIn(t.Sub(scheduleEpoch)) }101102// framesIn is how many output frames d lasts.103func framesIn(d time.Duration) int64 {104 return int64(d/time.Second)*loopFPS + int64(d%time.Second)*loopFPS/int64(time.Second)105}106107func frameDuration(frames int64) time.Duration {108 return time.Duration(frames) * time.Second / loopFPS109}110111// playing returns the index of the video on at t and how many frames into112// it, or -1 for an empty folder.113func playing(items []item, t time.Time) (int, int64) {114 var total int64115 for _, it := range items {116 total += it.Frames117 }118 if total == 0 {119 return -1, 0120 }121 n := framesSince(t) % total122 if n < 0 {123 n += total124 }125 for i, it := range items {126 if n < it.Frames {127 return i, n128 }129 n -= it.Frames130 }131 return 0, 0132}133134// programs is the guide for items looping on the clock: one entry per135// video, in loop order.136func programs(items []item, from, to time.Time) []Program {137 i, into := playing(items, from)138 if i < 0 {139 return nil140 }141 start := from.Add(-frameDuration(into))142 var out []Program143 for start.Before(to) && len(out) < maxPrograms {144 it := items[i]145 end := start.Add(frameDuration(it.Frames))146 out = append(out, programOf(it, start, end))147 start, i = end, (i+1)%len(items)148 }149 return out150}151152// programOf is an item's guide entry, airing from start to end.153func programOf(it item, start, end time.Time) Program {154 p := Program{155 Start: start, End: end, Title: it.Title, Subtitle: it.Subtitle, Description: it.Description,156 Category: it.Category, Season: it.Season, Episode: it.Episode, Image: it.Image,157 AudioLang: it.AudioLang, Captions: it.Captions != nil,158 Aired: it.Aired, Cast: it.Cast, Rating: it.Rating, Genres: it.Genres,159 }160 switch {161 case p.Date != "":162 case len(it.Aired) == len("2006-01-02"):163 p.Date = strings.ReplaceAll(it.Aired, "-", "")164 case it.Year > 0:165 p.Date = strconv.Itoa(it.Year)166 }167 return p168}169170// lineup gives a looping channel's items, and its schedule (nil for none),171// as they are now.172type lineup func() ([]item, *sched)173174// guideFor is the guide for a looping channel: on its schedule when it has175// one, else looping on the clock.176func guideFor(get lineup, from, to time.Time) []Program {177 items, s := get()178 if s == nil {179 return programs(items, from, to)180 }181 return scheduledPrograms(items, s, from, to)182}183184// scheduledPrograms is the guide for items airing on a schedule: one entry185// per airing, and "Off air" between, saying what's next and when.186func scheduledPrograms(items []item, s *sched, from, to time.Time) []Program {187 ls := lengths(items)188 g := s.guide(ls, from, to, maxPrograms)189 var out []Program190 for k, a := range g {191 if a.I >= 0 {192 out = append(out, programOf(items[a.I], a.Start, a.End))193 continue194 }195 // A gap ends with the next entry, or past the window, the next airing.196 next, ok := slot{}, false197 if k+1 < len(g) {198 next, ok = g[k+1], true199 } else {200 next, _, ok = s.at(ls, a.End)201 }202 if !ok {203 out = append(out, offAirProgram(a, "", time.Time{}, false))204 continue205 }206 out = append(out, offAirProgram(a, airingName(items[next.I]), next.Start, isPremiere(s, ls, next)))207 }208 return out209}210211// isPremiere reports whether a is the schedule's very first airing.212func isPremiere(s *sched, lengths []time.Duration, a slot) bool {213 first, ok := s.premiere(lengths)214 return ok && first.Start.Equal(a.Start)215}216217// lengths are the items' lengths, as a schedule takes them.218func lengths(items []item) []time.Duration {219 out := make([]time.Duration, len(items))220 for i, it := range items {221 out[i] = frameDuration(it.Frames)222 }223 return out224}225226// airingName names an item as an off-air note gives what's next: an227// episode by its series and number, "House of the Dragon, S1 E4".228func airingName(it item) string {229 if it.Season != "" && it.Episode != "" {230 return fmt.Sprintf("%s, S%s E%s", it.Title, it.Season, it.Episode)231 }232 return it.Title233}234235// sequence lists items for choosing where a schedule begins.236func sequence(items []item) []SequenceItem {237 out := make([]SequenceItem, len(items))238 for i, it := range items {239 out[i] = SequenceItem{Title: it.Title, Subtitle: it.Subtitle, Season: it.Season, Episode: it.Episode, Length: frameDuration(it.Frames)}240 }241 return out242}243244// play streams a looping channel to w until ctx ends: its items on the245// clock as they were when it started, or on its schedule, from when it has246// one (see scheduled). label names the channel in logs.247func play(ctx context.Context, w io.Writer, ffmpeg, label string, get lineup) error {248 items, s := get()249 i, skip := playing(items, time.Now())250 if i < 0 {251 return fmt.Errorf("%s: nothing to play", label)252 }253 var onSchedule cueFunc254 if s != nil {255 onSchedule = scheduled(label, get)256 }257 started, failures := false, 0258 return playCues(ctx, w, ffmpeg, label, func(now time.Time, ok bool) (item, int64, error) {259 // A schedule set meanwhile takes over as the video on ends.260 if onSchedule == nil && started {261 if _, s := get(); s != nil {262 onSchedule = scheduled(label, get)263 }264 }265 if onSchedule != nil {266 return onSchedule(now, ok)267 }268 if started {269 if ok {270 failures = 0271 } else if failures++; failures >= len(items) {272 return item{}, 0, fmt.Errorf("%s: no video would play", label)273 }274 i, skip = (i+1)%len(items), 0275 }276 started = true277 // The next video's captions load while this one plays.278 if next := items[(i+1)%len(items)].Captions; next != nil {279 next.start()280 }281 return items[i], skip, nil282 })283}284285const (286 // cueSlack is how far a scheduled stream may fall behind the clock287 // before it catches up.288 cueSlack = 5 * time.Second289 // maxCatchUp is how late a stream may be and still catch up as a290 // broadcaster does, playing the next airing from its top and cutting291 // its end (the credits), rather than cutting into it (its cold open).292 maxCatchUp = 2 * time.Minute293 // slateChunk is the longest a slate plays before the schedule is294 // looked at again, so changes to it show.295 slateChunk = time.Minute296)297298// scheduled cues a channel's items on its schedule: as each ends, the one299// on air then, from where it is, and while the channel is off the air, a300// slate saying when it's back, a minute at a time. The lineup is asked for301// each time, so changes to the items or the schedule apply as the stream302// goes on; with the schedule gone, the items loop on the clock. An airing303// that fails stands by for the rest of its time, and the stream ends once304// as many fail in a row as there are items.305func scheduled(label string, get lineup) cueFunc {306 // The airing cued last (I -1 for a slate), and the last that failed.307 last, failed := slot{I: -1}, slot{I: -1}308 var end time.Time // when what was cued last ends309 failures := 0310 return func(now time.Time, ok bool) (item, int64, error) {311 items, s := get()312 if len(items) == 0 {313 return item{}, 0, fmt.Errorf("%s: nothing to play", label)314 }315 if !end.IsZero() {316 switch {317 case !ok:318 if failures++; failures >= len(items) {319 return item{}, 0, fmt.Errorf("%s: no video would play", label)320 }321 if last.I >= 0 {322 failed = last323 }324 case last.I >= 0:325 failures = 0326 }327 }328 // What plays next picks up where the last left off, however early329 // its decoder finished, unless it failed or fell well behind.330 t := now331 if !end.IsZero() && ok && now.Sub(end) < cueSlack {332 t = end333 }334 var on slot335 onAir := true336 if s == nil {337 i, into := playing(items, t)338 start := t.Add(-frameDuration(into))339 on = slot{i, start, start.Add(frameDuration(items[i].Frames))}340 } else {341 ls := lengths(items)342 var found bool343 if on, onAir, found = s.at(ls, t); !found {344 return item{}, 0, fmt.Errorf("%s: nothing will air", label)345 }346 if !onAir {347 stop := minTime(on.Start, t.Add(slateChunk))348 last, end = slot{-1, t, stop}, stop349 note := offAirProgram(slot{-1, t, on.Start}, airingName(items[on.I]), on.Start, isPremiere(s, ls, on))350 return item{Slate: true, Title: strings.TrimSuffix(note.Description, "."), Frames: max(framesIn(stop.Sub(t)), 1)}, 0, nil351 }352 }353 if on.I == failed.I && on.Start.Equal(failed.Start) {354 stop := minTime(on.End, t.Add(slateChunk))355 last, end = slot{-1, t, stop}, stop356 return item{Slate: true, Title: standBy, Frames: max(framesIn(stop.Sub(t)), 1)}, 0, nil357 }358 // The next video's captions load while this one plays.359 if next := items[(on.I+1)%len(items)].Captions; next != nil {360 next.start()361 }362 it, skip := items[on.I], max(framesIn(t.Sub(on.Start)), 0)363 if late := t.Sub(on.Start); skip > 0 && on.Start.Equal(end) && late <= maxCatchUp && late < frameDuration(it.Frames)/2 {364 log.Printf("%s: %s late, cutting the end of %s to catch up", label, late.Round(time.Second), airingName(it))365 it.Frames, skip = it.Frames-skip, 0366 }367 last, end = on, on.End368 return it, skip, nil369 }370}371372func minTime(a, b time.Time) time.Time {373 if b.Before(a) {374 return b375 }376 return a377}378379// cueFunc says what to play at now, and how many frames into it to start.380// It is asked first when the stream starts, then each time an item ends;381// ok reports whether that item played without error. An error ends the382// stream.383type cueFunc func(now time.Time, ok bool) (item, int64, error)384385// playCues streams the items cue gives to w until ctx ends. One encoder386// runs for the whole viewing, fed raw picture and sound by a decoder per387// video in turn, so clients see a single unbroken stream across videos of388// any format. The videos' captions go into the picture as it leaves.389func playCues(ctx context.Context, w io.Writer, ffmpeg, label string, cue cueFunc) error {390 parent := ctx391 ctx, cancel := context.WithCancel(ctx)392 defer cancel()393394 // Decoders write picture and sound to pipes of their own; feeds carry395 // them to the encoder's pair through memory. A raw frame is far bigger396 // than a pipe, and ffmpeg holds back whichever input runs ahead, so397 // direct pipes deadlock: the decoder stuck writing picture while the398 // encoder waits for sound.399 p, err := openPipes(2)400 if err != nil {401 return err402 }403 encVideo, encAudio := p[0], p[1]404 app, script := forApp(ctx)405 out := &output{captions: &timeline{script: script}, app: app}406 captioned := cc.NewWriter(w, loopFPS, out.captions.pair)407 enc := exec.CommandContext(ctx, ffmpeg, encodeArgs()...)408 enc.ExtraFiles = []*os.File{encVideo.r, encAudio.r} // fds 3 and 4409 enc.Stdout = captioned410 var encErr bytes.Buffer411 enc.Stderr = &encErr412 err = enc.Start()413 encVideo.r.Close()414 encAudio.r.Close()415 if err != nil {416 encVideo.w.Close()417 encAudio.w.Close()418 return fmt.Errorf("start ffmpeg: %w", err)419 }420 encDone := make(chan struct{})421 go func() {422 _ = enc.Wait()423 close(encDone)424 cancel() // the viewer left, or the encoder failed425 }()426 // The picture sets the pace: the decoders run as fast as it lets them,427 // sound and all.428 out.video = newFeed(encVideo.w, 512, frameBytes, &pacer{per: loopFPS, lead: lead, quit: ctx.Done()}) // up to 32 MB, under a second429 out.audio = newFeed(encAudio.w, 128, 4, nil)430 out.video.first = func() { phase.Step(ctx, "decoding") }431 finish := func() {432 // Ending the feeds ends the encoder.433 out.video.close()434 out.audio.close()435 <-encDone436 }437 defer finish()438439 // What comes next is cued at the stream's own clock: where it was last440 // cued, plus the frames since. It runs ahead of the clock on the wall441 // by up to lead, which so holds from one video to the next; a stream442 // that fell behind is cued at the wall's.443 var at time.Time444 var cued int64 // frames when last cued445 ok := true446 for ctx.Err() == nil {447 now := time.Now()448 if s := at.Add(frameDuration(out.video.units - cued)); !at.IsZero() && s.After(now) {449 now = s450 }451 it, skip, err := cue(now, ok)452 if err != nil {453 return err454 }455 at, cued = now, out.video.units456 ok = decode(ctx, ffmpeg, label, it, skip, out)457 }458 finish()459 _ = captioned.Flush()460 if parent.Err() == nil && encErr.Len() > 0 {461 return fmt.Errorf("ffmpeg: %s", strings.TrimSpace(encErr.String()))462 }463 return nil464}465466// output is where decoders send a stream's picture, sound and captions.467type output struct {468 video, audio *feed469 captions *timeline470 app bool // for the Airwaves app: no subtitles burned in471}472473type osPipe struct{ r, w *os.File }474475func openPipes(n int) ([]osPipe, error) {476 var out []osPipe477 for range n {478 r, w, err := os.Pipe()479 if err != nil {480 for _, p := range out {481 p.r.Close()482 p.w.Close()483 }484 return nil, err485 }486 out = append(out, osPipe{r, w})487 }488 return out, nil489}490491const relayChunk = 64 << 10492493// frameBytes is the size of a raw frame.494const frameBytes = loopWidth * loopHeight * 3 / 2495496// feed carries raw picture or sound to one of the encoder's pipes through497// up to chunks*64 KB of memory, from one decoder after another, counting498// it in units (frames, or sound samples). If the encoder fails, the feed499// keeps reading, so decoders never block.500type feed struct {501 buf chan []byte502 unit int503 units int64 // from the decoders that have finished504 bytes atomic.Int64 // fed so far505 closed sync.Once506 first func() // called as the first unit arrives, when set507}508509// newFeed starts a feed to dst. With pace set, units go to dst no faster510// than pace allows; otherwise as soon as they arrive.511func newFeed(dst *os.File, chunks, unit int, pace *pacer) *feed {512 f := &feed{buf: make(chan []byte, chunks), unit: unit}513 go func() {514 defer dst.Close()515 failed := false516 var sent int64517 for b := range f.buf {518 if !failed && pace != nil {519 failed = !pace.wait(sent / int64(unit))520 }521 if !failed {522 _, err := dst.Write(b)523 failed = err != nil524 }525 sent += int64(len(b))526 }527 }()528 return f529}530531// pacer holds a feed to real time, but for a lead: unit n goes out lead532// before n/per seconds after the first did. Units that come late go out533// at once, so a stream that falls behind (a slow start, a network534// stall) catches up as fast as its decoder can.535type pacer struct {536 per int64 // units a second537 lead time.Duration538 quit <-chan struct{}539 start time.Time540 timer *time.Timer541}542543// wait waits until unit n is due, and reports false if quit came first.544func (p *pacer) wait(n int64) bool {545 now := time.Now()546 if p.start.IsZero() {547 p.start = now548 }549 due := p.start.Add(time.Duration(n/p.per)*time.Second + time.Duration(n%p.per)*time.Second/time.Duration(p.per) - p.lead)550 d := due.Sub(now)551 if d <= 0 {552 return true553 }554 if p.timer == nil {555 p.timer = time.NewTimer(d)556 } else {557 p.timer.Reset(d)558 }559 select {560 case <-p.timer.C:561 return true562 case <-p.quit:563 return false564 }565}566567// source returns a pipe for a decoder to write to, and done, which waits568// for what the decoder wrote to pass and returns how many units it was. A569// decoder cut off mid-frame has its frame made up, so the next decoder's570// frames start whole. Close the pipe once the decoder has it. process, when571// set, sees what passes, whole units at a time, and may change it in place.572// hold, when set, holds what the decoder writes back until it's closed.573func (f *feed) source(process func([]byte), hold <-chan struct{}) (w *os.File, done func() int64, err error) {574 r, w, err := os.Pipe()575 if err != nil {576 return nil, nil, err577 }578 result := make(chan int64, 1)579 go func() {580 defer r.Close()581 var n int64582 for {583 b := make([]byte, relayChunk)584 k, err := io.ReadFull(r, b)585 if k > 0 {586 if n == 0 && f.first != nil {587 f.first()588 }589 if n == 0 && hold != nil {590 <-hold591 }592 if process != nil {593 process(b[:k-k%f.unit])594 }595 f.buf <- b[:k]596 n += int64(k)597 f.bytes.Add(int64(k))598 }599 if err != nil {600 break601 }602 }603 if rest := int(n % int64(f.unit)); rest > 0 {604 pad := filler(f.unit, rest)605 f.buf <- pad606 n += int64(len(pad))607 f.bytes.Add(int64(len(pad)))608 }609 result <- n / int64(f.unit)610 }()611 var units int64612 once := sync.Once{}613 return w, func() int64 {614 once.Do(func() {615 units = <-result616 f.units += units617 })618 return units619 }, nil620}621622// filler completes a unit begun with have bytes: black for a frame (luma623// then chroma), silence for sound.624func filler(unit, have int) []byte {625 pad := make([]byte, unit-have)626 if unit == frameBytes {627 for i := range pad {628 if have+i < loopWidth*loopHeight {629 pad[i] = 0x10630 } else {631 pad[i] = 0x80632 }633 }634 }635 return pad636}637638// fed is how many units have gone into the feed.639func (f *feed) fed() int64 { return f.bytes.Load() / int64(f.unit) }640641// close ends the feed, once no decoder is writing.642func (f *feed) close() { f.closed.Do(func() { close(f.buf) }) }643644// encodeArgs reads raw 720p30 picture on fd 3 and 48 kHz stereo on fd 4645// and writes H.264 and AAC in MPEG-TS to stdout.646func encodeArgs() []string {647 return []string{648 "-hide_banner", "-loglevel", "error", "-nostdin",649 "-f", "rawvideo", "-pix_fmt", "yuv420p",650 "-video_size", fmt.Sprintf("%dx%d", loopWidth, loopHeight), "-framerate", strconv.Itoa(loopFPS), "-i", "pipe:3",651 "-f", "s16le", "-ar", strconv.Itoa(sampleRate), "-ac", "2", "-i", "pipe:4",652 "-map", "0:v", "-map", "1:a",653 "-c:v", "libx264", "-preset", "veryfast", "-g", "60",654 "-b:v", "3M", "-maxrate", "4M", "-bufsize", "6M",655 "-c:a", "aac", "-b:a", "160k",656 "-f", "mpegts", "pipe:1",657 }658}659660// decodeArgs plays one video from skip frames in, as raw661// picture (fd 3) and sound (fd 4): letterboxed to 720p at 30 fps, and padded662// or trimmed to exactly the frames the schedule gives it, so picture and663// sound stay in step from one video to the next.664func decodeArgs(it item, skip int64) []string {665 input, audio := it.inputs(skip)666 return decodeInputs(input, audio, it.Frames-skip, "")667}668669// inputs are ffmpeg's inputs for an item without Open, from skip frames670// in, and the sound's stream specifier, empty for silence.671func (it item) inputs(skip int64) (input []string, audio string) {672 if it.Input != nil {673 input = it.Input(frameDuration(skip))674 } else {675 if skip > 0 {676 input = append(input, "-ss", strconv.FormatFloat(float64(skip)/loopFPS, 'f', 3, 64))677 }678 input = append(input, "-i", it.Path)679 }680 if !it.Audio {681 return input, ""682 }683 return input, cmp.Or(it.AudioStream, "0:a:0")684}685686// decodeInputs plays ffmpeg inputs (picture from the first) as decodeArgs687// does, for frames. audio is the sound's stream specifier, empty for688// silence. subtitles, when set, is an SRT file to burn into the picture,689// timed from where the inputs start.690func decodeInputs(input []string, audio string, frames int64, subtitles string) []string {691 args := append([]string{"-hide_banner", "-loglevel", "error", "-nostdin"}, input...)692 if audio == "" {693 args = append(args, "-f", "lavfi", "-i", fmt.Sprintf("anullsrc=r=%d:cl=stereo", sampleRate))694 audio = fmt.Sprintf("%d:a", countInputs(input))695 }696 burn := ""697 if subtitles != "" {698 burn = "subtitles=filename=" + subtitles + ":force_style='FontName=DejaVu Sans,Outline=1.5,Shadow=0.5,MarginV=20',"699 }700 graph := fmt.Sprintf("[0:v:0]scale=%[1]d:%[2]d:force_original_aspect_ratio=decrease:force_divisible_by=2,"+701 "pad=%[1]d:%[2]d:-1:-1,setsar=1,%[8]sfps=%[3]d,format=yuv420p,tpad=stop=-1:stop_mode=clone,trim=end_frame=%[4]d[v];"+702 "[%[5]s]aresample=%[6]d,aformat=sample_fmts=s16:channel_layouts=stereo,apad,atrim=end_sample=%[7]d[a]",703 loopWidth, loopHeight, loopFPS, frames, audio, sampleRate, frames*sampleRate/loopFPS, burn)704 return append(args, "-filter_complex", graph,705 "-map", "[v]", "-f", "rawvideo", "pipe:3",706 "-map", "[a]", "-f", "s16le", "pipe:4")707}708709// countInputs counts the "-i" in ffmpeg options.710func countInputs(args []string) int {711 n := 0712 for i := 0; i < len(args); i++ {713 if args[i] == "-i" {714 n++715 i++716 }717 }718 return n719}720721// standBy captions the slate shown when a video can't be opened.722const standBy = "Please stand by"723724// decode plays one item into the stream and reports whether it played725// without error. An item whose Open fails is replaced by a slate for the726// rest of its time.727func decode(ctx context.Context, ffmpeg, label string, it item, skip int64, out *output) bool {728 name := it.Title729 var input []string730 var sound string731 switch {732 case it.Slate:733 case it.Open != nil:734 // Opened sources are logged by Path, an ID: their addresses are735 // long and may carry credentials.736 name = it.Path737 o, err := it.Open(ctx, frameDuration(skip))738 if err != nil {739 if ctx.Err() != nil {740 return true741 }742 log.Printf("%s: %s: %v", label, name, err)743 phase.Step(ctx, "open failed")744 // All of it: the time spent opening played nothing, so the745 // stream's clock didn't move (see playCues).746 if left := it.Frames - skip; left > 0 {747 decode(ctx, ffmpeg, label, item{Slate: true, Title: standBy, Frames: left}, 0, out)748 }749 return false750 }751 phase.Step(ctx, "opened")752 input, sound = o.Args, o.Audio753 it.AudioLang, it.Captions = o.AudioLang, o.Captions754 default:755 input, sound = it.inputs(skip)756 }757 // The picture waits for the video's captions to be in place before it758 // goes to the encoder (see subtitle), so the video opens meanwhile.759 hold := make(chan struct{})760 release := sync.OnceFunc(func() { close(hold) })761 defer release()762 video, videoDone, err := out.video.source(nil, hold)763 if err != nil {764 log.Printf("%s: %v", label, err)765 return false766 }767 // The sound is leveled on its way to the encoder: metered and turned768 // up or down, adding no delay.769 var level func([]byte)770 var lev *leveler771 if it.Level != nil && sound != "" && !it.Slate {772 lev = newLeveler(it.Level)773 level = lev.process774 }775 audio, audioDone, err := out.audio.source(level, nil)776 if err != nil {777 release()778 video.Close()779 videoDone()780 log.Printf("%s: %v", label, err)781 return false782 }783 span := out.captions.begin(out.video.units)784 var args []string785 if it.Slate {786 release()787 args = slateArgs(ffmpeg, it.Title, it.Frames-skip)788 } else {789 subs, settled := subtitle(ctx, ffmpeg, it, skip, span, out)790 if subs != "" {791 defer os.Remove(subs)792 }793 if settled == nil {794 release()795 } else {796 go func() {797 <-settled798 release()799 }()800 }801 args = decodeInputs(input, sound, it.Frames-skip, subs)802 }803 dec := exec.CommandContext(ctx, ffmpeg, args...)804 dec.ExtraFiles = []*os.File{video, audio}805 var decErr bytes.Buffer806 dec.Stderr = &decErr807 err = dec.Run()808 // The decoder has its own copies of the pipes; with these closed too,809 // the feeds see the end of its output.810 video.Close()811 audio.Close()812 videoDone()813 audioDone()814 if lev != nil {815 lev.finish() // what it measured is remembered for its next airing816 }817 if err != nil && ctx.Err() == nil {818 msg := strings.TrimSpace(decErr.String())819 if it.Open != nil {820 msg = urls.ReplaceAllString(msg, "<url>")821 }822 log.Printf("%s: %s: %v: %s", label, name, err, msg)823 return false824 }825 return true826}827828var urls = regexp.MustCompile(`https?://\S+`)829830// slateArgs plays a quiet dark picture for frames, with caption when831// ffmpeg can draw text.832func slateArgs(ffmpeg, caption string, frames int64) []string {833 src := fmt.Sprintf("color=c=0x101318:s=%dx%d:r=%d", loopWidth, loopHeight, loopFPS)834 if text := slateText(caption); text != "" && canDrawText(ffmpeg) {835 // Longer captions, like "Back at 8:00 PM with ...", get smaller to fit.836 size := 40837 if n := utf8.RuneCountInString(caption); n > 50 {838 size = max(24, 40*50/n)839 }840 src += fmt.Sprintf(",drawtext=text='%s':fontcolor=0x8c939f:fontsize=%d:x=(w-tw)/2:y=(h-th)/2", text, size)841 }842 return decodeInputs([]string{"-f", "lavfi", "-i", src}, "", frames, "")843}844845// slateText is a caption as drawtext's quoted text: letters, digits and846// a little punctuation, its colons escaped.847func slateText(caption string) string {848 var b strings.Builder849 for _, r := range caption {850 switch {851 case r == ':':852 b.WriteString(`\:`)853 case unicode.IsLetter(r) || unicode.IsDigit(r) || strings.ContainsRune(" .!,", r):854 b.WriteRune(r)855 }856 }857 return b.String()858}859860var drawText sync.Map // ffmpeg path: bool861862// canDrawText reports whether ffmpeg has drawtext and a font for it.863func canDrawText(ffmpeg string) bool {864 if ok, found := drawText.Load(ffmpeg); found {865 return ok.(bool)866 }867 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)868 defer cancel()869 err := exec.CommandContext(ctx, ffmpeg, "-hide_banner", "-loglevel", "error", "-nostdin",870 "-f", "lavfi", "-i", "color=s=64x64:d=0.1,drawtext=text=A", "-frames:v", "1", "-f", "null", "-").Run()871 drawText.Store(ffmpeg, err == nil)872 return err == nil873}