amigan/stillbox
1
0
Fork 0
Code Issues 16 Pull requests Projects Releases Packages Wiki Activity Actions

Scale scores during score operation. #25

Merged
amigan merged 1 commit from scorescale into trunk 2024-11-03 19:13:28 -05:00
Conversation 0 Commits 1 Files changed 9 +70 -23
9 changed files with 70 additions and 23 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
internal/trending/item.go
View file

@ -1,6 +1,7 @@
package trending
import (
"context"
"math"
"time"
)
@ -31,7 +32,7 @@ func newItem[K comparable](id K, options *options[K]) *item[K] {
}
}
func (i *item[K]) score() Score[K] {
func (i *item[K]) score(ctx context.Context, id K) Score[K] {
recentCount, count := i.computeCounts()
if recentCount < i.options.countThreshold {
return Score[K]{}
@ -58,7 +59,7 @@ func (i *item[K]) score() Score[K] {
}
i.decayMax()
mixedScore := 5 * (klScore + i.max)
mixedScore := 5 * (klScore + i.max) * i.options.weigher.Weight(ctx, id, i.options.clock.Now())
return Score[K]{
Score: mixedScore,

25
internal/trending/trending.go
View file

@ -1,6 +1,7 @@
package trending
import (
"context"
"sort"
"time"
@ -33,6 +34,7 @@ type options[K comparable] struct {
creator TimeSeriesCreator[K]
slidingWindowCreator SlidingWindowCreator[K]
clock timeseries.Clock
weigher Weigher[K]
halfLife time.Duration
@ -45,8 +47,24 @@ type options[K comparable] struct {
countThreshold float64
}
type Weigher[K comparable] interface {
Weight(context.Context, K, time.Time) float64
}
type Option[K comparable] func(*options[K])
type unityWeigher[K comparable] float64
func (u unityWeigher[K]) Weight(_ context.Context, _ K, _ time.Time) float64 {
return float64(u)
}
func WithWeigher[K comparable](w Weigher[K]) Option[K] {
return func(o *options[K]) {
o.weigher = w
}
}
func WithTimeSeries[K comparable](creator TimeSeriesCreator[K]) Option[K] {
return func(o *options[K]) {
o.creator = creator
@ -164,6 +182,9 @@ func NewScorer[K comparable](options ...Option[K]) Scorer[K] {
if scorer.options.clock == nil {
scorer.options.clock = timeseries.DefaultClock
}
if scorer.options.weigher == nil {
scorer.options.weigher = unityWeigher[K](1.0)
}
if scorer.options.slidingWindowCreator == nil {
scorer.options.slidingWindowCreator = func(id K) SlidingWindow {
return slidingwindow.NewSlidingWindow(
@ -189,10 +210,10 @@ func (s *Scorer[K]) addToItem(item *item[K], tm time.Time) {
item.eventSeries.IncreaseAtTime(1, tm)
}
func (s *Scorer[K]) Score() Scores[K] {
func (s *Scorer[K]) Score(ctx context.Context) Scores[K] {
var scores Scores[K]
for id, item := range s.items {
score := item.score()
score := item.score(ctx, id)
score.ID = id
scores = append(scores, score)
}

15
pkg/alerting/alerting.go
View file

@ -118,6 +118,7 @@ func New(cfg config.Alerting, tgCache talkgroups.Store, opts ...AlertOption) Ale
trending.WithScoreThreshold[talkgroups.ID](ScoreThreshold),
trending.WithCountThreshold[talkgroups.ID](CountThreshold),
trending.WithClock[talkgroups.ID](as.clock),
trending.WithWeigher[talkgroups.ID](as.tgCache),
)
return as
@ -130,13 +131,13 @@ func (as *alerter) Go(ctx context.Context) {
log.Error().Err(err).Msg("backfill")
}
as.score(time.Now())
as.score(ctx, time.Now())
ticker := time.NewTicker(alerterTickInterval)
for {
select {
case now := <-ticker.C:
as.score(now)
as.score(ctx, now)
err := as.notify(ctx)
if err != nil {
log.Error().Err(err).Msg("notify")
@ -178,7 +179,7 @@ func (as *alerter) eval(ctx context.Context, now time.Time, testMode bool) ([]Al
if s.Score > as.cfg.AlertThreshold || testMode {
if old, inCache := as.alertCache[s.ID]; !inCache || now.Sub(old.Timestamp) > as.renotify {
s.Score = as.tgCache.ApplyAlertRules(s, now)
s.Score *= as.tgCache.ApplyAlertRules(s.ID, now)
a, err := as.makeAlert(ctx, s, origScore)
if err != nil {
return nil, fmt.Errorf("makeAlert: %w", err)
@ -385,16 +386,16 @@ func (as *alerter) startBackfill(ctx context.Context) error {
if err != nil {
return err
}
log.Debug().Int("callsCount", count).Str("in", time.Since(now).String()).Int("tgCount", as.scorer.Score().Len()).Msg("backfill finished")
log.Debug().Int("callsCount", count).Str("in", time.Since(now).String()).Int("tgCount", as.scorer.Score(ctx).Len()).Msg("backfill finished")
return nil
}
func (as *alerter) score(now time.Time) {
func (as *alerter) score(ctx context.Context, now time.Time) {
as.Lock()
defer as.Unlock()
as.scores = as.scorer.Score()
as.scores = as.scorer.Score(ctx)
as.lastScore = now
sort.Sort(as.scores)
}
@ -420,7 +421,7 @@ func (as *alerter) backfill(ctx context.Context, since time.Time, until time.Tim
}
as.scorer.AddEvent(tg, callDate)
if as.sim != nil { // step the simulator if it is active
as.sim.stepClock(callDate)
as.sim.stepClock(ctx, callDate)
}
count++
}

6
pkg/alerting/simulate.go
View file

@ -46,12 +46,12 @@ func (s *Simulation) verify() error {
}
// stepClock is called by backfill during simulation operations.
func (s *Simulation) stepClock(t time.Time) {
func (s *Simulation) stepClock(ctx context.Context, t time.Time) {
now := s.clock.Now()
step := t.Sub(s.lastScore)
if step > time.Duration(s.SimInterval) {
s.clock += offsetClock(s.SimInterval)
s.scores = s.scorer.Score()
s.scores = s.scorer.Score(ctx)
s.lastScore = now
}
@ -85,7 +85,7 @@ func (s *Simulation) Simulate(ctx context.Context) (trending.Scores[talkgroups.I
}
// initial score
s.scores = s.scorer.Score()
s.scores = s.scorer.Score(ctx)
s.lastScore = time.Time(s.ScoreStart)
ssT := time.Time(s.ScoreStart)

4
pkg/alerting/stats.html
View file

@ -85,8 +85,8 @@
{{ range .Scores }}
{{ $tg := (index $.TGs .ID) }}
<tr>
<td>{{ $tg.Name_2}}</td>
<td>{{ $tg.Name}}</td>
<td>{{ $tg.System.Name}}</td>
<td>{{ $tg.Talkgroup.Name}}</td>
<td>{{ .ID.Talkgroup }}</td>
<td>{{ f .Count 0 }}</td>
<td>{{ f .RecentCount 0 }}</td>

1
pkg/server/signals.go
View file

@ -17,6 +17,7 @@ func (s *Server) huppers() []hupper {
return []hupper{
s.logger,
s.auth,
s.tgs,
}
}

9
pkg/talkgroups/alertconfig.go
View file

@ -6,7 +6,6 @@ import (
"time"
"dynatron.me/x/stillbox/internal/ruletime"
"dynatron.me/x/stillbox/internal/trending"
)
type AlertConfig struct {
@ -50,15 +49,15 @@ func (ac *AlertConfig) UnmarshalTGRules(tg ID, confBytes []byte) error {
return nil
}
func (ac *AlertConfig) ApplyAlertRules(score trending.Score[ID], t time.Time, coversOpts ...ruletime.CoversOption) float64 {
func (ac *AlertConfig) ApplyAlertRules(id ID, t time.Time, coversOpts ...ruletime.CoversOption) float64 {
ac.RLock()
s, has := ac.m[score.ID]
s, has := ac.m[id]
ac.RUnlock()
if !has {
return score.Score
return 1.0
}
final := score.Score
final := 1.0
for _, ar := range s {
if ar.MatchTime(t, coversOpts...) {

2
pkg/talkgroups/alertconfig_test.go
View file

@ -126,7 +126,7 @@ func TestAlertConfig(t *testing.T) {
ID: tc.tg,
Score: tc.origScore,
}
assert.Equal(t, tc.expectScore, toFixed(ac.ApplyAlertRules(cs, tc.t), 5))
assert.Equal(t, tc.expectScore, toFixed(cs.Score*ac.ApplyAlertRules(cs.ID, tc.t), 5))
})
}
}

26
pkg/talkgroups/cache.go
View file

@ -6,6 +6,7 @@ import (
"sync"
"time"
"dynatron.me/x/stillbox/pkg/config"
"dynatron.me/x/stillbox/pkg/database"
"dynatron.me/x/stillbox/internal/ruletime"
@ -25,7 +26,7 @@ type Store interface {
SystemName(ctx context.Context, id int) (string, bool)
// ApplyAlertRules applies the score's talkgroup alert rules to the call occurring at t and returns the weighted score.
ApplyAlertRules(score trending.Score[ID], t time.Time, coversOpts ...ruletime.CoversOption) float64
ApplyAlertRules(id ID, t time.Time, coversOpts ...ruletime.CoversOption) float64
// Hint hints the Store that the provided talkgroups will be asked for.
Hint(ctx context.Context, tgs []ID) error
@ -35,6 +36,12 @@ type Store interface {
// Invalidate invalidates any caching in the Store.
Invalidate()
// Include the trending Weigher interface
trending.Weigher[ID]
// Hupper
HUP(*config.Config)
}
type CtxStoreKeyT string
@ -54,6 +61,10 @@ func StoreFrom(ctx context.Context) Store {
return s
}
func (t *cache) HUP(_ *config.Config) {
t.Invalidate()
}
func (t *cache) Invalidate() {
t.Lock()
defer t.Unlock()
@ -145,6 +156,19 @@ func (t *cache) Load(ctx context.Context, tgs []int64) error {
var ErrNotFound = errors.New("talkgroup not found")
func (t *cache) Weight(ctx context.Context, id ID, tm time.Time) float64 {
tg, err := t.TG(ctx, id)
if err != nil {
return 1.0
}
m := float64(tg.Weight)
m *= t.AlertConfig.ApplyAlertRules(id, tm)
return float64(tg.Weight)
}
func (t *cache) TG(ctx context.Context, tg ID) (Talkgroup, error) {
t.RLock()
rec, has := t.tgs[tg]