From adec5c34ee81c3a6bc4169a6cf3ca25ecce0251d Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Thu, 22 Oct 2026 09:07:52 +0200 Subject: [PATCH 1/7] fix(ocpp): one heartbeat and meter loop per connection, DropConnection Each connection to a Central System started another heartbeat, MeterValues and simulation loop, and nothing ended them but Stop. After one reconnect a charger sent every heartbeat and MeterValues twice and integrated its energy twice a second. Heartbeats and MeterValues now end with their connection; the simulation runs once for the charger's life, connected or not, since a car charges through a network outage. Two hooks for scripted runs: DropConnection(d) closes the connection as a lost network does and keeps the charger away for d, with the session going on; SetFrameLog reports every OCPP-J frame the charger reads or writes. Co-Authored-By: Claude Opus 5.5 --- cmd/simulator/main.go | 85 ++++++++++++++++-------------- internal/ocpp/client_test.go | 58 +++++++++++++++++++++ internal/ocpp/ocpptest/csms.go | 22 +++++++- internal/ocpp/server.go | 95 ++++++++++++++++++++++++++-------- 4 files changed, 195 insertions(+), 65 deletions(-) diff --git a/cmd/simulator/main.go b/cmd/simulator/main.go index 77043c0..cffa94f 100644 --- a/cmd/simulator/main.go +++ b/cmd/simulator/main.go @@ -1953,46 +1953,7 @@ func handleProfile(w http.ResponseWriter, r *http.Request, sim *Simulator, profi return } - // Apply profile based on protocol - switch sim.Protocol { - case "modbus": - // Update RegisterSet if the profile specifies one - if profile.RegisterSetName != "" { - if rs := devices.GetRegisterSet(profile.RegisterSetName); rs != nil { - sim.RegisterSet = rs - sim.Server.RegisterSet = rs - sim.Server.InitDefaultsFromSet(rs) - } - } - // Apply profile-specific defaults - for addr, value := range profile.Defaults { - var def *modbus.RegisterDef - if sim.RegisterSet != nil { - def = sim.RegisterSet.Lookup[addr] - } - if def == nil { - def = modbus.RegisterLookup[addr] - } - if def != nil { - if err := sim.Server.WriteRegisterValue(def, value); err != nil { - log.Printf("[%s] Failed to write register %d: %v", sim.Serial, addr, err) - } - } - } - sim.mu.Lock() - sim.EMSResponse = commandResponse(profile) - sim.mu.Unlock() - case "ocpp": - if profile.OCPPConfig != nil && sim.OCPPServer != nil { - sim.OCPPServer.SetVendorModel(profile.OCPPConfig.Vendor, profile.OCPPConfig.Model) - sim.OCPPServer.ApplyConfiguration(profile.OCPPConfig.Configuration) - sim.mu.Lock() - sim.EMSResponse = commandResponse(profile) - sim.mu.Unlock() - sim.OCPPServer.SetProfileDelay(commandResponse(profile)) - log.Printf("[%s] Applied OCPP profile: %s (%s)", sim.Serial, profile.OCPPConfig.Model, profile.OCPPConfig.Vendor) - } - } + applyProfile(sim, profile) sim.ProfilesManager.SetActive(name) @@ -2010,6 +1971,50 @@ func handleProfile(w http.ResponseWriter, r *http.Request, sim *Simulator, profi json.NewEncoder(w).Encode(profile) } +// applyProfile puts a profile's register set, defaults, OCPP configuration +// and model parameters on a simulator. +func applyProfile(sim *Simulator, profile *profiles.Profile) { + switch sim.Protocol { + case "modbus": + // Update RegisterSet if the profile specifies one + if profile.RegisterSetName != "" { + if rs := devices.GetRegisterSet(profile.RegisterSetName); rs != nil { + sim.RegisterSet = rs + sim.Server.RegisterSet = rs + sim.Server.InitDefaultsFromSet(rs) + } + } + // Apply profile-specific defaults + for addr, value := range profile.Defaults { + var def *modbus.RegisterDef + if sim.RegisterSet != nil { + def = sim.RegisterSet.Lookup[addr] + } + if def == nil { + def = modbus.RegisterLookup[addr] + } + if def != nil { + if err := sim.Server.WriteRegisterValue(def, value); err != nil { + log.Printf("[%s] Failed to write register %d: %v", sim.Serial, addr, err) + } + } + } + sim.mu.Lock() + sim.EMSResponse = commandResponse(profile) + sim.mu.Unlock() + case "ocpp": + if profile.OCPPConfig != nil && sim.OCPPServer != nil { + sim.OCPPServer.SetVendorModel(profile.OCPPConfig.Vendor, profile.OCPPConfig.Model) + sim.OCPPServer.ApplyConfiguration(profile.OCPPConfig.Configuration) + sim.mu.Lock() + sim.EMSResponse = commandResponse(profile) + sim.mu.Unlock() + sim.OCPPServer.SetProfileDelay(commandResponse(profile)) + log.Printf("[%s] Applied OCPP profile: %s (%s)", sim.Serial, profile.OCPPConfig.Model, profile.OCPPConfig.Vendor) + } + } +} + func handleAutomation(w http.ResponseWriter, r *http.Request, sim *Simulator) { sim.Automation.mu.RLock() data := map[string]interface{}{ diff --git a/internal/ocpp/client_test.go b/internal/ocpp/client_test.go index 7684934..dec1320 100644 --- a/internal/ocpp/client_test.go +++ b/internal/ocpp/client_test.go @@ -3,6 +3,7 @@ package ocpp import ( "encoding/json" "math" + "sync" "testing" "time" @@ -268,3 +269,60 @@ func TestOnePhaseDrawIsCappedAtRatedCurrent(t *testing.T) { t.Fatalf("phase powers %v, want 16 A (3680 W) on L1", got) } } + +// A dropped connection keeps the charger away for the time asked, the car +// keeps charging through it, and after it comes back the charger sends one +// heartbeat per interval, not one per connection it ever had. +func TestDropConnectionKeepsTheSessionAndOneLoopSet(t *testing.T) { + csms := ocpptest.New(t) + csms.SetHeartbeatInterval(1) + s := newTestServer() + s.reconnectDelay = 50 * time.Millisecond + var mu sync.Mutex + frames := map[string]int{} + s.SetFrameLog(func(dir string, _ []byte) { mu.Lock(); frames[dir]++; mu.Unlock() }) + s.SetOCPPURL(csms.URL("")) + if err := s.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { s.Stop() }) + waitFor(t, "BootNotification", 5*time.Second, func() bool { return csms.Sent(ActionBootNotification) == 1 }) + if err := s.SetValue("evConnected", true); err != nil { + t.Fatal(err) + } + waitFor(t, "StartTransaction", 5*time.Second, func() bool { return csms.Sent(ActionStartTransaction) == 1 }) + + for i := 0; i < 2; i++ { + before := time.Now() + s.DropConnection(300 * time.Millisecond) + waitFor(t, "reconnect", 5*time.Second, func() bool { return csms.Connections() == i+2 }) + if away := time.Since(before); away < 300*time.Millisecond { + t.Fatalf("back after %v, want at least 300 ms", away) + } + } + waitFor(t, "BootNotification after the drops", 5*time.Second, func() bool { return csms.Sent(ActionBootNotification) == 3 }) + + beats := csms.Sent(ActionHeartbeat) + s.mu.RLock() + energy0 := s.connectors[1].EnergyWh + s.mu.RUnlock() + time.Sleep(3500 * time.Millisecond) + if n := csms.Sent(ActionHeartbeat) - beats; n < 2 || n > 4 { + t.Errorf("%d heartbeats in 3.5 s at a 1 s interval after two drops, want 3", n) + } + s.mu.RLock() + gained, txs, power := s.connectors[1].EnergyWh-energy0, len(s.transactions), s.connectors[1].CurrentPower + s.mu.RUnlock() + if txs != 1 || power <= 0 { + t.Fatalf("after the drops: %d transactions, %v W", txs, power) + } + // About 3.5 s at the charger's power, not a multiple of it. + if want := power * 3.5 / 3600; gained > 1.5*want { + t.Errorf("energy grew %.2f Wh in 3.5 s at %.0f W, want about %.2f Wh", gained, power, want) + } + mu.Lock() + defer mu.Unlock() + if frames["cs->cp"] == 0 || frames["cp->cs"] == 0 { + t.Errorf("frame log %v, want both directions", frames) + } +} diff --git a/internal/ocpp/ocpptest/csms.go b/internal/ocpp/ocpptest/csms.go index 7e465e9..da80237 100644 --- a/internal/ocpp/ocpptest/csms.go +++ b/internal/ocpp/ocpptest/csms.go @@ -23,6 +23,7 @@ type CSMS struct { srv *httptest.Server mu sync.Mutex + interval int // heartbeat interval in the BootNotification answer closed bool paths []string auths []string @@ -34,7 +35,7 @@ type CSMS struct { // New starts a Central System that stops with the test. func New(t testing.TB) *CSMS { - f := &CSMS{t: t} + f := &CSMS{t: t, interval: 60} up := websocket.Upgrader{Subprotocols: []string{"ocpp1.6"}} f.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { f.mu.Lock() @@ -71,9 +72,12 @@ func New(t testing.TB) *CSMS { } json.Unmarshal(frame[2], &action) var result interface{} = map[string]interface{}{} + f.mu.Lock() + interval := f.interval + f.mu.Unlock() switch action { case "BootNotification": - result = map[string]interface{}{"status": "Accepted", "currentTime": time.Now().UTC().Format(time.RFC3339), "interval": 60} + result = map[string]interface{}{"status": "Accepted", "currentTime": time.Now().UTC().Format(time.RFC3339), "interval": interval} case "StartTransaction": result = map[string]interface{}{"transactionId": 42, "idTagInfo": map[string]string{"status": "Accepted"}} } @@ -103,6 +107,13 @@ func (f *CSMS) URL(userinfo string) string { return u + "/ocpp" } +// SetHeartbeatInterval sets the interval the BootNotification answer gives. +func (f *CSMS) SetHeartbeatInterval(s int) { + f.mu.Lock() + defer f.mu.Unlock() + f.interval = s +} + // SetClosed makes the Central System refuse connections, as one that is down. func (f *CSMS) SetClosed(closed bool) { f.mu.Lock() @@ -157,6 +168,13 @@ func (f *CSMS) Actions() []string { return append([]string(nil), f.actions...) } +// Connections counts the charge point's connections so far. +func (f *CSMS) Connections() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.paths) +} + // Paths lists the request path of each connection. func (f *CSMS) Paths() []string { f.mu.Lock() diff --git a/internal/ocpp/server.go b/internal/ocpp/server.go index 64099cc..d91a08f 100644 --- a/internal/ocpp/server.go +++ b/internal/ocpp/server.go @@ -124,6 +124,8 @@ type Server struct { ocppURL string // Central System URL to connect to (client mode) clientMode bool // true = connect to URL, false = listen for connections reconnectDelay time.Duration + redialAt time.Time // DropConnection keeps the charger away until then + frameLog func(dir string, frame []byte) // 3-phase metering voltageL1 float64 @@ -252,6 +254,10 @@ func (s *Server) Start() error { // Recreate stop channel (may have been closed by previous Stop) s.stopChan = make(chan struct{}) + // The charger's power follows its profiles whether or not a Central + // System is connected. + go s.simulationLoop(s.stopChan) + if s.clientMode && s.ocppURL != "" { // Client mode: connect to Central System go s.clientLoop() @@ -331,7 +337,7 @@ func (s *Server) clientLoop() { select { case <-s.stopChan: return - case <-time.After(s.reconnectDelay): + case <-time.After(s.redialWait()): continue } } @@ -348,17 +354,14 @@ func (s *Server) clientLoop() { // Send BootNotification go s.sendBootNotification() - // Start heartbeat loop - go s.heartbeatLoop() - - // Start meter values loop - go s.meterValuesLoop() - - // Start simulation loop to update values - go s.simulationLoop() + // Heartbeats and meter values belong to this connection. + done := make(chan struct{}) + go s.heartbeatLoop(done) + go s.meterValuesLoop(done) // Read messages until disconnected s.readLoop(conn) + close(done) s.mu.Lock() s.wsConn = nil @@ -368,7 +371,7 @@ func (s *Server) clientLoop() { select { case <-s.stopChan: return - case <-time.After(s.reconnectDelay): + case <-time.After(s.redialWait()): log.Printf("[%s] Reconnecting to Central System...", serial) } } @@ -398,6 +401,46 @@ func dialTarget(raw, identity string) (string, http.Header, error) { return u.String(), header, nil } +// DropConnection closes the connection to the Central System, as a lost +// network does, and keeps the charger from dialling again for d. A session +// in progress goes on: the car keeps charging. +func (s *Server) DropConnection(d time.Duration) { + s.mu.Lock() + s.redialAt = time.Now().Add(d) + conn := s.wsConn + s.mu.Unlock() + if conn != nil { + conn.Close() + } +} + +// redialWait is how long the client waits before it dials again. +func (s *Server) redialWait() time.Duration { + s.mu.RLock() + defer s.mu.RUnlock() + if until := time.Until(s.redialAt); until > s.reconnectDelay { + return until + } + return s.reconnectDelay +} + +// SetFrameLog makes the charger report every OCPP-J frame it reads +// ("cs->cp") or writes ("cp->cs"). Nil turns it off. +func (s *Server) SetFrameLog(fn func(dir string, frame []byte)) { + s.mu.Lock() + defer s.mu.Unlock() + s.frameLog = fn +} + +func (s *Server) logFrame(dir string, frame []byte) { + s.mu.RLock() + fn := s.frameLog + s.mu.RUnlock() + if fn != nil { + fn(dir, frame) + } +} + // RedactURL hides the password in a Central System URL for logs. func RedactURL(raw string) string { u, err := url.Parse(raw) @@ -429,7 +472,7 @@ func (s *Server) readLoop(conn *websocket.Conn) { } // meterValuesLoop periodically sends meter values during active transactions -func (s *Server) meterValuesLoop() { +func (s *Server) meterValuesLoop(done <-chan struct{}) { for { intervalSeconds := s.getMeterValueSampleInterval() if intervalSeconds <= 0 { @@ -439,6 +482,8 @@ func (s *Server) meterValuesLoop() { select { case <-s.stopChan: return + case <-done: + return case <-time.After(time.Duration(intervalSeconds) * time.Second): s.sendMeterValuesForActiveTransactions() } @@ -560,14 +605,11 @@ func (s *Server) handleWebSocket(w http.ResponseWriter, r *http.Request) { // Send BootNotification go s.sendBootNotification() - // Start heartbeat loop - go s.heartbeatLoop() - - // Start meter values loop - go s.meterValuesLoop() - - // Start simulation loop to update values - go s.simulationLoop() + // Heartbeats and meter values belong to this connection. + done := make(chan struct{}) + defer close(done) + go s.heartbeatLoop(done) + go s.meterValuesLoop(done) // Read messages for { @@ -588,6 +630,7 @@ func (s *Server) handleWebSocket(w http.ResponseWriter, r *http.Request) { } func (s *Server) handleMessage(data []byte) { + s.logFrame("cs->cp", data) messageType, messageId, action, payload, err := ParseOCPPMessage(data) if err != nil { log.Printf("[%s] Failed to parse message: %v", s.serialNumber, err) @@ -1600,7 +1643,7 @@ func (s *Server) getHeartbeatInterval() int { return 60 } -func (s *Server) heartbeatLoop() { +func (s *Server) heartbeatLoop(done <-chan struct{}) { interval := s.getHeartbeatInterval() ticker := time.NewTicker(time.Duration(interval) * time.Second) defer ticker.Stop() @@ -1609,6 +1652,8 @@ func (s *Server) heartbeatLoop() { select { case <-s.stopChan: return + case <-done: + return case <-ticker.C: s.sendRequest(ActionHeartbeat, HeartbeatRequest{}) // Check if interval changed (e.g. from BootNotification response) @@ -1621,14 +1666,15 @@ func (s *Server) heartbeatLoop() { } } -// simulationLoop updates simulation values every second -func (s *Server) simulationLoop() { +// simulationLoop updates simulation values every second for as long as the +// charger runs, connected or not: a car charges through a network outage. +func (s *Server) simulationLoop(stop <-chan struct{}) { ticker := time.NewTicker(1 * time.Second) defer ticker.Stop() for { select { - case <-s.stopChan: + case <-stop: return case <-ticker.C: s.GenerateValues(time.Now()) @@ -1785,6 +1831,7 @@ func (s *Server) sendRequest(action string, payload interface{}) (json.RawMessag // Serialize WebSocket writes to prevent concurrent write panic s.writeMu.Lock() + s.logFrame("cp->cs", msg) err = conn.WriteMessage(websocket.TextMessage, msg) s.writeMu.Unlock() @@ -1831,6 +1878,7 @@ func (s *Server) sendResponse(messageId string, payload interface{}) { // Serialize WebSocket writes s.writeMu.Lock() + s.logFrame("cp->cs", msg) if err := conn.WriteMessage(websocket.TextMessage, msg); err != nil { log.Printf("[%s] Failed to send response: %v", s.serialNumber, err) } @@ -1853,6 +1901,7 @@ func (s *Server) sendError(messageId, errorCode, errorDescription string, errorD // Serialize WebSocket writes s.writeMu.Lock() + s.logFrame("cp->cs", msg) if err := conn.WriteMessage(websocket.TextMessage, msg); err != nil { log.Printf("[%s] Failed to send error response: %v", s.serialNumber, err) } From 83c11df7d812d3f27a619ef2fd0532153e7d8308 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Thu, 22 Oct 2026 09:08:46 +0200 Subject: [PATCH 2/7] feat(ocpp): the car's own limits (SetCar) A car takes at most its own current per phase and charges on its own phases: a one-phase car under a three-phase order draws on L1 only, and a car that takes 10 A draws 10 A of a 16 A offer. The charger's offer (effectivePowerW) is unchanged; its draw follows the car. Co-Authored-By: Claude Opus 5.5 --- internal/ocpp/client_test.go | 24 ++++++++++++++++++++++++ internal/ocpp/server.go | 29 ++++++++++++++++++++++++++++- 2 files changed, 52 insertions(+), 1 deletion(-) diff --git a/internal/ocpp/client_test.go b/internal/ocpp/client_test.go index dec1320..abef579 100644 --- a/internal/ocpp/client_test.go +++ b/internal/ocpp/client_test.go @@ -326,3 +326,27 @@ func TestDropConnectionKeepsTheSessionAndOneLoopSet(t *testing.T) { t.Errorf("frame log %v, want both directions", frames) } } + +// The car decides too: a one-phase car under a three-phase order draws on +// L1 only, and a car that takes 10 A takes 10 A of a 16 A offer. +func TestCarTakesWhatItCan(t *testing.T) { + // Amps per phase at the 227-230 V the simulator varies over. + amps := func(w float64) float64 { return w / 229 } + near := func(w, a float64) bool { return math.Abs(amps(w)-a) < 0.2 } + s := charging(t, 22.08, false, txProfileA(16, 3)) + s.SetCar(0, 1) + s.GenerateValues(time.Now()) + if got := s.PhasePowers(); !near(got[0], 16) || got[1] != 0 || got[2] != 0 { + t.Fatalf("one-phase car: %v, want 16 A on L1 only", got) + } + s.SetCar(10, 0) + s.GenerateValues(time.Now()) + if got := s.PhasePowers(); !near(got[0], 10) || !near(got[2], 10) { + t.Fatalf("10 A car: %v, want 10 A on each phase", got) + } + s.SetCar(0, 0) + s.GenerateValues(time.Now()) + if got := s.PhasePowers(); !near(got[1], 16) { + t.Fatalf("car without limits: %v, want the 16 A offer", got) + } +} diff --git a/internal/ocpp/server.go b/internal/ocpp/server.go index d91a08f..4bf23f7 100644 --- a/internal/ocpp/server.go +++ b/internal/ocpp/server.go @@ -137,6 +137,8 @@ type Server struct { phaseW [3]float64 // what the chargers draw on L1, L2, L3 // Simulation control + carMaxA float64 // most the car draws per phase; 0 = whatever is offered + carPhases int // phases the car charges on; 0 = as many as offered evConnected bool // Whether an EV is plugged in nextTxId int // Counter for local transaction IDs } @@ -212,6 +214,31 @@ func (s *Server) ProfileDelay() time.Duration { return s.profileDelay } +// SetCar describes the car on the charger: the most it draws per phase and +// on how many phases it charges. A one-phase car draws on L1 whatever the +// charger offers. Zeros mean the car takes what it is offered. +func (s *Server) SetCar(maxA float64, phases int) { + s.mu.Lock() + defer s.mu.Unlock() + s.carMaxA, s.carPhases = maxA, phases +} + +// carDrawLocked is what the car takes of an offer of w watts on phases. +func (s *Server) carDrawLocked(w float64, phases int) (float64, int) { + if w <= 0 || (s.carMaxA <= 0 && (s.carPhases <= 0 || s.carPhases >= phases)) { + return w, phases + } + v := s.averagePhaseVoltageLocked() + a := w / (float64(phases) * v) + if s.carMaxA > 0 { + a = math.Min(a, s.carMaxA) + } + if s.carPhases > 0 && s.carPhases < phases { + phases = s.carPhases + } + return a * float64(phases) * v, phases +} + // GetOCPPURL returns the configured Central System URL func (s *Server) GetOCPPURL() string { s.mu.RLock() @@ -2429,7 +2456,7 @@ func (s *Server) GenerateValues(simTime time.Time) { for _, tx := range s.transactions { conn := s.connectors[tx.ConnectorID] if conn.Status == StatusCharging || conn.Status == StatusSuspendedEVSE { - powerW, phases := s.computeEffectiveLocked(tx.ConnectorID, simTime) + powerW, phases := s.carDrawLocked(s.computeEffectiveLocked(tx.ConnectorID, simTime)) conn.Phases = phases if powerW <= 0 && conn.Status == StatusCharging { From 296d8749d452108ca930cdda094877e5c93787bf Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Thu, 22 Oct 2026 09:15:40 +0200 Subject: [PATCH 3/7] feat(run): headless runs from a house spec, with a timeline and a record simulator run -spec house.json -port-base N -out DIR builds one house: devices by the device-drivers driver they emulate (a hybrid whose battery follows its EMS writes, OCPP chargers with their car), the site meter behind a fuse per phase, and the house load. It serves them on ports from the base, applies a timeline of events (loads, pv, soc, plug, car, offline, online, reconnect, mark, stop), and takes the same events over POST /api/run/event while it runs. It records a sample a second of what flows through each fuse, the fuse's damage and each device's state to plant.jsonl, every OCPP frame per charger, ports.json and result.json. Self-update endpoints are off in a run. The site loop read Running without the simulator's lock while a stop wrote it; it now takes the lock. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 9 + README.md | 6 + cmd/simulator/main.go | 25 +- cmd/simulator/run.go | 753 ++++++++++++++++++++++++++++++ cmd/simulator/run_test.go | 208 +++++++++ cmd/simulator/static/openapi.json | 32 ++ docs/HEADLESS_RUN.md | 81 ++++ internal/mcp/server.go | 41 ++ internal/profiles/loader.go | 14 + 9 files changed, 1165 insertions(+), 4 deletions(-) create mode 100644 cmd/simulator/run.go create mode 100644 cmd/simulator/run_test.go create mode 100644 docs/HEADLESS_RUN.md diff --git a/CLAUDE.md b/CLAUDE.md index 6031bdc..92b3b64 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -47,6 +47,15 @@ make build-linux - MQTT: 1883+ - OCPP: 9000+ +### Headless Run + +```bash +go run ./cmd/simulator run -spec house.json -port-base 18700 -out run/ +``` + +One house from a spec, a timeline of events, a record a second +(`cmd/simulator/run.go`, [docs/HEADLESS_RUN.md](docs/HEADLESS_RUN.md)). + ### Run Tests ```bash diff --git a/README.md b/README.md index 4327881..9c23309 100644 --- a/README.md +++ b/README.md @@ -103,6 +103,12 @@ POST /api/simulators/{id}/automation/config # Configure time multiplier GET /health # Health check (always 42... I mean OK) ``` +## Headless runs + +`simulator run -spec house.json -port-base 18700 -out run/` builds one house +from a JSON spec, runs a timeline of events and records it once a second, for +an EMS's automated tests. See [docs/HEADLESS_RUN.md](docs/HEADLESS_RUN.md). + ## Configuration | Environment Variable | Default | Description | diff --git a/cmd/simulator/main.go b/cmd/simulator/main.go index cffa94f..0b68c8f 100644 --- a/cmd/simulator/main.go +++ b/cmd/simulator/main.go @@ -149,6 +149,7 @@ type Site struct { FuseOverload FuseOverload `json:"-"` FuseHeat [3]fuse.Phase `json:"-"` // each phase's thermal state (L1, L2, L3) FuseAmbientC *float64 `json:"-"` // the fuse cabinet's air; nil is 25 °C + lastPhaseW PhasePowers // what flowed through each phase's fuse on the last tick BoundMeter bool `json:"bound_meter"` // True if meter was auto-created with site P1Buffer *P1IntervalBuffer // P1 port interval buffer @@ -189,6 +190,11 @@ var ( ) func main() { + // `simulator run -spec house.json` runs one house headless (run.go). + if len(os.Args) > 1 && os.Args[1] == "run" { + os.Exit(runHeadless(os.Args[2:])) + } + // Base port configuration modbusBasePort := flag.Int("modbus-base", 5000, "Base port for Modbus servers") mqttBasePort := flag.Int("mqtt-base", 1883, "Base port for MQTT brokers") @@ -4122,6 +4128,14 @@ func getPowerContribution(sim *Simulator) float64 { return 0 } +// isRunning reads Running under the simulator's lock; a stop from the API +// or a headless run writes it while the site loop ticks. +func isRunning(sim *Simulator) bool { + sim.mu.RLock() + defer sim.mu.RUnlock() + return sim.Running +} + // getPhaseContribution returns a charger's draw per phase. A car on a // 1-phase profile loads L1 only, which phase factors cannot express. func getPhaseContribution(sim *Simulator) ([3]float64, bool) { @@ -4618,6 +4632,9 @@ func calculateAndUpdateMeter(site *Site, simDelta time.Duration) { // Calculate total power from phase powers totalPower := phasePowers.L1 + phasePowers.L2 + phasePowers.L3 + site.mu.Lock() + site.lastPhaseW = phasePowers + site.mu.Unlock() // Update BaseLoadW for backward compatibility (used for display) // This is the sum of SitePhaseLoads @@ -4741,7 +4758,7 @@ func calculateAndUpdateMeter(site *Site, simDelta time.Duration) { // Direct mode: write immediately to all meters for _, mID := range meterIDs { mSim := getSimulator(mID) - if mSim != nil && mSim.Protocol == "modbus" && mSim.Server != nil && mSim.Running { + if mSim != nil && mSim.Protocol == "modbus" && mSim.Server != nil && isRunning(mSim) { writeMeterRegisters(mSim, totalPower, phasePowers, fuseBlown) updateMeterEnergy(mSim, totalPower, energyDeltaKWh, energyDeltaWh) } @@ -4754,7 +4771,7 @@ func calculateAndUpdateMeter(site *Site, simDelta time.Duration) { // Propagate meter values to inverter sims that have built-in meter registers for _, simID := range simIDs { sim := getSimulator(simID) - if sim == nil || sim.Protocol != "modbus" || sim.Server == nil || !sim.Running { + if sim == nil || sim.Protocol != "modbus" || sim.Server == nil || !isRunning(sim) { continue } if sim.Category == "inverter" && sim.RegisterSet != nil && sim.RegisterSet.Semantic["meter_power"] != nil { @@ -4793,7 +4810,7 @@ func calculateAndUpdateMeter(site *Site, simDelta time.Duration) { // Write instant power values to all meters for _, mID := range meterIDs { mSim := getSimulator(mID) - if mSim == nil || mSim.Protocol != "modbus" || mSim.Server == nil || !mSim.Running { + if mSim == nil || mSim.Protocol != "modbus" || mSim.Server == nil || !isRunning(mSim) { continue } writeMeterRegisters(mSim, instantPower, instantPhasePowers, fuseBlown) @@ -4807,7 +4824,7 @@ func calculateAndUpdateMeter(site *Site, simDelta time.Duration) { // Propagate meter values to inverter sims that have built-in meter registers for _, simID := range simIDs { sim := getSimulator(simID) - if sim == nil || sim.Protocol != "modbus" || sim.Server == nil || !sim.Running { + if sim == nil || sim.Protocol != "modbus" || sim.Server == nil || !isRunning(sim) { continue } if sim.Category == "inverter" && sim.RegisterSet != nil && sim.RegisterSet.Semantic["meter_power"] != nil { diff --git a/cmd/simulator/run.go b/cmd/simulator/run.go new file mode 100644 index 0000000..81a0dd0 --- /dev/null +++ b/cmd/simulator/run.go @@ -0,0 +1,753 @@ +package main + +// A headless run: one house built from a JSON spec, served on ports from a +// base, driven by a timeline of events and recorded once a second into an +// output directory. An EMS under test talks to the house over Modbus and +// OCPP; its test harness reads the record and can post events while the +// house runs. +// +// simulator run -spec house.json -port-base 18700 -out run/ +// +// Ports: base+0 HTTP API, from base+10 Modbus, from base+30 OCPP, from +// base+50 MQTT. Two runs on one machine need bases 100 apart. +// +// The output directory gets ports.json (where each device listens, written +// once the house is up), plant.jsonl (a sample a second plus every event and +// mark), ocpp-.jsonl (every OCPP frame each charger reads or writes) +// and result.json (written at the end). + +import ( + "bytes" + "encoding/json" + "flag" + "fmt" + "log" + "math" + "net" + "net/http" + "os" + "os/signal" + "path/filepath" + "sort" + "strings" + "sync" + "syscall" + "time" + + simmdns "github.com/srcfl/device-simulator/internal/mdns" + "github.com/srcfl/device-simulator/internal/ocpp" + "github.com/srcfl/device-simulator/internal/ports" + "github.com/srcfl/device-simulator/internal/settings" +) + +// runSpec is the house of a headless run. +type runSpec struct { + Name string `json:"name"` + DurationS float64 `json:"duration_s"` // 0 runs until stopped + Site runSiteSpec `json:"site"` + Devices []runDeviceSpec `json:"devices"` + Timeline []runEvent `json:"timeline"` +} + +type runSiteSpec struct { + FuseA int `json:"fuse_a"` + FuseProfile string `json:"fuse_profile,omitempty"` // default iec_gg + HouseW [3]float64 `json:"house_w"` // per phase +} + +// runDeviceSpec is one device, by the device-drivers driver it emulates. +// The meter (an SDM630) comes with the site. +type runDeviceSpec struct { + ID string `json:"id"` // the run's name for it, used by events + Driver string `json:"driver"` // sungrow, pixii... or ocpp for a charger + + // Overrides of the model's parameters. + CommandResponseS *float64 `json:"command_response_s,omitempty"` + + // A hybrid inverter. + EMSControl *bool `json:"ems_control,omitempty"` // default on + BatteryKWh float64 `json:"battery_kwh,omitempty"` + SoCPct *float64 `json:"soc_pct,omitempty"` + MinSoCPct *float64 `json:"min_soc_pct,omitempty"` + MaxSoCPct *float64 `json:"max_soc_pct,omitempty"` + PVW float64 `json:"pv_w,omitempty"` + + // An OCPP 1.6 charger in client mode, and the car on it. + OCPPURL string `json:"ocpp_url,omitempty"` // user:pass@ is sent as Basic auth + MaxKW float64 `json:"max_kw,omitempty"` // on three phases; default 11.04 + Vendor string `json:"vendor,omitempty"` + Model string `json:"model,omitempty"` + PhaseSwitch bool `json:"phase_switch,omitempty"` + OCPPConfig map[string]string `json:"ocpp_config,omitempty"` // configuration keys over the defaults + CarMaxA float64 `json:"car_max_a,omitempty"` + CarPhases int `json:"car_phases,omitempty"` + Plugged bool `json:"plugged,omitempty"` // a car plugged in at the start +} + +// runEvent changes the house: at at_s seconds into the run on the timeline, +// or at once when posted to /api/run/event. +// +// mark label a named moment in the record +// loads w or a (per phase) house load +// pv pv_w, device PV power of an inverter (default the first) +// soc pct, device battery state of charge +// plug on, device car plugged in or out +// car max_a, phases, device what the car takes +// offline device, for_s the device stops answering and drops out +// of the site; back after for_s if given +// online device back again +// reconnect device, for_s a charger loses its connection for for_s; +// the car keeps charging +// stop end the run +type runEvent struct { + AtS float64 `json:"at_s,omitempty"` + Do string `json:"do"` + Device string `json:"device,omitempty"` + Label string `json:"label,omitempty"` + W *[3]float64 `json:"w,omitempty"` + A *[3]float64 `json:"a,omitempty"` + PVW *float64 `json:"pv_w,omitempty"` + Pct *float64 `json:"pct,omitempty"` + On *bool `json:"on,omitempty"` + MaxA *float64 `json:"max_a,omitempty"` + Phases *int `json:"phases,omitempty"` + ForS float64 `json:"for_s,omitempty"` +} + +type headlessRun struct { + spec runSpec + out string + base int + start time.Time + + site *Site + meter *Simulator + devs map[string]*Simulator + ids []string // spec order + + recMu sync.Mutex + rec *os.File + frameMu sync.Mutex + frames map[string]*os.File + events int + samples int + + stop chan struct{} + stopOnce sync.Once +} + +// runHeadless is `simulator run`. It returns the process exit code. +func runHeadless(args []string) int { + fs := flag.NewFlagSet("run", flag.ContinueOnError) + specPath := fs.String("spec", "", "house spec (JSON)") + base := fs.Int("port-base", 18700, "first of 100 ports on 127.0.0.1") + out := fs.String("out", "run", "output directory") + if err := fs.Parse(args); err != nil { + return 2 + } + b, err := os.ReadFile(*specPath) + if err != nil { + log.Printf("run: %v", err) + return 2 + } + var spec runSpec + if err := json.Unmarshal(b, &spec); err != nil { + log.Printf("run: %s: %v", *specPath, err) + return 2 + } + r, err := startRun(spec, *base, *out) + if err != nil { + log.Printf("run: %v", err) + return 1 + } + sig := make(chan os.Signal, 1) + signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM) + select { + case <-sig: + case <-r.stop: + } + r.end() + return 0 +} + +// startRun builds the house and starts serving, recording and the timeline. +func startRun(spec runSpec, base int, out string) (*headlessRun, error) { + if err := os.MkdirAll(out, 0o755); err != nil { + return nil, err + } + r := &headlessRun{spec: spec, out: out, base: base, devs: map[string]*Simulator{}, + frames: map[string]*os.File{}, stop: make(chan struct{})} + // A run is self-contained: its own ports and state file, none of the + // user's settings or saved simulators. + os.Setenv("DEVICE_SIMULATOR_STATE_PATH", filepath.Join(out, "state.json")) + runtimeSettings = &settings.AppSettings{ModbusBasePort: base + 10, MQTTBasePort: base + 50, OCPPBasePort: base + 30, SiteTickMs: 500} + portAllocator = ports.NewAllocator(base+10, base+50, base+30) + if mdnsAdvertiser == nil { + mdnsAdvertiser = simmdns.New() + } + f, err := os.Create(filepath.Join(out, "plant.jsonl")) + if err != nil { + return nil, err + } + r.rec = f + r.start = time.Now() + if err := r.build(); err != nil { + r.end() + return nil, err + } + ln, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", base)) + if err != nil { + r.end() + return nil, err + } + go http.Serve(ln, corsMiddleware(r.handler())) + if err := r.writePorts(); err != nil { + r.end() + return nil, err + } + go r.record() + go r.timeline() + if spec.DurationS > 0 { + time.AfterFunc(time.Duration(spec.DurationS*float64(time.Second)), r.finish) + } + log.Printf("run %q on ports %d+, recording to %s", spec.Name, base, out) + return r, nil +} + +func (r *headlessRun) build() error { + var members []int + for _, d := range r.spec.Devices { + if d.ID == "" || r.devs[d.ID] != nil { + return fmt.Errorf("device %q: every device needs its own id", d.ID) + } + var sim *Simulator + var err error + if d.Driver == "ocpp" { + sim, err = r.buildCharger(d) + } else { + sim, err = r.buildInverter(d) + } + if err != nil { + return fmt.Errorf("device %s: %w", d.ID, err) + } + r.devs[d.ID] = sim + r.ids = append(r.ids, d.ID) + members = append(members, sim.ID) + } + site, err := createSite(r.spec.Name, 0, members, 0) + if err != nil { + return err + } + r.site = site + r.meter = getSimulator(site.MeterID) + fuseProfile := r.spec.Site.FuseProfile + if fuseProfile == "" { + fuseProfile = FuseProfileIecGg + } + if _, ok := fuseProfiles[fuseProfile]; !ok { + return fmt.Errorf("fuse profile %q", fuseProfile) + } + site.mu.Lock() + if r.spec.Site.FuseA > 0 { + site.FuseSize = r.spec.Site.FuseA + } + site.FuseProfile = fuseProfile + site.PhaseFactors = PhaseFactors{L1: 1.0 / 3, L2: 1.0 / 3, L3: 1.0 / 3} + site.LoadScenario = "static" + site.mu.Unlock() + r.setLoads(r.spec.Site.HouseW) + for _, d := range r.spec.Devices { + if d.Driver == "ocpp" && d.Plugged { + if err := r.devs[d.ID].OCPPServer.SetValue("evConnected", true); err != nil { + return err + } + } + } + return nil +} + +func (r *headlessRun) buildInverter(d runDeviceSpec) (*Simulator, error) { + sim, err := createSimulator("inverter") + if err != nil { + return nil, err + } + profile := sim.ProfilesManager.ForDriver(d.Driver) + if profile == nil || profile.Protocol != "modbus" { + return nil, fmt.Errorf("no Modbus model emulates driver %q", d.Driver) + } + applyProfile(sim, profile) + sim.ProfilesManager.SetActive(profile.Name) + sim.mu.Lock() + sim.EMSControl = d.EMSControl == nil || *d.EMSControl + if d.CommandResponseS != nil { + sim.EMSResponse = seconds(*d.CommandResponseS) + } + if d.BatteryKWh > 0 { + sim.BatteryCapacityKWh = d.BatteryKWh + } + sim.mu.Unlock() + for name, v := range map[string]*float64{"battery_soc": d.SoCPct, "battery_min_soc": d.MinSoCPct, "battery_max_soc": d.MaxSoCPct} { + if v != nil { + setSemanticValue(sim, name, *v) + } + } + setSemanticValue(sim, "pv_power", math.Max(0, d.PVW)) + return sim, startSimulatorServer(sim) +} + +func (r *headlessRun) buildCharger(d runDeviceSpec) (*Simulator, error) { + if d.OCPPURL == "" { + return nil, fmt.Errorf("a charger needs ocpp_url") + } + sim, err := createSimulator("ocpp_charger") + if err != nil { + return nil, err + } + kw := d.MaxKW + if kw <= 0 { + kw = 11.04 + } + vendor, model := d.Vendor, d.Model + if vendor == "" { + vendor = "SimulatorCo" + } + if model == "" { + model = fmt.Sprintf("Sim%.0f", kw) + } + srv := ocpp.NewServer(sim.ID, 0, vendor, model, 1, kw) + cfg := map[string]string{ + "HeartbeatInterval": "60", + "MeterValueSampleInterval": "10", + "NumberOfConnectors": "1", + "ConnectorSwitch3to1PhaseSupported": fmt.Sprint(d.PhaseSwitch), + "ChargingScheduleAllowedChargingRateUnit": "Current,Power", + "SupportedFeatureProfiles": "Core,SmartCharging,RemoteTrigger,LocalAuthListManagement", + } + for k, v := range d.OCPPConfig { + cfg[k] = v + } + srv.ApplyConfiguration(cfg) + srv.SetOCPPURL(d.OCPPURL) + delay := commandResponse(sim.ProfilesManager.ForDriver("ocpp")) + if d.CommandResponseS != nil { + delay = seconds(*d.CommandResponseS) + } + srv.SetProfileDelay(delay) + srv.SetCar(d.CarMaxA, d.CarPhases) + frames, err := os.Create(filepath.Join(r.out, "ocpp-"+d.ID+".jsonl")) + if err != nil { + return nil, err + } + r.frames[d.ID] = frames + srv.SetFrameLog(func(dir string, frame []byte) { r.writeFrame(d.ID, dir, frame) }) + sim.mu.Lock() + sim.OCPPServer, sim.Device, sim.Serial, sim.EMSResponse = srv, srv, srv.SerialNumber(), delay + sim.mu.Unlock() + return sim, startSimulatorServer(sim) +} + +// handler is the simulator's API plus /api/run, without self-update. +func (r *headlessRun) handler() http.Handler { + mux := buildHTTPMux() + mux.HandleFunc("/api/run", func(w http.ResponseWriter, req *http.Request) { + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{"name": r.spec.Name, "elapsed_s": time.Since(r.start).Seconds(), + "events": r.eventCount(), "samples": r.sampleCount()}) + }) + mux.HandleFunc("/api/run/event", func(w http.ResponseWriter, req *http.Request) { + var ev runEvent + if req.Method != http.MethodPost || json.NewDecoder(req.Body).Decode(&ev) != nil { + http.Error(w, "POST a runEvent", http.StatusBadRequest) + return + } + if err := r.apply(ev); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + w.WriteHeader(http.StatusNoContent) + }) + mux.HandleFunc("/api/run/mark", func(w http.ResponseWriter, req *http.Request) { + var m struct { + Label string `json:"label"` + } + if req.Method != http.MethodPost || json.NewDecoder(req.Body).Decode(&m) != nil || m.Label == "" { + http.Error(w, `POST {"label": ...}`, http.StatusBadRequest) + return + } + r.apply(runEvent{Do: "mark", Label: m.Label}) + w.WriteHeader(http.StatusNoContent) + }) + return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + if strings.HasPrefix(req.URL.Path, "/api/update") || req.URL.Path == "/api/restart" { + http.Error(w, "not in a headless run", http.StatusForbidden) + return + } + mux.ServeHTTP(w, req) + }) +} + +// timeline applies the spec's events at their times. +func (r *headlessRun) timeline() { + evs := append([]runEvent(nil), r.spec.Timeline...) + sort.SliceStable(evs, func(i, j int) bool { return evs[i].AtS < evs[j].AtS }) + for _, ev := range evs { + select { + case <-r.stop: + return + case <-time.After(time.Until(r.start.Add(time.Duration(ev.AtS * float64(time.Second))))): + } + if err := r.apply(ev); err != nil { + log.Printf("run: event at %.1f s: %v", ev.AtS, err) + } + } +} + +func (r *headlessRun) device(id string, wantCharger bool) (*Simulator, error) { + sim := r.devs[id] + if sim == nil { + return nil, fmt.Errorf("no device %q", id) + } + if wantCharger && sim.OCPPServer == nil { + return nil, fmt.Errorf("%s is not a charger", id) + } + return sim, nil +} + +func (r *headlessRun) firstInverter() string { + for _, id := range r.ids { + if r.devs[id].Protocol == "modbus" { + return id + } + } + return "" +} + +// apply changes the house now and records the event. +func (r *headlessRun) apply(ev runEvent) error { + var err error + switch ev.Do { + case "mark": + if ev.Label == "" { + err = fmt.Errorf("mark needs a label") + } + case "loads": + switch { + case ev.W != nil: + r.setLoads(*ev.W) + case ev.A != nil: + a := *ev.A + r.setLoads([3]float64{a[0] * 230, a[1] * 230, a[2] * 230}) + default: + err = fmt.Errorf("loads needs w or a") + } + case "pv", "soc": + id := ev.Device + if id == "" { + id = r.firstInverter() + } + var sim *Simulator + if sim, err = r.device(id, false); err != nil { + break + } + switch { + case ev.Do == "pv" && ev.PVW != nil: + setSemanticValue(sim, "pv_power", math.Max(0, *ev.PVW)) + case ev.Do == "soc" && ev.Pct != nil: + setSemanticValue(sim, "battery_soc", *ev.Pct) + sim.mu.Lock() + sim.socInitialized = false + sim.mu.Unlock() + default: + err = fmt.Errorf("%s needs a value", ev.Do) + } + case "plug": + var sim *Simulator + if sim, err = r.device(ev.Device, true); err == nil { + err = sim.OCPPServer.SetValue("evConnected", ev.On == nil || *ev.On) + } + case "car": + var sim *Simulator + if sim, err = r.device(ev.Device, true); err == nil { + maxA, phases := 0.0, 0 + if ev.MaxA != nil { + maxA = *ev.MaxA + } + if ev.Phases != nil { + phases = *ev.Phases + } + sim.OCPPServer.SetCar(maxA, phases) + } + case "offline": + var sim *Simulator + if sim, err = r.device(ev.Device, false); err == nil { + stopSimulatorServer(sim) + if ev.ForS > 0 { + id := ev.Device + time.AfterFunc(time.Duration(ev.ForS*float64(time.Second)), func() { r.apply(runEvent{Do: "online", Device: id}) }) + } + } + case "online": + var sim *Simulator + if sim, err = r.device(ev.Device, false); err == nil { + err = startSimulatorServer(sim) + } + case "reconnect": + var sim *Simulator + if sim, err = r.device(ev.Device, true); err == nil { + sim.OCPPServer.DropConnection(time.Duration(ev.ForS * float64(time.Second))) + } + case "stop": + defer r.finish() + default: + err = fmt.Errorf("unknown event %q", ev.Do) + } + if err != nil { + return err + } + ev.AtS = round3(time.Since(r.start).Seconds()) + r.recMu.Lock() + r.events++ + r.recMu.Unlock() + line := map[string]any{"t": unixSeconds(time.Now()), "elapsed_s": round3(ev.AtS)} + if ev.Do == "mark" { + line["mark"] = ev.Label + } else { + line["event"] = ev + } + r.writeLine(line) + return nil +} + +func (r *headlessRun) setLoads(w [3]float64) { + r.site.mu.Lock() + defer r.site.mu.Unlock() + r.site.SitePhaseLoads = PhaseLoads{L1: w[0], L2: w[1], L3: w[2]} + r.site.BaseLoadW = w[0] + w[1] + w[2] +} + +// plantSample is one second of the house. Power is + into the site. +type plantSample struct { + T float64 `json:"t"` + Elapsed float64 `json:"elapsed_s"` + PhaseW [3]float64 `json:"phase_w"` // through each phase's fuse + PhaseA [3]float64 `json:"phase_a"` // the same, signed amperes + MeterW float64 `json:"meter_w"` // what the site meter reads + Fuse fuseSample `json:"fuse"` + Devices map[string]any `json:"devices"` +} + +type fuseSample struct { + RatingA int `json:"rating_a"` + Profile string `json:"profile"` + Damage [3]float64 `json:"damage"` + Blown [3]bool `json:"blown"` +} + +func (r *headlessRun) record() { + tick := time.NewTicker(time.Second) + defer tick.Stop() + for { + select { + case <-r.stop: + return + case now := <-tick.C: + r.writeLine(r.sample(now)) + r.recMu.Lock() + r.samples++ + r.recMu.Unlock() + } + } +} + +func (r *headlessRun) sample(now time.Time) plantSample { + r.site.mu.RLock() + pw := r.site.lastPhaseW + fs := fuseSample{RatingA: r.site.FuseSize, Profile: r.site.FuseProfile, + Damage: [3]float64{r.site.FuseOverload.L1.Damage, r.site.FuseOverload.L2.Damage, r.site.FuseOverload.L3.Damage}, + Blown: [3]bool{r.site.FuseBlown.L1, r.site.FuseBlown.L2, r.site.FuseBlown.L3}} + r.site.mu.RUnlock() + s := plantSample{T: unixSeconds(now), Elapsed: round3(now.Sub(r.start).Seconds()), + PhaseW: [3]float64{pw.L1, pw.L2, pw.L3}, Fuse: fs, Devices: map[string]any{}} + for k, w := range s.PhaseW { + s.PhaseA[k] = round3(w / 230) + } + if r.meter != nil { + s.MeterW = getSemanticValue(r.meter, "meter_power") + } + for _, id := range r.ids { + sim := r.devs[id] + sim.mu.RLock() + running := sim.Running + sim.mu.RUnlock() + if sim.OCPPServer != nil { + st := sim.OCPPServer.GetState() + pp := sim.OCPPServer.PhasePowers() + status, phases := "", 0 + if cs, ok := st["connectors"].([]map[string]interface{}); ok { + for _, c := range cs { + if c["id"] == 1 { + status, _ = c["status"].(string) + phases, _ = c["phases"].(int) + } + } + } + s.Devices[id] = map[string]any{"running": running, "connected": st["connected"], "ev_connected": st["evConnected"], + "status": status, "w": st["totalPower"], "phase_w": pp, "phases": phases, "offer_w": st["effectivePowerW"], + "transaction": st["hasActiveTransaction"]} + continue + } + s.Devices[id] = map[string]any{"running": running, "pv_w": getSemanticValue(sim, "pv_power"), + "battery_w": getBatterySignedPower(sim), "soc_pct": getSemanticValue(sim, "battery_soc"), + "ems_mode": getSemanticValue(sim, "grid_mode"), "ems_cmd": getSemanticValue(sim, "battery_cmd"), + "ems_setpoint_w": getSemanticValue(sim, "battery_setpoint")} + } + return s +} + +func (r *headlessRun) writeLine(v any) { + b, err := marshalLine(v) + if err != nil { + log.Printf("run: record: %v", err) + return + } + r.recMu.Lock() + defer r.recMu.Unlock() + if r.rec != nil { + r.rec.Write(b) + } +} + +// marshalLine is one JSON line, with "->" and "<" left as they are. +func marshalLine(v any) ([]byte, error) { + var buf bytes.Buffer + enc := json.NewEncoder(&buf) + enc.SetEscapeHTML(false) + err := enc.Encode(v) + return buf.Bytes(), err +} + +func (r *headlessRun) writeFrame(id, dir string, frame []byte) { + rec := map[string]any{"t": unixSeconds(time.Now()), "dir": dir, "frame": json.RawMessage(frame)} + if !json.Valid(frame) { + rec["frame"] = string(frame) + } + b, _ := marshalLine(rec) + r.frameMu.Lock() + defer r.frameMu.Unlock() + if f := r.frames[id]; f != nil { + f.Write(b) + } +} + +// runPorts is ports.json: where the EMS finds each device. +type runPorts struct { + HTTP int `json:"http"` + Devices map[string]runDevicePorts `json:"devices"` +} + +type runDevicePorts struct { + Driver string `json:"driver"` + Modbus int `json:"modbus_port,omitempty"` + UnitID uint8 `json:"unit_id,omitempty"` + Serial string `json:"serial,omitempty"` // an OCPP charger's identity +} + +func (r *headlessRun) writePorts() error { + p := runPorts{HTTP: r.base, Devices: map[string]runDevicePorts{}} + add := func(id, driver string, sim *Simulator) { + sim.mu.RLock() + defer sim.mu.RUnlock() + dp := runDevicePorts{Driver: driver} + if sim.Protocol == "modbus" { + dp.Modbus, dp.UnitID = sim.Port, sim.SlaveID + } else { + dp.Serial = sim.Serial + } + p.Devices[id] = dp + } + for i, id := range r.ids { + add(id, r.spec.Devices[i].Driver, r.devs[id]) + } + if r.meter != nil { + add("meter", "sdm630", r.meter) + } + b, _ := json.MarshalIndent(p, "", " ") + return os.WriteFile(filepath.Join(r.out, "ports.json"), b, 0o644) +} + +func (r *headlessRun) eventCount() int { r.recMu.Lock(); defer r.recMu.Unlock(); return r.events } +func (r *headlessRun) sampleCount() int { r.recMu.Lock(); defer r.recMu.Unlock(); return r.samples } + +// finish ends the run; runHeadless then calls end. +func (r *headlessRun) finish() { r.stopOnce.Do(func() { close(r.stop) }) } + +// end stops the house, writes result.json and closes the record. +func (r *headlessRun) end() { + r.finish() + res := map[string]any{"name": r.spec.Name, "duration_s": round3(time.Since(r.start).Seconds()), + "samples": r.sampleCount(), "events": r.eventCount()} + if r.site != nil { + r.site.mu.RLock() + res["fuse"] = map[string]any{"rating_a": r.site.FuseSize, "profile": r.site.FuseProfile, + "blown": [3]bool{r.site.FuseBlown.L1, r.site.FuseBlown.L2, r.site.FuseBlown.L3}} + r.site.mu.RUnlock() + stopSiteSimulation(r.site) + stopSiteAggregation(r.site) + } + for _, id := range r.ids { + stopSimulatorServer(r.devs[id]) + } + if r.meter != nil { + stopSimulatorServer(r.meter) + } + b, _ := json.MarshalIndent(res, "", " ") + os.WriteFile(filepath.Join(r.out, "result.json"), b, 0o644) + r.recMu.Lock() + if r.rec != nil { + r.rec.Close() + r.rec = nil + } + r.recMu.Unlock() + r.frameMu.Lock() + for id, f := range r.frames { + f.Close() + delete(r.frames, id) + } + r.frameMu.Unlock() + // Leave the process's simulator lists as they were before the run. + var ids []int + for _, sim := range r.devs { + ids = append(ids, sim.ID) + } + if r.meter != nil { + ids = append(ids, r.meter.ID) + } + simulatorsMu.Lock() + kept := simulators[:0] + for _, sim := range simulators { + drop := false + for _, id := range ids { + drop = drop || sim.ID == id + } + if !drop { + kept = append(kept, sim) + } + } + simulators = kept + simulatorsMu.Unlock() + if r.site != nil { + sitesMu.Lock() + keptSites := sites[:0] + for _, s := range sites { + if s != r.site { + keptSites = append(keptSites, s) + } + } + sites = keptSites + sitesMu.Unlock() + } +} + +func unixSeconds(t time.Time) float64 { return float64(t.UnixMicro()) / 1e6 } + +func round3(x float64) float64 { return math.Round(x*1000) / 1000 } diff --git a/cmd/simulator/run_test.go b/cmd/simulator/run_test.go new file mode 100644 index 0000000..2b041e9 --- /dev/null +++ b/cmd/simulator/run_test.go @@ -0,0 +1,208 @@ +package main + +import ( + "bufio" + "bytes" + "encoding/json" + "net" + "net/http" + "os" + "path/filepath" + "testing" + "time" + + "github.com/srcfl/device-simulator/internal/ocpp/ocpptest" +) + +// freeBase finds a port whose next 60 are free on 127.0.0.1. +func freeBase(t *testing.T) int { + t.Helper() + for try := 0; try < 50; try++ { + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + base := l.Addr().(*net.TCPAddr).Port + l.Close() + if base > 65000 { + continue + } + ok := true + for p := base; p < base+60 && ok; p++ { + l, err := net.Listen("tcp", net.JoinHostPort("127.0.0.1", itoa(p))) + if err != nil { + ok = false + continue + } + l.Close() + } + if ok { + return base + } + } + t.Fatal("no free port range") + return 0 +} + +func itoa(n int) string { return string(json.Number(json.Number(jsonInt(n)))) } + +func jsonInt(n int) string { b, _ := json.Marshal(n); return string(b) } + +type planted struct { + samples []plantSample + marks map[string]float64 + events []runEvent +} + +func readPlant(t *testing.T, out string) planted { + t.Helper() + f, err := os.Open(filepath.Join(out, "plant.jsonl")) + if err != nil { + t.Fatal(err) + } + defer f.Close() + p := planted{marks: map[string]float64{}} + sc := bufio.NewScanner(f) + for sc.Scan() { + var line struct { + plantSample + Mark string `json:"mark"` + Event *runEvent `json:"event"` + } + if err := json.Unmarshal(sc.Bytes(), &line); err != nil { + t.Fatalf("plant.jsonl: %v: %s", err, sc.Bytes()) + } + switch { + case line.Mark != "": + p.marks[line.Mark] = line.Elapsed + case line.Event != nil: + p.events = append(p.events, *line.Event) + default: + p.samples = append(p.samples, line.plantSample) + } + } + return p +} + +func runFor(t *testing.T, spec runSpec) (*headlessRun, string) { + t.Helper() + out := t.TempDir() + r, err := startRun(spec, freeBase(t), out) + if err != nil { + t.Fatal(err) + } + t.Cleanup(r.end) + return r, out +} + +func waitStop(t *testing.T, r *headlessRun, within time.Duration) { + t.Helper() + select { + case <-r.stop: + r.end() + case <-time.After(within): + t.Fatal("the run did not end") + } +} + +// A run builds its house from the spec, applies its timeline, records a +// sample a second and says where each device listens. +func TestRunRecordsTheHouse(t *testing.T) { + spec := runSpec{Name: "t", DurationS: 4, + Site: runSiteSpec{FuseA: 25, HouseW: [3]float64{690, 690, 690}}, + Devices: []runDeviceSpec{{ID: "inv", Driver: "sungrow", PVW: 1500, + SoCPct: ptr(50.0), MinSoCPct: ptr(5.0), EMSControl: ptr(false)}}, + Timeline: []runEvent{{AtS: 1.5, Do: "loads", A: &[3]float64{10, 3, 3}}, {AtS: 2, Do: "mark", Label: "after"}}} + r, out := runFor(t, spec) + waitStop(t, r, 10*time.Second) + + var ports runPorts + b, err := os.ReadFile(filepath.Join(out, "ports.json")) + if err != nil || json.Unmarshal(b, &ports) != nil { + t.Fatalf("ports.json: %v %s", err, b) + } + if ports.Devices["inv"].Modbus == 0 || ports.Devices["meter"].Modbus == 0 || ports.HTTP == 0 { + t.Fatalf("ports %+v", ports) + } + p := readPlant(t, out) + if len(p.samples) < 3 || len(p.events) != 1 || p.marks["after"] < 1.9 { + t.Fatalf("%d samples, events %v, marks %v", len(p.samples), p.events, p.marks) + } + first, last := p.samples[0], p.samples[len(p.samples)-1] + // House 690 W per phase less 1 500 W of PV spread evenly. + if first.PhaseW[0] < 180 || first.PhaseW[0] > 200 { + t.Errorf("before the step: %v W", first.PhaseW) + } + if last.PhaseW[0]-last.PhaseW[1] < 1600 || last.PhaseA[0] < 7 || last.Fuse.RatingA != 25 || last.Fuse.Profile != "iec_gg" { + t.Errorf("after 10/3/3 A: %+v", last) + } + if _, err := os.Stat(filepath.Join(out, "result.json")); err != nil { + t.Error(err) + } + if n := len(simulators); n != 0 { + t.Errorf("%d simulators left after the run", n) + } +} + +// The harness can change the house while it runs, and stop it. +func TestRunTakesEventsOverHTTP(t *testing.T) { + spec := runSpec{Name: "t", Site: runSiteSpec{FuseA: 16, HouseW: [3]float64{230, 230, 230}}, + Devices: []runDeviceSpec{{ID: "inv", Driver: "sungrow", EMSControl: ptr(false)}}} + r, out := runFor(t, spec) + post := func(path, body string) { + t.Helper() + resp, err := http.Post("http://127.0.0.1:"+itoa(r.base)+path, "application/json", bytes.NewBufferString(body)) + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusNoContent { + t.Fatalf("%s: %s", path, resp.Status) + } + } + time.Sleep(1200 * time.Millisecond) + post("/api/run/mark", `{"label": "step"}`) + post("/api/run/event", `{"do": "loads", "w": [2300, 230, 230]}`) + time.Sleep(1500 * time.Millisecond) + post("/api/run/event", `{"do": "stop"}`) + waitStop(t, r, 5*time.Second) + p := readPlant(t, out) + if _, ok := p.marks["step"]; !ok || len(p.events) != 2 { + t.Fatalf("marks %v, events %v", p.marks, p.events) + } + if last := p.samples[len(p.samples)-1]; last.PhaseW[0] < 2200 { + t.Fatalf("after the posted step: %v W", last.PhaseW) + } + resp, err := http.Get("http://127.0.0.1:" + itoa(r.base) + "/api/update/check") + if err == nil { + resp.Body.Close() + } +} + +// A charger dials the Central System in its spec, logs its frames, and a +// reconnect event drops the connection without ending the session. +func TestRunChargerReconnects(t *testing.T) { + csms := ocpptest.New(t) + spec := runSpec{Name: "t", Site: runSiteSpec{FuseA: 25}, + Devices: []runDeviceSpec{{ID: "garage", Driver: "ocpp", OCPPURL: csms.URL("ftw:pw"), Plugged: true, CarPhases: 1}}} + r, out := runFor(t, spec) + ocpptest.WaitFor(t, "StartTransaction", 10*time.Second, func() bool { return csms.Sent("StartTransaction") == 1 }) + ocpptest.WaitFor(t, "charging", 5*time.Second, func() bool { return r.devs["garage"].OCPPServer.PhasePowers()[0] > 3000 }) + if p := r.devs["garage"].OCPPServer.PhasePowers(); p[1] != 0 { + t.Fatalf("a one-phase car: %v", p) + } + if err := r.apply(runEvent{Do: "reconnect", Device: "garage", ForS: 0.2}); err != nil { + t.Fatal(err) + } + ocpptest.WaitFor(t, "a second connection", 15*time.Second, func() bool { return csms.Connections() == 2 }) + if p := r.devs["garage"].OCPPServer.PhasePowers(); p[0] < 3000 { + t.Fatalf("after the reconnect the car draws %v", p) + } + r.end() + b, err := os.ReadFile(filepath.Join(out, "ocpp-garage.jsonl")) + if err != nil || !bytes.Contains(b, []byte(`"cp->cs"`)) || !bytes.Contains(b, []byte("BootNotification")) { + t.Fatalf("frame log: %v %.300s", err, b) + } +} + +func ptr[T any](v T) *T { return &v } diff --git a/cmd/simulator/static/openapi.json b/cmd/simulator/static/openapi.json index c6eb456..bb907cc 100644 --- a/cmd/simulator/static/openapi.json +++ b/cmd/simulator/static/openapi.json @@ -422,6 +422,38 @@ "responses": { "200": { "description": "Updated" } } } }, + "/api/run": { + "get": { + "summary": "Headless run only (simulator run): name, elapsed time, events and samples so far", + "responses": { "200": { "description": "OK" } } + } + }, + "/api/run/event": { + "post": { + "summary": "Headless run only: change the house now and record the event", + "requestBody": { "content": { "application/json": { "schema": { "type": "object", "required": ["do"], "properties": { + "do": { "type": "string", "enum": ["mark", "loads", "pv", "soc", "plug", "car", "offline", "online", "reconnect", "stop"] }, + "device": { "type": "string", "description": "Device id from the run's spec" }, + "label": { "type": "string" }, + "w": { "type": "array", "items": { "type": "number" }, "description": "House load per phase, W" }, + "a": { "type": "array", "items": { "type": "number" }, "description": "House load per phase, A" }, + "pv_w": { "type": "number" }, + "pct": { "type": "number", "description": "Battery state of charge" }, + "on": { "type": "boolean", "description": "Car plugged in" }, + "max_a": { "type": "number", "description": "Most the car draws per phase" }, + "phases": { "type": "integer", "description": "Phases the car charges on" }, + "for_s": { "type": "number", "description": "How long the device stays offline or disconnected" } + } } } } }, + "responses": { "204": { "description": "Applied" }, "400": { "description": "Unknown event or device" } } + } + }, + "/api/run/mark": { + "post": { + "summary": "Headless run only: put a named mark in the record", + "requestBody": { "content": { "application/json": { "schema": { "type": "object", "required": ["label"], "properties": { "label": { "type": "string" } } } } } }, + "responses": { "204": { "description": "Marked" } } + } + }, "/api/load-scenarios": { "get": { "summary": "List load scenarios", diff --git a/docs/HEADLESS_RUN.md b/docs/HEADLESS_RUN.md new file mode 100644 index 0000000..4fc43de --- /dev/null +++ b/docs/HEADLESS_RUN.md @@ -0,0 +1,81 @@ +# Headless runs + +`simulator run` builds one house from a JSON spec, serves it on ports from a +base, applies a timeline of events and records the house once a second. An +EMS under test talks to it over Modbus and OCPP like a real site; its test +harness reads the record and can post events while the house runs. FTW's +site test (`go/test/site` in srcfl/ftw) uses it. + +```bash +simulator run -spec house.json -port-base 18700 -out run/ +``` + +The run ends after `duration_s`, on SIGINT or SIGTERM, or on a `stop` event. + +## Ports + +| Port | What | +|---|---| +| base | HTTP API (the usual one, plus `/api/run`), on 127.0.0.1 | +| base + 10 … | Modbus devices, in spec order, then the site meter | +| base + 30 … | OCPP servers (unused by chargers in client mode) | +| base + 50 … | MQTT | + +Two runs on one machine need bases at least 100 apart. `ports.json` in the +output directory says where each device is. + +## Spec + +```json +{ + "name": "chargenow-25A", + "duration_s": 0, + "site": {"fuse_a": 25, "fuse_profile": "iec_gg", "house_w": [690, 690, 690]}, + "devices": [ + {"id": "inverter", "driver": "sungrow", "battery_kwh": 10, "soc_pct": 50, "min_soc_pct": 5, "pv_w": 0}, + {"id": "garage", "driver": "ocpp", "ocpp_url": "ws://ftw:secret@127.0.0.1:8887/", + "max_kw": 11.04, "phase_switch": false, "command_response_s": 60, + "ocpp_config": {"MeterValueSampleInterval": "10"}, "car_phases": 3, "plugged": false} + ], + "timeline": [{"at_s": 120, "do": "loads", "a": [13, 13, 13]}] +} +``` + +- `driver` is the srcfl/device-drivers driver the device emulates (the file + name without `.lua`), or `ocpp` for an OCPP 1.6 charger. A Modbus device + takes the model (profile) that names that driver; its timing comes from + the model's `model` block and can be overridden per device + (`command_response_s`). +- A hybrid inverter's battery follows its EMS registers (`ems_control`, on by + default in a run). +- The site meter is an SDM630 behind a fuse per phase (`fuse_a`, + `fuse_profile` from package `fuse`). +- A charger dials `ocpp_url`; `user:password@` goes out as Basic auth. + `car_max_a` and `car_phases` are what the car takes. + +## Events + +Each has `do` and, on the timeline, `at_s`. Post one to `/api/run/event` to +apply it now, or `{"label": "..."}` to `/api/run/mark`. + +| `do` | Fields | Effect | +|---|---|---| +| `mark` | `label` | a named moment in the record | +| `loads` | `w` or `a` | house load per phase | +| `pv` | `pv_w`, `device` | PV power of an inverter (default the first) | +| `soc` | `pct`, `device` | battery state of charge | +| `plug` | `on`, `device` | car plugged in or out | +| `car` | `max_a`, `phases`, `device` | what the car takes | +| `offline` | `device`, `for_s` | the device stops answering and drops out of the site; back after `for_s` if given | +| `online` | `device` | back again | +| `reconnect` | `device`, `for_s` | a charger loses its connection for `for_s`; the car keeps charging | +| `stop` | | end the run | + +## Output + +| File | What | +|---|---| +| `ports.json` | the HTTP port, and each device's Modbus port and unit or OCPP identity | +| `plant.jsonl` | a sample a second: `phase_w` and `phase_a` through each fuse (+ into the site), `meter_w`, `fuse` (`damage`, `blown`, `rating_a`), and each device's state; plus every event (`event`) and mark (`mark`) as it happened | +| `ocpp-.jsonl` | every OCPP-J frame a charger read (`cs->cp`) or wrote (`cp->cs`), with its time | +| `result.json` | at the end: samples, events, whether a fuse blew | diff --git a/internal/mcp/server.go b/internal/mcp/server.go index ae6e9cc..89d0049 100644 --- a/internal/mcp/server.go +++ b/internal/mcp/server.go @@ -895,6 +895,47 @@ func buildToolSpecs() ([]Tool, map[string]ToolSpec) { Method: http.MethodGet, PathTemplate: "/api/load-scenarios", }, + { + Tool: Tool{ + Name: "run.status", + Description: "Headless run only: name, elapsed time, events and samples so far", + InputSchema: schemaObject(nil, nil), + }, + Method: http.MethodGet, + PathTemplate: "/api/run", + }, + { + Tool: Tool{ + Name: "run.event", + Description: "Headless run only: change the house now (do: mark, loads, pv, soc, plug, car, offline, online, reconnect, stop)", + InputSchema: schemaObject(map[string]interface{}{ + "do": schemaString("Event kind"), + "device": schemaString("Device id from the run's spec"), + "label": schemaString("Mark label"), + "w": schemaArray("House load per phase in watts", "number"), + "a": schemaArray("House load per phase in amperes", "number"), + "pv_w": schemaNumber("PV power"), + "pct": schemaNumber("State of charge"), + "on": schemaBoolean("Car plugged in"), + "max_a": schemaNumber("Most the car draws per phase"), + "phases": schemaInteger("Phases the car charges on"), + "for_s": schemaNumber("How long the device stays offline or disconnected"), + }, []string{"do"}), + }, + Method: http.MethodPost, + PathTemplate: "/api/run/event", + BodyParams: []string{"do", "device", "label", "w", "a", "pv_w", "pct", "on", "max_a", "phases", "for_s"}, + }, + { + Tool: Tool{ + Name: "run.mark", + Description: "Headless run only: put a named mark in the record", + InputSchema: schemaObject(map[string]interface{}{"label": schemaString("Mark label")}, []string{"label"}), + }, + Method: http.MethodPost, + PathTemplate: "/api/run/mark", + BodyParams: []string{"label"}, + }, { Tool: Tool{ Name: "system.info", diff --git a/internal/profiles/loader.go b/internal/profiles/loader.go index d1a7ea7..a35d3d6 100644 --- a/internal/profiles/loader.go +++ b/internal/profiles/loader.go @@ -142,6 +142,20 @@ func (p *Profile) Params() ModelParams { return *p.Model } +// ForDriver returns the first profile, by name, that emulates the +// device-drivers driver id, or nil. +func (m *Manager) ForDriver(id string) *Profile { + m.mu.RLock() + defer m.mu.RUnlock() + var best *Profile + for _, p := range m.profiles { + if p.DriverID == id && (best == nil || p.Name < best.Name) { + best = p + } + } + return best +} + // Get returns a profile by name func (m *Manager) Get(name string) *Profile { m.mu.RLock() From 57448ffba854adf81d784f3b1fc890408cc4ce57 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Thu, 22 Oct 2026 10:55:19 +0200 Subject: [PATCH 4/7] fix(run): an outage asked for during a retry keeps its length; "meter" is reserved Codex's review of #10: - A reconnect event while the charger was already waiting to retry only moved a deadline nobody read; the retry dialled after the usual delay. Each dial now waits out DropConnection's deadline first. - A device called "meter" lost its ports.json entry to the site meter. The spec now refuses that id. Co-Authored-By: Claude Opus 5.5 --- cmd/simulator/run.go | 3 +++ cmd/simulator/run_test.go | 10 ++++++++++ docs/HEADLESS_RUN.md | 3 ++- internal/ocpp/client_test.go | 22 ++++++++++++++++++++++ internal/ocpp/server.go | 18 ++++++++++++++++++ 5 files changed, 55 insertions(+), 1 deletion(-) diff --git a/cmd/simulator/run.go b/cmd/simulator/run.go index 81a0dd0..cd84799 100644 --- a/cmd/simulator/run.go +++ b/cmd/simulator/run.go @@ -220,6 +220,9 @@ func (r *headlessRun) build() error { if d.ID == "" || r.devs[d.ID] != nil { return fmt.Errorf("device %q: every device needs its own id", d.ID) } + if d.ID == "meter" { + return fmt.Errorf(`device "meter": that id is the site meter's`) + } var sim *Simulator var err error if d.Driver == "ocpp" { diff --git a/cmd/simulator/run_test.go b/cmd/simulator/run_test.go index 2b041e9..4ecf598 100644 --- a/cmd/simulator/run_test.go +++ b/cmd/simulator/run_test.go @@ -105,6 +105,16 @@ func waitStop(t *testing.T, r *headlessRun, within time.Duration) { } } +// "meter" names the site meter in ports.json, so no device may take it. +func TestRunRefusesTheMeterID(t *testing.T) { + spec := runSpec{Name: "meter-id", Site: runSiteSpec{FuseA: 25}, + Devices: []runDeviceSpec{{ID: "meter", Driver: "sungrow", BatteryKWh: 10}}} + if r, err := startRun(spec, freeBase(t), t.TempDir()); err == nil { + r.end() + t.Fatal("a device called meter was accepted") + } +} + // A run builds its house from the spec, applies its timeline, records a // sample a second and says where each device listens. func TestRunRecordsTheHouse(t *testing.T) { diff --git a/docs/HEADLESS_RUN.md b/docs/HEADLESS_RUN.md index 4fc43de..e6371bc 100644 --- a/docs/HEADLESS_RUN.md +++ b/docs/HEADLESS_RUN.md @@ -49,7 +49,8 @@ output directory says where each device is. - A hybrid inverter's battery follows its EMS registers (`ems_control`, on by default in a run). - The site meter is an SDM630 behind a fuse per phase (`fuse_a`, - `fuse_profile` from package `fuse`). + `fuse_profile` from package `fuse`). Its id is `meter`, which no device + may take. - A charger dials `ocpp_url`; `user:password@` goes out as Basic auth. `car_max_a` and `car_phases` are what the car takes. diff --git a/internal/ocpp/client_test.go b/internal/ocpp/client_test.go index abef579..c60d323 100644 --- a/internal/ocpp/client_test.go +++ b/internal/ocpp/client_test.go @@ -273,6 +273,28 @@ func TestOnePhaseDrawIsCappedAtRatedCurrent(t *testing.T) { // A dropped connection keeps the charger away for the time asked, the car // keeps charging through it, and after it comes back the charger sends one // heartbeat per interval, not one per connection it ever had. +// An outage that starts while the charger is already retrying lasts its +// full length. +func TestDropConnectionWhileRetrying(t *testing.T) { + csms := ocpptest.New(t) + csms.SetClosed(true) + s := newTestServer() + s.reconnectDelay = 50 * time.Millisecond + s.SetOCPPURL(csms.URL("")) + if err := s.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { s.Stop() }) + time.Sleep(20 * time.Millisecond) // the first dial fails; a retry waits + before := time.Now() + s.DropConnection(600 * time.Millisecond) + csms.SetClosed(false) + waitFor(t, "BootNotification", 5*time.Second, func() bool { return csms.Sent(ActionBootNotification) == 1 }) + if away := time.Since(before); away < 600*time.Millisecond { + t.Fatalf("back after %v, want at least 600 ms", away) + } +} + func TestDropConnectionKeepsTheSessionAndOneLoopSet(t *testing.T) { csms := ocpptest.New(t) csms.SetHeartbeatInterval(1) diff --git a/internal/ocpp/server.go b/internal/ocpp/server.go index 4bf23f7..87bec85 100644 --- a/internal/ocpp/server.go +++ b/internal/ocpp/server.go @@ -335,6 +335,17 @@ func (s *Server) clientLoop() { return } + // An outage asked for while a retry was already waiting ends at its + // own deadline, not the retry's. + if wait := s.untilRedial(); wait > 0 { + select { + case <-s.stopChan: + return + case <-time.After(wait): + continue + } + } + // Build the full WebSocket URL with charge point identity // OCPP 1.6 expects: ws://host/path/ChargePointId wsURL, header, err := dialTarget(url, serial) @@ -441,6 +452,13 @@ func (s *Server) DropConnection(d time.Duration) { } } +// untilRedial is how long DropConnection still keeps the charger away. +func (s *Server) untilRedial() time.Duration { + s.mu.RLock() + defer s.mu.RUnlock() + return time.Until(s.redialAt) +} + // redialWait is how long the client waits before it dials again. func (s *Server) redialWait() time.Duration { s.mu.RLock() From 053e2edcfb0507e6cdd8807aa610ed19ea93d7b2 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Thu, 22 Oct 2026 11:00:00 +0200 Subject: [PATCH 5/7] test(run): draw port bases below the ephemeral range freeBase took a base from the kernel's next ephemeral port and skipped those over 65000. macOS hands ephemeral ports out in turn, so once its cursor sat near the top every try was skipped and every run test failed with "no free port range". Co-Authored-By: Claude Opus 5.5 --- cmd/simulator/run_test.go | 24 +++++++++--------------- 1 file changed, 9 insertions(+), 15 deletions(-) diff --git a/cmd/simulator/run_test.go b/cmd/simulator/run_test.go index 4ecf598..b1041bd 100644 --- a/cmd/simulator/run_test.go +++ b/cmd/simulator/run_test.go @@ -4,32 +4,28 @@ import ( "bufio" "bytes" "encoding/json" + "math/rand" "net" "net/http" "os" "path/filepath" + "strconv" "testing" "time" "github.com/srcfl/device-simulator/internal/ocpp/ocpptest" ) -// freeBase finds a port whose next 60 are free on 127.0.0.1. +// freeBase finds a base whose next 60 ports are free on 127.0.0.1. It draws +// bases below the ephemeral range, which the kernel hands out in turn and so +// may have used up near its top. func freeBase(t *testing.T) int { t.Helper() - for try := 0; try < 50; try++ { - l, err := net.Listen("tcp", "127.0.0.1:0") - if err != nil { - t.Fatal(err) - } - base := l.Addr().(*net.TCPAddr).Port - l.Close() - if base > 65000 { - continue - } + for try := 0; try < 100; try++ { + base := 20000 + rand.Intn(300)*100 ok := true for p := base; p < base+60 && ok; p++ { - l, err := net.Listen("tcp", net.JoinHostPort("127.0.0.1", itoa(p))) + l, err := net.Listen("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(p))) if err != nil { ok = false continue @@ -44,9 +40,7 @@ func freeBase(t *testing.T) int { return 0 } -func itoa(n int) string { return string(json.Number(json.Number(jsonInt(n)))) } - -func jsonInt(n int) string { b, _ := json.Marshal(n); return string(b) } +func itoa(n int) string { return strconv.Itoa(n) } type planted struct { samples []plantSample From 72044c04ba3a45e30c0d5e3d770f5b7696e8881b Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Fri, 23 Oct 2026 07:14:59 +0200 Subject: [PATCH 6/7] feat(run): the fuse cabinet's air and a warm start MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit With the thermal fuse on main (#9), a run's spec can set the cabinet's air (site.fuse_ambient_c, default 25 °C) and start the fuses warm (site.fuse_preheat_a, a current per phase they have carried long enough to settle), as a house is in the evening. A fuse size the profile does not model is refused before the run touches the simulator's settings. The site loop now reads its tick before it starts: a run sets the tick before it builds its site, and the next run could set it again while the last loop was still reading it (the race detector caught it once runs followed each other within milliseconds). Co-Authored-By: Claude Opus 5.5 --- cmd/simulator/main.go | 6 +++++- cmd/simulator/run.go | 44 ++++++++++++++++++++++++++++++++------- cmd/simulator/run_test.go | 30 ++++++++++++++++++++++++++ docs/HEADLESS_RUN.md | 9 +++++--- 4 files changed, 78 insertions(+), 11 deletions(-) diff --git a/cmd/simulator/main.go b/cmd/simulator/main.go index 0b68c8f..fd9fa78 100644 --- a/cmd/simulator/main.go +++ b/cmd/simulator/main.go @@ -4166,8 +4166,12 @@ func startSiteAggregation(site *Site) { log.Printf("[Site %d] Aggregation started", site.ID) + // Read here, not in the loop: a headless run sets the tick before it + // builds its site, and the next run may set it again while this loop + // winds down. + interval := getSiteTickInterval() go func() { - ticker := time.NewTicker(getSiteTickInterval()) + ticker := time.NewTicker(interval) defer ticker.Stop() lastTick := time.Now() diff --git a/cmd/simulator/run.go b/cmd/simulator/run.go index cd84799..91fdf02 100644 --- a/cmd/simulator/run.go +++ b/cmd/simulator/run.go @@ -53,6 +53,12 @@ type runSiteSpec struct { FuseA int `json:"fuse_a"` FuseProfile string `json:"fuse_profile,omitempty"` // default iec_gg HouseW [3]float64 `json:"house_w"` // per phase + // FuseAmbientC is the cabinet's air around the fuses; omitted is 25 °C. + FuseAmbientC *float64 `json:"fuse_ambient_c,omitempty"` + // FusePreheatA is a current per phase the fuses have carried long + // enough to settle, so the run starts warm, as a house does in the + // evening; omitted starts cold. + FusePreheatA *[3]float64 `json:"fuse_preheat_a,omitempty"` } // runDeviceSpec is one device, by the device-drivers driver it emulates. @@ -172,6 +178,11 @@ func runHeadless(args []string) int { // startRun builds the house and starts serving, recording and the timeline. func startRun(spec runSpec, base int, out string) (*headlessRun, error) { + // A refused spec changes nothing: checked before the run takes the + // simulator's settings. + if _, err := runFuseProfile(spec.Site); err != nil { + return nil, err + } if err := os.MkdirAll(out, 0o755); err != nil { return nil, err } @@ -214,7 +225,23 @@ func startRun(spec runSpec, base int, out string) (*headlessRun, error) { return r, nil } +// runFuseProfile is the spec's fuse profile, checked with its size. +func runFuseProfile(site runSiteSpec) (string, error) { + profile := site.FuseProfile + if profile == "" { + profile = FuseProfileIecGg + } + if _, ok := fuseProfiles[profile]; !ok { + return "", fmt.Errorf("fuse profile %q", profile) + } + return profile, checkFuse(site.FuseA, profile) +} + func (r *headlessRun) build() error { + fuseProfile, err := runFuseProfile(r.spec.Site) + if err != nil { + return err + } var members []int for _, d := range r.spec.Devices { if d.ID == "" || r.devs[d.ID] != nil { @@ -243,18 +270,21 @@ func (r *headlessRun) build() error { } r.site = site r.meter = getSimulator(site.MeterID) - fuseProfile := r.spec.Site.FuseProfile - if fuseProfile == "" { - fuseProfile = FuseProfileIecGg - } - if _, ok := fuseProfiles[fuseProfile]; !ok { - return fmt.Errorf("fuse profile %q", fuseProfile) - } site.mu.Lock() if r.spec.Site.FuseA > 0 { site.FuseSize = r.spec.Site.FuseA } site.FuseProfile = fuseProfile + site.FuseAmbientC = r.spec.Site.FuseAmbientC + if pre := r.spec.Site.FusePreheatA; pre != nil { + for i, a := range pre { + site.FuseHeat[i].AmbientC = site.FuseAmbientC + if err := fuseProfiles[fuseProfile].Preheat(&site.FuseHeat[i], a, float64(site.FuseSize)); err != nil { + site.mu.Unlock() + return fmt.Errorf("fuse_preheat_a L%d: %w", i+1, err) + } + } + } site.PhaseFactors = PhaseFactors{L1: 1.0 / 3, L2: 1.0 / 3, L3: 1.0 / 3} site.LoadScenario = "static" site.mu.Unlock() diff --git a/cmd/simulator/run_test.go b/cmd/simulator/run_test.go index b1041bd..1be8fe3 100644 --- a/cmd/simulator/run_test.go +++ b/cmd/simulator/run_test.go @@ -99,6 +99,36 @@ func waitStop(t *testing.T, r *headlessRun, within time.Duration) { } } +// fuse_preheat_a starts the fuses warm and fuse_ambient_c sets the +// cabinet; a size the fuse profile does not model is refused. One run at a +// time: runs share the simulator's settings. +func TestRunFuseStartsWarm(t *testing.T) { + t.Run("warm", func(t *testing.T) { + r, _ := runFor(t, runSpec{Name: "warm", Site: runSiteSpec{FuseA: 25, HouseW: [3]float64{230, 230, 230}, + FuseAmbientC: ptr(30.0), FusePreheatA: &[3]float64{25, 12.5, 0}}}) + r.site.mu.RLock() + heat, ambient := r.site.FuseHeat, r.site.FuseAmbientC + r.site.mu.RUnlock() + if heat[0].Damage < 0.4 || heat[1].Damage >= heat[0].Damage || heat[2].Damage != 0 || ambient == nil || *ambient != 30 { + t.Fatalf("preheated at 25/12.5/0 A in 30 °C: damage %.2f/%.2f/%.2f, ambient %v", heat[0].Damage, heat[1].Damage, heat[2].Damage, ambient) + } + }) + t.Run("cold", func(t *testing.T) { + r, _ := runFor(t, runSpec{Name: "cold", Site: runSiteSpec{FuseA: 25}}) + r.site.mu.RLock() + defer r.site.mu.RUnlock() + if d := r.site.FuseHeat[0].Damage; d != 0 { + t.Errorf("a cold start has damage %.2f", d) + } + }) + t.Run("10 A gG", func(t *testing.T) { + if r, err := startRun(runSpec{Name: "ten", Site: runSiteSpec{FuseA: 10}}, freeBase(t), t.TempDir()); err == nil { + r.end() + t.Error("a 10 A gG fuse was accepted") + } + }) +} + // "meter" names the site meter in ports.json, so no device may take it. func TestRunRefusesTheMeterID(t *testing.T) { spec := runSpec{Name: "meter-id", Site: runSiteSpec{FuseA: 25}, diff --git a/docs/HEADLESS_RUN.md b/docs/HEADLESS_RUN.md index e6371bc..4d23b13 100644 --- a/docs/HEADLESS_RUN.md +++ b/docs/HEADLESS_RUN.md @@ -49,8 +49,11 @@ output directory says where each device is. - A hybrid inverter's battery follows its EMS registers (`ems_control`, on by default in a run). - The site meter is an SDM630 behind a fuse per phase (`fuse_a`, - `fuse_profile` from package `fuse`). Its id is `meter`, which no device - may take. + `fuse_profile` from package `fuse`: a thermal model, gG for 16-63 A). Its + id is `meter`, which no device may take. `fuse_ambient_c` is the + cabinet's air (default 25 °C); `fuse_preheat_a`, a current per phase the + fuses have carried long enough to settle, starts the run warm instead of + cold. A size the profile does not model is refused. - A charger dials `ocpp_url`; `user:password@` goes out as Basic auth. `car_max_a` and `car_phases` are what the car takes. @@ -77,6 +80,6 @@ apply it now, or `{"label": "..."}` to `/api/run/mark`. | File | What | |---|---| | `ports.json` | the HTTP port, and each device's Modbus port and unit or OCPP identity | -| `plant.jsonl` | a sample a second: `phase_w` and `phase_a` through each fuse (+ into the site), `meter_w`, `fuse` (`damage`, `blown`, `rating_a`), and each device's state; plus every event (`event`) and mark (`mark`) as it happened | +| `plant.jsonl` | a sample a second: `phase_w` and `phase_a` through each fuse (+ into the site), `meter_w`, `fuse` (`damage`, the element's way from ambient to melting, 0 to 1; `blown`; `rating_a`), and each device's state; plus every event (`event`) and mark (`mark`) as it happened | | `ocpp-.jsonl` | every OCPP-J frame a charger read (`cs->cp`) or wrote (`cp->cs`), with its time | | `result.json` | at the end: samples, events, whether a fuse blew | From 5a6b8479c822510c8ba08bb7dc3af75c41a961cb Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Fri, 23 Oct 2026 08:03:16 +0200 Subject: [PATCH 7/7] minor: faults and seeded random timelines for headless runs (#11) * feat(run): faults and seeded random timelines Faults for an EMS under test, as events in a headless run and over /api/run/event: a Modbus device that keeps its connections and answers nothing (silence), a site meter whose readings stand still (meter_late), one reading whose total disagrees with its phases (meter_glitch), phase currents that carry a sign (signed_currents), and a charger's own cap per phase (load_balancer). New model parameters, overridable per device: startup_delay_s (the car draws that long after an offer rises from zero) and load_balancer_a for an OCPP charger, signed_phase_currents for a meter. A spec's random block draws a timeline from a seed within the house's bounds (house load per phase under the fuse less 2 A), picks the open timing of each device, and writes timeline.json, a spec that replays it. Co-Authored-By: Claude Opus 5.5 * fix(run): each session waits out the startup delay; faults keep the latest; a late meter keeps its energy Codex's review of #11: - offerSince now resets when a connector leaves its session, so every session waits out startup_delay_s, not only the first. - An earlier silence's or outage's timer no longer ends a later one: each device counts its faults and a timer acts only for the latest. - Energy that flows during meter_late shows on the counters when the readings move again; before, it was lost. Co-Authored-By: Claude Opus 5.5 * feat(ocpp): a charger wired on one phase wired_phases: 1 in a run spec (SetWiredPhases) wires a charger on L1. It gives an order's current there whatever numberPhases says, and max_kw is its rating as wired, so a 16 A order on a 3.68 kW charger draws 16 A on L1. Before, the order became 16 A x 3 x 230 W, was capped to 3.68 kW and spread over three phases, and a one-phase car took 5.3 A of it. FTW's site test needs it for a charger wired on one phase, under Charge now and planned charging. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- cmd/simulator/main.go | 61 ++++++++- cmd/simulator/run.go | 121 +++++++++++++++++- cmd/simulator/run_random.go | 197 ++++++++++++++++++++++++++++++ cmd/simulator/run_test.go | 145 +++++++++++++++++++++- cmd/simulator/static/openapi.json | 2 +- docs/HEADLESS_RUN.md | 37 +++++- internal/mcp/server.go | 2 +- internal/modbus/server.go | 19 ++- internal/modbus/server_test.go | 38 ++++++ internal/ocpp/client_test.go | 91 ++++++++++++++ internal/ocpp/server.go | 90 +++++++++++++- internal/profiles/loader.go | 9 ++ 12 files changed, 794 insertions(+), 18 deletions(-) create mode 100644 cmd/simulator/run_random.go diff --git a/cmd/simulator/main.go b/cmd/simulator/main.go index fd9fa78..c14cbfb 100644 --- a/cmd/simulator/main.go +++ b/cmd/simulator/main.go @@ -150,6 +150,12 @@ type Site struct { FuseHeat [3]fuse.Phase `json:"-"` // each phase's thermal state (L1, L2, L3) FuseAmbientC *float64 `json:"-"` // the fuse cabinet's air; nil is 25 °C lastPhaseW PhasePowers // what flowed through each phase's fuse on the last tick + + // What the site meter reports, beside the truth (headless runs). + meterFrozenUntil time.Time // a late meter: its registers stand still until then + meterHeldKWh [2]float64 // import, export that flowed while they stood still + meterGlitch bool // the next reading's total disagrees with its phases + signedCurrents bool // phase currents carry the sign of their power BoundMeter bool `json:"bound_meter"` // True if meter was auto-created with site P1Buffer *P1IntervalBuffer // P1 port interval buffer @@ -4757,19 +4763,44 @@ func calculateAndUpdateMeter(site *Site, simDelta time.Duration) { } } + // A late meter keeps its last reading; a glitch shows once. + site.mu.Lock() + frozen := time.Now().Before(site.meterFrozenUntil) + q := meterQuirks{signedCurrents: site.signedCurrents, glitch: site.meterGlitch && !frozen} + if q.glitch { + site.meterGlitch = false + } + // The energy goes on while the readings stand still, and shows when + // they move again. + var held [2]float64 + if frozen { + if totalPower > 0 { + site.meterHeldKWh[0] += energyDeltaKWh + } else if totalPower < 0 { + site.meterHeldKWh[1] += energyDeltaKWh + } + } else { + held, site.meterHeldKWh = site.meterHeldKWh, [2]float64{} + } + site.mu.Unlock() + // Handle P1 interval mode - if p1Buffer == nil || p1Buffer.Mode == P1ModeDirect { + if frozen { + // The meter stands still; the physics above went on. + } else if p1Buffer == nil || p1Buffer.Mode == P1ModeDirect { // Direct mode: write immediately to all meters for _, mID := range meterIDs { mSim := getSimulator(mID) if mSim != nil && mSim.Protocol == "modbus" && mSim.Server != nil && isRunning(mSim) { - writeMeterRegisters(mSim, totalPower, phasePowers, fuseBlown) + writeMeterRegisters(mSim, totalPower, phasePowers, fuseBlown, q) + accumulateMeterEnergy(mSim, held[0], held[1]) updateMeterEnergy(mSim, totalPower, energyDeltaKWh, energyDeltaWh) } } // Also write to primary meter if not in meterIDs (backward compat) if len(meterIDs) == 0 { - writeMeterRegisters(meterSim, totalPower, phasePowers, fuseBlown) + writeMeterRegisters(meterSim, totalPower, phasePowers, fuseBlown, q) + accumulateMeterEnergy(meterSim, held[0], held[1]) updateMeterEnergy(meterSim, totalPower, energyDeltaKWh, energyDeltaWh) } // Propagate meter values to inverter sims that have built-in meter registers @@ -4779,7 +4810,7 @@ func calculateAndUpdateMeter(site *Site, simDelta time.Duration) { continue } if sim.Category == "inverter" && sim.RegisterSet != nil && sim.RegisterSet.Semantic["meter_power"] != nil { - writeMeterRegisters(sim, totalPower, phasePowers, fuseBlown) + writeMeterRegisters(sim, totalPower, phasePowers, fuseBlown, q) } } } else { @@ -4796,6 +4827,8 @@ func calculateAndUpdateMeter(site *Site, simDelta time.Duration) { } else if totalPower < 0 { p1Buffer.ExportEnergyDelta += energyDeltaKWh } + p1Buffer.ImportEnergyDelta += held[0] + p1Buffer.ExportEnergyDelta += held[1] // Check if interval has elapsed intervalDuration := getP1IntervalDuration(p1Buffer.Mode) @@ -4844,7 +4877,17 @@ func calculateAndUpdateMeter(site *Site, simDelta time.Duration) { // writeMeterRegisters writes power values to meter registers. // Writes to semantic names on the meter's RegisterSet, and also to global-lookup // addresses for dual-format support (SDM630 + Sungrow on same meter). -func writeMeterRegisters(meterSim *Simulator, totalPower float64, phasePowers PhasePowers, fuseBlown PhaseFuses) { +// meterQuirks are what a meter reports beside the truth. +type meterQuirks struct { + signedCurrents bool // currents carry the sign of their phase's power + glitch bool // the total reads 0 W while the phases are right +} + +func writeMeterRegisters(meterSim *Simulator, totalPower float64, phasePowers PhasePowers, fuseBlown PhaseFuses, quirks ...meterQuirks) { + var q meterQuirks + if len(quirks) > 0 { + q = quirks[0] + } if fuseBlown.L1 { phasePowers.L1 = 0 } @@ -4855,6 +4898,9 @@ func writeMeterRegisters(meterSim *Simulator, totalPower float64, phasePowers Ph phasePowers.L3 = 0 } totalPower = phasePowers.L1 + phasePowers.L2 + phasePowers.L3 + if q.glitch { + totalPower = 0 + } // Write total power via semantic name (hits the correct address for any device) setSemanticValue(meterSim, "meter_power", totalPower) @@ -4868,6 +4914,11 @@ func writeMeterRegisters(meterSim *Simulator, totalPower float64, phasePowers Ph l1Current := math.Abs(phasePowers.L1) / 230.0 l2Current := math.Abs(phasePowers.L2) / 230.0 l3Current := math.Abs(phasePowers.L3) / 230.0 + if q.signedCurrents { + l1Current = math.Copysign(l1Current, phasePowers.L1) + l2Current = math.Copysign(l2Current, phasePowers.L2) + l3Current = math.Copysign(l3Current, phasePowers.L3) + } setSemanticValue(meterSim, "meter_l1_current", l1Current) setSemanticValue(meterSim, "meter_l2_current", l2Current) diff --git a/cmd/simulator/run.go b/cmd/simulator/run.go index 91fdf02..da5e51c 100644 --- a/cmd/simulator/run.go +++ b/cmd/simulator/run.go @@ -47,6 +47,9 @@ type runSpec struct { Site runSiteSpec `json:"site"` Devices []runDeviceSpec `json:"devices"` Timeline []runEvent `json:"timeline"` + // Random adds a timeline drawn from a seed (run_random.go). The run + // writes the result to timeline.json, a spec that replays it. + Random *runRandom `json:"random,omitempty"` } type runSiteSpec struct { @@ -59,6 +62,9 @@ type runSiteSpec struct { // enough to settle, so the run starts warm, as a house does in the // evening; omitted starts cold. FusePreheatA *[3]float64 `json:"fuse_preheat_a,omitempty"` + // SignedPhaseCurrents overrides the SDM630 model's: the meter's phase + // currents carry the sign of their power. + SignedPhaseCurrents *bool `json:"signed_phase_currents,omitempty"` } // runDeviceSpec is one device, by the device-drivers driver it emulates. @@ -69,6 +75,8 @@ type runDeviceSpec struct { // Overrides of the model's parameters. CommandResponseS *float64 `json:"command_response_s,omitempty"` + StartupDelayS *float64 `json:"startup_delay_s,omitempty"` + LoadBalancerA *float64 `json:"load_balancer_a,omitempty"` // A hybrid inverter. EMSControl *bool `json:"ems_control,omitempty"` // default on @@ -80,11 +88,12 @@ type runDeviceSpec struct { // An OCPP 1.6 charger in client mode, and the car on it. OCPPURL string `json:"ocpp_url,omitempty"` // user:pass@ is sent as Basic auth - MaxKW float64 `json:"max_kw,omitempty"` // on three phases; default 11.04 + MaxKW float64 `json:"max_kw,omitempty"` // as wired; default 11.04 on three phases Vendor string `json:"vendor,omitempty"` Model string `json:"model,omitempty"` PhaseSwitch bool `json:"phase_switch,omitempty"` - OCPPConfig map[string]string `json:"ocpp_config,omitempty"` // configuration keys over the defaults + WiredPhases int `json:"wired_phases,omitempty"` // 1: on L1 only, and max_kw is its rating there + OCPPConfig map[string]string `json:"ocpp_config,omitempty"` // configuration keys over the defaults CarMaxA float64 `json:"car_max_a,omitempty"` CarPhases int `json:"car_phases,omitempty"` Plugged bool `json:"plugged,omitempty"` // a car plugged in at the start @@ -104,6 +113,13 @@ type runDeviceSpec struct { // online device back again // reconnect device, for_s a charger loses its connection for for_s; // the car keeps charging +// silence device, for_s a Modbus device keeps its connections but +// answers nothing for for_s (0: until online) +// meter_late for_s the site meter's readings stand still +// meter_glitch the next reading's total reads 0 W while +// its phases are right +// signed_currents on the meter's phase currents carry a sign +// load_balancer device, max_a the charger's own cap per phase; 0 none // stop end the run type runEvent struct { AtS float64 `json:"at_s,omitempty"` @@ -140,6 +156,28 @@ type headlessRun struct { stop chan struct{} stopOnce sync.Once + + // faultGen counts the outages and silences applied to each device, so + // the timer of an earlier one cannot end a later one. + faultMu sync.Mutex + faultGen map[string]int +} + +// fault starts an outage or silence of a device and returns a check that +// holds while it is still the latest one. +func (r *headlessRun) fault(id string) func() bool { + r.faultMu.Lock() + defer r.faultMu.Unlock() + if r.faultGen == nil { + r.faultGen = map[string]int{} + } + r.faultGen[id]++ + gen := r.faultGen[id] + return func() bool { + r.faultMu.Lock() + defer r.faultMu.Unlock() + return r.faultGen[id] == gen + } } // runHeadless is `simulator run`. It returns the process exit code. @@ -186,6 +224,18 @@ func startRun(spec runSpec, base int, out string) (*headlessRun, error) { if err := os.MkdirAll(out, 0o755); err != nil { return nil, err } + if spec.Random != nil { + evs := randomize(&spec) + log.Printf("run: %s", describeRandom(spec.Random, len(evs))) + spec.Timeline = append(spec.Timeline, evs...) + sort.SliceStable(spec.Timeline, func(i, j int) bool { return spec.Timeline[i].AtS < spec.Timeline[j].AtS }) + replay := spec + replay.Random = nil + b, _ := json.MarshalIndent(replay, "", " ") + if err := os.WriteFile(filepath.Join(out, "timeline.json"), b, 0o644); err != nil { + return nil, err + } + } r := &headlessRun{spec: spec, out: out, base: base, devs: map[string]*Simulator{}, frames: map[string]*os.File{}, stop: make(chan struct{})} // A run is self-contained: its own ports and state file, none of the @@ -287,6 +337,12 @@ func (r *headlessRun) build() error { } site.PhaseFactors = PhaseFactors{L1: 1.0 / 3, L2: 1.0 / 3, L3: 1.0 / 3} site.LoadScenario = "static" + if r.meter != nil { + site.signedCurrents = r.meter.ProfilesManager.Get("sdm630").Params().SignedPhaseCurrents + } + if v := r.spec.Site.SignedPhaseCurrents; v != nil { + site.signedCurrents = *v + } site.mu.Unlock() r.setLoads(r.spec.Site.HouseW) for _, d := range r.spec.Devices { @@ -361,12 +417,23 @@ func (r *headlessRun) buildCharger(d runDeviceSpec) (*Simulator, error) { } srv.ApplyConfiguration(cfg) srv.SetOCPPURL(d.OCPPURL) - delay := commandResponse(sim.ProfilesManager.ForDriver("ocpp")) + params := sim.ProfilesManager.ForDriver("ocpp").Params() + delay := seconds(params.CommandResponseS) if d.CommandResponseS != nil { delay = seconds(*d.CommandResponseS) } srv.SetProfileDelay(delay) + startup, lb := params.StartupDelayS, params.LoadBalancerA + if d.StartupDelayS != nil { + startup = *d.StartupDelayS + } + if d.LoadBalancerA != nil { + lb = *d.LoadBalancerA + } + srv.SetStartupDelay(seconds(startup)) + srv.SetLoadBalancer(lb) srv.SetCar(d.CarMaxA, d.CarPhases) + srv.SetWiredPhases(d.WiredPhases) frames, err := os.Create(filepath.Join(r.out, "ocpp-"+d.ID+".jsonl")) if err != nil { return nil, err @@ -514,14 +581,23 @@ func (r *headlessRun) apply(ev runEvent) error { var sim *Simulator if sim, err = r.device(ev.Device, false); err == nil { stopSimulatorServer(sim) + latest := r.fault(ev.Device) if ev.ForS > 0 { id := ev.Device - time.AfterFunc(time.Duration(ev.ForS*float64(time.Second)), func() { r.apply(runEvent{Do: "online", Device: id}) }) + time.AfterFunc(time.Duration(ev.ForS*float64(time.Second)), func() { + if latest() { + r.apply(runEvent{Do: "online", Device: id}) + } + }) } } case "online": var sim *Simulator if sim, err = r.device(ev.Device, false); err == nil { + r.fault(ev.Device) + if sim.Server != nil { + sim.Server.SetSilent(false) + } err = startSimulatorServer(sim) } case "reconnect": @@ -529,6 +605,43 @@ func (r *headlessRun) apply(ev runEvent) error { if sim, err = r.device(ev.Device, true); err == nil { sim.OCPPServer.DropConnection(time.Duration(ev.ForS * float64(time.Second))) } + case "silence": + var sim *Simulator + if sim, err = r.device(ev.Device, false); err == nil && sim.Server == nil { + err = fmt.Errorf("%s is not a Modbus device", ev.Device) + } + if err == nil { + sim.Server.SetSilent(true) + latest := r.fault(ev.Device) + if ev.ForS > 0 { + time.AfterFunc(time.Duration(ev.ForS*float64(time.Second)), func() { + if latest() { + sim.Server.SetSilent(false) + } + }) + } + } + case "meter_late": + r.site.mu.Lock() + r.site.meterFrozenUntil = time.Now().Add(time.Duration(ev.ForS * float64(time.Second))) + r.site.mu.Unlock() + case "meter_glitch": + r.site.mu.Lock() + r.site.meterGlitch = true + r.site.mu.Unlock() + case "signed_currents": + r.site.mu.Lock() + r.site.signedCurrents = ev.On == nil || *ev.On + r.site.mu.Unlock() + case "load_balancer": + var sim *Simulator + if sim, err = r.device(ev.Device, true); err == nil { + a := 0.0 + if ev.MaxA != nil { + a = *ev.MaxA + } + sim.OCPPServer.SetLoadBalancer(a) + } case "stop": defer r.finish() default: diff --git a/cmd/simulator/run_random.go b/cmd/simulator/run_random.go new file mode 100644 index 0000000..612428c --- /dev/null +++ b/cmd/simulator/run_random.go @@ -0,0 +1,197 @@ +package main + +import ( + "fmt" + "math" + "math/rand" + "sort" +) + +// runRandom draws a timeline for a headless run from a seed, within the +// house's physical bounds: house load per phase stays under the fuse less +// 2 A, so the house alone never overloads it and anything over the limit +// is the EMS's doing or a step it must answer. The same seed and spec give +// the same timeline; timeline.json in the output holds it for replay. +type runRandom struct { + Seed int64 `json:"seed"` + DurationS float64 `json:"duration_s"` // events fall in [30 s, duration_s - 30 s] + GapS float64 `json:"gap_s,omitempty"` // mean time between events; default 40 + Kinds []string `json:"kinds,omitempty"` // which kinds may happen; default all + MaxPVW float64 `json:"max_pv_w,omitempty"` // PV ceiling; default 8 000 W + MinSoCPct float64 `json:"min_soc_pct,omitempty"` // the battery's floor; default 5 +} + +// randomKinds are the events a random timeline draws, with their weights. +var randomKinds = []struct { + name string + weight int +}{ + {"loads_step", 6}, {"loads_drift", 3}, {"loads_spike", 4}, {"pv_cloud", 3}, {"soc_floor", 1}, + {"car_limit", 2}, {"silence", 1}, {"offline", 1}, {"reconnect", 2}, {"meter_late", 2}, + {"meter_glitch", 1}, {"signed_currents", 1}, {"load_balancer", 1}, +} + +// randomize fills the unset timing and quirks of the spec's devices from the +// seed and returns the events to add to its timeline. +func randomize(spec *runSpec) []runEvent { + rr := spec.Random + rng := rand.New(rand.NewSource(rr.Seed)) + gap := rr.GapS + if gap <= 0 { + gap = 40 + } + maxPV := rr.MaxPVW + if maxPV <= 0 { + maxPV = 8000 + } + minSoC := rr.MinSoCPct + if minSoC <= 0 { + minSoC = 5 + } + fuse := float64(spec.Site.FuseA) + if fuse <= 0 { + fuse = 16 + } + maxA := fuse - 2 + allowed := map[string]bool{} + for _, k := range rr.Kinds { + allowed[k] = true + } + var kinds []string + total := 0 + for _, k := range randomKinds { + if len(allowed) == 0 || allowed[k.name] { + for i := 0; i < k.weight; i++ { + kinds = append(kinds, k.name) + } + total += k.weight + } + } + + // The devices' timing: whatever the spec leaves open. + var chargers, inverters []string + for i := range spec.Devices { + d := &spec.Devices[i] + if d.Driver == "ocpp" { + chargers = append(chargers, d.ID) + if d.CommandResponseS == nil { + d.CommandResponseS = ptr([]float64{0, 5, 40, 60}[rng.Intn(4)]) + } + if d.StartupDelayS == nil { + d.StartupDelayS = ptr([]float64{0, 0, 5, 30}[rng.Intn(4)]) + } + if d.CarPhases == 0 && rng.Intn(5) == 0 { + d.CarPhases = 1 + } + } else { + inverters = append(inverters, d.ID) + } + } + if spec.Site.SignedPhaseCurrents == nil { + spec.Site.SignedPhaseCurrents = ptr(rng.Intn(2) == 0) + } + if len(kinds) == 0 { + return nil + } + + // House load per phase, in amperes, as the timeline moves it. + var level [3]float64 + for k, w := range spec.Site.HouseW { + level[k] = math.Min(maxA, math.Max(0.3, w/230)) + } + pick := func(xs []string) string { + if len(xs) == 0 { + return "" + } + return xs[rng.Intn(len(xs))] + } + between := func(lo, hi float64) float64 { return math.Round((lo+rng.Float64()*(hi-lo))*10) / 10 } + // later is when a follow-up lands: never after the run. + later := func(at float64) float64 { return math.Min(at, rr.DurationS) } + loads := func(at float64) runEvent { + a := level + return runEvent{AtS: at, Do: "loads", A: &a} + } + + var evs []runEvent + end := rr.DurationS - 30 + for t := 30 + rng.ExpFloat64()*gap; t < end; t += math.Min(3*gap, math.Max(5, rng.ExpFloat64()*gap)) { + t = math.Round(t*10) / 10 + switch kinds[rng.Intn(total)] { + case "loads_step": + if rng.Intn(2) == 0 { + level[rng.Intn(3)] = between(0.3, maxA) + } else { + for k := range level { + level[k] = between(0.3, maxA) + } + } + evs = append(evs, loads(t)) + case "loads_drift": + for i := 0; i < 3+rng.Intn(4); i++ { + for k := range level { + level[k] = math.Min(maxA, math.Max(0.3, level[k]+between(-1.5, 1.5))) + } + evs = append(evs, loads(later(t+float64(10*i)))) + } + case "loads_spike": + k := rng.Intn(3) + if room := maxA - level[k]; room > 1 { + before := level[k] + level[k] += between(1, room) + evs = append(evs, loads(t)) + level[k] = before + evs = append(evs, loads(later(t+between(5, 30)))) + } + case "pv_cloud": + if inv := pick(inverters); inv != "" { + sun := between(0.4, 1) * maxPV + evs = append(evs, runEvent{AtS: t, Do: "pv", Device: inv, PVW: ptr(math.Round(sun * between(0.1, 0.6)))}, + runEvent{AtS: later(t + between(10, 90)), Do: "pv", Device: inv, PVW: ptr(math.Round(sun))}) + } + case "soc_floor": + if inv := pick(inverters); inv != "" { + evs = append(evs, runEvent{AtS: t, Do: "soc", Device: inv, Pct: ptr(minSoC + between(0.5, 3))}) + } + case "car_limit": + if cp := pick(chargers); cp != "" { + a := 0.0 + if rng.Intn(3) > 0 { + a = between(6, 16) + } + evs = append(evs, runEvent{AtS: t, Do: "car", Device: cp, MaxA: ptr(a)}) + } + case "silence": + if inv := pick(inverters); inv != "" { + evs = append(evs, runEvent{AtS: t, Do: "silence", Device: inv, ForS: between(5, 60)}) + } + case "offline": + if inv := pick(inverters); inv != "" { + evs = append(evs, runEvent{AtS: t, Do: "offline", Device: inv, ForS: between(20, 90)}) + } + case "reconnect": + if cp := pick(chargers); cp != "" { + evs = append(evs, runEvent{AtS: t, Do: "reconnect", Device: cp, ForS: between(5, 30)}) + } + case "meter_late": + evs = append(evs, runEvent{AtS: t, Do: "meter_late", ForS: between(5, 40)}) + case "meter_glitch": + evs = append(evs, runEvent{AtS: t, Do: "meter_glitch"}) + case "signed_currents": + evs = append(evs, runEvent{AtS: t, Do: "signed_currents", On: ptr(rng.Intn(2) == 0)}) + case "load_balancer": + if cp := pick(chargers); cp != "" { + evs = append(evs, runEvent{AtS: t, Do: "load_balancer", Device: cp, MaxA: ptr(between(6, 16))}, + runEvent{AtS: later(t + between(30, 180)), Do: "load_balancer", Device: cp, MaxA: ptr(0.0)}) + } + } + } + sort.SliceStable(evs, func(i, j int) bool { return evs[i].AtS < evs[j].AtS }) + return evs +} + +func describeRandom(rr *runRandom, n int) string { + return fmt.Sprintf("random seed %d: %d events over %.0f s", rr.Seed, n, rr.DurationS) +} + +func ptr[T any](v T) *T { return &v } diff --git a/cmd/simulator/run_test.go b/cmd/simulator/run_test.go index 1be8fe3..dd33cde 100644 --- a/cmd/simulator/run_test.go +++ b/cmd/simulator/run_test.go @@ -129,6 +129,17 @@ func TestRunFuseStartsWarm(t *testing.T) { }) } +// wired_phases reaches the charger. +func TestRunWiresAOnePhaseCharger(t *testing.T) { + spec := runSpec{Name: "t", Site: runSiteSpec{FuseA: 25}, + Devices: []runDeviceSpec{{ID: "cp", Driver: "ocpp", OCPPURL: "ws://127.0.0.1:9/", MaxKW: 3.68, WiredPhases: 1}}} + r, _ := runFor(t, spec) + srv := r.devs["cp"].OCPPServer + if srv == nil || !srv.WiredOnOnePhase() { + t.Fatal("the charger is not wired on one phase") + } +} + // "meter" names the site meter in ports.json, so no device may take it. func TestRunRefusesTheMeterID(t *testing.T) { spec := runSpec{Name: "meter-id", Site: runSiteSpec{FuseA: 25}, @@ -239,4 +250,136 @@ func TestRunChargerReconnects(t *testing.T) { } } -func ptr[T any](v T) *T { return &v } +func randomSpec(seed int64) runSpec { + return runSpec{Name: "r", Site: runSiteSpec{FuseA: 20, HouseW: [3]float64{690, 690, 690}}, + Devices: []runDeviceSpec{{ID: "inv", Driver: "sungrow", PVW: 3000}, + {ID: "garage", Driver: "ocpp", OCPPURL: "ws://127.0.0.1:1/"}}, + Random: &runRandom{Seed: seed, DurationS: 1800}} +} + +// A seed gives the same timeline every time, within the house's bounds. +func TestRandomTimelinesAreSeededAndBounded(t *testing.T) { + a, b, c := randomSpec(7), randomSpec(7), randomSpec(8) + ea, eb, ec := randomize(&a), randomize(&b), randomize(&c) + ja, _ := json.Marshal(ea) + jb, _ := json.Marshal(eb) + jc, _ := json.Marshal(ec) + if !bytes.Equal(ja, jb) || bytes.Equal(ja, jc) { + t.Fatal("seed 7 twice differs, or seed 8 is the same") + } + if *a.Devices[1].CommandResponseS != *b.Devices[1].CommandResponseS || a.Site.SignedPhaseCurrents == nil { + t.Fatal("the seed did not fix the devices' timing") + } + if len(ea) < 20 { + t.Fatalf("%d events in 30 min", len(ea)) + } + kinds := map[string]int{} + for seed := int64(1); seed <= 20; seed++ { + s := randomSpec(seed) + for _, ev := range randomize(&s) { + kinds[ev.Do]++ + if ev.AtS < 30 || ev.AtS > 1800 { + t.Fatalf("seed %d: event at %.0f s outside the run", seed, ev.AtS) + } + if ev.A != nil { + for _, x := range ev.A { + if x < 0.3 || x > 18 { + t.Fatalf("seed %d: house %.1f A on a 20 A fuse", seed, x) + } + } + } + } + } + for _, k := range []string{"loads", "pv", "soc", "car", "silence", "offline", "reconnect", "meter_late", "meter_glitch", "signed_currents", "load_balancer"} { + if kinds[k] == 0 { + t.Errorf("20 seeds never drew %s", k) + } + } + only := randomSpec(1) + only.Random.Kinds = []string{"meter_glitch"} + for _, ev := range randomize(&only) { + if ev.Do != "meter_glitch" { + t.Fatalf("kinds filter let %s through", ev.Do) + } + } +} + +// The meter's faults reach what an EMS reads: a late meter stands still, +// a glitch puts 0 W in the total once, and signed currents carry the sign. +// A later silence or outage is not ended by an earlier one's timer. +func TestOverlappingFaultsKeepTheLatest(t *testing.T) { + spec := runSpec{Name: "t", Site: runSiteSpec{FuseA: 25}, + Devices: []runDeviceSpec{{ID: "inv", Driver: "sungrow", EMSControl: ptr(false)}}} + r, _ := runFor(t, spec) + inv := r.devs["inv"] + for _, ev := range []runEvent{{Do: "silence", Device: "inv", ForS: 0.3}, {Do: "silence", Device: "inv"}} { + if err := r.apply(ev); err != nil { + t.Fatal(err) + } + } + time.Sleep(600 * time.Millisecond) + if !inv.Server.Silent() { + t.Fatal("an indefinite silence ended at the earlier silence's deadline") + } + for _, ev := range []runEvent{{Do: "online", Device: "inv"}, {Do: "offline", Device: "inv", ForS: 0.3}, {Do: "offline", Device: "inv", ForS: 5}} { + if err := r.apply(ev); err != nil { + t.Fatal(err) + } + } + time.Sleep(600 * time.Millisecond) + if isRunning(inv) { + t.Fatal("a 5 s outage ended at the earlier outage's deadline") + } +} + +// Energy that flows while the meter's readings stand still shows on its +// counters once they move again. +func TestLateMeterKeepsTheEnergy(t *testing.T) { + spec := runSpec{Name: "t", Site: runSiteSpec{FuseA: 25, HouseW: [3]float64{3000, 3000, 3000}}, + Devices: []runDeviceSpec{{ID: "inv", Driver: "sungrow", EMSControl: ptr(false)}}} + r, _ := runFor(t, spec) + kwh := func() float64 { return getSemanticValue(r.meter, "total_import") } + ocpptest.WaitFor(t, "import on the meter", 5*time.Second, func() bool { return getSemanticValue(r.meter, "meter_power") > 8000 }) + t0, e0 := time.Now(), kwh() + if err := r.apply(runEvent{Do: "meter_late", ForS: 2}); err != nil { + t.Fatal(err) + } + time.Sleep(3 * time.Second) + w := getSemanticValue(r.meter, "meter_power") + got, want := kwh()-e0, w/1000*time.Since(t0).Hours() + if got < 0.8*want || got > 1.2*want { + t.Fatalf("%.5f kWh counted over %v at %.0f W, want about %.5f", got, time.Since(t0).Round(time.Millisecond), w, want) + } +} + +func TestRunMeterFaults(t *testing.T) { + spec := runSpec{Name: "t", Site: runSiteSpec{FuseA: 25, HouseW: [3]float64{690, 690, 690}}, + Devices: []runDeviceSpec{{ID: "inv", Driver: "sungrow", PVW: 5000, EMSControl: ptr(false)}}} + r, _ := runFor(t, spec) + meterW := func() float64 { return getSemanticValue(r.meter, "meter_power") } + l1A := func() float64 { return getSemanticValue(r.meter, "meter_l1_current") } + ocpptest.WaitFor(t, "export on the meter", 5*time.Second, func() bool { return meterW() < -2000 }) + if l1A() < 0 { + t.Fatalf("an SDM630 reports magnitudes, got %v A", l1A()) + } + if err := r.apply(runEvent{Do: "signed_currents", On: ptr(true)}); err != nil { + t.Fatal(err) + } + ocpptest.WaitFor(t, "a signed current", 3*time.Second, func() bool { return l1A() < 0 }) + + if err := r.apply(runEvent{Do: "meter_late", ForS: 2}); err != nil { + t.Fatal(err) + } + r.setLoads([3]float64{3000, 3000, 3000}) + time.Sleep(1200 * time.Millisecond) + if w := meterW(); w > -2000 { + t.Fatalf("a late meter moved to %v W", w) + } + ocpptest.WaitFor(t, "the meter again", 5*time.Second, func() bool { return meterW() > 0 }) + + if err := r.apply(runEvent{Do: "meter_glitch"}); err != nil { + t.Fatal(err) + } + ocpptest.WaitFor(t, "a 0 W total", 3*time.Second, func() bool { return meterW() == 0 && getSemanticValue(r.meter, "meter_l1_power") > 0 }) + ocpptest.WaitFor(t, "the total again", 3*time.Second, func() bool { return meterW() > 0 }) +} diff --git a/cmd/simulator/static/openapi.json b/cmd/simulator/static/openapi.json index bb907cc..070da58 100644 --- a/cmd/simulator/static/openapi.json +++ b/cmd/simulator/static/openapi.json @@ -432,7 +432,7 @@ "post": { "summary": "Headless run only: change the house now and record the event", "requestBody": { "content": { "application/json": { "schema": { "type": "object", "required": ["do"], "properties": { - "do": { "type": "string", "enum": ["mark", "loads", "pv", "soc", "plug", "car", "offline", "online", "reconnect", "stop"] }, + "do": { "type": "string", "enum": ["mark", "loads", "pv", "soc", "plug", "car", "offline", "online", "reconnect", "silence", "meter_late", "meter_glitch", "signed_currents", "load_balancer", "stop"] }, "device": { "type": "string", "description": "Device id from the run's spec" }, "label": { "type": "string" }, "w": { "type": "array", "items": { "type": "number" }, "description": "House load per phase, W" }, diff --git a/docs/HEADLESS_RUN.md b/docs/HEADLESS_RUN.md index 4d23b13..9c2c4e0 100644 --- a/docs/HEADLESS_RUN.md +++ b/docs/HEADLESS_RUN.md @@ -55,7 +55,9 @@ output directory says where each device is. fuses have carried long enough to settle, starts the run warm instead of cold. A size the profile does not model is refused. - A charger dials `ocpp_url`; `user:password@` goes out as Basic auth. - `car_max_a` and `car_phases` are what the car takes. + `car_max_a` and `car_phases` are what the car takes. `wired_phases: 1` + wires a charger on L1 only: it gives an order's current there, whatever + `numberPhases` says, up to `max_kw` (its rating as wired). ## Events @@ -73,8 +75,41 @@ apply it now, or `{"label": "..."}` to `/api/run/mark`. | `offline` | `device`, `for_s` | the device stops answering and drops out of the site; back after `for_s` if given | | `online` | `device` | back again | | `reconnect` | `device`, `for_s` | a charger loses its connection for `for_s`; the car keeps charging | +| `silence` | `device`, `for_s` | a Modbus device keeps its connections but answers nothing (timeouts, not refusals); 0 until `online` | +| `meter_late` | `for_s` | the site meter's readings stand still; the house goes on | +| `meter_glitch` | | the next reading's total reads 0 W while its phases are right | +| `signed_currents` | `on` | the meter's phase currents carry the sign of their power (an SDM630 reports magnitudes) | +| `load_balancer` | `device`, `max_a` | the charger's own cap per phase, whatever it is offered; 0 none | | `stop` | | end the run | +## Model parameters + +Each model's timing and quirks sit in its profile's `model` block and can +be overridden per device in the spec: + +| Parameter | Device | Meaning | +|---|---|---| +| `command_response_s` | hybrid, OCPP charger | the battery follows an EMS write, or the charger applies an accepted profile, this much later | +| `startup_delay_s` | OCPP charger | the car draws this long after an offer rises from zero; the charger says Charging meanwhile | +| `load_balancer_a` | OCPP charger | the charger's own cap per phase | +| `signed_phase_currents` | meter | currents carry a sign (`site.signed_phase_currents` in a spec) | + +## Random timelines + +```json +"random": {"seed": 42, "duration_s": 1800, "gap_s": 40, "kinds": ["loads_step", "meter_late"]} +``` + +draws events between 30 s and `duration_s` − 30 s, on average `gap_s` +apart: house load steps, drift and one-phase spikes (each phase stays under +the fuse less 2 A), PV cloud edges, the battery near its floor, the car +taking less than offered, device silence and outages, charger reconnects, +a late or glitched meter, signed currents and a load-balancer cap. It also +picks each charger's `command_response_s` (0, 5, 40 or 60 s) and +`startup_delay_s`, a one-phase car now and then, and the meter's sign, +where the spec leaves them open. `kinds` limits the draw. The run writes +`timeline.json`: the spec with the drawn timeline, which replays it. + ## Output | File | What | diff --git a/internal/mcp/server.go b/internal/mcp/server.go index 89d0049..38341cb 100644 --- a/internal/mcp/server.go +++ b/internal/mcp/server.go @@ -907,7 +907,7 @@ func buildToolSpecs() ([]Tool, map[string]ToolSpec) { { Tool: Tool{ Name: "run.event", - Description: "Headless run only: change the house now (do: mark, loads, pv, soc, plug, car, offline, online, reconnect, stop)", + Description: "Headless run only: change the house now (do: mark, loads, pv, soc, plug, car, offline, online, reconnect, silence, meter_late, meter_glitch, signed_currents, load_balancer, stop)", InputSchema: schemaObject(map[string]interface{}{ "do": schemaString("Event kind"), "device": schemaString("Device id from the run's spec"), diff --git a/internal/modbus/server.go b/internal/modbus/server.go index 49a46f2..db72f62 100644 --- a/internal/modbus/server.go +++ b/internal/modbus/server.go @@ -47,6 +47,23 @@ type Server struct { clientsMu sync.RWMutex nextClientID int RegisterSet *RegisterSet // Device-specific register definitions (nil = uses global Sungrow defs) + silent bool // read requests but never answer +} + +// SetSilent makes the server keep its connections and read every request +// but answer none, as a device that hangs does: a client sees timeouts, +// not a refused connection. +func (s *Server) SetSilent(on bool) { + s.mu.Lock() + defer s.mu.Unlock() + s.silent = on +} + +// Silent says whether the server is ignoring requests. +func (s *Server) Silent() bool { + s.mu.RLock() + defer s.mu.RUnlock() + return s.silent } // NewServer creates a new Modbus server with the specified register count @@ -375,7 +392,7 @@ func (s *Server) handleConnection(conn net.Conn) { response = s.makeException(transactionID, unitID, functionCode, 0x01) } - if response != nil { + if response != nil && !s.Silent() { conn.Write(response) } } diff --git a/internal/modbus/server_test.go b/internal/modbus/server_test.go index d5671e1..34281aa 100644 --- a/internal/modbus/server_test.go +++ b/internal/modbus/server_test.go @@ -1,6 +1,7 @@ package modbus import ( + "net" "testing" "time" ) @@ -203,3 +204,40 @@ func TestServerBoundaryConditions(t *testing.T) { t.Errorf("Boundary read: got length %d", len(regs)) } } + +// A silent server keeps the connection and answers nothing until it speaks +// again. +func TestSilentServerAnswersNothing(t *testing.T) { + s := NewServer(65536) + if err := s.Listen("127.0.0.1:0"); err != nil { + t.Fatal(err) + } + go s.Serve() + defer s.Close() + conn, err := net.Dial("tcp", s.listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + ask := func() bool { + req := []byte{0, 1, 0, 0, 0, 6, 1, 0x03, 0x13, 0x89, 0, 1} // read holding 5001 + if _, err := conn.Write(req); err != nil { + t.Fatal(err) + } + conn.SetReadDeadline(time.Now().Add(300 * time.Millisecond)) + buf := make([]byte, 64) + n, err := conn.Read(buf) + return err == nil && n > 0 + } + if !ask() { + t.Fatal("no answer before silence") + } + s.SetSilent(true) + if ask() { + t.Fatal("answered while silent") + } + s.SetSilent(false) + if !ask() { + t.Fatal("no answer after silence") + } +} diff --git a/internal/ocpp/client_test.go b/internal/ocpp/client_test.go index c60d323..f23f6f8 100644 --- a/internal/ocpp/client_test.go +++ b/internal/ocpp/client_test.go @@ -262,6 +262,35 @@ func TestCompositeScheduleKeepsThePhases(t *testing.T) { } } +// A charger wired on one phase gives an order's current on L1 whatever its +// numberPhases, and no more than its rating. +func TestOnePhaseWiring(t *testing.T) { + wired := func(p ChargingProfile) *Server { + s := NewServer(1, 0, "TestVendor", "TestModel", 1, 3.68) + s.SetWiredPhases(1) + s.transactions[1] = &Transaction{ID: 1, ConnectorID: 1, StartTime: time.Now()} + s.connectors[1].Status = StatusCharging + s.applyChargingProfile(1, &p) + s.GenerateValues(time.Now()) + return s + } + for _, c := range []struct { + name string + p ChargingProfile + want float64 + }{ + {"16 A on three phases", txProfileA(16, 3), 3680}, + {"10 A on one phase", txProfileA(10, 1), 2300}, + {"32 A", txProfileA(32, 3), 3680}, + {"3 kW", txProfileW(1, 3000), 3000}, + } { + s := wired(c.p) + if got := s.PhasePowers(); math.Abs(got[0]-c.want) > 1 || got[1] != 0 || got[2] != 0 { + t.Errorf("%s: phase powers %v, want %.0f W on L1", c.name, got, c.want) + } + } +} + // On one phase a charger gives at most its rated current per phase. func TestOnePhaseDrawIsCappedAtRatedCurrent(t *testing.T) { s := charging(t, 11.04, true, txProfileA(24, 1)) // a 16 A charger @@ -372,3 +401,65 @@ func TestCarTakesWhatItCan(t *testing.T) { t.Fatalf("car without limits: %v, want the 16 A offer", got) } } + +// The charger's own load balancer caps the draw per phase below the offer. +func TestLoadBalancerCapsTheDraw(t *testing.T) { + s := charging(t, 22.08, false, txProfileA(16, 3)) + s.SetLoadBalancer(10) + s.GenerateValues(time.Now()) + if got := s.PhasePowers(); got[0]/229 > 10.2 || got[0]/229 < 9.8 { + t.Fatalf("load balancer at 10 A: %v", got) + } + s.mu.RLock() + offer := s.computeEffectivePowerLocked(1, time.Now()) + s.mu.RUnlock() + if offer < 10900 { + t.Fatalf("the offer itself is %v W, want the profile's 16 A", offer) + } +} + +// After an offer rises from zero the car waits the start delay, while the +// charger says Charging. +// Every session waits out the delay, not only the first. +func TestStartupDelayEachSession(t *testing.T) { + s := charging(t, 22.08, false, txProfileA(16, 3)) + s.SetStartupDelay(time.Minute) + s.connectors[1].offerSince = time.Time{} + t0 := time.Now() + s.GenerateValues(t0) + s.GenerateValues(t0.Add(61 * time.Second)) + if got := s.connectors[1].CurrentPower; got < 11000 { + t.Fatalf("first session after the delay: %v W", got) + } + delete(s.transactions, 1) + s.GenerateValues(t0.Add(62 * time.Second)) + s.transactions[2] = &Transaction{ID: 2, ConnectorID: 1, StartTime: t0.Add(63 * time.Second)} + s.connectors[1].Status = StatusCharging + s.GenerateValues(t0.Add(63 * time.Second)) + if got := s.connectors[1].CurrentPower; got != 0 { + t.Fatalf("second session at once: %v W, want 0 until the delay passes", got) + } + s.GenerateValues(t0.Add(124 * time.Second)) + if got := s.connectors[1].CurrentPower; got < 11000 { + t.Fatalf("second session after the delay: %v W", got) + } +} + +func TestStartupDelay(t *testing.T) { + s := charging(t, 22.08, false, txProfileA(16, 3)) + s.SetStartupDelay(time.Minute) + s.connectors[1].offerSince = time.Time{} + t0 := time.Now() + s.GenerateValues(t0) + if got := s.connectors[1].CurrentPower; got != 0 || s.connectors[1].Status != StatusCharging { + t.Fatalf("at the offer: %v W, %s", got, s.connectors[1].Status) + } + s.GenerateValues(t0.Add(30 * time.Second)) + if got := s.connectors[1].CurrentPower; got != 0 { + t.Fatalf("after 30 s: %v W", got) + } + s.GenerateValues(t0.Add(61 * time.Second)) + if got := s.connectors[1].CurrentPower; got < 11000 { + t.Fatalf("after the delay: %v W", got) + } +} diff --git a/internal/ocpp/server.go b/internal/ocpp/server.go index 87bec85..eae1c38 100644 --- a/internal/ocpp/server.go +++ b/internal/ocpp/server.go @@ -67,6 +67,8 @@ type Connector struct { CurrentPower float64 `json:"currentPower"` EnergyWh float64 `json:"energyWh"` Phases int `json:"phases"` // 1 or 3: phases the car draws on + + offerSince time.Time // when the offer last rose from zero } // Transaction represents an active charging transaction @@ -137,8 +139,11 @@ type Server struct { phaseW [3]float64 // what the chargers draw on L1, L2, L3 // Simulation control - carMaxA float64 // most the car draws per phase; 0 = whatever is offered + carMaxA float64 // most the car draws per phase; 0 = whatever is offered + loadBalancerA float64 // the charger's own cap per phase; 0 = none + startupDelay time.Duration // from an offer rising from zero to the car drawing carPhases int // phases the car charges on; 0 = as many as offered + wiredPhases int // phases the charger is wired on (from L1); 0 = three evConnected bool // Whether an EV is plugged in nextTxId int // Counter for local transaction IDs } @@ -223,6 +228,53 @@ func (s *Server) SetCar(maxA float64, phases int) { s.carMaxA, s.carPhases = maxA, phases } +// SetLoadBalancer caps what the charger gives per phase, as a charger with +// its own load balancer (Easee) does whatever the Central System offers. +// Zero turns it off. +func (s *Server) SetLoadBalancer(a float64) { + s.mu.Lock() + defer s.mu.Unlock() + s.loadBalancerA = a +} + +// SetWiredPhases wires the charger on its first n phases from L1; 0 or 3 +// is three. A one-phase charger gives an order's current on L1, whatever +// numberPhases says, and at most its rating there. +func (s *Server) SetWiredPhases(n int) { + s.mu.Lock() + defer s.mu.Unlock() + s.wiredPhases = n +} + +// WiredOnOnePhase says whether SetWiredPhases(1) wired the charger on L1. +func (s *Server) WiredOnOnePhase() bool { + s.mu.RLock() + defer s.mu.RUnlock() + return s.wiredLocked() == 1 +} + +func (s *Server) wiredLocked() int { + if s.wiredPhases > 0 && s.wiredPhases < 3 { + return s.wiredPhases + } + return 3 +} + +// SetStartupDelay is how long the car takes to draw after an offer rises +// from zero. +func (s *Server) SetStartupDelay(d time.Duration) { + s.mu.Lock() + defer s.mu.Unlock() + s.startupDelay = d +} + +func (s *Server) loadBalancedLocked(w float64, phases int) float64 { + if s.loadBalancerA <= 0 { + return w + } + return math.Min(w, s.loadBalancerA*float64(phases)*s.averagePhaseVoltageLocked()) +} + // carDrawLocked is what the car takes of an offer of w watts on phases. func (s *Server) carDrawLocked(w float64, phases int) (float64, int) { if w <= 0 || (s.carMaxA <= 0 && (s.carPhases <= 0 || s.carPhases >= phases)) { @@ -1466,6 +1518,8 @@ func (s *Server) evaluateProfile(profile *ChargingProfile, now time.Time, txStar if activePeriod.NumberPhases != nil && *activePeriod.NumberPhases > 0 { phases = *activePeriod.NumberPhases } + // A current flows on the phases the charger has. + phases = min(phases, s.wiredLocked()) // Convert to watts unit := strings.ToUpper(strings.TrimSpace(schedule.ChargingRateUnit)) switch unit { @@ -1579,6 +1633,11 @@ func (s *Server) computeEffectiveLocked(connectorId int, now time.Time) (float64 phases = 1 maxW /= 3 } + // A charger wired on fewer phases draws on those; max_kw is its rating + // as wired. + if w := s.wiredLocked(); w < phases { + phases = w + } // Cap at hardware max effectiveW = math.Min(effectiveW, maxW) @@ -2471,18 +2530,34 @@ func (s *Server) GenerateValues(simTime time.Time) { // Simulate power consumption for active transactions var totalPower float64 var phaseW [3]float64 + offered := map[int]bool{} // connectors in a session that may draw for _, tx := range s.transactions { conn := s.connectors[tx.ConnectorID] if conn.Status == StatusCharging || conn.Status == StatusSuspendedEVSE { - powerW, phases := s.carDrawLocked(s.computeEffectiveLocked(tx.ConnectorID, simTime)) + offered[tx.ConnectorID] = true + offerW, offerPhases := s.computeEffectiveLocked(tx.ConnectorID, simTime) + offerW = s.loadBalancedLocked(offerW, offerPhases) + powerW, phases := s.carDrawLocked(offerW, offerPhases) conn.Phases = phases + // The car takes a while to start drawing after an offer rises + // from zero; the charger meanwhile reports Charging at 0 W. + if offerW > 0 { + if conn.offerSince.IsZero() { + conn.offerSince = simTime + } + if simTime.Sub(conn.offerSince) < s.startupDelay { + powerW = 0 + } + } else { + conn.offerSince = time.Time{} + } - if powerW <= 0 && conn.Status == StatusCharging { + if offerW <= 0 && conn.Status == StatusCharging { // Power dropped to 0 — transition to SuspendedEVSE conn.Status = StatusSuspendedEVSE conn.CurrentPower = 0 statusChanges = append(statusChanges, statusChange{tx.ConnectorID, StatusSuspendedEVSE}) - } else if powerW > 0 && conn.Status == StatusSuspendedEVSE { + } else if offerW > 0 && conn.Status == StatusSuspendedEVSE { // Power restored — transition back to Charging conn.Status = StatusCharging conn.CurrentPower = powerW @@ -2506,6 +2581,13 @@ func (s *Server) GenerateValues(simTime time.Time) { tx.MeterNow = int(conn.EnergyWh) } } + // A connector out of session waits out the startup delay again on its + // next offer. + for id, conn := range s.connectors { + if !offered[id] && conn != nil { + conn.offerSince = time.Time{} + } + } // Update 3-phase metering from the power on each phase s.phaseW = phaseW diff --git a/internal/profiles/loader.go b/internal/profiles/loader.go index a35d3d6..22fd728 100644 --- a/internal/profiles/loader.go +++ b/internal/profiles/loader.go @@ -48,6 +48,15 @@ type ModelParams struct { // command: a hybrid's battery after an EMS write, an OCPP charger's // output after a charging profile. CommandResponseS float64 `json:"command_response_s,omitempty"` + // StartupDelayS is how long a car on an OCPP charger takes to draw + // after an offer rises from zero. + StartupDelayS float64 `json:"startup_delay_s,omitempty"` + // LoadBalancerA is an OCPP charger's own cap per phase, whatever the + // Central System offers (Easee's load balancer). Zero is none. + LoadBalancerA float64 `json:"load_balancer_a,omitempty"` + // SignedPhaseCurrents: a meter's phase currents carry the sign of their + // phase's power. An SDM630 reports magnitudes. + SignedPhaseCurrents bool `json:"signed_phase_currents,omitempty"` } // Profile defines a device simulation profile