WIP

internal/trending/item.go

@@ -5,23 +5,23 @@ import (
 	"time"
 )
 
-type item struct {
+type item[K comparable] struct {
 	eventSeries TimeSeries
 	maxSeries   SlidingWindow
 
 	max     float64
 	maxTime time.Time
-	options *options
+	options *options[K]
 
 	// TODO: move outside of item because it's the same for all items
 	defaultExpectation float64
 	defaultHourlyCount float64
 }
 
-func newItem(id string, options *options) *item {
+func newItem[K comparable](id K, options *options[K]) *item[K] {
 	defaultHourlyCount := float64(options.baseCount) * float64(options.storageDuration/time.Hour)
 	defaultExpectation := float64(options.baseCount) / float64(time.Hour/options.recentDuration)
-	return &item{
+	return &item[K]{
 		eventSeries: options.creator(id),
 		maxSeries:   options.slidingWindowCreator(id),
 		options:     options,
@@ -31,10 +31,10 @@ func newItem(id string, options *options) *item {
 	}
 }
 
-func (i *item) score() score {
+func (i *item[K]) score() score[K] {
 	recentCount, count := i.computeCounts()
 	if recentCount < i.options.countThreshold {
-		return score{}
+		return score[K]{}
 	}
 	if recentCount == count {
 		// we see this for the first time so there is no historical data
@@ -60,7 +60,7 @@ func (i *item) score() score {
 
 	mixedScore := 0.5 * (klScore + i.max)
 
-	return score{
+	return score[K]{
 		Score:       mixedScore,
 		Probability: probability,
 		Expectation: expectation,
@@ -69,27 +69,27 @@ func (i *item) score() score {
 	}
 }
 
-func (i *item) computeCounts() (float64, float64) {
+func (i *item[K]) computeCounts() (float64, float64) {
 	now := time.Now()
 	totalCount, _ := i.eventSeries.Range(now.Add(-i.options.storageDuration), now)
 	count, _ := i.eventSeries.Range(now.Add(-i.options.recentDuration), now)
 	return count, totalCount
 }
 
-func (i *item) computeRecentMax() float64 {
+func (i *item[K]) computeRecentMax() float64 {
 	return i.maxSeries.Max()
 }
 
-func (i *item) decayMax() {
+func (i *item[K]) decayMax() {
 	i.updateMax(i.max * i.computeExponentialDecayMultiplier())
 }
 
-func (i *item) updateMax(score float64) {
+func (i *item[K]) updateMax(score float64) {
 	i.max = score
 	i.maxTime = time.Now()
 }
 
-func (i *item) computeExponentialDecayMultiplier() float64 {
+func (i *item[K]) computeExponentialDecayMultiplier() float64 {
 	return math.Pow(0.5, float64(time.Now().Unix()-i.maxTime.Unix())/i.options.halfLife.Seconds())
 }
 

internal/trending/score.go

@@ -1,7 +1,7 @@
 package trending
 
-type score struct {
+type score[K comparable] struct {
-	ID          string
+	ID          K
 	Score       float64
 	Probability float64
 	Expectation float64
@@ -9,28 +9,28 @@ type score struct {
 	KLScore     float64
 }
 
-type Scores []score
+type Scores[K comparable] []score[K]
 
-func (s Scores) Len() int {
+func (s Scores[K]) Len() int {
 	return len(s)
 }
 
-func (s Scores) Swap(i, j int) {
+func (s Scores[K]) Swap(i, j int) {
 	s[i], s[j] = s[j], s[i]
 }
 
-func (s Scores) Less(i, j int) bool {
+func (s Scores[K]) Less(i, j int) bool {
 	return s[i].Score > s[j].Score
 }
 
-func (s Scores) take(count int) Scores {
+func (s Scores[K]) take(count int) Scores[K] {
 	if count >= len(s) {
 		return s
 	}
 	return s[0 : count-1]
 }
 
-func (s Scores) threshold(t float64) Scores {
+func (s Scores[K]) threshold(t float64) Scores[K] {
 	for i := range s {
 		if s[i].Score < t {
 			return s[0:i]

internal/trending/trending.go

@@ -29,9 +29,9 @@ var defaultBaseCount = 3
 var defaultScoreThreshold = 0.01
 var defaultCountThreshold = 3.0
 
-type options struct {
+type options[K comparable] struct {
-	creator              TimeSeriesCreator
+	creator              TimeSeriesCreator[K]
-	slidingWindowCreator SlidingWindowCreator
+	slidingWindowCreator SlidingWindowCreator[K]
 
 	halfLife time.Duration
 
@@ -44,59 +44,59 @@ type options struct {
 	countThreshold float64
 }
 
-type Option func(*options)
+type Option[K comparable] func(*options[K])
 
-func WithTimeSeries(creator TimeSeriesCreator) Option {
+func WithTimeSeries[K comparable](creator TimeSeriesCreator[K]) Option[K] {
-	return func(o *options) {
+	return func(o *options[K]) {
 		o.creator = creator
 	}
 }
 
-func WithSlidingWindow(creator SlidingWindowCreator) Option {
+func WithSlidingWindow[K comparable](creator SlidingWindowCreator[K]) Option[K] {
-	return func(o *options) {
+	return func(o *options[K]) {
 		o.slidingWindowCreator = creator
 	}
 }
 
-func WithHalfLife(halfLife time.Duration) Option {
+func WithHalfLife[K comparable](halfLife time.Duration) Option[K] {
-	return func(o *options) {
+	return func(o *options[K]) {
 		o.halfLife = halfLife
 	}
 }
 
-func WithRecentDuration(recentDuration time.Duration) Option {
+func WithRecentDuration[K comparable](recentDuration time.Duration) Option[K] {
-	return func(o *options) {
+	return func(o *options[K]) {
 		o.recentDuration = recentDuration
 	}
 }
 
-func WithStorageDuration(storageDuration time.Duration) Option {
+func WithStorageDuration[K comparable](storageDuration time.Duration) Option[K] {
-	return func(o *options) {
+	return func(o *options[K]) {
 		o.storageDuration = storageDuration
 	}
 }
 
-func WithMaxResults(maxResults int) Option {
+func WithMaxResults[K comparable](maxResults int) Option[K] {
-	return func(o *options) {
+	return func(o *options[K]) {
 		o.maxResults = maxResults
 	}
 }
 
-func WithScoreThreshold(threshold float64) Option {
+func WithScoreThreshold[K comparable](threshold float64) Option[K] {
-	return func(o *options) {
+	return func(o *options[K]) {
 		o.scoreThreshold = threshold
 	}
 }
 
-func WithCountThreshold(threshold float64) Option {
+func WithCountThreshold[K comparable](threshold float64) Option[K] {
-	return func(o *options) {
+	return func(o *options[K]) {
 		o.countThreshold = threshold
 	}
 }
 
-type Scorer struct {
+type Scorer[K comparable] struct {
-	options options
+	options options[K]
-	items   map[string]*item
+	items   map[K]*item[K]
 }
 
 type SlidingWindow interface {
@@ -104,16 +104,16 @@ type SlidingWindow interface {
 	Max() float64
 }
 
-type SlidingWindowCreator func(string) SlidingWindow
+type SlidingWindowCreator[K comparable] func(K) SlidingWindow
 
 type TimeSeries interface {
 	IncreaseAtTime(amount int, time time.Time)
 	Range(start, end time.Time) (float64, error)
 }
 
-type TimeSeriesCreator func(string) TimeSeries
+type TimeSeriesCreator[K comparable] func(K) TimeSeries
 
-func NewMemoryTimeSeries(id string) TimeSeries {
+func NewMemoryTimeSeries[K comparable](id K) TimeSeries {
 	ts, _ := timeseries.NewTimeSeries(timeseries.WithGranularities(
 		[]timeseries.Granularity{
 			{Granularity: time.Second, Count: 60},
@@ -125,13 +125,13 @@ func NewMemoryTimeSeries(id string) TimeSeries {
 	return ts
 }
 
-func NewScorer(options ...Option) Scorer {
+func NewScorer[K comparable](options ...Option[K]) Scorer[K] {
-	scorer := Scorer{items: make(map[string]*item)}
+	scorer := Scorer[K]{items: make(map[K]*item[K])}
 	for _, o := range options {
 		o(&scorer.options)
 	}
 	if scorer.options.creator == nil {
-		scorer.options.creator = NewMemoryTimeSeries
+		scorer.options.creator = NewMemoryTimeSeries[K]
 	}
 	if scorer.options.halfLife == 0 {
 		scorer.options.halfLife = defaultHalfLife
@@ -155,7 +155,7 @@ func NewScorer(options ...Option) Scorer {
 		scorer.options.baseCount = defaultBaseCount
 	}
 	if scorer.options.slidingWindowCreator == nil {
-		scorer.options.slidingWindowCreator = func(id string) SlidingWindow {
+		scorer.options.slidingWindowCreator = func(id K) SlidingWindow {
 			return slidingwindow.NewSlidingWindow(
 				slidingwindow.WithStep(time.Hour*24),
 				slidingwindow.WithDuration(scorer.options.storageDuration),
@@ -165,7 +165,7 @@ func NewScorer(options ...Option) Scorer {
 	return scorer
 }
 
-func (s *Scorer) AddEvent(id string, time time.Time) {
+func (s *Scorer[K]) AddEvent(id K, time time.Time) {
 	item := s.items[id]
 	if item == nil {
 		item = newItem(id, &s.options)
@@ -174,12 +174,12 @@ func (s *Scorer) AddEvent(id string, time time.Time) {
 	s.addToItem(item, time)
 }
 
-func (s *Scorer) addToItem(item *item, time time.Time) {
+func (s *Scorer[K]) addToItem(item *item[K], time time.Time) {
 	item.eventSeries.IncreaseAtTime(1, time)
 }
 
-func (s *Scorer) Score() Scores {
+func (s *Scorer[K]) Score() Scores[K] {
-	var scores Scores
+	var scores Scores[K]
 	for id, item := range s.items {
 		score := item.score()
 		score.ID = id

pkg/gordio/server/server.go

@@ -11,6 +11,7 @@ import (
 	"dynatron.me/x/stillbox/pkg/gordio/database"
 	"dynatron.me/x/stillbox/pkg/gordio/nexus"
 	"dynatron.me/x/stillbox/pkg/gordio/sinks"
+	"dynatron.me/x/stillbox/pkg/gordio/sinks/alerting"
 	"dynatron.me/x/stillbox/pkg/gordio/sources"
 	"github.com/go-chi/chi/v5"
 	"github.com/go-chi/chi/v5/middleware"
@@ -43,6 +44,8 @@ func New(ctx context.Context, cfg *config.Config) (*Server, error) {
 		return nil, err
 	}
 
+	ctx = database.CtxWithDB(ctx, db)
+
 	r := chi.NewRouter()
 	authenticator := auth.NewAuthenticator(cfg.Auth)
 	srv := &Server{
@@ -56,6 +59,7 @@ func New(ctx context.Context, cfg *config.Config) (*Server, error) {
 
 	srv.sinks.Register("database", sinks.NewDatabaseSink(srv.db), true)
 	srv.sinks.Register("nexus", sinks.NewNexusSink(srv.nex), false)
+	srv.sinks.Register("alerting", alerting.NewSink(ctx), false)
 	srv.sources.Register("rdio-http", sources.NewRdioHTTP(authenticator, srv))
 
 	r.Use(middleware.RequestID)

pkg/gordio/sinks/alerting/alerting.go

@@ -0,0 +1,122 @@
+package alerting
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	cl "dynatron.me/x/stillbox/pkg/calls"
+	"dynatron.me/x/stillbox/pkg/gordio/database"
+
+	"dynatron.me/x/stillbox/internal/timeseries"
+	"dynatron.me/x/stillbox/internal/trending"
+
+	"github.com/rs/zerolog/log"
+)
+
+const (
+	StorageLookbackDays = 2
+	HalfLife            = time.Hour
+	RecentDuration      = time.Hour
+)
+
+type AlertSink struct {
+	sync.RWMutex
+	scorer trending.Scorer[cl.Talkgroup]
+}
+
+func NewSink(ctx context.Context) *AlertSink {
+	as := &AlertSink{
+		scorer: trending.NewScorer[cl.Talkgroup](
+			trending.WithTimeSeries(newTimeSeries),
+			trending.WithStorageDuration[cl.Talkgroup](StorageLookbackDays*24*time.Hour),
+			trending.WithRecentDuration[cl.Talkgroup](RecentDuration),
+			trending.WithHalfLife[cl.Talkgroup](HalfLife),
+		),
+	}
+
+	go as.startBackfill(ctx)
+
+	return as
+}
+
+func (as *AlertSink) startBackfill(ctx context.Context) {
+	since := time.Now().Add(StorageLookbackDays * -24 * time.Hour)
+	log.Debug().Time("since", since).Msg("starting stats backfill")
+	count, err := as.backfill(ctx, since)
+	if err != nil {
+		log.Error().Err(err).Msg("backfill failed")
+		return
+	}
+	log.Debug().Int("count", count).Int("len", as.scorer.Score().Len()).Msg("backfill finished")
+	as.printScores()
+}
+
+type score[K comparable] struct {
+	ID          K
+	Score       float64
+	Probability float64
+	Expectation float64
+	Maximum     float64
+	KLScore     float64
+}
+
+func (as *AlertSink) printScores() {
+	scores := as.scorer.Score()
+	fmt.Printf("score len is %d\n", scores.Len())
+	for _, s := range scores {
+		fmt.Printf("%d:%d score %f prob %f exp %f max %f kl %f", s.ID.System, s.ID.Talkgroup, s.Score,
+			s.Probability, s.Expectation, s.Maximum, s.KLScore)
+	}
+}
+
+func (as *AlertSink) backfill(ctx context.Context, since time.Time) (count int, err error) {
+	db := database.FromCtx(ctx)
+	const backfillStatsQuery = `SELECT system, talkgroup, call_date FROM calls WHERE call_date > $1`
+
+	rows, err := db.Query(ctx, backfillStatsQuery, since)
+	if err != nil {
+		return count, err
+	}
+	defer rows.Close()
+
+	as.Lock()
+	defer as.Unlock()
+
+	for rows.Next() {
+		var tg cl.Talkgroup
+		var callDate time.Time
+		if err := rows.Scan(&tg.System, &tg.Talkgroup, &callDate); err != nil {
+			return count, err
+		}
+		as.scorer.AddEvent(tg, callDate)
+		count++
+	}
+
+	if err := rows.Err(); err != nil {
+		return count, err
+	}
+
+	return count, nil
+}
+
+func newTimeSeries(id cl.Talkgroup) trending.TimeSeries {
+	ts, _ := timeseries.NewTimeSeries(timeseries.WithGranularities(
+		[]timeseries.Granularity{
+			{Granularity: time.Second, Count: 60},
+			{Granularity: time.Minute, Count: 10},
+			{Granularity: time.Hour, Count: 24},
+			{Granularity: time.Hour * 24, Count: StorageLookbackDays},
+		},
+	))
+	return ts
+}
+
+func (as *AlertSink) SinkType() string {
+	return "alerting"
+}
+
+func (ns *AlertSink) Call(ctx context.Context, call *cl.Call) error {
+	return nil
+}