From 6e35f49d2a5cc4cac8b8d6c20cdf88b87f8715e9 Mon Sep 17 00:00:00 2001 From: CarterPerez-dev Date: Sun, 3 May 2026 19:39:41 -0400 Subject: [PATCH] feat(monitor/backend): swpc emits rich space_weather payload (kp + bz_gsm + speed/density + xray flux/class) --- .../internal/collectors/swpc/collector.go | 123 ++++++++++++++++-- .../collectors/swpc/collector_test.go | 55 ++++++++ 2 files changed, 167 insertions(+), 11 deletions(-) diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/swpc/collector.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/swpc/collector.go index 3c9dab9d..b92177ac 100644 --- a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/swpc/collector.go +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/swpc/collector.go @@ -6,7 +6,10 @@ package swpc import ( "context" "encoding/json" + "fmt" "log/slog" + "math" + "strconv" "time" "github.com/carterperez-dev/monitor-the-situation/backend/internal/events" @@ -22,6 +25,11 @@ const ( keyKp = "swpc:kp" keyXray = "swpc:xray" keyAlerts = "swpc:alerts" + + xrayBaseB = 1e-7 + xrayBaseC = 1e-6 + xrayBaseM = 1e-5 + xrayBaseX = 1e-4 ) type Fetcher interface { @@ -58,6 +66,11 @@ type CollectorConfig struct { type Collector struct { cfg CollectorConfig logger *slog.Logger + + latestKp *KpTick + latestPlasma *PlasmaTick + latestMag *MagTick + latestXray *XrayTick } func NewCollector(cfg CollectorConfig) *Collector { @@ -130,16 +143,7 @@ func (c *Collector) tickFast(ctx context.Context) { } if pushed > 0 { - body, _ := json.Marshal(map[string]any{ - "ts": time.Now().UTC(), - "pushed": pushed, - }) - c.cfg.Emitter.Emit(events.Event{ - Topic: events.TopicSpaceWeather, - Timestamp: time.Now().UTC(), - Source: Name, - Payload: json.RawMessage(body), - }) + c.emitCombined(pushed) } if !hadError { @@ -148,10 +152,55 @@ func (c *Collector) tickFast(ctx context.Context) { } func (c *Collector) tickSlow(ctx context.Context) { - if _, err := c.pushKp(ctx); err != nil { + n, err := c.pushKp(ctx) + if err != nil { c.logger.Warn("swpc kp", "err", err) _ = c.cfg.State.RecordError(ctx, Name, err.Error()) + return } + if n > 0 { + c.emitCombined(n) + } +} + +func (c *Collector) emitCombined(pushed int64) { + body, _ := json.Marshal(c.buildPayload(pushed)) + c.cfg.Emitter.Emit(events.Event{ + Topic: events.TopicSpaceWeather, + Timestamp: time.Now().UTC(), + Source: Name, + Payload: json.RawMessage(body), + }) +} + +func (c *Collector) buildPayload(pushed int64) map[string]any { + out := map[string]any{ + "ts": time.Now().UTC(), + "pushed": pushed, + } + if c.latestKp != nil { + out["kp"] = c.latestKp.Kp + } + if c.latestPlasma != nil { + if v, ok := parseSWPCFloat(c.latestPlasma.Speed); ok { + out["speed_kms"] = v + } + if v, ok := parseSWPCFloat(c.latestPlasma.Density); ok { + out["density"] = v + } + } + if c.latestMag != nil { + if v, ok := parseSWPCFloat(c.latestMag.BzGSM); ok { + out["bz_gsm"] = v + } + } + if c.latestXray != nil && c.latestXray.Flux > 0 { + out["xray_flux"] = c.latestXray.Flux + if cls := classifyXray(c.latestXray.Flux); cls != "" { + out["xray_class"] = cls + } + } + return out } func (c *Collector) pushPlasma(ctx context.Context) (int64, error) { @@ -159,6 +208,9 @@ func (c *Collector) pushPlasma(ctx context.Context) (int64, error) { if err != nil { return 0, err } + if latest := lastNonZero(rows, func(r PlasmaTick) bool { return !r.TimeTag.IsZero() }); latest != nil { + c.latestPlasma = latest + } return pushAll(ctx, c.cfg.Ring, keyPlasma, rows, func(r PlasmaTick) int64 { return r.TimeTag.UnixMilli() }) } @@ -167,6 +219,9 @@ func (c *Collector) pushMag(ctx context.Context) (int64, error) { if err != nil { return 0, err } + if latest := lastNonZero(rows, func(r MagTick) bool { return !r.TimeTag.IsZero() }); latest != nil { + c.latestMag = latest + } return pushAll(ctx, c.cfg.Ring, keyMag, rows, func(r MagTick) int64 { return r.TimeTag.UnixMilli() }) } @@ -175,6 +230,9 @@ func (c *Collector) pushKp(ctx context.Context) (int64, error) { if err != nil { return 0, err } + if latest := lastNonZero(rows, func(r KpTick) bool { return !r.TimeTag.IsZero() }); latest != nil { + c.latestKp = latest + } return pushAll(ctx, c.cfg.Ring, keyKp, rows, func(r KpTick) int64 { return r.TimeTag.UnixMilli() }) } @@ -183,6 +241,9 @@ func (c *Collector) pushXray(ctx context.Context) (int64, error) { if err != nil { return 0, err } + if latest := lastNonZero(rows, func(r XrayTick) bool { return !r.TimeTag.IsZero() }); latest != nil { + c.latestXray = latest + } return pushAll(ctx, c.cfg.Ring, keyXray, rows, func(r XrayTick) int64 { return r.TimeTag.UnixMilli() }) } @@ -209,3 +270,43 @@ func pushAll[T any](ctx context.Context, ring Ring, key string, rows []T, score } return pushed, nil } + +func lastNonZero[T any](rows []T, ok func(T) bool) *T { + for i := len(rows) - 1; i >= 0; i-- { + if ok(rows[i]) { + r := rows[i] + return &r + } + } + return nil +} + +func parseSWPCFloat(s string) (float64, bool) { + if s == "" { + return 0, false + } + f, err := strconv.ParseFloat(s, 64) + if err != nil || math.IsNaN(f) { + return 0, false + } + return f, true +} + +func classifyXray(flux float64) string { + if flux <= 0 { + return "" + } + if flux >= xrayBaseX { + return fmt.Sprintf("X%.1f", flux/xrayBaseX) + } + if flux >= xrayBaseM { + return fmt.Sprintf("M%.1f", flux/xrayBaseM) + } + if flux >= xrayBaseC { + return fmt.Sprintf("C%.1f", flux/xrayBaseC) + } + if flux >= xrayBaseB { + return fmt.Sprintf("B%.1f", flux/xrayBaseB) + } + return "" +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/swpc/collector_test.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/swpc/collector_test.go index 11bedc5f..1a950a13 100644 --- a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/swpc/collector_test.go +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/swpc/collector_test.go @@ -5,6 +5,7 @@ package swpc_test import ( "context" + "encoding/json" "errors" "sync" "testing" @@ -135,6 +136,60 @@ func TestCollector_FastTickPushesToRingsAndEmits(t *testing.T) { require.Equal(t, 0, st.failures) } +func TestCollector_EmitsRichPayloadWithLatestReadings(t *testing.T) { + now := time.Now().UTC() + ftch := &fakeFetcher{ + plasma: []swpc.PlasmaTick{ + {TimeTag: now.Add(-2 * time.Minute), Density: "1.0", Speed: "300"}, + {TimeTag: now, Density: "5.42", Speed: "487"}, + }, + mag: []swpc.MagTick{ + {TimeTag: now.Add(-2 * time.Minute), BzGSM: "1.5"}, + {TimeTag: now, BzGSM: "-3.21"}, + }, + xray: []swpc.XrayTick{ + {TimeTag: now.Add(-2 * time.Minute), Flux: 1e-7}, + {TimeTag: now, Flux: 2.5e-5}, + }, + alerts: []swpc.AlertItem{}, + kp: []swpc.KpTick{{TimeTag: now, Kp: 4.0}}, + } + ring := &fakeRing{} + emt := &fakeEmitter{} + st := &recordingState{} + + c := swpc.NewCollector(swpc.CollectorConfig{ + FastInterval: 20 * time.Millisecond, + SlowInterval: 25 * time.Millisecond, + Fetcher: ftch, + Ring: ring, + Emitter: emt, + State: st, + }) + + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Millisecond) + defer cancel() + _ = c.Run(ctx) + + require.GreaterOrEqual(t, emt.Count(), 1) + + emt.mu.Lock() + last := emt.events[len(emt.events)-1] + emt.mu.Unlock() + + raw, ok := last.Payload.(json.RawMessage) + require.True(t, ok, "expected payload to be json.RawMessage") + var payload map[string]any + require.NoError(t, json.Unmarshal(raw, &payload)) + + require.InDelta(t, 487.0, payload["speed_kms"], 0.001) + require.InDelta(t, 5.42, payload["density"], 0.001) + require.InDelta(t, -3.21, payload["bz_gsm"], 0.001) + require.InDelta(t, 2.5e-5, payload["xray_flux"], 1e-9) + require.Equal(t, "M2.5", payload["xray_class"]) + require.InDelta(t, 4.0, payload["kp"], 0.001) +} + func TestCollector_FetchErrorsRecordsState(t *testing.T) { ftch := &fakeFetcher{err: errors.New("upstream 503")} ring := &fakeRing{}