cat / airwaves
clone

a desktop tv and program guide for over-the-air broadcasts: what's on air, what your tuner gets, live tv, recordings

Fix the signal sweep on real Tvheadend, and show plain call signs

FelineStateMachine committed 9f05269
Tvheadend lists idle tuners in its input status too, so a tuner now
counts as busy only while something subscribes to it. A reading's lock
comes from the subscription state, and on an HDHomeRun a quality of 0
means no lock. The sweep measures the RF channels the scan found first,
waits up to 6 s for a lock, and records "no lock" with the strength it
saw otherwise. Readings add transport errors per second, and channels a
summary of their latest readings. Channels show the station's call sign
("KWGN"), with the listings' ("KWGNDT") kept for guide matching.
commit
9f05269fb69e064a61e38abcf15e117843ffdc73
parent
cae5de7e3791442145b7fb1cb4f3a671a4292084
author
FelineStateMachine <226808055+FelineStateMachine@users.noreply.github.com>
kind
fix risk score 2.2 of 3, Jev's estimate
15 files changed, +800 -225
Minternal/lineup/tuner.go43+++-
Minternal/lineup/tuner_test.go32++-
Minternal/service/antenna.go36++-
Minternal/service/antenna_test.go62++++-
Minternal/service/hdhr.go4+-
Minternal/service/measure.go333++++++++++++++++++++----------
Minternal/service/service.go15+-
Minternal/signal/signal.go53+++++-
Minternal/signal/signal_test.go88++++++++-
Minternal/tvh/status.go38+++-
Minternal/tvh/status_test.go40++++-
Ainternal/tvh/tvhtest/testdata/status_inputs_idle.json43++++
Ainternal/tvh/tvhtest/testdata/status_inputs_measuring.json63++++++
Ainternal/tvh/tvhtest/testdata/status_subscriptions_measuring.json51+++++
Minternal/tvh/tvhtest/tvhtest.go124++++++++---
M internal/lineup/tuner.go +30 -13
@@ -3,6 +3,7 @@ package lineup
33 import (44 "cmp"55 "fmt"6+ "regexp"67 "slices"78 "strings"89
@@ -28,16 +29,22 @@ type TunerChannel struct {
2829 Number string `json:"number"`2930 Major int `json:"major"`3031 Minor int `json:"minor"`31- // Name is the broadcast's own name for the channel.32+ // Name is the broadcast's own name for the channel (its PSIP short33+ // name): "KTVD-HD", "H & I", "StartTV".3234 Name string `json:"name"`33- // GuideID, Network and Logo come from the listings, when they have34- // the channel. CallSign is theirs too, or else the call sign of the35- // station whose virtual channel it is, or else Name.36- GuideID string `json:"guideId,omitempty"`35+ // CallSign is the station's call sign to show, "KWGN", the same for36+ // all its subchannels: from the listings, else the station whose37+ // virtual channel it is, else the broadcast's name. BaseCall is the38+ // same call sign as a key ("" when the name isn't a call sign).3739 CallSign string `json:"callSign"`3840 BaseCall string `json:"baseCall"`39- Network string `json:"network"`40- Logo string `json:"logo,omitempty"`41+ // GuideID, GuideCallSign ("KWGNDT", "KCNCDT2"), Network ("CW",42+ // "Start TV") and Logo come from the listings, when they have the43+ // channel.44+ GuideID string `json:"guideId,omitempty"`45+ GuideCallSign string `json:"guideCallSign,omitempty"`46+ Network string `json:"network"`47+ Logo string `json:"logo,omitempty"`4148 // RF is the RF channel the tuner receives it on; 0 when unknown.4249 RF int `json:"rf"`4350 // FacilityID and Transmitter are the licensed transmitter on RF that
@@ -63,11 +70,20 @@ type NextGen struct {
6370 }6471 6572 // nameBase is a broadcast name's call sign base, for names that are call66-// signs: "KTVD-HD" and "KRMADT1" give "KTVD" and "KRMA".73+// signs: "KTVD-HD" and "KRMADT1" give "KTVD" and "KRMA"; "" for names that74+// aren't ("H & I", "StartTV").6775 func nameBase(name string) string {68- return guide.BaseCall(fcc.BaseCall(strings.TrimSpace(name)))76+ b := guide.BaseCall(fcc.BaseCall(strings.TrimSpace(name)))77+ if !callSign.MatchString(b) {78+ return ""79+ }80+ return b6981 }7082 83+// callSign matches a US broadcast call sign's base: "KWGN", "WGN", and a84+// low-power station's "K14QW".85+var callSign = regexp.MustCompile(`^(?:[KW][A-Z]{2,3}|[KW]\d{2}[A-Z]{2})$`)86+7187 // Match describes the tuner's channels from rep's transmitters and ATSC 3.07288 // hosts and from the listings g (nil without them), in channel order.7389 func Match(rep *Report, g *guide.Guide, found []Scanned) []TunerChannel {
@@ -139,14 +155,15 @@ func Match(rep *Report, g *guide.Guide, found []Scanned) []TunerChannel {
139155 ch.FacilityID, ch.Transmitter = t.FacilityID, t.CallSign140156 }141157 if gc, ok := pickListing(listed[f.Number], nameBase(f.Name), t); ok {142- ch.GuideID, ch.CallSign, ch.BaseCall, ch.Network, ch.Logo = gc.ID, gc.CallSign, guide.BaseCall(gc.CallSign), gc.Network, gc.Logo158+ ch.GuideID, ch.GuideCallSign, ch.BaseCall, ch.Network, ch.Logo = gc.ID, gc.CallSign, guide.BaseCall(gc.CallSign), gc.Network, gc.Logo143159 } else if t != nil && t.VirtualChannel == major {144- ch.CallSign, ch.BaseCall = t.CallSign, t.BaseCall160+ ch.BaseCall = t.BaseCall145161 } else if s := byVirtual[major]; s != nil {146- ch.CallSign, ch.BaseCall = s.CallSign, s.BaseCall162+ ch.BaseCall = s.BaseCall147163 } else {148- ch.CallSign, ch.BaseCall = ch.Name, nameBase(f.Name)164+ ch.BaseCall = nameBase(f.Name)149165 }166+ ch.CallSign = cmp.Or(ch.BaseCall, ch.Name)150167 if t != nil && t.BaseCall != ch.BaseCall && t.VirtualChannel != major {151168 ch.Via = t.CallSign152169 }
M internal/lineup/tuner_test.go +20 -12
@@ -37,28 +37,35 @@ func TestMatch(t *testing.T) {
3737 if want := []string{"2.1", "6.1", "9.4", "9.7", "14.1", "20.1", "20.2", "31.1", "88.1"}; !slices.Equal(numbers, want) {3838 t.Fatalf("channels %v, want %v", numbers, want)3939 }40- check := func(number, call, base, tx, via, guideID string, rf int) {40+ // The call sign shown is the station's, the same for its subchannels;41+ // the listings' own ("KUSADT4") is kept apart.42+ check := func(number, call, guideCall, tx, via, guideID string, rf int) {4143 t.Helper()4244 c := by[number]43- if c.CallSign != call || c.BaseCall != base || c.Transmitter != tx || c.Via != via || c.GuideID != guideID || c.RF != rf {45+ if c.CallSign != call || c.BaseCall != call || c.GuideCallSign != guideCall || c.Transmitter != tx || c.Via != via || c.GuideID != guideID || c.RF != rf {4446 t.Errorf("%s = %+v", number, c)4547 }4648 }4749 // From the listings, on its own transmitter (not the nearer4850 // low-power one on the same RF channel).49- check("20.1", "KTVDDT", "KTVD", "KTVD", "", "g20", 31)51+ check("20.1", "KTVD", "KTVDDT", "KTVD", "", "g20", 31)5052 // Not listed: the transmitter's own virtual channel.51- check("20.2", "KTVD", "KTVD", "KTVD", "", "", 31)53+ check("20.2", "KTVD", "", "KTVD", "", "", 31)5254 // KUSA's subchannels ride KTVD's transmitter.53- check("9.4", "KUSADT4", "KUSA", "KTVD", "KTVD", "g94", 31)54- check("9.7", "KUSA", "KUSA", "KTVD", "KTVD", "", 31)55+ check("9.4", "KUSA", "KUSADT4", "KTVD", "KTVD", "g94", 31)56+ check("9.7", "KUSA", "", "KTVD", "KTVD", "", 31)5557 // KWGN's ATSC 1.0 signal is on KDVR's transmitter; it broadcasts in5658 // ATSC 3.0 itself.57- check("2.1", "KWGNDT", "KWGN", "KDVR", "KDVR", "g2", 36)58- check("6.1", "KRMADT", "KRMA", "KRMA-TV", "", "g6", 33)59- check("14.1", "KCECDT", "KCEC", "KCEC", "", "g14", 32)60- // Nothing on record: only what the broadcast says.61- check("88.1", "Mystery", "MYSTERY", "", "", "", 0)59+ check("2.1", "KWGN", "KWGNDT", "KDVR", "KDVR", "g2", 36)60+ check("6.1", "KRMA", "KRMADT", "KRMA-TV", "", "g6", 33)61+ check("14.1", "KCEC", "KCECDT", "KCEC", "", "g14", 32)62+ // Nothing on record, and a name that isn't a call sign: the name.63+ if c := by["88.1"]; c.CallSign != "Mystery" || c.BaseCall != "" || c.Transmitter != "" || c.GuideID != "" {64+ t.Errorf("88.1 = %+v", c)65+ }66+ if c := by["20.2"]; c.Name != "H & I" || c.Network != "" {67+ t.Errorf("20.2 = %+v", c)68+ }6269 if c := by["2.1"]; c.NextGen == nil || c.NextGen.HostCall != "KWGN-TV" || c.NextGen.RF != "34" || c.Network != "CW" || c.Name != "KWGN-DT" {6370 t.Errorf("2.1 = %+v, %+v", c, c.NextGen)6471 }
@@ -71,7 +78,8 @@ func TestMatch(t *testing.T) {
7178 7279 // Without listings or records: every channel still, as broadcast.7380 bare := lineup.Match(&lineup.Report{}, nil, found)74- if len(bare) != len(found) || bare[0].Number != "2.1" || bare[0].CallSign != "KWGN-DT" || bare[0].Transmitter != "" {81+ if len(bare) != len(found) || bare[0].Number != "2.1" || bare[0].CallSign != "KWGN" || bare[0].Transmitter != "" ||82+ bare[6].Number != "20.2" || bare[6].CallSign != "H & I" || bare[6].BaseCall != "" {7583 t.Errorf("bare = %+v", bare)7684 }7785 }
M internal/service/antenna.go +21 -15
@@ -62,9 +62,12 @@ type TunerInfo struct {
6262 // AntennaChannel is a channel the tuner receives.6363 type AntennaChannel struct {6464 lineup.TunerChannel65- // Signal is the latest reading of the channel's multiplex; null when66- // it has never been measured.65+ // Signal is the latest reading of the channel's multiplex, and Recent66+ // the latest stretch of readings (the last sweep, or the last while67+ // it was watched) with lows and means, steadier than one reading on a68+ // marginal signal; null when it has never been measured.6769 Signal *signal.Reading `json:"signal"`70+ Recent *signal.Period `json:"recent"`6871 // Demo marks a generated test pattern (demo mode).6972 Demo bool `json:"demo,omitempty"`7073 }
@@ -111,11 +114,13 @@ type MuxStation struct {
111114 ATSC3 bool `json:"atsc3"`112115 }113116 114-// ChannelSignal is a channel's latest reading, for the app to poll.117+// ChannelSignal is a channel's latest reading and stretch of readings,118+// for the app to poll.115119 type ChannelSignal struct {116120 Number string `json:"number"`117121 RF int `json:"rf"`118122 Signal *signal.Reading `json:"signal"`123+ Recent *signal.Period `json:"recent"`119124 }120125 121126 // SignalReport is the measurements alone, light enough to poll while
@@ -196,8 +201,8 @@ func (s *Service) tunerInfo(ctx context.Context, l *tvh.Lineup) TunerInfo {
196201 if len(fes) > 0 {197202 ti.Standards = []string{"ATSC 1.0"}198203 }199- if busy, ok := s.inputsNow(); ok {200- ti.InUse = busyTuners(busy)204+ if st, ok := s.statusNow(); ok {205+ ti.InUse = st.busy()201206 }202207 switch {203208 case l == nil:
@@ -244,7 +249,7 @@ func (s *Service) antenna(ctx context.Context, rep *lineup.Report, g *guide.Guid
244249 for i, c := range chans {245250 ac := AntennaChannel{TunerChannel: c, Demo: tuned[i].Mux.Network == tvh.DemoNetwork}246251 if m, ok := s.signals.Get(tuned[i].Mux.FrequencyHz); ok {247- ac.Signal = m.Latest252+ ac.Signal, ac.Recent = m.Latest, m.Recent()248253 }249254 a.Channels = append(a.Channels, ac)250255 onMux[tuned[i].Mux.UUID] = append(onMux[tuned[i].Mux.UUID], c.Number)
@@ -285,10 +290,10 @@ func (s *Service) matched(ctx context.Context, rep *lineup.Report, g *guide.Guid
285290 // muxSignals lists the antenna network's multiplexes (or, without286291 // Tvheadend's lineup, those measured before) with what's known of each.287292 func (s *Service) muxSignals(l *tvh.Lineup, rep *lineup.Report, chans map[string][]string) []MuxSignal {288- busy, _ := s.inputsNow()293+ st, _ := s.statusNow()289294 tunedTo := map[string]bool{}290- for _, in := range busy {291- if mux, net, ok := in.Mux(); ok && net == tvh.ATSCNetwork {295+ for _, in := range st.inputs {296+ if mux, net, ok := in.Mux(); ok && net == tvh.ATSCNetwork && in.InUse() {292297 tunedTo[mux] = true293298 }294299 }
@@ -362,7 +367,8 @@ func hostsRF(s string, rf int) bool {
362367 363368 // compatChannels is the tuner's lineup in the report's older Channel364369 // shape, for app versions from before the measured lineup, which list a365-// channel only when its "tier" reads as receivable. Its tier is measured,370+// channel only when its "tier" reads as receivable, and key favorites by371+// the listings' call sign ("KWGNDT"), as that shape had it. Its tier is measured,366372 // not estimated, and the same for every antenna: "good" when the latest367373 // reading of its multiplex (or, before any, Tvheadend's scan) locked,368374 // "weak" when it didn't.
@@ -383,7 +389,7 @@ func compatChannels(a *Antenna) []lineup.Channel {
383389 t = reception.Good384390 }385391 lc := lineup.Channel{386- Number: c.Number, Major: c.Major, Minor: c.Minor, GuideID: c.GuideID, CallSign: c.CallSign, BaseCall: c.BaseCall,392+ Number: c.Number, Major: c.Major, Minor: c.Minor, GuideID: c.GuideID, CallSign: cmp.Or(c.GuideCallSign, c.CallSign), BaseCall: c.BaseCall,387393 Network: c.Network, Logo: c.Logo, FacilityID: c.FacilityID, Via: c.Via,388394 Tier: map[string]reception.Tier{"indoor": t, "attic": t, "rooftop": t},389395 }
@@ -395,12 +401,12 @@ func compatChannels(a *Antenna) []lineup.Channel {
395401 return out396402 }397403 398-// dvrChannels is the tuner's lineup for recording: number, call sign and399-// listings.404+// dvrChannels is the tuner's lineup for recording: number, call sign (the405+// listings', as recording rules have it) and listings.400406 func dvrChannels(chans []lineup.TunerChannel) []lineup.Channel {401407 out := make([]lineup.Channel, 0, len(chans))402408 for _, c := range chans {403- out = append(out, lineup.Channel{Number: c.Number, Major: c.Major, Minor: c.Minor, GuideID: c.GuideID, CallSign: c.CallSign, BaseCall: c.BaseCall})409+ out = append(out, lineup.Channel{Number: c.Number, Major: c.Major, Minor: c.Minor, GuideID: c.GuideID, CallSign: cmp.Or(c.GuideCallSign, c.CallSign), BaseCall: c.BaseCall})404410 }405411 return out406412 }
@@ -445,7 +451,7 @@ func (s *Service) Signal(ctx context.Context) (*SignalReport, error) {
445451 a := s.antenna(ctx, rep, g)446452 out := &SignalReport{Tuner: a.Tuner, Channels: make([]ChannelSignal, 0, len(a.Channels)), Muxes: a.Muxes, Sweep: a.Sweep}447453 for _, c := range a.Channels {448- out.Channels = append(out.Channels, ChannelSignal{Number: c.Number, RF: c.RF, Signal: c.Signal})454+ out.Channels = append(out.Channels, ChannelSignal{Number: c.Number, RF: c.RF, Signal: c.Signal, Recent: c.Recent})449455 }450456 return out, nil451457 }
M internal/service/antenna_test.go +49 -13
@@ -27,7 +27,7 @@ import (
2727 // fastTiming measures in milliseconds rather than seconds.2828 var fastTiming = measureTiming{2929 sample: 20 * time.Millisecond,30- appear: 2 * time.Second, settle: 20 * time.Millisecond, interval: 20 * time.Millisecond, readings: 3,30+ appear: 2 * time.Second, acquire: 300 * time.Millisecond, settle: 20 * time.Millisecond, interval: 20 * time.Millisecond, readings: 3,3131 idleWait: 300 * time.Millisecond,3232 poll: 20 * time.Millisecond, noLock: 200 * time.Millisecond,3333 }
@@ -93,7 +93,7 @@ func TestTunerLineup(t *testing.T) {
9393 t.Errorf("%s listed without the tuner having it", n)9494 }9595 }96- if c := by["20.1"]; c.CallSign != "KTVDDT" || c.Transmitter != "KTVD" || c.RF != 31 || c.Signal != nil || c.Name != "KTVD-HD" || c.Demo {96+ if c := by["20.1"]; c.CallSign != "KTVD" || c.GuideCallSign != "KTVDDT" || c.Transmitter != "KTVD" || c.RF != 31 || c.Signal != nil || c.Name != "KTVD-HD" || c.Demo {9797 t.Errorf("20.1 = %+v", c)9898 }9999 if c := by["2.1"]; c.Via != "KDVR" || c.NextGen == nil || c.NextGen.HostCall != "KWGN-TV" {
@@ -132,6 +132,11 @@ func TestTunerLineup(t *testing.T) {
132132 var compat []string133133 for _, c := range snap.Report.Channels {134134 compat = append(compat, c.Number)135+ // The listings' call sign, which that shape had (and app136+ // versions key favorites by).137+ if c.Number == "20.1" && (c.CallSign != "KTVDDT" || c.BaseCall != "KTVD") {138+ t.Errorf("report 20.1 = %+v", c)139+ }135140 if c.Tier["indoor"] != reception.Good || c.Tier["rooftop"] != reception.Good {136141 t.Errorf("%s tier %v", c.Number, c.Tier)137142 }
@@ -165,7 +170,7 @@ func TestTunerLineup(t *testing.T) {
165170 if info.Tuner.Tuners != 2 || info.Tuner.Model != "hdhomerun_dvr_atsc" || !slices.Equal(info.Tuner.Standards, []string{"ATSC 1.0"}) || info.Tuner.ATSC3 {166171 t.Errorf("info tuner = %+v", info.Tuner)167172 }168- if got := s.AntennaChannels(ctx); len(got) != 38 || got["20.1"] != "KTVDDT" || got["9.7"] != "KUSA" || got["7.1"] != "" {173+ if got := s.AntennaChannels(ctx); len(got) != 38 || got["20.1"] != "KTVD" || got["9.7"] != "KUSA" || got["7.1"] != "" {169174 t.Errorf("antenna channels = %v", got)170175 }171176 }
@@ -292,15 +297,18 @@ func TestDemo(t *testing.T) {
292297 func TestSampleInputs(t *testing.T) {293298 s, fake := newTunerService(t)294299 ctx := t.Context()300+ fake.SetSignal("575MHz", tvhtest.Signal{Lock: true, Strength: 85, Quality: 83, Errors: 4})295301 fake.Busy(tvhtest.Tuner0, "575MHz", 150)296302 s.sampleInputs(ctx)297303 if _, ok := s.signals.Get(575_000_000); ok {298304 t.Fatal("read a tuner just tuned")299305 }306+ time.Sleep(50 * time.Millisecond)300307 s.sampleInputs(ctx)301308 m, ok := s.signals.Get(575_000_000)302- if !ok || m.RF != 31 || m.Latest == nil || !m.Latest.Lock || *m.Latest.StrengthPct != 85 || *m.Latest.QualityPct != 83 || m.Latest.Source != signal.Active {303- t.Fatalf("RF 31 = %+v", m)309+ if !ok || m.RF != 31 || m.Latest == nil || !m.Latest.Lock || *m.Latest.StrengthPct != 85 || *m.Latest.QualityPct != 83 || m.Latest.Source != signal.Active ||310+ m.Latest.ErrorsPerSec == nil || *m.Latest.ErrorsPerSec <= 0 {311+ t.Fatalf("RF 31 = %+v, %+v", m, m.Latest)304312 }305313 306314 sig, err := s.Signal(ctx)
@@ -311,8 +319,11 @@ func TestSampleInputs(t *testing.T) {
311319 t.Errorf("in use: %d", sig.Tuner.InUse)312320 }313321 for _, c := range sig.Channels {314- if (c.RF == 31) != (c.Signal != nil) {315- t.Errorf("%s (RF %d) signal %+v", c.Number, c.RF, c.Signal)322+ if (c.RF == 31) != (c.Signal != nil) || (c.RF == 31) != (c.Recent != nil) {323+ t.Errorf("%s (RF %d) signal %+v, recent %+v", c.Number, c.RF, c.Signal, c.Recent)324+ }325+ if c.Recent != nil && (c.Recent.Samples != 1 || c.Recent.Source != signal.Active || c.Recent.QualityPct.Avg != 83) {326+ t.Errorf("%s recent %+v", c.Number, c.Recent)316327 }317328 }318329 for _, m := range sig.Muxes {
@@ -321,7 +332,7 @@ func TestSampleInputs(t *testing.T) {
321332 }322333 }323334 raw, _ := json.Marshal(sig)324- if !strings.Contains(string(raw), `"number":"4.1","rf":35,"signal":null`) {335+ if !strings.Contains(string(raw), `"number":"4.1","rf":35,"signal":null,"recent":null`) {325336 t.Errorf("unmeasured channel: %s", raw)326337 }327338
@@ -357,6 +368,10 @@ func waitSweep(t *testing.T, s *Service) SweepStatus {
357368 func TestMeasure(t *testing.T) {358369 s, fake := newTunerService(t)359370 ctx := t.Context()371+ // KMGH's VHF signal arrives strong, but the tuner can't lock to it;372+ // KRMA's locks, with errors.373+ fake.SetSignal("177MHz", tvhtest.Signal{Strength: 81})374+ fake.SetSignal("587MHz", tvhtest.Signal{Lock: true, Strength: 79, Quality: 51, Errors: 2})360375 st, err := s.Measure(ctx)361376 if err != nil || !st.Running || st.StartedAt == nil {362377 t.Fatalf("measure: %+v, %v", st, err)
@@ -367,12 +382,18 @@ func TestMeasure(t *testing.T) {
367382 done := waitSweep(t, s)368383 // The 8 multiplexes with channels, and RF 7 (KMGH), 16 (KUSA) and 34369384 // (KWGN's ATSC 3.0), licensed but not found.370- if done.Total != 11 || done.Done != 11 || done.Note != "" || done.FinishedAt == nil {385+ if done.Total != 11 || done.Found != 8 || done.Done != 11 || done.Note != "" || done.FinishedAt == nil {371386 t.Errorf("sweep = %+v", done)372387 }388+ // The found ones first, by RF; then the others.373389 opened, most := fake.Opened()374- if len(opened) != 11 || most != 1 {375- t.Errorf("opened %v, %d at once", opened, most)390+ var muxes []string391+ for _, o := range opened {392+ muxes = append(muxes, o[strings.LastIndex(o, " ")+1:])393+ }394+ want := []string{"479MHz", "557MHz", "563MHz", "575MHz", "581MHz", "587MHz", "599MHz", "605MHz", "177MHz", "485MHz", "593MHz"}395+ if !slices.Equal(muxes, want) || most != 1 {396+ t.Errorf("opened %v, %d at once", muxes, most)376397 }377398 for _, r := range fake.Requests() {378399 if strings.HasPrefix(r, "GET /stream/") && !strings.HasSuffix(r, "?pids=0&weight=10") {
@@ -381,13 +402,28 @@ func TestMeasure(t *testing.T) {
381402 }382403 for rf, freq := range map[int]int64{31: 575_000_000, 36: 605_000_000, 7: 177_000_000, 34: 593_000_000} {383404 m, ok := s.signals.Get(freq)384- if !ok || m.Latest == nil || m.Latest.Source != signal.Sweep || m.History().Samples != 3 || m.Latest.Lock != (rf == 31 || rf == 36) {405+ // Three readings with a lock; without one, the one after giving406+ // the tuner its time to lock.407+ locked := rf == 31 || rf == 36408+ if want := map[bool]int{true: 3, false: 1}[locked]; !ok || m.Latest == nil || m.Latest.Source != signal.Sweep || m.History().Samples != want || m.Latest.Lock != locked {385409 t.Errorf("RF %d = %+v", rf, m)386410 }387411 }388412 if _, ok := s.signals.Get(473_000_000); ok {389413 t.Error("measured RF 14, which nothing uses")390414 }415+ // No lock is a reading too, with the strength the tuner measured.416+ if m, _ := s.signals.Get(177_000_000); m.Latest.Lock || m.Latest.StrengthPct == nil || *m.Latest.StrengthPct != 81 || *m.Latest.QualityPct != 0 {417+ t.Errorf("RF 7 = %+v", m.Latest)418+ }419+ // A lock, read once the tuner has a quality to give.420+ if m, _ := s.signals.Get(575_000_000); !m.Latest.Lock || m.History().QualityPct == nil || m.History().QualityPct.Min != 83 ||421+ m.Latest.ErrorsPerSec == nil || *m.Latest.ErrorsPerSec != 0 {422+ t.Errorf("RF 31 = %+v, %+v", m.History(), m.Latest)423+ }424+ if m, _ := s.signals.Get(587_000_000); !m.Latest.Lock || m.Latest.ErrorsPerSec == nil || *m.Latest.ErrorsPerSec <= 0 {425+ t.Errorf("RF 33 = %+v", m.Latest)426+ }391427 }392428 393429 // TestMeasureLeavesTunersAlone: the sweep reads a viewer's tuner rather
@@ -533,7 +569,7 @@ func TestHDHomeRunAntenna(t *testing.T) {
533569 numbers = append(numbers, n)534570 names[n] = e["GuideName"].(string)535571 }536- if !slices.Equal(numbers, tvhNumbers(fake)) || names["20.1"] != "KTVDDT" || names["20.2"] != "KTVD" || names["9.7"] != "KUSA" {572+ if !slices.Equal(numbers, tvhNumbers(fake)) || names["20.1"] != "KTVDDT" || names["20.2"] != "H & I" || names["9.7"] != "TBD" {537573 t.Errorf("lineup %v, names %v", numbers, names)538574 }539575 if s.TunerCount() != 2 {
M internal/service/hdhr.go +3 -1
@@ -70,7 +70,9 @@ func (b hdhrBackend) Lineup(ctx context.Context) ([]hdhr.Entry, error) {
7070 if custom[vchan.NumberKey(c.Number)] {7171 continue7272 }73- out = append(out, hdhr.Entry{Number: c.Number, Name: cmp.Or(c.CallSign, c.Name), HD: true, Group: "Antenna", Logo: c.Logo})73+ // Players match guide data by the listings' call signs, one per74+ // subchannel ("KTVDDT2"); without listings, the broadcast's name.75+ out = append(out, hdhr.Entry{Number: c.Number, Name: cmp.Or(c.GuideCallSign, c.Name), HD: true, Group: "Antenna", Logo: c.Logo})7476 }7577 slices.SortStableFunc(out, func(a, b hdhr.Entry) int { return guide.CompareNumbers(a.Number, b.Number) })7678 return out, nil
M internal/service/measure.go +218 -115
@@ -5,12 +5,12 @@ import (
55 "context"66 "errors"77 "fmt"8+ "io"89 "log"910 "net/http"1011 "slices"1112 "strings"1213 "sync"13- "sync/atomic"1414 "time"1515 1616 "airwaves/internal/fcc"
@@ -30,10 +30,11 @@ type measureTiming struct {
3030 // sample is how often the inputs are read while tuners are in use.3131 sample time.Duration3232 // appear is how long the sweep waits for its tuner to show the33- // multiplex, settle how long it then waits before reading, and34- // interval the time between its readings, readings of them.35- appear, settle, interval time.Duration36- readings int33+ // multiplex, acquire how long it then gives the tuner to lock, settle34+ // how long it waits after a lock before reading, and interval the time35+ // between its readings, readings of them (one without a lock).36+ appear, acquire, settle, interval time.Duration37+ readings int3738 // idleWait is how long the sweep waits for an idle tuner before it3839 // gives up.3940 idleWait time.Duration
@@ -44,9 +45,13 @@ type measureTiming struct {
4445 4546 var defaultTiming = measureTiming{4647 sample: 10 * time.Second,47- appear: 6 * time.Second, settle: 1500 * time.Millisecond, interval: time.Second, readings: 3,48+ // An HDHomeRun locks to a good signal in a second or two, to a weak49+ // one in three or four; its quality reads 0 for its first second on a50+ // multiplex. So about 5 s for an RF channel with a lock, 7 s for one51+ // without.52+ appear: 5 * time.Second, acquire: 6 * time.Second, settle: time.Second, interval: time.Second, readings: 3,4853 idleWait: 30 * time.Second,49- poll: time.Second, noLock: 10 * time.Second,54+ poll: 500 * time.Millisecond, noLock: 10 * time.Second,5055 }5156 5257 // SweepStatus is how Measure now is going, or went.
@@ -55,7 +60,10 @@ type SweepStatus struct {
5560 StartedAt *time.Time `json:"startedAt"`5661 FinishedAt *time.Time `json:"finishedAt"`5762 // Total is how many RF channels it measures, Done how many it has.63+ // The first Found of them are those Tvheadend's scan found, measured64+ // first; the rest are those a licensed station nearby uses.5865 Total int `json:"total"`66+ Found int `json:"found"`5967 Done int `json:"done"`6068 // RF is the RF channel being measured now.6169 RF int `json:"rf,omitempty"`
@@ -65,35 +73,98 @@ type SweepStatus struct {
6573 6674 // ---------- inputs ----------6775 68-// inputsNow returns the tuned inputs, as read in the last few seconds;69-// ok is false when Tvheadend can't be read.70-func (s *Service) inputsNow() ([]tvh.InputStatus, bool) {76+// tunerStatus is what Tvheadend says of its inputs at one moment: every77+// input, tuned or idle, and the subscriptions using them.78+type tunerStatus struct {79+ inputs []tvh.InputStatus80+ subs []tvh.Subscription81+}82+83+// busy counts the inputs something subscribes to.84+func (t tunerStatus) busy() int {85+ seen := map[string]bool{}86+ for _, in := range t.inputs {87+ if in.InUse() {88+ seen[in.Input] = true89+ }90+ }91+ return len(seen)92+}93+94+// on finds an input in use on the antenna multiplex named mux.95+func (t tunerStatus) on(mux string) *tvh.InputStatus {96+ for i := range t.inputs {97+ if name, net, ok := t.inputs[i].Mux(); ok && name == mux && net == tvh.ATSCNetwork && t.inputs[i].InUse() {98+ return &t.inputs[i]99+ }100+ }101+ return nil102+}103+104+// input finds an input by name.105+func (t tunerStatus) input(name string) *tvh.InputStatus {106+ for i := range t.inputs {107+ if t.inputs[i].Input == name {108+ return &t.inputs[i]109+ }110+ }111+ return nil112+}113+114+// reading is what in measures now.115+func (t tunerStatus) reading(in tvh.InputStatus, at time.Time, source string) signal.Reading {116+ return signal.FromInput(in, t.subs, at, source)117+}118+119+// ownScansOnly reports whether everything using the inputs is Tvheadend's120+// own scanning (weights below a user's).121+func (t tunerStatus) ownScansOnly() bool {122+ own := false123+ for _, in := range t.inputs {124+ if in.InUse() {125+ if in.Weight >= sweepWeight {126+ return false127+ }128+ own = true129+ }130+ }131+ return own132+}133+134+// statusNow returns the inputs and subscriptions as read in the last few135+// seconds; ok is false when Tvheadend can't be read.136+func (s *Service) statusNow() (tunerStatus, bool) {71137 if s.opt.Tvheadend == nil {72- return nil, false138+ return tunerStatus{}, false73139 }74140 s.mu.Lock()75- ins, at := s.inputs, s.inputsAt141+ st, at := s.status, s.statusAt76142 s.mu.Unlock()77143 if !at.IsZero() && time.Since(at) < 5*time.Second {78- return ins, true144+ return st, true79145 }80146 ctx, cancel := context.WithTimeout(s.ctx, 5*time.Second)81147 defer cancel()82- ins, err := s.readInputs(ctx)83- return ins, err == nil148+ st, err := s.readStatus(ctx)149+ return st, err == nil84150 }85151 86-// readInputs reads the tuned inputs from Tvheadend and keeps them for87-// inputsNow.88-func (s *Service) readInputs(ctx context.Context) ([]tvh.InputStatus, error) {152+// readStatus reads the inputs and subscriptions from Tvheadend, and keeps153+// them for statusNow.154+func (s *Service) readStatus(ctx context.Context) (tunerStatus, error) {89155 ins, err := s.opt.Tvheadend.Inputs(ctx)90156 if err != nil {91- return nil, err157+ return tunerStatus{}, err158+ }159+ subs, err := s.opt.Tvheadend.Subscriptions(ctx)160+ if err != nil {161+ return tunerStatus{}, err92162 }163+ st := tunerStatus{inputs: ins, subs: subs}93164 s.mu.Lock()94- s.inputs, s.inputsAt = ins, time.Now()165+ s.status, s.statusAt = st, time.Now()95166 s.mu.Unlock()96- return ins, nil167+ return st, nil97168 }98169 99170 // antennaMuxes maps the antenna network's multiplexes by name.
@@ -114,7 +185,7 @@ func (s *Service) antennaMuxes(ctx context.Context) (map[string]tvh.Mux, error)
114185 // Tvheadend's guide scan. An input just tuned isn't read yet, as it may115186 // not have locked; the sweep reads its own.116187 func (s *Service) sampleInputs(ctx context.Context) {117- ins, err := s.readInputs(ctx)188+ st, err := s.readStatus(ctx)118189 if err != nil {119190 return120191 }
@@ -123,25 +194,28 @@ func (s *Service) sampleInputs(ctx context.Context) {
123194 return124195 }125196 s.mu.Lock()126- prev := s.lastStreams197+ prev, prevAt := s.lastSample, s.lastSampleAt127198 sweeping := s.sweepMux128199 s.mu.Unlock()129200 now := time.Now()130- seen := map[string]string{}131- for _, in := range ins {201+ seen := map[string]tvh.InputStatus{}202+ for _, in := range st.inputs {132203 name, net, ok := in.Mux()133- if !ok || net != tvh.ATSCNetwork || in.Subs == 0 {204+ if !ok || net != tvh.ATSCNetwork || !in.InUse() {134205 continue135206 }136- seen[in.Input] = in.Stream207+ seen[in.Input] = in137208 m, ok := muxes[name]138- if !ok || prev[in.Input] != in.Stream || name == sweeping {209+ was, stable := prev[in.Input]210+ if !ok || !stable || was.Stream != in.Stream || name == sweeping {139211 continue140212 }141- s.signals.Record(m.FrequencyHz, signal.RF(m.FrequencyHz), signal.FromInput(in, now, signal.Active))213+ r := st.reading(in, now, signal.Active)214+ r.SetErrors(was, in, now.Sub(prevAt).Seconds())215+ s.signals.Record(m.FrequencyHz, signal.RF(m.FrequencyHz), r)142216 }143217 s.mu.Lock()144- s.lastStreams = seen218+ s.lastSample, s.lastSampleAt = seen, now145219 s.mu.Unlock()146220 }147221
@@ -237,39 +311,52 @@ func (s *Service) setSweep(f func(*SweepStatus)) {
237311 s.mu.Unlock()238312 }239313 240-// sweepTargets lists the antenna multiplexes worth measuring, by RF:241-// those with services, and those a licensed station within the report's242-// radius broadcasts on. It also names a channel on each, for servers that243-// won't stream a whole multiplex.244-func (s *Service) sweepTargets(ctx context.Context) ([]tvh.Mux, map[string]string, error) {314+// sweepTargets lists the antenna multiplexes worth measuring: first those315+// Tvheadend's scan found (by RF), then those a licensed station within the316+// report's radius broadcasts on (nearest station first); found is how317+// many of them are of the first kind. It also names a channel on each318+// found one, for servers that won't stream a whole multiplex.319+func (s *Service) sweepTargets(ctx context.Context) (muxes []tvh.Mux, found int, channelOn map[string]string, err error) {245320 s.tvhTuner.Invalidate()246321 l, err := s.tvhTuner.Lineup(ctx)247322 if err != nil {248- return nil, nil, err323+ return nil, 0, nil, err249324 }250325 s.mu.Lock()251326 rep := s.report252327 s.mu.Unlock()253- licensed := map[int]bool{}328+ nearest := map[int]float64{} // km to the nearest licensed station, by RF254329 if rep != nil {255330 for _, st := range rep.Stations {256- licensed[st.RFChannel] = true331+ if d, ok := nearest[st.RFChannel]; !ok || st.DistanceKm < d {332+ nearest[st.RFChannel] = st.DistanceKm333+ }257334 }258335 }259- channelOn := map[string]string{}336+ channelOn = map[string]string{}260337 for _, c := range l.Channels {261338 if _, m, ok := l.Source(c); ok && c.Enabled && channelOn[m.UUID] == "" {262339 channelOn[m.UUID] = c.UUID263340 }264341 }265- var out []tvh.Mux342+ var scanned, licensed []tvh.Mux266343 for _, m := range l.NetworkMuxes(tvh.ATSCNetwork) {267- if bool(m.Enabled) && m.FrequencyHz > 0 && (m.NumSvc > 0 || licensed[signal.RF(m.FrequencyHz)]) {268- out = append(out, m)344+ if !bool(m.Enabled) || m.FrequencyHz <= 0 {345+ continue346+ }347+ _, near := nearest[signal.RF(m.FrequencyHz)]348+ switch {349+ case m.NumSvc > 0 || m.ScanResult == tvh.ScanOK || m.ScanResult == tvh.ScanPartial:350+ scanned = append(scanned, m)351+ case near:352+ licensed = append(licensed, m)269353 }270354 }271- slices.SortFunc(out, func(a, b tvh.Mux) int { return cmp.Compare(a.FrequencyHz, b.FrequencyHz) })272- return out, channelOn, nil355+ slices.SortFunc(scanned, func(a, b tvh.Mux) int { return cmp.Compare(a.FrequencyHz, b.FrequencyHz) })356+ slices.SortFunc(licensed, func(a, b tvh.Mux) int {357+ return cmp.Or(cmp.Compare(nearest[signal.RF(a.FrequencyHz)], nearest[signal.RF(b.FrequencyHz)]), cmp.Compare(a.FrequencyHz, b.FrequencyHz))358+ })359+ return append(scanned, licensed...), len(scanned), channelOn, nil273360 }274361 275362 func (s *Service) runSweep(ctx context.Context, tuners int) {
@@ -287,14 +374,15 @@ func (s *Service) runSweep(ctx context.Context, tuners int) {
287374 s.sweepMux = ""288375 s.sweep.Running, s.sweep.FinishedAt, s.sweep.RF, s.sweep.Note = false, &now, 0, note289376 s.mu.Unlock()377+ log.Printf("measuring done: %+v", s.sweepStatus())290378 }()291- muxes, channelOn, err := s.sweepTargets(ctx)379+ muxes, found, channelOn, err := s.sweepTargets(ctx)292380 if err != nil {293381 note = "Tvheadend is not reachable: " + err.Error()294382 return295383 }296- s.setSweep(func(st *SweepStatus) { st.Total = len(muxes) })297- log.Printf("measuring %d RF channels", len(muxes))384+ s.setSweep(func(st *SweepStatus) { st.Total, st.Found = len(muxes), found })385+ log.Printf("measuring %d RF channels, the %d found first", len(muxes), found)298386 for _, m := range muxes {299387 if ctx.Err() != nil {300388 note = "stopped: the server is shutting down"
@@ -302,16 +390,18 @@ func (s *Service) runSweep(ctx context.Context, tuners int) {
302390 }303391 rf := signal.RF(m.FrequencyHz)304392 s.setSweep(func(st *SweepStatus) { st.RF = rf })305- ins, err := s.readInputs(ctx)306- if on := onMux(ins, m.Name); err == nil && on != nil {307- // A tuner is on it already (a viewer, a recording): read that308- // rather than take another.309- s.signals.Record(m.FrequencyHz, rf, signal.FromInput(*on, time.Now(), signal.Sweep))393+ st, err := s.readStatus(ctx)394+ if on := st.on(m.Name); err == nil && on != nil {395+ // A tuner is on it already (a viewer, a recording): read that,396+ // once it has locked, rather than take another.397+ if r := st.reading(*on, time.Now(), signal.Sweep); r.Lock {398+ s.signals.Record(m.FrequencyHz, rf, r)399+ }310400 s.setSweep(func(st *SweepStatus) { st.Done++ })311401 continue312402 }313- if ins, ok := s.waitIdle(ctx, tuners); !ok {314- note = fmt.Sprintf("stopped at RF %d: %s", rf, noTunerFree(ins))403+ if st, ok := s.waitIdle(ctx, tuners); !ok {404+ note = fmt.Sprintf("stopped at RF %d: %s", rf, noTunerFree(st))315405 return316406 }317407 err = s.measureMux(ctx, m, channelOn[m.UUID])
@@ -321,7 +411,7 @@ func (s *Service) runSweep(ctx context.Context, tuners int) {
321411 note = fmt.Sprintf("stopped at RF %d: %v", rf, err)322412 return323413 case errors.As(err, &se) && se.Status == http.StatusServiceUnavailable:324- note = fmt.Sprintf("stopped at RF %d: %s", rf, noTunerFree(nil))414+ note = fmt.Sprintf("stopped at RF %d: %s", rf, noTunerFree(tunerStatus{}))325415 return326416 case errors.As(err, &se):327417 skipped++
@@ -330,30 +420,19 @@ func (s *Service) runSweep(ctx context.Context, tuners int) {
330420 }331421 s.setSweep(func(st *SweepStatus) { st.Done++ })332422 }333- log.Printf("measured %d RF channels", len(muxes))334423 }335424 336-// onMux finds an input tuned to the antenna multiplex named mux.337-func onMux(ins []tvh.InputStatus, mux string) *tvh.InputStatus {338- for i := range ins {339- if name, net, ok := ins[i].Mux(); ok && name == mux && net == tvh.ATSCNetwork {340- return &ins[i]341- }342- }343- return nil344-}345-346-// waitIdle waits up to timing.idleWait for a tuner nobody uses, and347-// returns the inputs in use then; ok is false when none came free.348-func (s *Service) waitIdle(ctx context.Context, tuners int) ([]tvh.InputStatus, bool) {425+// waitIdle waits up to timing.idleWait for a tuner nobody subscribes to,426+// and returns the status then; ok is false when none came free.427+func (s *Service) waitIdle(ctx context.Context, tuners int) (tunerStatus, bool) {349428 deadline := time.Now().Add(s.timing.idleWait)350429 for {351- ins, err := s.readInputs(ctx)352- if err == nil && busyTuners(ins) < tuners {353- return ins, true430+ st, err := s.readStatus(ctx)431+ if err == nil && st.busy() < tuners {432+ return st, true354433 }355434 if time.Now().After(deadline) || sleep(ctx, s.timing.poll) != nil {356- return ins, false435+ return st, false357436 }358437 }359438 }
@@ -361,26 +440,13 @@ func (s *Service) waitIdle(ctx context.Context, tuners int) ([]tvh.InputStatus,
361440 // noTunerFree says why the sweep found no tuner: in use by viewers or362441 // recordings, which it leaves alone, or by Tvheadend's own guide or363442 // channel scans (weights below a user's), which end on their own.364-func noTunerFree(ins []tvh.InputStatus) string {365- own := len(ins) > 0366- for _, in := range ins {367- own = own && in.Weight < sweepWeight368- }369- if own {443+func noTunerFree(st tunerStatus) string {444+ if st.ownScansOnly() {370445 return "every tuner is busy with Tvheadend's own guide or channel scan; try again in a few minutes"371446 }372447 return "no tuner was free (they're left to viewers and recordings)"373448 }374449 375-// busyTuners counts the inputs in use.376-func busyTuners(ins []tvh.InputStatus) int {377- seen := map[string]bool{}378- for _, in := range ins {379- seen[in.Input] = true380- }381- return len(seen)382-}383-384450 func sleep(ctx context.Context, d time.Duration) error {385451 t := time.NewTimer(d)386452 defer t.Stop()
@@ -393,7 +459,8 @@ func sleep(ctx context.Context, d time.Duration) error {
393459 }394460 395461 // measureMux subscribes to a multiplex at the sweep's weight, waits for a396-// tuner to take it, and records timing.readings readings of it. A462+// tuner to take it, and records timing.readings readings of it: with a463+// lock, or without one, with whatever strength the tuner measures. A397464 // subscription Tvheadend refuses (a server that won't stream whole398465 // multiplexes) is retried on channelUUID, a channel of the multiplex,399466 // when it has one.
@@ -405,7 +472,9 @@ func (s *Service) measureMux(ctx context.Context, m tvh.Mux, channelUUID string)
405472 s.mu.Unlock()406473 rf := signal.RF(m.FrequencyHz)407474 408- var received atomic.Int64475+ // Tvheadend answers a multiplex it can't lock with no headers at all,476+ // so this waits on its own while the tuner is read; what comes is477+ // drained, so the stream keeps flowing.409478 opened := make(chan error, 1)410479 go func() {411480 body, err := s.opt.Tvheadend.OpenMux(ctx, m.UUID, sweepWeight)
@@ -418,14 +487,7 @@ func (s *Service) measureMux(ctx context.Context, m tvh.Mux, channelUUID string)
418487 return419488 }420489 defer body.Close()421- buf := make([]byte, 32*1024)422- for {423- n, err := body.Read(buf)424- received.Add(int64(n))425- if err != nil {426- return427- }428- }490+ _, _ = io.Copy(io.Discard, body)429491 }()430492 431493 // Wait for a tuner to show the multiplex.
@@ -439,8 +501,8 @@ func (s *Service) measureMux(ctx context.Context, m tvh.Mux, channelUUID string)
439501 }440502 default:441503 }442- if ins, err := s.readInputs(ctx); err == nil {443- if on := onMux(ins, m.Name); on != nil {504+ if st, err := s.readStatus(ctx); err == nil {505+ if on := st.on(m.Name); on != nil {444506 input = on.Input445507 break446508 }
@@ -452,27 +514,64 @@ func (s *Service) measureMux(ctx context.Context, m tvh.Mux, channelUUID string)
452514 return err453515 }454516 }517+ // Give the tuner time to lock: a weak signal can take seconds. What it518+ // measures meanwhile, past its first second on the multiplex (which519+ // reads what it measured before, or nothing), is the reading when it520+ // doesn't lock.521+ locked := false522+ var tried *signal.Reading523+ start := time.Now()524+ for !locked && time.Since(start) < s.timing.acquire {525+ if err := sleep(ctx, s.timing.poll); err != nil {526+ return err527+ }528+ st, err := s.readStatus(ctx)529+ if err != nil {530+ continue531+ }532+ on := st.on(m.Name)533+ if on == nil {534+ if in := st.input(input); in != nil && in.InUse() {535+ return errTakenOver536+ }537+ break // Tvheadend gave up on it538+ }539+ input = on.Input540+ r := st.reading(*on, time.Now(), signal.Sweep)541+ locked = r.Lock542+ if !locked && time.Since(start) >= s.timing.settle {543+ tried = &r544+ }545+ }546+ if !locked {547+ if tried == nil {548+ return fmt.Errorf("RF %d: the tuner left it before a reading", rf)549+ }550+ s.signals.Record(m.FrequencyHz, rf, *tried)551+ return nil552+ }455553 if err := sleep(ctx, s.timing.settle); err != nil {456554 return err457555 }458- for i := range s.timing.readings {556+ readings := s.timing.readings557+ var prev *tvh.InputStatus558+ var prevAt time.Time559+ for i := range readings {459560 if i > 0 {460561 if err := sleep(ctx, s.timing.interval); err != nil {461562 return err462563 }463564 }464- ins, err := s.readInputs(ctx)565+ st, err := s.readStatus(ctx)465566 if err != nil {466567 return err467568 }468- on := onMux(ins, m.Name)569+ on := st.on(m.Name)469570 if on == nil {470- for _, in := range ins {471- if in.Input == input {472- // Its tuner is on another multiplex now: someone473- // else's subscription took it over.474- return errTakenOver475- }571+ if in := st.input(input); in != nil && in.InUse() {572+ // Its tuner is on another multiplex now: someone else's573+ // subscription took it over.574+ return errTakenOver476575 }477576 // Tvheadend let the multiplex go (or is between478577 // subscriptions): there's nothing to read, and nothing is
@@ -482,8 +581,12 @@ func (s *Service) measureMux(ctx context.Context, m tvh.Mux, channelUUID string)
482581 // Tvheadend may move a subscription that can't lock to another483582 // tuner; follow it there.484583 input = on.Input485- r := signal.FromInput(*on, time.Now(), signal.Sweep)486- r.Lock = r.Lock || received.Load() > 0584+ now := time.Now()585+ r := st.reading(*on, now, signal.Sweep)586+ if prev != nil {587+ r.SetErrors(*prev, *on, now.Sub(prevAt).Seconds())588+ }589+ prev, prevAt = on, now487590 s.signals.Record(m.FrequencyHz, rf, r)488591 }489592 return nil
@@ -519,14 +622,14 @@ func (s *Service) watchTune(ctx context.Context, cancel context.CancelCauseFunc,
519622 if sleep(ctx, s.timing.poll) != nil {520623 return521624 }522- ins, err := s.readInputs(ctx)625+ st, err := s.readStatus(ctx)523626 if err != nil {524627 continue525628 }526- on := onMux(ins, mux.Name)629+ on := st.on(mux.Name)527630 w.mu.Lock()528631 if on == nil {529- w.busy = w.busy || tuners > 0 && busyTuners(ins) >= tuners632+ w.busy = w.busy || tuners > 0 && st.busy() >= tuners530633 w.mu.Unlock()531634 continue532635 }
@@ -534,7 +637,7 @@ func (s *Service) watchTune(ctx context.Context, cancel context.CancelCauseFunc,
534637 if w.since.IsZero() {535638 w.since = now536639 }537- r := signal.FromInput(*on, now, signal.Active)640+ r := st.reading(*on, now, signal.Active)538641 w.reading, w.last = &r, now539642 w.mu.Unlock()540643 switch {
M internal/service/service.go +8 -7
@@ -224,13 +224,14 @@ type Service struct {
224224 fes []tvh.Frontend225225 fesAt time.Time226226 fesKnown bool227- // inputs are the tuned inputs as of inputsAt; lastStreams what each228- // was tuned to at the last sample.229- inputs []tvh.InputStatus230- inputsAt time.Time231- lastStreams map[string]string232- sweep SweepStatus233- sweepMux string // the multiplex the sweep is on227+ // status is the inputs and subscriptions as of statusAt; lastSample228+ // the inputs at the last sample, and when.229+ status tunerStatus230+ statusAt time.Time231+ lastSample map[string]tvh.InputStatus232+ lastSampleAt time.Time233+ sweep SweepStatus234+ sweepMux string // the multiplex the sweep is on234235 }235236 236237 // New builds a Service. Streaming failures (no ffmpeg) are reported by
M internal/signal/signal.go +51 -2
@@ -52,11 +52,38 @@ type Reading struct {
5252 // tuners that count them.5353 BER *float64 `json:"ber"`5454 UNC *int64 `json:"unc"`55+ // ErrorsPerSec is the rate of transport and continuity errors in what56+ // arrived since the reading before, on the same tuning: a picture57+ // breaking up. Null for a first reading.58+ ErrorsPerSec *float64 `json:"errorsPerSec"`5559 }5660 57-// FromInput reads an input's status as Tvheadend reports it.58-func FromInput(in tvh.InputStatus, at time.Time, source string) Reading {61+// FromInput reads an input's status as Tvheadend reports it, with the62+// subscriptions now. Whether the tuner holds a lock is first what its63+// subscriptions say: "Running" once data has arrived, "Testing" while none64+// has. The input's data rate says less: it reads 0 now and then on an65+// input with little to send, and can show the last tuning's for a moment;66+// it decides only for an input with no subscription listed. Then a tuner67+// that reports quality on the relative scale (an HDHomeRun) has none, 0,68+// exactly while its demodulator has no lock (its own status page reads69+// "Modulation Lock none, Signal Quality none" then), even while a70+// subscription that had data still reads "Running".71+func FromInput(in tvh.InputStatus, subs []tvh.Subscription, at time.Time, source string) Reading {5972 r := Reading{At: at.UTC().Truncate(time.Second), Source: source, Lock: in.BPS > 0}73+ if mux, net, ok := in.Mux(); ok {74+ listed, running := false, false75+ for _, sub := range subs {76+ if sub.On(in.Input, net, mux) {77+ listed, running = true, running || sub.Running()78+ }79+ }80+ if listed {81+ r.Lock = running82+ }83+ }84+ if in.SNRScale == tvh.ScaleRelative && in.SNR == 0 {85+ r.Lock = false86+ }6087 switch in.SignalScale {6188 case tvh.ScaleRelative:6289 r.StrengthPct = ptr(round1(float64(in.Signal) * 100 / 65535))
@@ -83,6 +110,17 @@ func FromInput(in tvh.InputStatus, at time.Time, source string) Reading {
83110 return r84111 }85112 113+// SetErrors sets the error rate from the input's status at an earlier114+// reading of the same tuning, secs before.115+func (r *Reading) SetErrors(prev, now tvh.InputStatus, secs float64) {116+ r.ErrorsPerSec = nil117+ was, is := prev.TE+prev.CC, now.TE+now.CC118+ if secs <= 0 || is < was || prev.Stream != now.Stream || prev.Input != now.Input {119+ return120+ }121+ r.ErrorsPerSec = ptr(round1(float64(is-was) / secs))122+}123+86124 func ptr[T any](v T) *T { return &v }87125 88126 func round1(v float64) float64 { return math.Round(v*10) / 10 }
@@ -254,6 +292,17 @@ type History struct {
254292 Windows []Period `json:"windows"`255293 }256294 295+// Recent summarizes the latest window of readings (the last sweep of the296+// multiplex, or the last stretch of watching it), steadier than any one297+// reading; nil when there are none.298+func (m *Mux) Recent() *Period {299+ if len(m.Windows) == 0 {300+ return nil301+ }302+ p := period(m.Windows[len(m.Windows)-1])303+ return &p304+}305+257306 // History summarizes the multiplex's readings; nil when there are none.258307 func (m *Mux) History() *History {259308 if len(m.Windows) == 0 {
M internal/signal/signal_test.go +84 -4
@@ -23,13 +23,13 @@ func TestFromInput(t *testing.T) {
2323 t.Fatal(err)2424 }2525 at := time.Date(2026, 10, 4, 18, 0, 0, 0, time.UTC)26- r := FromInput(body.Entries[0], at, Active)26+ r := FromInput(body.Entries[0], nil, at, Active)2727 if !r.Lock || *r.StrengthPct != 85 || *r.QualityPct != 83 || r.StrengthDBm != nil || r.SNRdB != nil ||2828 r.BER != nil || r.UNC != nil || r.Source != Active || !r.At.Equal(at) {2929 t.Errorf("reading = %+v", r)3030 }3131 raw, _ := json.Marshal(r)32- for _, want := range []string{`"lock":true`, `"strengthPct":85`, `"qualityPct":83`, `"snrDb":null`, `"strengthDbm":null`, `"ber":null`, `"unc":null`} {32+ for _, want := range []string{`"lock":true`, `"strengthPct":85`, `"qualityPct":83`, `"snrDb":null`, `"strengthDbm":null`, `"ber":null`, `"unc":null`, `"errorsPerSec":null`} {3333 if !strings.Contains(string(raw), want) {3434 t.Errorf("JSON lacks %s: %s", want, raw)3535 }
@@ -37,17 +37,97 @@ func TestFromInput(t *testing.T) {
3737 3838 // A tuner reporting decibels, counting bits and blocks, with no lock.3939 r = FromInput(tvh.InputStatus{Signal: -48250, SignalScale: tvh.ScaleDecibel, SNR: 21400, SNRScale: tvh.ScaleDecibel,40- ECBit: 3, TCBit: 1000, ECBlock: 7, TCBlock: 500}, at, Sweep)40+ ECBit: 3, TCBit: 1000, ECBlock: 7, TCBlock: 500}, nil, at, Sweep)4141 if r.Lock || *r.StrengthDBm != -48.3 || *r.SNRdB != 21.4 || r.StrengthPct != nil || r.QualityPct != nil || *r.BER != 0.003 || *r.UNC != 7 {4242 t.Errorf("decibel reading = %+v", r)4343 }4444 // Nothing reported: nothing claimed.45- r = FromInput(tvh.InputStatus{}, at, Sweep)45+ r = FromInput(tvh.InputStatus{}, nil, at, Sweep)4646 if r.StrengthPct != nil || r.StrengthDBm != nil || r.QualityPct != nil || r.SNRdB != nil || r.UNC != nil || r.Lock {4747 t.Errorf("empty reading = %+v", r)4848 }4949 }5050 51+// TestFromInputMeasuring reads two tuners measuring, as a real Tvheadend52+// reported them: RF 31 locked (its subscription "Running", though its53+// input sends little), RF 7 not (strength, but no quality, no data, and54+// its subscription "Testing"). Idle inputs say nothing.55+func TestFromInputMeasuring(t *testing.T) {56+ var ins struct {57+ Entries []tvh.InputStatus `json:"entries"`58+ }59+ var subs struct {60+ Entries []tvh.Subscription `json:"entries"`61+ }62+ if err := json.Unmarshal(tvhtest.Raw(t, "status_inputs_measuring.json"), &ins); err != nil {63+ t.Fatal(err)64+ }65+ if err := json.Unmarshal(tvhtest.Raw(t, "status_subscriptions_measuring.json"), &subs); err != nil {66+ t.Fatal(err)67+ }68+ at := time.Date(2026, 10, 4, 18, 0, 0, 0, time.UTC)69+ rf7, rf31 := ins.Entries[0], ins.Entries[1]70+ rf31.BPS = 0 // as it reads, now and then71+ r := FromInput(rf31, subs.Entries, at, Sweep)72+ if !r.Lock || *r.StrengthPct != 84 || *r.QualityPct != 67 {73+ t.Errorf("RF 31 = %+v", r)74+ }75+ r = FromInput(rf7, subs.Entries, at, Sweep)76+ if r.Lock || *r.StrengthPct != 81 || *r.QualityPct != 0 {77+ t.Errorf("RF 7 = %+v", r)78+ }79+ // Data seems to arrive on RF 7 for a moment (the last tuning's rate),80+ // but its subscription is still "Testing": no lock.81+ rf7.BPS = 21056082+ if r := FromInput(rf7, subs.Entries, at, Sweep); r.Lock {83+ t.Errorf("RF 7 with a stale rate = %+v", r)84+ }85+ // Without a subscription listed, the rate decides.86+ if r := FromInput(rf31, nil, at, Sweep); r.Lock {87+ t.Errorf("RF 31 without subscriptions or data = %+v", r)88+ }89+ rf31.BPS = 15792090+ if r := FromInput(rf31, nil, at, Sweep); !r.Lock {91+ t.Errorf("RF 31 without subscriptions, with data = %+v", r)92+ }93+ // A marginal multiplex losing its lock for a moment: its subscription94+ // still "Running", the HDHomeRun's quality none.95+ gone := rf3196+ gone.SNR = 097+ if r := FromInput(gone, subs.Entries, at, Sweep); r.Lock || *r.QualityPct != 0 {98+ t.Errorf("RF 31 between locks = %+v", r)99+ }100+ // A marginal multiplex: transport and continuity errors climbing.101+ later := rf31102+ later.TE, later.CC = rf31.TE+60, rf31.CC+30103+ r = FromInput(later, subs.Entries, at, Sweep)104+ if r.SetErrors(rf31, later, 2); r.ErrorsPerSec == nil || *r.ErrorsPerSec != 45 {105+ t.Errorf("errors = %v", r.ErrorsPerSec)106+ }107+ if r := FromInput(rf31, subs.Entries, at, Sweep); r.ErrorsPerSec != nil {108+ t.Errorf("first reading errors = %v", *r.ErrorsPerSec)109+ }110+ r.SetErrors(rf7, rf31, 2) // another tuning: no rate111+ if r.ErrorsPerSec != nil {112+ t.Errorf("errors across tunings = %v", *r.ErrorsPerSec)113+ }114+115+ var idle struct {116+ Entries []tvh.InputStatus `json:"entries"`117+ }118+ if err := json.Unmarshal(tvhtest.Raw(t, "status_inputs_idle.json"), &idle); err != nil {119+ t.Fatal(err)120+ }121+ for _, in := range idle.Entries {122+ if _, _, ok := in.Mux(); ok || in.InUse() {123+ t.Errorf("idle input %+v", in)124+ }125+ if r := FromInput(in, nil, at, Active); r.Lock || r.StrengthPct != nil || r.QualityPct != nil {126+ t.Errorf("idle reading %+v", r)127+ }128+ }129+}130+51131 func TestScanOf(t *testing.T) {52132 seen := time.Date(2026, 10, 4, 18, 0, 0, 0, time.UTC)53133 if sc, ok := ScanOf(tvh.Mux{ScanResult: tvh.ScanOK, ScanLast: 1791129355}, seen); !ok || !sc.Lock || sc.Partial || sc.At.Unix() != 1791129355 {
M internal/tvh/status.go +30 -8
@@ -10,13 +10,14 @@ import (
1010 "strings"1111 )1212 13-// InputStatus is a tuner input that is tuned now, as /api/status/inputs14-// reports it. Idle inputs aren't listed there.13+// InputStatus is a tuner input as /api/status/inputs reports it. Tvheadend14+// 4.3 lists idle inputs too, with no Stream, no subscriptions and weight 0.1515 type InputStatus struct {1616 UUID string `json:"uuid"`1717 // Input is the input's name, as Frontend.Name has it.1818 Input string `json:"input"`19- // Stream is what the input is tuned to: "605MHz in Airwaves antenna".19+ // Stream is what the input is tuned to: "605MHz in Airwaves antenna";20+ // empty while idle.2021 Stream string `json:"stream"`2122 Subs int `json:"subs"`2223 Weight int `json:"weight"`
@@ -29,7 +30,9 @@ type InputStatus struct {
2930 BER int64 `json:"ber"`3031 UNC int64 `json:"unc"`3132 // BPS is the data rate the input receives, in bits per second: above32- // zero only while the tuner holds a lock.33+ // zero only while the tuner holds a lock, though a locked input with34+ // little to send (a guide scan, a measurement) can read 0 for seconds35+ // at a time.3336 BPS int64 `json:"bps"`3437 // TE and CC count transport and continuity errors in what arrived.3538 TE int64 `json:"te"`
@@ -53,6 +56,11 @@ const (
5356 ScaleDecibel Scale = 2 // thousandths of a dB (dBm for strength)5457 )5558 59+// InUse reports whether something subscribes to the input. An idle input,60+// or one Tvheadend keeps tuned for a moment after its last subscription61+// ended, can be had by any subscription.62+func (in InputStatus) InUse() bool { return in.Subs > 0 }63+5664 // Mux names the multiplex and network the input is tuned to, from Stream;5765 // ok is false for an input that isn't on a multiplex.5866 func (in InputStatus) Mux() (mux, network string, ok bool) {
@@ -77,14 +85,28 @@ func (c *Client) Inputs(ctx context.Context) ([]InputStatus, error) {
7785 // Subscription is something using a tuner (or another source) now: a7886 // viewer, a recording, or Tvheadend's own guide and channel scans.7987 type Subscription struct {80- ID int64 `json:"id"`81- Start int64 `json:"start"`82- State string `json:"state"` // "Running", "Testing", "Bad"...83- Title string `json:"title"` // "epggrab", "DVR: ...", the HTTP client88+ ID int64 `json:"id"`89+ Start int64 `json:"start"`90+ // State is "Testing" while Tvheadend waits for data (a tuner that91+ // can't lock stays there), "Running" once data arrives, "Bad" when92+ // the service failed.93+ State string `json:"state"`94+ Title string `json:"title"` // "epggrab", "DVR: ...", "HTTP"8495 Service string `json:"service"` // "<input>/<network>/<mux>/<service>"8596 Errors int `json:"errors"`97+ // TotalIn counts the bytes it has received.98+ TotalIn int64 `json:"total_in"`99+}100+101+// On reports whether the subscription is to the multiplex mux of network102+// on input.103+func (s Subscription) On(input, network, mux string) bool {104+ return strings.HasPrefix(s.Service, input+"/"+network+"/"+mux+"/")86105 }87106 107+// Running reports whether data has arrived for it: the tuner locked.108+func (s Subscription) Running() bool { return s.State == "Running" }109+88110 // Subscriptions lists what is using Tvheadend's inputs now.89111 func (c *Client) Subscriptions(ctx context.Context) ([]Subscription, error) {90112 var body struct {
M internal/tvh/status_test.go +39 -1
@@ -86,6 +86,35 @@ func TestLineup(t *testing.T) {
8686 }8787 }8888 89+// TestInputsIdle reads the inputs as a real Tvheadend 4.3 lists them when90+// nothing uses them: listed, with no stream and no subscriptions.91+func TestInputsIdle(t *testing.T) {92+ var body struct {93+ Entries []tvh.InputStatus `json:"entries"`94+ }95+ if err := json.Unmarshal(tvhtest.Raw(t, "status_inputs_idle.json"), &body); err != nil {96+ t.Fatal(err)97+ }98+ if len(body.Entries) != 2 {99+ t.Fatalf("entries = %d", len(body.Entries))100+ }101+ for _, in := range body.Entries {102+ if _, _, ok := in.Mux(); ok || in.InUse() || in.Weight != 0 || in.Input == "" {103+ t.Errorf("idle input %+v", in)104+ }105+ }106+ var subs struct {107+ Entries []tvh.Subscription `json:"entries"`108+ }109+ if err := json.Unmarshal(tvhtest.Raw(t, "status_subscriptions_measuring.json"), &subs); err != nil {110+ t.Fatal(err)111+ }112+ if len(subs.Entries) != 2 || !subs.Entries[0].Running() || subs.Entries[1].Running() || subs.Entries[1].State != "Testing" ||113+ !subs.Entries[1].On(tvhtest.Tuner0, tvh.ATSCNetwork, "177MHz") || subs.Entries[0].TotalIn == 0 {114+ t.Errorf("subscriptions %+v", subs.Entries)115+ }116+}117+89118 // TestFrontendsFromTree reads the tuners, their names as the input status90119 // gives them and the model, from the hardware tree as Tvheadend 4.3 sends91120 // it, class and parameters included, so without a lookup per node.
@@ -131,9 +160,18 @@ func TestOpenMux(t *testing.T) {
131160 t.Fatalf("read %x, %v", buf[:1], err)132161 }133162 ins, err := c.Inputs(ctx)134- if err != nil || len(ins) != 1 || ins[0].Stream != "575MHz in Airwaves antenna" || ins[0].Weight != 10 || ins[0].BPS == 0 {163+ if err != nil || len(ins) != 2 || ins[0].Stream != "575MHz in Airwaves antenna" || ins[0].Weight != 10 || !ins[0].InUse() ||164+ ins[1].InUse() || ins[1].Stream != "" {135165 t.Fatalf("inputs %+v, %v", ins, err)136166 }167+ // Locked by the next look.168+ if _, err := c.Inputs(ctx); err != nil {169+ t.Fatal(err)170+ }171+ subs, err := c.Subscriptions(ctx)172+ if err != nil || len(subs) != 1 || !subs[0].Running() || !subs[0].On(tvhtest.Tuner0, tvh.ATSCNetwork, "575MHz") || subs[0].On(tvhtest.Tuner1, tvh.ATSCNetwork, "575MHz") {173+ t.Fatalf("subscriptions %+v, %v", subs, err)174+ }137175 body.Close()138176 deadline := time.Now().Add(2 * time.Second)139177 for len(srv.Streams()) > 0 && time.Now().Before(deadline) {
A internal/tvh/tvhtest/testdata/status_inputs_idle.json +43 -0
@@ -0,0 +1,43 @@
1+{2+ "entries": [3+ {4+ "uuid": "966aa49ec18feb963e17f7fdb1b78db0",5+ "input": "HDHomeRun ATSC-T Tuner #0 (169.254.1.2)",6+ "subs": 0,7+ "weight": 0,8+ "signal": 0,9+ "signal_scale": 0,10+ "ber": 0,11+ "snr": 0,12+ "snr_scale": 0,13+ "unc": 0,14+ "bps": 0,15+ "te": 43638,16+ "cc": 39293,17+ "ec_bit": 0,18+ "tc_bit": 0,19+ "ec_block": 0,20+ "tc_block": 021+ },22+ {23+ "uuid": "9fa2dd024c6683a288ebb77929612508",24+ "input": "HDHomeRun ATSC-T Tuner #1 (169.254.1.2)",25+ "subs": 0,26+ "weight": 0,27+ "signal": 0,28+ "signal_scale": 0,29+ "ber": 0,30+ "snr": 0,31+ "snr_scale": 0,32+ "unc": 0,33+ "bps": 0,34+ "te": 40476,35+ "cc": 28623,36+ "ec_bit": 0,37+ "tc_bit": 0,38+ "ec_block": 0,39+ "tc_block": 040+ }41+ ],42+ "totalCount": 243+}
A internal/tvh/tvhtest/testdata/status_inputs_measuring.json +63 -0
@@ -0,0 +1,63 @@
1+{2+ "entries": [3+ {4+ "uuid": "0bdd522a23f39bd94dd62fcc63aabe95",5+ "input": "HDHomeRun ATSC-T Tuner #0 (169.254.1.2)",6+ "stream": "177MHz in Airwaves antenna",7+ "subs": 1,8+ "weight": 10,9+ "pids": [10+ 0,11+ 1,12+ 818713+ ],14+ "signal": 53083,15+ "signal_scale": 1,16+ "ber": 0,17+ "snr": 0,18+ "snr_scale": 1,19+ "unc": 0,20+ "bps": 0,21+ "te": 0,22+ "cc": 0,23+ "ec_bit": 0,24+ "tc_bit": 0,25+ "ec_block": 0,26+ "tc_block": 027+ },28+ {29+ "uuid": "f9511eb993ea1c714dfa0c4c258e64bf",30+ "input": "HDHomeRun ATSC-T Tuner #1 (169.254.1.2)",31+ "stream": "575MHz in Airwaves antenna",32+ "subs": 1,33+ "weight": 10,34+ "pids": [35+ 0,36+ 1,37+ 7424,38+ 7425,39+ 7426,40+ 7427,41+ 7680,42+ 7681,43+ 7682,44+ 7683,45+ 818746+ ],47+ "signal": 55049,48+ "signal_scale": 1,49+ "ber": 0,50+ "snr": 43908,51+ "snr_scale": 1,52+ "unc": 0,53+ "bps": 200032,54+ "te": 2715,55+ "cc": 1220,56+ "ec_bit": 0,57+ "tc_bit": 0,58+ "ec_block": 0,59+ "tc_block": 060+ }61+ ],62+ "totalCount": 263+}
A internal/tvh/tvhtest/testdata/status_subscriptions_measuring.json +51 -0
@@ -0,0 +1,51 @@
1+{2+ "entries": [3+ {4+ "id": 582,5+ "start": 1791134354,6+ "errors": 0,7+ "state": "Running",8+ "hostname": "127.0.0.1",9+ "client": "airwavesd",10+ "title": "HTTP",11+ "service": "HDHomeRun ATSC-T Tuner #1 (169.254.1.2)/Airwaves antenna/575MHz/Raw PID Subscription",12+ "pids": [13+ 0,14+ 1,15+ 7424,16+ 7425,17+ 7426,18+ 7427,19+ 7680,20+ 7681,21+ 7682,22+ 7683,23+ 818724+ ],25+ "in": 2256,26+ "out": 2256,27+ "total_in": 4324,28+ "total_out": 432429+ },30+ {31+ "id": 581,32+ "start": 1791134354,33+ "errors": 0,34+ "state": "Testing",35+ "hostname": "127.0.0.1",36+ "client": "airwavesd",37+ "title": "HTTP",38+ "service": "HDHomeRun ATSC-T Tuner #0 (169.254.1.2)/Airwaves antenna/177MHz/Raw PID Subscription",39+ "pids": [40+ 0,41+ 1,42+ 818743+ ],44+ "in": 0,45+ "out": 0,46+ "total_in": 0,47+ "total_out": 048+ }49+ ],50+ "totalCount": 251+}
M internal/tvh/tvhtest/tvhtest.go +90 -34
@@ -34,10 +34,14 @@ const Network = "Airwaves antenna"
3434 3535 // Signal is how a fake tuner receives a multiplex.3636 type Signal struct {37+ // Lock is whether the tuner locks to it, which takes it a status read:38+ // its subscription is "Testing" at first, then "Running".3739 Lock bool3840 // Strength and Quality are percentages, reported on Tvheadend's3941 // relative scale as an HDHomeRun's are.4042 Strength, Quality float6443+ // Errors is how many transport errors each status read finds more.44+ Errors int4145 }4246 4347 // Server is a fake Tvheadend.
@@ -48,9 +52,8 @@ type Server struct {
4852 grids map[string][]map[string]any // by grid: "channel", "mux", "service", "network"4953 tree map[string][]map[string]any // hardware tree by parent UUID5054 classes map[string]string51- inputs []map[string]any // inputs busy with something else52- subs []map[string]any53- signals map[string]Signal // by multiplex name55+ busy map[string]*stream // inputs busy with something else, by name56+ signals map[string]Signal // by multiplex name5457 open map[int]*stream5558 nextID int5659 requests []string
@@ -70,6 +73,8 @@ type stream struct {
7073 input string7174 mux string7275 weight int76+ title string77+ polls int // input status reads since it tuned7378 done chan struct{}7479 once sync.Once7580 }
@@ -84,7 +89,7 @@ func New(t testing.TB) *Server {
8489 t.Helper()8590 s := &Server{8691 grids: map[string][]map[string]any{}, tree: map[string][]map[string]any{}, classes: map[string]string{},87- signals: map[string]Signal{}, open: map[int]*stream{}, Tuners: 2,92+ signals: map[string]Signal{}, open: map[int]*stream{}, busy: map[string]*stream{}, Tuners: 2,8893 }8994 for grid, file := range map[string]string{"channel": "channel_grid.json", "mux": "mux_grid.json", "service": "service_grid.json", "network": "network_grid.json"} {9095 var body struct {
@@ -167,19 +172,23 @@ func (s *Server) SetSignal(mux string, sig Signal) {
167172 }168173 169174 // Busy has an input tuned to a multiplex for something other than170-// Airwaves' measuring: a viewer, a recording or a guide scan, at weight.175+// Airwaves' measuring: a viewer or a recording, or with a weight below 10176+// Tvheadend's guide scan.171177 func (s *Server) Busy(input, mux string, weight int) {172178 s.mu.Lock()173179 defer s.mu.Unlock()174- s.inputs = slices.DeleteFunc(s.inputs, func(in map[string]any) bool { return in["input"] == input })175- s.inputs = append(s.inputs, s.inputStatus(input, mux, weight))180+ title := "HTTP"181+ if weight < 10 {182+ title = "epggrab"183+ }184+ s.busy[input] = &stream{input: input, mux: mux, weight: weight, title: title, polls: 5}176185 }177186 178187 // Free ends what Busy started on an input.179188 func (s *Server) Free(input string) {180189 s.mu.Lock()181190 defer s.mu.Unlock()182- s.inputs = slices.DeleteFunc(s.inputs, func(in map[string]any) bool { return in["input"] == input })191+ delete(s.busy, input)183192 }184193 185194 // TakeOver has a viewer take every tuner a measuring stream holds, as a
@@ -253,19 +262,57 @@ func (s *Server) Opened() ([]string, int) {
253262 return slices.Clone(s.streams), s.maxOpen254263 }255264 256-// inputStatus is an input's status on a multiplex, as /api/status/inputs257-// writes it. The caller holds mu.258-func (s *Server) inputStatus(input, mux string, weight int) map[string]any {259- sig := s.signals[mux]260- bps := 0261- if sig.Lock {262- bps = 231616265+// tuned lists what the inputs are tuned to, by input. The caller holds266+// mu.267+func (s *Server) tuned() map[string]*stream {268+ out := maps.Clone(s.busy)269+ for _, st := range s.open {270+ out[st.input] = st271+ }272+ return out273+}274+275+// inputStatus is an input's status, as /api/status/inputs writes it: idle,276+// with no stream; or on a multiplex, where a lock shows as data arriving277+// (though the rate reads 0 now and then, as a real one's does with little278+// to send) and the quality reads 0 for the first read after tuning. The279+// caller holds mu.280+func (s *Server) inputStatus(input string, st *stream) map[string]any {281+ uuid := "fe-" + strings.Fields(input)[3]282+ if st == nil {283+ return map[string]any{284+ "uuid": uuid, "input": input, "subs": 0, "weight": 0, "signal": 0, "signal_scale": 0, "snr": 0, "snr_scale": 0,285+ "ber": 0, "unc": 0, "bps": 0, "te": 43638, "cc": 39293, "ec_bit": 0, "tc_bit": 0, "ec_block": 0, "tc_block": 0,286+ }287+ }288+ st.polls++289+ sig := s.signals[st.mux]290+ bps, snr := 0, int(sig.Quality*655.35)291+ if sig.Lock && st.polls%3 != 0 {292+ bps = 157920293+ }294+ if st.polls == 1 {295+ snr = 0263296 }264297 return map[string]any{265- "uuid": fmt.Sprintf("mmi-%s-%s", strings.Fields(input)[3], mux), "input": input, "stream": mux + " in " + Network,266- "subs": 1, "weight": weight, "pids": []int{0},267- "signal": int(sig.Strength * 655.35), "signal_scale": 1, "snr": int(sig.Quality * 655.35), "snr_scale": 1,268- "ber": 0, "unc": 0, "bps": bps, "te": 0, "cc": 0, "ec_bit": 0, "tc_bit": 0, "ec_block": 0, "tc_block": 0,298+ "uuid": fmt.Sprintf("mmi-%s-%s", strings.Fields(input)[3], st.mux), "input": input, "stream": st.mux + " in " + Network,299+ "subs": 1, "weight": st.weight, "pids": []int{0},300+ "signal": int(sig.Strength * 655.35), "signal_scale": 1, "snr": snr, "snr_scale": 1,301+ "ber": 0, "unc": 0, "bps": bps, "te": st.polls * sig.Errors, "cc": 0, "ec_bit": 0, "tc_bit": 0, "ec_block": 0, "tc_block": 0,302+ }303+}304+305+// subscription is what /api/status/subscriptions writes for a stream:306+// "Running" once data arrives, "Testing" while none does. The caller holds307+// mu.308+func (s *Server) subscription(id int, st *stream) map[string]any {309+ state := "Testing"310+ if s.signals[st.mux].Lock && st.polls >= 2 {311+ state = "Running"312+ }313+ return map[string]any{314+ "id": id, "start": 1791134354, "errors": 0, "state": state, "title": st.title,315+ "service": st.input + "/" + Network + "/" + st.mux + "/Raw PID Subscription", "in": 0, "out": 0, "total_in": 0, "total_out": 0,269316 }270317 }271318
@@ -291,15 +338,21 @@ func (s *Server) serve(w http.ResponseWriter, r *http.Request) {
291338 s.writeGrid(w, "network")292339 case path == "/api/status/inputs":293340 s.mu.Lock()294- list := slices.Clone(s.inputs)295- for _, st := range s.open {296- list = append(list, s.inputStatus(st.input, st.mux, st.weight))341+ var list []map[string]any342+ tuned := s.tuned()343+ for _, name := range []string{Tuner0, Tuner1}[:min(s.Tuners, 2)] {344+ list = append(list, s.inputStatus(name, tuned[name]))297345 }298346 s.mu.Unlock()299347 writeJSON(w, map[string]any{"entries": list, "totalCount": len(list)})300348 case path == "/api/status/subscriptions":301349 s.mu.Lock()302- list := slices.Clone(s.subs)350+ list := []map[string]any{}351+ for i, name := range []string{Tuner0, Tuner1} {352+ if st := s.tuned()[name]; st != nil {353+ list = append(list, s.subscription(500+i, st))354+ }355+ }303356 s.mu.Unlock()304357 writeJSON(w, map[string]any{"entries": list, "totalCount": len(list)})305358 case path == "/api/hardware/tree":
@@ -450,11 +503,8 @@ func (s *Server) stream(w http.ResponseWriter, r *http.Request, mux string) {
450503 weight, _ := strconv.Atoi(r.URL.Query().Get("weight"))451504 s.mu.Lock()452505 used := map[string]bool{}453- for _, in := range s.inputs {454- used[in["input"].(string)] = true455- }456- for _, st := range s.open {457- used[st.input] = true506+ for name := range s.tuned() {507+ used[name] = true458508 }459509 input := ""460510 for _, name := range []string{Tuner0, Tuner1}[:min(s.Tuners, 2)] {
@@ -468,7 +518,7 @@ func (s *Server) stream(w http.ResponseWriter, r *http.Request, mux string) {
468518 http.Error(w, "No input source available", http.StatusServiceUnavailable)469519 return470520 }471- st := &stream{input: input, mux: mux, weight: weight, done: make(chan struct{})}521+ st := &stream{input: input, mux: mux, weight: weight, title: "HTTP", done: make(chan struct{})}472522 s.nextID++473523 id := s.nextID474524 s.open[id] = st
@@ -488,6 +538,14 @@ func (s *Server) stream(w http.ResponseWriter, r *http.Request, mux string) {
488538 s.mu.Unlock()489539 }()490540 541+ // Without a lock Tvheadend sends nothing, headers included.542+ if !lock {543+ select {544+ case <-r.Context().Done():545+ case <-st.done:546+ }547+ return548+ }491549 w.Header().Set("Content-Type", "video/mp2t")492550 w.WriteHeader(http.StatusOK)493551 w.(http.Flusher).Flush()
@@ -502,12 +560,10 @@ func (s *Server) stream(w http.ResponseWriter, r *http.Request, mux string) {
502560 case <-st.done:503561 return504562 case <-tick.C:505- if lock {506- if _, err := w.Write(packet); err != nil {507- return508- }509- w.(http.Flusher).Flush()563+ if _, err := w.Write(packet); err != nil {564+ return510565 }566+ w.(http.Flusher).Flush()511567 }512568 }513569 }