diff --git a/.changeset/ocpp-available-zero-power.md b/.changeset/ocpp-available-zero-power.md new file mode 100644 index 00000000..63e911c6 --- /dev/null +++ b/.changeset/ocpp-available-zero-power.md @@ -0,0 +1,5 @@ +--- +"ftw": patch +--- + +Home use is no longer marked estimated because an OCPP charger has no car. When the charger reports Available, no car is connected and no current flows, so its power counts as a measured 0 W in live Home use and in history. Each heartbeat or other message from the charger keeps that reading current, and it ends as soon as a car arrives. Unavailable and Faulted still leave the charger's power unknown. The Available a charger sends for itself (connector 0) at boot or reconnect no longer reads as an unplugged car. Core now asks OCPP chargers for a heartbeat every 30 s instead of every 60 s, so one late heartbeat no longer marks an idle charger offline: one more small request and reply a minute per charger. A heartbeat_interval_s set in the config is kept. diff --git a/config.example.yaml b/config.example.yaml index c2ea888b..249d228f 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -192,7 +192,7 @@ ocpp: path: / username: ftw password: "" # required when enabled; use a long random string - heartbeat_interval_s: 60 + heartbeat_interval_s: 30 # Serve wss:// instead of ws://. client_ca_file additionally requires each # charger to present a certificate signed by that CA whose CN or DNS SAN # matches the identity in the URL — OCPP 2.0.1 security profile 3. diff --git a/config.local.example.yaml b/config.local.example.yaml index 2e5f36f8..a5af8fa5 100644 --- a/config.local.example.yaml +++ b/config.local.example.yaml @@ -52,7 +52,7 @@ ocpp: port_v201: 8888 username: ftw password: sim-ocpp - heartbeat_interval_s: 60 + heartbeat_interval_s: 30 state: # `make dev` starts the Go process from go/, so keep disposable state in diff --git a/docs/ocpp.md b/docs/ocpp.md index 0b400173..9955b75b 100644 --- a/docs/ocpp.md +++ b/docs/ocpp.md @@ -135,7 +135,7 @@ ocpp: path: / # default / username: ftw password: - heartbeat_interval_s: 60 + heartbeat_interval_s: 30 ``` **Credentials are mandatory.** FTW refuses to start with `enabled: true` and an diff --git a/go/cmd/ftw/history_source_load_test.go b/go/cmd/ftw/history_source_load_test.go index fb290398..7ee07e27 100644 --- a/go/cmd/ftw/history_source_load_test.go +++ b/go/cmd/ftw/history_source_load_test.go @@ -2,11 +2,15 @@ package main import ( "context" + "encoding/json" "math" "path/filepath" "testing" "time" + "github.com/lorenzodonini/ocpp-go/ocpp1.6/core" + + "github.com/srcfl/ftw/go/internal/ocpp" "github.com/srcfl/ftw/go/internal/state" "github.com/srcfl/ftw/go/internal/telemetry" ) @@ -120,3 +124,186 @@ func TestHistorySourceQualificationDoesNotInventMissingFlows(t *testing.T) { } } } + +// An OCPP charger with no car sends no MeterValues. Its Available status is +// the evidence that nothing flows, so the household stays measured. +func TestHistoryCountsAnEmptyOCPPChargerAsMeasured(t *testing.T) { + st, err := state.Open(filepath.Join(t.TempDir(), "state.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + if err := st.EnableHistoryAggregation(); err != nil { + t.Fatal(err) + } + tel := telemetry.NewStore() + h := ocpp.NewHandler(tel, 60) + h.SetApprovedIDs([]string{"charger"}) + h.OnConnect("charger") + if _, err := h.OnStatusNotification("charger", &core.StatusNotificationRequest{ConnectorId: 1, Status: core.ChargePointStatusAvailable, ErrorCode: core.NoError}); err != nil { + t.Fatal(err) + } + tel.Update("grid", telemetry.DerMeter, 2000, nil, nil) + tel.RecordDriverSuccess("grid") + if live := tel.LiveMeasurementNow("grid", telemetry.ForecastOptions{}); live.LoadQuality != telemetry.LoadMeasured || live.LoadW != 2000 { + t.Fatalf("live Home use %s (%s), %v W", live.LoadQuality, live.LoadReason, live.LoadW) + } + base := time.Now().Add(-10 * time.Minute).Truncate(5 * time.Minute).UnixMilli() + for i := int64(0); i <= 30; i++ { + p, _ := buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "grid"}, base+i*10000, time.Minute) + if p.Sources.Missing != 0 || p.Missing&^state.MissingSoC != 0 { + t.Fatalf("empty charger marked history %b, sources %b", p.Missing, p.Sources.Missing) + } + p.Sources.AtMS = base + i*10000 + if err := st.RecordTickWithEnergy(p, nil, nil); err != nil { + t.Fatal(err) + } + } + if err := st.FlushHistory(context.Background()); err != nil { + t.Fatal(err) + } + d, err := st.DailyEnergy(base, base+300000) + if err != nil { + t.Fatal(err) + } + if d.Missing&state.MissingLoad != 0 || math.Abs(d.LoadWh-2000.0/12) > 0.001 { + t.Fatalf("household %v Wh, quality %s", d.LoadWh, d.Missing.Quality()) + } +} + +// Between heartbeats an Available charger's 0 W is up to a minute old. The +// time-aligned balance calls that time_skew; the chart's quality comes from +// the source rows, so it must not say the measurements do not balance. +func TestHistoryChartKeepsSourceQualityForAnAgeingOCPPZero(t *testing.T) { + st, err := state.Open(filepath.Join(t.TempDir(), "state.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + if err := st.EnableHistoryAggregation(); err != nil { + t.Fatal(err) + } + tel := telemetry.NewStore() + tel.Update("grid", telemetry.DerMeter, 2000, nil, nil) + tel.RecordDriverSuccess("grid") + // What the OCPP handler publishes 40 s after the last heartbeat. + at := time.Now().Add(-40 * time.Second).UnixMilli() + data, _ := json.Marshal(map[string]any{"type": "ev", "w": 0.0, "connected": false, "forecast_power": telemetry.ForecastPowerSample{Version: 1, Known: true, MeasuredAtMS: at, ReceivedAtMS: at}}) + tel.Update("charger", telemetry.DerEV, 0, nil, data) + tel.RecordDriverSuccess("charger") + base := time.Now().Add(-10 * time.Minute).Truncate(5 * time.Minute).UnixMilli() + for i := int64(0); i <= 30; i++ { + p, _ := buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "grid"}, base+i*10000, time.Minute) + if p.Missing&state.BalanceMismatch == 0 || p.Sources.Missing != 0 { + t.Fatalf("want a skewed aligned balance and complete sources, got %b and %b", p.Missing, p.Sources.Missing) + } + p.Sources.AtMS = base + i*10000 + if err := st.RecordTickWithEnergy(p, nil, nil); err != nil { + t.Fatal(err) + } + } + if err := st.FlushHistory(context.Background()); err != nil { + t.Fatal(err) + } + points, err := st.LoadHistoryContext(context.Background(), base, base+300000, 0) + if err != nil { + t.Fatal(err) + } + if len(points) == 0 { + t.Fatal("no chart points") + } + for _, p := range points { + if q := p.Missing.Quality(); q == "balance_mismatch" || q == "incomplete" { + t.Fatalf("chart point at %d says %s", p.TsMs, q) + } + } +} + +// An idle charger often sends only heartbeats. At Core's default interval, +// with ±1 s jitter, its Available 0 W is at most 31 s old at any 2 s history +// tick: inside the 60 s limit, and its health never lapses. The old 60 s +// default loses ticks to the same jitter. +func TestHistoryKeepsAnIdleOCPPChargerMeasuredBetweenHeartbeats(t *testing.T) { + var c ocpp.Config + c.Defaults() + if lost := idleOCPPStaleTicks(t, time.Duration(c.HeartbeatIntervalS)*time.Second, true); lost != 0 { + t.Fatalf("default heartbeat %d s: %d ticks lost the charger", c.HeartbeatIntervalS, lost) + } + if lost := idleOCPPStaleTicks(t, time.Minute, false); lost == 0 { + t.Fatal("a 60 s heartbeat lost no tick; the test cannot see the edge") + } +} + +// idleOCPPStaleTicks plays ten heartbeats at interval ±1 s through the real +// handler and counts history ticks whose source row misses a flow. Between +// beats the charger's reading and health are aged by the time since the beat. +func idleOCPPStaleTicks(t *testing.T, interval time.Duration, record bool) int { + t.Helper() + const watchdog = time.Minute // site default; history uses it as its age limit + st, err := state.Open(filepath.Join(t.TempDir(), "state.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + if err := st.EnableHistoryAggregation(); err != nil { + t.Fatal(err) + } + tel := telemetry.NewStore() + h := ocpp.NewHandler(tel, int(interval/time.Second)) + h.SetApprovedIDs([]string{"charger"}) + h.OnConnect("charger") + if _, err := h.OnStatusNotification("charger", &core.StatusNotificationRequest{ConnectorId: 1, Status: core.ChargePointStatusAvailable, ErrorCode: core.NoError}); err != nil { + t.Fatal(err) + } + age := func(beat time.Time, since time.Duration) { + r := tel.Get("charger", telemetry.DerEV) + var d map[string]any + if err := json.Unmarshal(r.Data, &d); err != nil { + t.Fatal(err) + } + at := beat.Add(-since) + d["forecast_power"] = telemetry.ForecastPowerSample{Version: 1, Known: true, MeasuredAtMS: at.UnixMilli(), ReceivedAtMS: at.UnixMilli()} + r.Data, _ = json.Marshal(d) + r.UpdatedAt = at + tel.DriverHealthMut("charger").LastSuccess = &at + } + base := time.Now().Add(-time.Hour).Truncate(5 * time.Minute).UnixMilli() + tick, lost := base, 0 + for beat := 0; beat < 10; beat++ { + gap := interval + []time.Duration{-time.Second, 0, time.Second}[beat%3] + if _, err := h.OnHeartbeat("charger", &core.HeartbeatRequest{}); err != nil { + t.Fatal(err) + } + beatAt := time.Now() + for since := time.Duration(beat%4) * 500 * time.Millisecond; since < gap; since += 2 * time.Second { + age(beatAt, since) + tel.Update("grid", telemetry.DerMeter, 2000, nil, nil) + tel.RecordDriverSuccess("grid") + tel.WatchdogScan(watchdog) + p, _ := buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "grid"}, tick, watchdog) + if p.Sources.Missing != 0 { + lost++ + } + if record { + p.Sources.AtMS = tick + if err := st.RecordTickWithEnergy(p, nil, nil); err != nil { + t.Fatal(err) + } + } + tick += 2000 + } + } + if record { + if err := st.FlushHistory(context.Background()); err != nil { + t.Fatal(err) + } + d, err := st.DailyEnergy(base, tick) + if err != nil { + t.Fatal(err) + } + if d.Missing&(state.MissingLoad|state.MissingEV) != 0 || d.LoadWh <= 0 { + t.Fatalf("history over %d s: %v Wh, quality %s", (tick-base)/1000, d.LoadWh, d.Missing.Quality()) + } + } + return lost +} diff --git a/go/internal/config/ocpp_test.go b/go/internal/config/ocpp_test.go index abe403e3..11a5d431 100644 --- a/go/internal/config/ocpp_test.go +++ b/go/internal/config/ocpp_test.go @@ -1,6 +1,11 @@ package config -import "testing" +import ( + "os" + "path/filepath" + "strings" + "testing" +) // The OCPP listener cannot be pinned to one interface — the library builds its // listen address from the port alone. Basic auth is therefore the only thing @@ -142,3 +147,39 @@ func TestConfigValidateCoversOCPP(t *testing.T) { t.Fatalf("config with a credentialed ocpp section should validate, got %v", err) } } + +// Only the default heartbeat changed (60 s to 30 s). An unset value must stay +// unset through a save, and a saved 60 must stay 60. +func TestOCPPHeartbeatSurvivesSave(t *testing.T) { + for _, tc := range []struct { + yaml string + want int + }{ + {"ocpp:\n enabled: false\n", 0}, + {"ocpp:\n enabled: false\n heartbeat_interval_s: 60\n", 60}, + } { + dir := t.TempDir() + path := filepath.Join(dir, "c.yaml") + c, err := Parse([]byte(minimalYAML+tc.yaml), dir) + if err != nil { + t.Fatal(err) + } + if err := SaveAtomic(path, c); err != nil { + t.Fatal(err) + } + saved, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if tc.want == 0 && strings.Contains(string(saved), "heartbeat_interval_s") { + t.Fatalf("an unset heartbeat was written:\n%s", saved) + } + c2, err := Load(path) + if err != nil { + t.Fatal(err) + } + if c2.OCPP == nil || c2.OCPP.HeartbeatIntervalS != tc.want { + t.Fatalf("heartbeat after save: got %+v, want %d", c2.OCPP, tc.want) + } + } +} diff --git a/go/internal/ocpp/charging_needs.go b/go/internal/ocpp/charging_needs.go index 3a2a76dd..3e0f2f69 100644 --- a/go/internal/ocpp/charging_needs.go +++ b/go/internal/ocpp/charging_needs.go @@ -198,7 +198,7 @@ func (h *handlerV201) OnNotifyEVChargingNeeds(id string, req *smartcharging.Noti "departure", n.DepartureTime) h.noteChargingNeeds(id, n) - h.telSuccess(id) + h.telHealth(id) // from the car, so no proof of Available return smartcharging.NewNotifyEVChargingNeedsResponse(smartcharging.EVChargingNeedsStatusAccepted), nil } @@ -207,7 +207,7 @@ func (h *handlerV201) OnNotifyEVChargingNeeds(id string, req *smartcharging.Noti // acknowledged and dropped — accepting it costs nothing and refusing would // make the charger retry forever. func (h *handlerV201) OnNotifyEVChargingSchedule(id string, _ *smartcharging.NotifyEVChargingScheduleRequest) (*smartcharging.NotifyEVChargingScheduleResponse, error) { - h.telSuccess(id) + h.telHealth(id) return smartcharging.NewNotifyEVChargingScheduleResponse(types201.GenericStatusAccepted), nil } diff --git a/go/internal/ocpp/config.go b/go/internal/ocpp/config.go index c6708d17..9a6fa9fa 100644 --- a/go/internal/ocpp/config.go +++ b/go/internal/ocpp/config.go @@ -128,8 +128,11 @@ func (c *Config) Defaults() { if c.Path == "" { c.Path = "/" } + // Half the 60 s watchdog and history age limit: an idle charger often + // sends nothing else, so one late heartbeat would otherwise mark it + // offline and its Available 0 W stale. if c.HeartbeatIntervalS == 0 { - c.HeartbeatIntervalS = 60 + c.HeartbeatIntervalS = 30 } } diff --git a/go/internal/ocpp/config_test.go b/go/internal/ocpp/config_test.go new file mode 100644 index 00000000..05adf19c --- /dev/null +++ b/go/internal/ocpp/config_test.go @@ -0,0 +1,26 @@ +package ocpp + +import ( + "testing" + + "github.com/lorenzodonini/ocpp-go/ocpp1.6/core" + + "github.com/srcfl/ftw/go/internal/telemetry" +) + +// Core asks chargers for a heartbeat at half the 60 s watchdog unless the +// owner set one; a saved value wins. +func TestHeartbeatDefaultIsHalfTheWatchdog(t *testing.T) { + for _, tc := range []struct{ set, want int }{{0, 30}, {60, 60}, {120, 120}} { + c := Config{HeartbeatIntervalS: tc.set} + c.Defaults() + if c.HeartbeatIntervalS != tc.want { + t.Fatalf("heartbeat_interval_s %d: got %d, want %d", tc.set, c.HeartbeatIntervalS, tc.want) + } + h := NewHandler(telemetry.NewStore(), c.HeartbeatIntervalS) + conf, err := h.OnBootNotification("charger", &core.BootNotificationRequest{ChargePointVendor: "Rig", ChargePointModel: "One"}) + if err != nil || conf.Interval != tc.want { + t.Fatalf("heartbeat_interval_s %d: boot asked for %d s (%v)", tc.set, conf.Interval, err) + } + } +} diff --git a/go/internal/ocpp/forecast_power_test.go b/go/internal/ocpp/forecast_power_test.go index c0f61cc3..53016e05 100644 --- a/go/internal/ocpp/forecast_power_test.go +++ b/go/internal/ocpp/forecast_power_test.go @@ -9,8 +9,10 @@ import ( "github.com/lorenzodonini/ocpp-go/ocpp1.6/core" types16 "github.com/lorenzodonini/ocpp-go/ocpp1.6/types" + "github.com/lorenzodonini/ocpp-go/ocpp2.0.1/authorization" "github.com/lorenzodonini/ocpp-go/ocpp2.0.1/availability" "github.com/lorenzodonini/ocpp-go/ocpp2.0.1/meter" + "github.com/lorenzodonini/ocpp-go/ocpp2.0.1/provisioning" "github.com/lorenzodonini/ocpp-go/ocpp2.0.1/transactions" types201 "github.com/lorenzodonini/ocpp-go/ocpp2.0.1/types" "github.com/srcfl/ftw/go/internal/telemetry" @@ -23,8 +25,17 @@ type forecastOCPPFixture struct { energy func() status func() available func() - stop func() - phases func(l1, l2, l3 float64, at time.Time) + // availableAt sends Available with the charger's time; zero sends none. + availableAt func(time.Time) + // unavailable and faulted leave power unknown; heartbeat is the charger's. + unavailable func() + faulted func() + heartbeat func() + authorize func() + boot func() + start func() + stop func() + phases func(l1, l2, l3 float64, at time.Time) } func newForecastOCPPFixture(t *testing.T, version string) forecastOCPPFixture { @@ -44,9 +55,25 @@ func newForecastOCPPFixture(t *testing.T, version string) forecastOCPPFixture { f.status = func() { h.OnStatusNotification("charger", &core.StatusNotificationRequest{ConnectorId: 1, Status: core.ChargePointStatusCharging}) } - f.available = func() { - h.OnStatusNotification("charger", &core.StatusNotificationRequest{ConnectorId: 1, Status: core.ChargePointStatusAvailable}) + f.availableAt = func(at time.Time) { + req := &core.StatusNotificationRequest{ConnectorId: 1, Status: core.ChargePointStatusAvailable} + if !at.IsZero() { + req.Timestamp = types16.NewDateTime(at) + } + h.OnStatusNotification("charger", req) + } + f.unavailable = func() { + h.OnStatusNotification("charger", &core.StatusNotificationRequest{ConnectorId: 1, Status: core.ChargePointStatusUnavailable}) + } + f.authorize = func() { h.OnAuthorize("charger", &core.AuthorizeRequest{IdTag: "card"}) } + f.boot = func() { + h.OnBootNotification("charger", &core.BootNotificationRequest{ChargePointVendor: "Rig", ChargePointModel: "One"}) + } + f.start = func() { h.OnStartTransaction("charger", &core.StartTransactionRequest{ConnectorId: 1, IdTag: "card"}) } + f.faulted = func() { + h.OnStatusNotification("charger", &core.StatusNotificationRequest{ConnectorId: 1, Status: core.ChargePointStatusFaulted, ErrorCode: core.GroundFailure}) } + f.heartbeat = func() { h.OnHeartbeat("charger", &core.HeartbeatRequest{}) } f.stop = func() { h.OnStopTransaction("charger", &core.StopTransactionRequest{}) } f.phases = func(l1, l2, l3 float64, at time.Time) { h.OnMeterValues("charger", &core.MeterValuesRequest{ConnectorId: 1, MeterValue: []types16.MeterValue{{Timestamp: types16.NewDateTime(at), SampledValue: []types16.SampledValue{ @@ -66,9 +93,29 @@ func newForecastOCPPFixture(t *testing.T, version string) forecastOCPPFixture { f.status = func() { v.OnStatusNotification("charger", &availability.StatusNotificationRequest{EvseID: 1, ConnectorID: 1, ConnectorStatus: availability.ConnectorStatusOccupied}) } - f.available = func() { - v.OnStatusNotification("charger", &availability.StatusNotificationRequest{EvseID: 1, ConnectorID: 1, ConnectorStatus: availability.ConnectorStatusAvailable}) + f.availableAt = func(at time.Time) { + req := &availability.StatusNotificationRequest{EvseID: 1, ConnectorID: 1, ConnectorStatus: availability.ConnectorStatusAvailable} + if !at.IsZero() { + req.Timestamp = types201.NewDateTime(at) + } + v.OnStatusNotification("charger", req) + } + f.authorize = func() { + v.OnAuthorize("charger", &authorization.AuthorizeRequest{IdToken: types201.IdToken{IdToken: "card", Type: types201.IdTokenTypeISO14443}}) + } + f.boot = func() { + v.OnBootNotification("charger", &provisioning.BootNotificationRequest{ChargingStation: provisioning.ChargingStationType{VendorName: "Rig", Model: "One"}}) } + f.start = func() { + v.OnTransactionEvent("charger", &transactions.TransactionEventRequest{EventType: transactions.TransactionEventStarted}) + } + f.unavailable = func() { + v.OnStatusNotification("charger", &availability.StatusNotificationRequest{EvseID: 1, ConnectorID: 1, ConnectorStatus: availability.ConnectorStatusUnavailable}) + } + f.faulted = func() { + v.OnStatusNotification("charger", &availability.StatusNotificationRequest{EvseID: 1, ConnectorID: 1, ConnectorStatus: availability.ConnectorStatusFaulted}) + } + f.heartbeat = func() { v.OnHeartbeat("charger", &availability.HeartbeatRequest{}) } f.stop = func() { v.OnTransactionEvent("charger", &transactions.TransactionEventRequest{EventType: transactions.TransactionEventEnded}) } @@ -80,6 +127,8 @@ func newForecastOCPPFixture(t *testing.T, version string) forecastOCPPFixture { }}}}) } } + // Chargers stamp their status; 2.0.1 requires it. + f.available = func() { f.availableAt(time.Now()) } return f } func (f forecastOCPPFixture) reading() telemetry.ForecastReading { @@ -218,8 +267,8 @@ func TestOCPPLastPowerSharedAcceptRule(t *testing.T) { if raw := f.tel.Get("charger", telemetry.DerEV); raw == nil || raw.RawW != 0 { t.Fatalf("Available left lastPowerW set: %+v", raw) } - if r := f.reading(); r.Valid { - t.Fatalf("Available published a measured zero: %+v", r) + if r := f.reading(); !r.Valid || r.EVW != 0 { + t.Fatalf("Available is not a measured zero: %+v", r) } f2 := newForecastOCPPFixture(t, version) @@ -251,3 +300,251 @@ func TestMeterPowerWPrefersUnphasedThenSum(t *testing.T) { t.Fatal("negative/NaN samples accepted") } } + +// live is the balance every client shows: the site meter, PV and the charger. +func (f forecastOCPPFixture) live() telemetry.ForecastReading { + f.tel.Update("site", telemetry.DerMeter, 5000, nil, nil) + f.tel.RecordDriverSuccess("site") + f.tel.Update("pv", telemetry.DerPV, -1000, nil, nil) + f.tel.RecordDriverSuccess("pv") + return f.tel.LiveMeasurementNow("site", telemetry.ForecastOptions{ExpectedFlows: []telemetry.ForecastFlow{{Driver: "charger", DerType: telemetry.DerEV}}}) +} + +// age moves the published charger reading back, as if no message came for d. +func (f forecastOCPPFixture) age(d time.Duration) { + f.h.mu.Lock() + s := f.h.chargers["charger"] + s.forecastPower.MeasuredAtMS -= d.Milliseconds() + s.forecastPower.ReceivedAtMS -= d.Milliseconds() + f.h.mu.Unlock() + f.h.pushReading("charger", s) +} + +func (f forecastOCPPFixture) wantMeasuredZero(t *testing.T, what string) { + t.Helper() + if r := f.live(); r.LoadQuality != telemetry.LoadMeasured || r.EVW != 0 || r.LoadW != 6000 { + t.Fatalf("%s: live Home use %s (%s), EV %v W, load %v W", what, r.LoadQuality, r.LoadReason, r.EVW, r.LoadW) + } + if r := f.reading(); !r.Valid || !r.Known[telemetry.DerEV] || r.EVW != 0 { + t.Fatalf("%s: forecast balance %+v", what, r) + } + if raw := f.tel.Get("charger", telemetry.DerEV); raw == nil || raw.RawW != 0 { + t.Fatalf("%s: dispatch power %+v", what, raw) + } +} + +func TestOCPPAvailableIsAMeasuredZeroInBothVersions(t *testing.T) { + for _, version := range []string{"1.6", "2.0.1"} { + t.Run(version, func(t *testing.T) { + f := newForecastOCPPFixture(t, version) + f.available() + f.wantMeasuredZero(t, "Available after boot") + + // No message for two minutes: the zero is as old as its evidence. + f.age(2 * time.Minute) + if r := f.live(); r.LoadQuality == telemetry.LoadMeasured || !strings.HasSuffix(r.LoadReason, ":charger:ev") { + t.Fatalf("an old Available stayed current: %s (%s)", r.LoadQuality, r.LoadReason) + } + f.heartbeat() + f.wantMeasuredZero(t, "heartbeat") + + // A car arrives. Its draw is unknown until the meter reports. + f.status() + f.heartbeat() + if r := f.live(); r.LoadQuality == telemetry.LoadMeasured { + t.Fatalf("the zero outlived Available: %s", r.LoadReason) + } + f.available() + f.wantMeasuredZero(t, "Available again") + + // A new socket says nothing about the connector until it reports. + f.h.OnDisconnect("charger") + f.h.OnConnect("charger") + f.heartbeat() + if r := f.live(); r.LoadQuality == telemetry.LoadMeasured { + t.Fatalf("a reconnect kept the zero before any status: %s", r.LoadReason) + } + }) + } +} + +func TestOCPPAvailableAfterSessionIsAMeasuredZero(t *testing.T) { + for _, version := range []string{"1.6", "2.0.1"} { + for _, order := range []string{"stop first", "Available first"} { + t.Run(version+" "+order, func(t *testing.T) { + f := newForecastOCPPFixture(t, version) + f.status() + f.power(3000, time.Now()) + if r := f.live(); r.LoadQuality != telemetry.LoadMeasured || r.EVW != 3000 || r.LoadW != 3000 { + t.Fatalf("charging with MeterValues: %s (%s), EV %v W, load %v W", r.LoadQuality, r.LoadReason, r.EVW, r.LoadW) + } + if r := f.reading(); !r.Valid || r.EVW != 3000 { + t.Fatalf("charging forecast balance: %+v", r) + } + lastCharge := time.Now() + time.Sleep(2 * time.Millisecond) + if order == "stop first" { + f.stop() + f.available() + } else { + f.available() + f.stop() + } + f.wantMeasuredZero(t, "after the session") + + // A meter sample from before the car left cannot bring its draw back. + f.power(3000, lastCharge.Add(time.Millisecond)) + f.wantMeasuredZero(t, "late session sample") + if version == "2.0.1" { + // A periodic update with no car keeps the zero until Ended. + v := &handlerV201{f.h} + v.OnTransactionEvent("charger", &transactions.TransactionEventRequest{EventType: transactions.TransactionEventUpdated, TransactionInfo: transactions.Transaction{ChargingState: transactions.ChargingStateIdle}}) + f.wantMeasuredZero(t, "idle transaction update") + } + f.heartbeat() + f.wantMeasuredZero(t, "heartbeat after the session") + }) + } + } +} + +func TestOCPPUnavailableAndFaultedStayUnknown(t *testing.T) { + for _, version := range []string{"1.6", "2.0.1"} { + for name, status := range map[string]func(forecastOCPPFixture){ + "Unavailable": func(f forecastOCPPFixture) { f.unavailable() }, + "Faulted": func(f forecastOCPPFixture) { f.faulted() }, + } { + t.Run(version+" "+name, func(t *testing.T) { + f := newForecastOCPPFixture(t, version) + f.available() + status(f) + f.heartbeat() + r := f.live() + if r.LoadQuality != telemetry.LoadEstimated || !strings.HasPrefix(r.LoadReason, "unknown_power:charger:") { + t.Fatalf("%s became known power: %s (%s)", name, r.LoadQuality, r.LoadReason) + } + if r := f.reading(); r.Valid || !strings.HasPrefix(r.Reason, "unknown_power:charger:") { + t.Fatalf("%s became a measured balance: %+v", name, r) + } + }) + } + } +} + +// Connector 0 is the charge point. Its Available is not a cable status. +func TestOCPPChargePointAvailableIsNotACableStatus(t *testing.T) { + f := newForecastOCPPFixture(t, "1.6") + chargePointAvailable := func() { + f.h.OnStatusNotification("charger", &core.StatusNotificationRequest{ConnectorId: 0, Status: core.ChargePointStatusAvailable}) + } + chargePointAvailable() + if r := f.live(); r.LoadQuality == telemetry.LoadMeasured { + t.Fatal("connector 0 Available became a measured zero") + } + if raw := f.tel.Get("charger", telemetry.DerEV); raw == nil || !strings.Contains(string(raw.Data), `"connection_unknown":true`) { + t.Fatalf("connector 0 Available reported the cable: %+v", raw) + } + + f.status() + f.power(3000, time.Now()) + chargePointAvailable() + if r := f.live(); r.LoadQuality != telemetry.LoadMeasured || r.EVW != 3000 { + t.Fatalf("connector 0 Available cleared a charging car: %s (%s), EV %v W", r.LoadQuality, r.LoadReason, r.EVW) + } + if v := f.h.Snapshot()["charger"]; !v.Connected || !v.Charging { + t.Fatalf("connector 0 Available unplugged the car: %+v", v) + } + + time.Sleep(2 * time.Millisecond) + f.available() + chargePointAvailable() + f.age(2 * time.Minute) + f.heartbeat() + f.wantMeasuredZero(t, "connector 0 after connector 1 Available") +} + +func TestOCPPAvailableConfirmsOnlyAZero(t *testing.T) { + for _, version := range []string{"1.6", "2.0.1"} { + t.Run(version, func(t *testing.T) { + f := newForecastOCPPFixture(t, version) + f.available() + time.Sleep(2 * time.Millisecond) + f.power(3000, time.Now()) + f.age(2 * time.Minute) + f.heartbeat() + if r := f.live(); r.LoadQuality == telemetry.LoadMeasured || !strings.HasSuffix(r.LoadReason, ":charger:ev") { + t.Fatalf("a heartbeat kept an old 3000 W current: %s (%s), EV %v W", r.LoadQuality, r.LoadReason, r.EVW) + } + }) + } +} + +func TestOCPPAvailableGateFollowsTheChargerClock(t *testing.T) { + for _, version := range []string{"1.6", "2.0.1"} { + for name, stamp := range map[string]func(time.Time) time.Time{ + "status time": func(now time.Time) time.Time { return now }, + "no status time": func(time.Time) time.Time { return time.Time{} }, + } { + t.Run(version+" "+name, func(t *testing.T) { + f := newForecastOCPPFixture(t, version) + f.h.mu.Lock() + f.h.chargers["charger"].powerConnectedAt = time.Now().Add(-time.Hour) + f.h.mu.Unlock() + // The charger's clock is 30 s behind Core's. The car plugs in at once. + behind := time.Now().Add(-30 * time.Second) + f.availableAt(stamp(behind)) + f.status() + f.power(3000, behind.Add(time.Second)) + if r := f.live(); r.LoadQuality != telemetry.LoadMeasured || r.EVW != 3000 { + t.Fatalf("the car's first sample was rejected: %s (%s), EV %v W", r.LoadQuality, r.LoadReason, r.EVW) + } + }) + } + } +} + +func TestOCPPZeroEndsWhenTheCarArrives(t *testing.T) { + for _, version := range []string{"1.6", "2.0.1"} { + for name, arrive := range map[string]func(forecastOCPPFixture){ + "status": func(f forecastOCPPFixture) { f.status() }, + "transaction": func(f forecastOCPPFixture) { f.start() }, + } { + t.Run(version+" "+name, func(t *testing.T) { + f := newForecastOCPPFixture(t, version) + f.available() + arrive(f) + r := f.live() + if r.LoadQuality == telemetry.LoadMeasured || !strings.HasPrefix(r.LoadReason, "unknown_power:charger:") { + t.Fatalf("the car arrived and 0 W stayed measured: %s (%s)", r.LoadQuality, r.LoadReason) + } + if r := f.reading(); r.Valid { + t.Fatalf("the car arrived and 0 W stayed a training balance: %+v", r) + } + time.Sleep(2 * time.Millisecond) + f.power(3000, time.Now()) + if r := f.live(); r.LoadQuality != telemetry.LoadMeasured || r.EVW != 3000 { + t.Fatalf("the car's sample after Available: %s (%s), EV %v W", r.LoadQuality, r.LoadReason, r.EVW) + } + }) + } + } +} + +func TestOCPPAnyMessageConfirmsAvailable(t *testing.T) { + for _, version := range []string{"1.6", "2.0.1"} { + t.Run(version, func(t *testing.T) { + f := newForecastOCPPFixture(t, version) + f.available() + f.age(2 * time.Minute) + f.authorize() + f.wantMeasuredZero(t, "Authorize") + + // A boot may follow a reset; the charger reports its statuses again. + f.age(2 * time.Minute) + f.boot() + if r := f.live(); r.LoadQuality == telemetry.LoadMeasured { + t.Fatalf("a boot confirmed Available: %s", r.LoadReason) + } + }) + } +} diff --git a/go/internal/ocpp/handlers.go b/go/internal/ocpp/handlers.go index c06dd443..64b9ccd7 100644 --- a/go/internal/ocpp/handlers.go +++ b/go/internal/ocpp/handlers.go @@ -79,7 +79,16 @@ type chargerState struct { sessionMeterWh float64 lastPowerW float64 forecastPower telemetry.ForecastPowerSample - powerConnectedAt time.Time + // meterAtMS is the newest meter sample time accepted. A heartbeat moves + // forecastPower's time to Core's clock; this gate stays on the meter's. + meterAtMS int64 + // available: the connector reported Available on this socket. With no + // vehicle no current flows, so its power is a known 0 W. OCPP reports a + // status change when it happens, so any later message from the charger + // confirms the reading. State is per charger, not per connector: on a + // two-connector charger an Available on one claims 0 W for both. + available bool + powerConnectedAt time.Time // lastAmps is the most recent per-phase limit this charger accepted. // A resume with no rate of its own restores it. lastAmps float64 @@ -203,7 +212,16 @@ func (h *Handler) isApproved(id string) bool { // The tel* wrappers are the quarantine choke points: every telemetry write // for a charger goes through one of them, so a pending charger is dropped // here once rather than gated at each call site. +// telSuccess records a message from the charger. While the connector is +// Available, any message also confirms its 0 W. func (h *Handler) telSuccess(id string) { + h.telHealth(id) + h.confirmAvailable(id) +} + +// telHealth records a message without confirming the connector status: a +// new socket or a boot says nothing about it until the charger reports. +func (h *Handler) telHealth(id string) { if h.isApproved(id) { h.tel.RecordDriverSuccess(id) } @@ -413,7 +431,7 @@ func (h *Handler) OnConnect(id string) { h.scheduleIdentityProbeLocked(id, s) h.mu.Unlock() h.pushReading(id, s) - h.telSuccess(id) + h.telHealth(id) h.maybeProbeCapability(id) } @@ -464,7 +482,7 @@ func (h *Handler) OnBootNotification(id string, req *core.BootNotificationReques s.firmware = req.FirmwareVersion h.mu.Unlock() h.noteIdentity(id) - h.telSuccess(id) + h.telHealth(id) h.maybeProbeCapability(id) return core.NewBootNotificationConfirmation( types.NewDateTime(time.Now()), @@ -478,10 +496,23 @@ func (h *Handler) OnHeartbeat(id string, _ *core.HeartbeatRequest) (*core.Heartb return core.NewHeartbeatConfirmation(types.NewDateTime(time.Now())), nil } +// confirmAvailable publishes the 0 W of an Available connector again at the +// time of a message from the charger. telSuccess calls it for both dialects. +func (h *Handler) confirmAvailable(id string) { + s := h.state(id) + h.mu.Lock() + confirmed := s.confirmAvailablePower(time.Now()) + h.mu.Unlock() + if confirmed { + h.pushReading(id, s) + } +} + func (h *Handler) OnAuthorize(id string, req *core.AuthorizeRequest) (*core.AuthorizeConfirmation, error) { // Phase 1: auto-authorize every tag. RFID gating is a Phase 2 feature // alongside RemoteStartTransaction. slog.Debug("OCPP authorize", "charger", id, "tag", req.IdTag) + h.confirmAvailable(id) return core.NewAuthorizationConfirmation( types.NewIdTagInfo(types.AuthorizationStatusAccepted), ), nil @@ -490,15 +521,32 @@ func (h *Handler) OnAuthorize(id string, req *core.AuthorizeRequest) (*core.Auth func (h *Handler) OnDataTransfer(id string, req *core.DataTransferRequest) (*core.DataTransferConfirmation, error) { // Vendor-specific extensions — accept gracefully but do nothing. slog.Debug("OCPP DataTransfer ignored", "charger", id, "vendor", req.VendorId) + h.confirmAvailable(id) return core.NewDataTransferConfirmation(core.DataTransferStatusUnknownVendorId), nil } func (h *Handler) OnStatusNotification(id string, req *core.StatusNotificationRequest) (*core.StatusNotificationConfirmation, error) { + // Connector 0 is the charge point itself. Its Available, sent at boot and + // on reconnect, says the controller is up, not that a cable is out. + // Unavailable and Faulted there cover every connector. + if req.ConnectorId == 0 && req.Status == core.ChargePointStatusAvailable { + slog.Info("OCPP status", "charger", id, "connector", req.ConnectorId, "status", req.Status) + h.telSuccess(id) + return core.NewStatusNotificationConfirmation(), nil + } s := h.state(id) h.mu.Lock() s.connectedKnown = true + if req.Status != core.ChargePointStatusAvailable { + s.leaveAvailable() + } switch req.Status { - case core.ChargePointStatusAvailable, core.ChargePointStatusUnavailable: + case core.ChargePointStatusAvailable: + s.connected = false + s.charging = false + s.holdAvailablePower(time.Now(), statusTime(req.Timestamp)) + case core.ChargePointStatusUnavailable: + // Out of service, maybe broken: no evidence that nothing flows. s.connected = false s.charging = false s.clearMeasuredPower() @@ -590,6 +638,7 @@ func (h *Handler) OnStartTransaction(id string, req *core.StartTransactionReques s.sessionMeterWh = 0 s.connected = true s.connectedKnown = true + s.leaveAvailable() s.charging = true h.mu.Unlock() @@ -612,7 +661,10 @@ func (h *Handler) OnStopTransaction(id string, req *core.StopTransactionRequest) sessionWh := float64(req.MeterStop) - s.sessionStartMeterWh s.transactionID = -1 s.charging = false - s.clearMeasuredPower() + // A charger may report Available before the stop. The car is gone then. + if !s.available { + s.clearMeasuredPower() + } s.sessionMeterWh = sessionWh h.mu.Unlock() @@ -710,16 +762,66 @@ func meterPowerW(samples []powerSample) (float64, bool) { // accepted timestamp. Samples from before this socket or in the future cannot // revive a stale or synthesized reading. The caller holds h.mu. func (s *chargerState) recordPower(w float64, measured, received time.Time) bool { - if math.IsNaN(w) || math.IsInf(w, 0) || w < 0 || measured.IsZero() || measured.After(received) || measured.Before(s.powerConnectedAt) || measured.UnixMilli() <= s.forecastPower.MeasuredAtMS { + if math.IsNaN(w) || math.IsInf(w, 0) || w < 0 || measured.IsZero() || measured.After(received) || measured.Before(s.powerConnectedAt) || measured.UnixMilli() <= s.meterAtMS { return false } s.lastPowerW = w + s.meterAtMS = measured.UnixMilli() s.forecastPower = telemetry.ForecastPowerSample{Version: 1, Known: true, Watts: w, MeasuredAtMS: measured.UnixMilli(), ReceivedAtMS: received.UnixMilli()} return true } +// statusTime is the charger's own time of a status, when it sent one. +func statusTime(t *types.DateTime) time.Time { + if t == nil { + return time.Time{} + } + return t.Time +} + +// holdAvailablePower records the 0 W of an empty connector. A meter sample +// the charger took before the status cannot bring back the car's draw; a +// newer one wins. The gate uses the charger's clock, so a clock that runs +// behind Core's does not reject the next session's samples. Without a +// status time it stays where it was. The caller holds h.mu. +func (s *chargerState) holdAvailablePower(now, at time.Time) { + ms := now.UnixMilli() + s.available = true + s.lastPowerW = 0 + if !at.IsZero() { + s.meterAtMS = max(s.meterAtMS, min(at.UnixMilli(), ms)) + } + s.forecastPower = telemetry.ForecastPowerSample{Version: 1, Known: true, Watts: 0, MeasuredAtMS: ms, ReceivedAtMS: ms} +} + +// confirmAvailablePower moves a 0 W reading to now while the connector stays +// Available on a live socket. A meter that read more than 0 W is not +// confirmed; it ages out. At most once a second. The caller holds h.mu. +func (s *chargerState) confirmAvailablePower(now time.Time) bool { + ms := now.UnixMilli() + if !s.available || !s.online || !s.connectedKnown || !s.forecastPower.Known || s.forecastPower.Watts != 0 || ms-s.forecastPower.MeasuredAtMS < 1000 { + return false + } + s.forecastPower.MeasuredAtMS, s.forecastPower.ReceivedAtMS = ms, ms + return true +} + +// leaveAvailable ends the hold when a car may be there. A 0 W reading is no +// longer evidence then; power is unknown until the meter reports. The caller +// holds h.mu. +func (s *chargerState) leaveAvailable() { + if !s.available { + return + } + s.available = false + if s.forecastPower.Watts == 0 { + s.forecastPower.Known = false + } +} + // clearMeasuredPower zeros dispatch watts without recording a measured sample. func (s *chargerState) clearMeasuredPower() { s.lastPowerW = 0 s.forecastPower.Known = false + s.available = false } diff --git a/go/internal/ocpp/handlers_v201.go b/go/internal/ocpp/handlers_v201.go index 3e1b1009..b89cbfa1 100644 --- a/go/internal/ocpp/handlers_v201.go +++ b/go/internal/ocpp/handlers_v201.go @@ -66,7 +66,7 @@ func (h *handlerV201) OnBootNotification(id string, req *provisioning.BootNotifi h.mu.Unlock() h.setVersion(id, Version201) h.noteIdentity(id) - h.telSuccess(id) + h.telHealth(id) return provisioning.NewBootNotificationResponse( types201.NewDateTime(time.Now()), @@ -98,8 +98,19 @@ func (h *handlerV201) OnStatusNotification(id string, req *availability.StatusNo s := h.state(id) h.mu.Lock() s.connectedKnown = true + if req.ConnectorStatus != availability.ConnectorStatusAvailable { + s.leaveAvailable() + } switch req.ConnectorStatus { - case availability.ConnectorStatusAvailable, availability.ConnectorStatusUnavailable: + case availability.ConnectorStatusAvailable: + s.connected = false + s.charging = false + var at time.Time + if req.Timestamp != nil { + at = req.Timestamp.Time + } + s.holdAvailablePower(time.Now(), at) + case availability.ConnectorStatusUnavailable: s.connected = false s.charging = false s.clearMeasuredPower() @@ -143,6 +154,10 @@ func (h *handlerV201) OnTransactionEvent(id string, req *transactions.Transactio h.mu.Lock() acceptedPower := false if req.EventType != transactions.TransactionEventEnded { + // A live transaction may mean a car, unless it says none is there. + if req.TransactionInfo.ChargingState != transactions.ChargingStateIdle { + s.leaveAvailable() + } acceptedPower = s.recordPowerV201(req.MeterValue, time.Now()) } switch req.EventType { @@ -178,7 +193,10 @@ func (h *handlerV201) OnTransactionEvent(id string, req *transactions.Transactio s.transactionID = -1 s.transactionRef = "" s.charging = false - s.clearMeasuredPower() + // Available may come first; the car has gone then. + if !s.available { + s.clearMeasuredPower() + } } powerW := s.lastPowerW