1package service23import (4 "cmp"5 "context"6 "errors"7 "fmt"8 "io"9 "log"10 "slices"11 "strings"12 "time"1314 "airwaves/internal/fcc"15 "airwaves/internal/hdhr"16 "airwaves/internal/lineup"17 "airwaves/internal/signal"18 "airwaves/internal/tuner"19)2021// measureTiming holds the measuring's pauses; tests shorten them.22type measureTiming struct {23 // sample is how often the tuners are read while in use.24 sample time.Duration25 // acquire is how long the sweep gives a tuner to lock, settle how26 // long it waits after a lock before reading, and interval the time27 // between its readings, readings of them (one without a lock).28 acquire, settle, interval time.Duration29 readings int30 // idleWait is how long the sweep waits for a free tuner before it31 // gives up.32 idleWait time.Duration33 // poll is how often the tuner is read while a channel starts or the34 // sweep waits, and noLock how long a channel may take to start.35 poll, noLock time.Duration36}3738var defaultTiming = measureTiming{39 sample: 10 * time.Second,40 // An HDHomeRun locks to a good signal in a second or two, to a weak41 // one in three or four; its quality reads 0 for its first second on a42 // multiplex. So about 5 s for an RF channel with a lock, 7 s for one43 // without.44 acquire: 6 * time.Second, settle: time.Second, interval: time.Second, readings: 3,45 idleWait: 30 * time.Second,46 poll: 500 * time.Millisecond, noLock: 10 * time.Second,47}4849// SweepStatus is how Measure now is going, or went.50type SweepStatus struct {51 Running bool `json:"running"`52 StartedAt *time.Time `json:"startedAt"`53 FinishedAt *time.Time `json:"finishedAt"`54 // Total is how many RF channels it measures, Done how many it has.55 // The first Found of them are those the tuner's scan found channels56 // on, measured first; the rest are those a licensed station nearby57 // uses.58 Total int `json:"total"`59 Found int `json:"found"`60 Done int `json:"done"`61 // RF is the RF channel being measured now.62 RF int `json:"rf,omitempty"`63 // Note says why it stopped early or what it left out.64 Note string `json:"note,omitempty"`65}6667// ---------- tuners in use ----------6869// sampleTuners records a reading of every RF channel a tuner has been on70// since the last call, for whoever tuned it, with the errors in what71// arrived meanwhile. A tuning just started isn't read yet, as it may not72// have locked; the sweep reads its own.73func (s *Service) sampleTuners(ctx context.Context) {74 tuners, err := s.tuners.Tuners(ctx)75 if err != nil {76 return77 }78 s.mu.Lock()79 prev, prevAt := s.lastSample, s.lastSampleAt80 sweeping := s.sweepFreq81 s.mu.Unlock()82 now := time.Now()83 seen := map[int]tuner.State{}84 for _, t := range tuners {85 if t.FrequencyHz == 0 || t.Elsewhere {86 continue87 }88 seen[t.Number] = t89 was, stable := prev[t.Number]90 if !stable || was.FrequencyHz != t.FrequencyHz || !was.Since.Equal(t.Since) || t.FrequencyHz == sweeping {91 continue92 }93 r := signal.FromTuner(t, now, signal.Active)94 r.SetErrors(was, t, now.Sub(prevAt).Seconds())95 s.signals.Record(t.FrequencyHz, signal.RF(t.FrequencyHz), r)96 }97 s.mu.Lock()98 s.lastSample, s.lastSampleAt = seen, now99 s.mu.Unlock()100}101102// sample reads the tuners every timing.sample until ctx ends, saving the103// measurements every minute.104func (s *Service) sample(ctx context.Context) {105 tick := time.NewTicker(s.timing.sample)106 defer tick.Stop()107 for {108 select {109 case <-ctx.Done():110 if err := s.signals.Save(); err != nil {111 log.Printf("signal measurements: %v", err)112 }113 return114 case <-tick.C:115 }116 s.sampleTuners(ctx)117 if err := s.signals.SaveEvery(time.Minute); err != nil {118 log.Printf("signal measurements: %v", err)119 }120 }121}122123// ---------- Measure now ----------124125// errTakenOver ends a measurement whose tuner someone needed.126var errTakenOver = errors.New("a viewer or a recording needed the tuner")127128// Measure starts a sweep that measures every RF channel worth measuring,129// one at a time, on a free tuner: those the scan found channels on, and130// those a licensed station nearby uses. It never takes a tuner that's in131// use, and stops as soon as a viewer or a recording needs the one it132// uses. It returns at once; Signal reports progress and the readings.133func (s *Service) Measure(ctx context.Context) (*SweepStatus, error) {134 if s.opt.NoAntenna {135 return nil, ErrNoAntenna136 }137 if s.device() == nil {138 return nil, ErrNoTuner139 }140 s.mu.Lock()141 defer s.mu.Unlock()142 if s.sweep.Running {143 st := s.sweep144 return &st, nil145 }146 now := time.Now().UTC().Truncate(time.Second)147 s.sweep = SweepStatus{Running: true, StartedAt: &now}148 st := s.sweep149 go s.runSweep(s.ctx)150 return &st, nil151}152153func (s *Service) sweepStatus() SweepStatus {154 s.mu.Lock()155 defer s.mu.Unlock()156 return s.sweep157}158159func (s *Service) setSweep(f func(*SweepStatus)) {160 s.mu.Lock()161 f(&s.sweep)162 s.mu.Unlock()163}164165// sweepTargets lists the RF channels worth measuring, by frequency: first166// those the tuner's scan found channels on, then those a licensed station167// within the report's radius broadcasts on (nearest station first); found168// is how many are of the first kind.169func (s *Service) sweepTargets(ctx context.Context) (freqs []int64, found int, err error) {170 s.mu.Lock()171 s.lineupAt = time.Time{}172 rep := s.report173 s.mu.Unlock()174 l, err := s.tunerLineup(ctx)175 if err != nil {176 return nil, 0, err177 }178 have := map[int64]bool{}179 for _, c := range l {180 if !have[c.FrequencyHz] {181 have[c.FrequencyHz] = true182 freqs = append(freqs, c.FrequencyHz)183 }184 }185 slices.Sort(freqs)186 found = len(freqs)187 nearest := map[int64]float64{} // km to the nearest licensed station, by frequency188 if rep != nil {189 for _, st := range rep.Stations {190 f := signal.Frequency(st.RFChannel)191 if d, ok := nearest[f]; f > 0 && !have[f] && (!ok || st.DistanceKm < d) {192 nearest[f] = st.DistanceKm193 }194 }195 }196 licensed := slices.Collect(func(yield func(int64) bool) {197 for f := range nearest {198 if !yield(f) {199 return200 }201 }202 })203 slices.SortFunc(licensed, func(a, b int64) int { return cmp.Or(cmp.Compare(nearest[a], nearest[b]), cmp.Compare(a, b)) })204 return append(freqs, licensed...), found, nil205}206207func (s *Service) runSweep(ctx context.Context) {208 note := ""209 defer func() {210 if err := s.signals.Save(); err != nil {211 log.Printf("signal measurements: %v", err)212 }213 now := time.Now().UTC().Truncate(time.Second)214 s.mu.Lock()215 s.sweepFreq = 0216 s.sweep.Running, s.sweep.FinishedAt, s.sweep.RF, s.sweep.Note = false, &now, 0, note217 s.mu.Unlock()218 log.Printf("measuring done: %+v", s.sweepStatus())219 }()220 freqs, found, err := s.sweepTargets(ctx)221 if err != nil {222 note = "the tuner's lineup can't be read: " + err.Error()223 return224 }225 s.setSweep(func(st *SweepStatus) { st.Total, st.Found = len(freqs), found })226 log.Printf("measuring %d RF channels, the %d found first", len(freqs), found)227 for _, f := range freqs {228 if ctx.Err() != nil {229 note = "stopped: the server is shutting down"230 return231 }232 rf := signal.RF(f)233 s.setSweep(func(st *SweepStatus) { st.RF = rf })234 err := s.measureRF(ctx, f)235 var busy *tuner.BusyError236 switch {237 case errors.As(err, &busy):238 note = fmt.Sprintf("stopped at RF %d: no tuner was free (they're left to viewers and recordings)", rf)239 return240 case errors.Is(err, errTakenOver):241 note = fmt.Sprintf("stopped at RF %d: %v", rf, err)242 return243 case err != nil:244 log.Printf("measure RF %d: %v", rf, err)245 }246 s.setSweep(func(st *SweepStatus) { st.Done++ })247 }248}249250// measureRF records timing.readings readings of the RF channel at freq:251// with a lock, or without one, with whatever strength the tuner measures.252// A tuner already on it is read where it is; otherwise it waits up to253// timing.idleWait for a free one, which any viewer or recording takes254// back.255func (s *Service) measureRF(ctx context.Context, freq int64) error {256 rf := signal.RF(freq)257 if t, ok := s.tunerOn(ctx, freq); ok {258 if r := signal.FromTuner(t, time.Now(), signal.Sweep); r.Lock {259 s.signals.Record(freq, rf, r)260 }261 return nil262 }263 var sub *tuner.Sub264 deadline := time.Now().Add(s.timing.idleWait)265 for {266 var err error267 sub, err = s.tuners.Open(ctx, tuner.Request{FrequencyHz: freq, Priority: tuner.Measuring, Who: fmt.Sprintf("RF %d", rf)})268 var busy *tuner.BusyError269 if err == nil || !errors.As(err, &busy) || time.Now().After(deadline) {270 if err != nil {271 return err272 }273 break274 }275 if err := sleep(ctx, s.timing.poll); err != nil {276 return err277 }278 }279 defer sub.Close()280 s.mu.Lock()281 s.sweepFreq = freq282 s.mu.Unlock()283284 // Give the tuner time to lock: a weak signal can take seconds. What it285 // measures meanwhile, past its first second on the RF channel, is the286 // reading when it doesn't lock.287 read := func() (tuner.State, error) {288 select {289 case <-sub.Done():290 return tuner.State{}, errTakenOver291 default:292 }293 tuners, err := s.tuners.Tuners(ctx)294 if err != nil {295 return tuner.State{}, err296 }297 return tuners[sub.Tuner()], nil298 }299 var tried *signal.Reading300 start := time.Now()301 for !sub.Locked() && time.Since(start) < s.timing.acquire {302 if err := sleep(ctx, s.timing.poll); err != nil {303 return err304 }305 t, err := read()306 if err != nil {307 return err308 }309 if r := signal.FromTuner(t, time.Now(), signal.Sweep); time.Since(start) >= s.timing.settle {310 tried = &r311 }312 }313 if !sub.Locked() {314 if tried == nil {315 return fmt.Errorf("RF %d: no reading", rf)316 }317 s.signals.Record(freq, rf, *tried)318 return nil319 }320 if err := sleep(ctx, s.timing.settle); err != nil {321 return err322 }323 var prev *tuner.State324 var prevAt time.Time325 for i := range s.timing.readings {326 if i > 0 {327 if err := sleep(ctx, s.timing.interval); err != nil {328 return err329 }330 }331 t, err := read()332 if err != nil {333 return err334 }335 now := time.Now()336 r := signal.FromTuner(t, now, signal.Sweep)337 if prev != nil {338 r.SetErrors(*prev, t, now.Sub(prevAt).Seconds())339 }340 prev, prevAt = &t, now341 s.signals.Record(freq, rf, r)342 }343 return nil344}345346// tunerOn finds a tuner of ours already on the RF channel at freq.347func (s *Service) tunerOn(ctx context.Context, freq int64) (tuner.State, bool) {348 tuners, err := s.tuners.Tuners(ctx)349 if err != nil {350 return tuner.State{}, false351 }352 for _, t := range tuners {353 if t.FrequencyHz == freq && !t.Elsewhere {354 return t, true355 }356 }357 return tuner.State{}, false358}359360func sleep(ctx context.Context, d time.Duration) error {361 t := time.NewTimer(d)362 defer t.Stop()363 select {364 case <-ctx.Done():365 return ctx.Err()366 case <-t.C:367 return nil368 }369}370371// ---------- tuning ----------372373// NoSignalError is a channel that didn't start because its RF channel had374// no lock.375type NoSignalError struct {376 Channel string377 RF int378 Label string // "KTVD 20.x"379 Reading *signal.Reading380}381382func (e *NoSignalError) Error() string {383 msg := fmt.Sprintf("No signal on RF %d", e.RF)384 if e.Label != "" {385 msg += " (" + e.Label + ")"386 }387 msg += " right now"388 if r := e.Reading; r != nil {389 var parts []string390 if r.StrengthPct != nil {391 parts = append(parts, fmt.Sprintf("strength %.0f%%", *r.StrengthPct))392 }393 if r.QualityPct != nil {394 parts = append(parts, fmt.Sprintf("quality %.0f%%", *r.QualityPct))395 }396 if len(parts) > 0 {397 msg += ": the tuner measured " + strings.Join(parts, ", ") + " and couldn't lock"398 }399 }400 return msg401}402403// rfLabel names what a channel's RF channel carries: "KTVD 20.x", or404// "KUSA 9.x on KTVD" when another station's transmitter carries it.405func rfLabel(c lineup.TunerChannel) string {406 call := c.BaseCall407 if call == "" {408 call = fcc.BaseCall(c.Transmitter)409 }410 label := strings.TrimSpace(fmt.Sprintf("%s %d.x", call, c.Major))411 if c.Via != "" {412 label += " on " + fcc.BaseCall(c.Via)413 }414 return label415}416417// tuneAntenna starts an antenna channel for client: a tuner (client's own418// last one, when every tuner is busy), its program, and start, which419// reads it. A channel that won't lock fails with what was measured.420func (s *Service) tuneAntenna(ctx context.Context, client, number string, start func(context.Context, tuner.Input) error) error {421 freq, program, err := s.tuning(ctx, number)422 if err != nil {423 return err424 }425 sub, err := s.tuners.Open(ctx, tuner.Request{FrequencyHz: freq, Program: program, Priority: tuner.Watching, Who: number, Owner: client})426 var busy *tuner.BusyError427 if errors.As(err, &busy) {428 return fmt.Errorf("every tuner is in use right now (%s), so %s can't be tuned", strings.Join(busy.Users, "; "), number)429 }430 if err != nil {431 return err432 }433 wait, cancel := context.WithTimeout(ctx, s.timing.noLock)434 p, err := sub.Wait(wait)435 cancel()436 if err != nil {437 defer sub.Close()438 return s.tuneFailed(ctx, sub, number, freq, err)439 }440 audio, note := tuner.AudioTracks(p)441 in := tuner.Input{442 Args: []string{"-fflags", "+genpts", "-f", "mpegts", "-i", "pipe:0"},443 Video: "0:v:0", Audio: audio, Note: note, Broadcast: true,444 Source: func(ctx context.Context, w io.Writer) error {445 defer sub.Close()446 return sub.Copy(ctx, w)447 },448 }449 if err := start(ctx, in); err != nil {450 sub.Close()451 return err452 }453 return nil454}455456// tuneFailed says why a channel didn't start, with what the tuner measured457// of its RF channel.458func (s *Service) tuneFailed(ctx context.Context, sub *tuner.Sub, number string, freq int64, err error) error {459 rf := signal.RF(freq)460 var reading *signal.Reading461 if tuners, terr := s.tuners.Tuners(ctx); terr == nil && sub.Tuner() < len(tuners) {462 if t := tuners[sub.Tuner()]; t.FrequencyHz == freq && t.Device.StrengthPct != nil {463 r := signal.FromTuner(t, time.Now(), signal.Active)464 reading = &r465 s.signals.Record(freq, rf, r)466 }467 }468 var he *hdhr.Error469 switch {470 case errors.As(err, &he) && he.Code == 803:471 return errors.New("the tuner is scanning for channels; try again in a few minutes")472 case errors.As(err, &he) && he.Busy():473 return fmt.Errorf("the tuner is busy with something else, so %s can't be tuned", number)474 case errors.Is(err, tuner.ErrNoProgram):475 return fmt.Errorf("RF %d no longer carries %s: the tuner needs to scan for channels again", rf, number)476 case errors.Is(err, context.DeadlineExceeded) && ctx.Err() == nil && !sub.Locked():477 label := ""478 s.mu.Lock()479 rep, g := s.report, s.guide480 s.mu.Unlock()481 if chans, _, merr := s.matched(ctx, rep, g); merr == nil {482 for _, c := range chans {483 if c.Number == number {484 label = rfLabel(c)485 }486 }487 }488 return &NoSignalError{Channel: number, RF: rf, Label: label, Reading: reading}489 }490 return fmt.Errorf("%s (RF %d) didn't start: %w", number, rf, err)491}