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 77043c0..c14cbfb 100644 --- a/cmd/simulator/main.go +++ b/cmd/simulator/main.go @@ -149,6 +149,13 @@ 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 + + // 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 @@ -189,6 +196,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") @@ -1953,46 +1965,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 +1983,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{}{ @@ -4117,6 +4134,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) { @@ -4147,8 +4172,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() @@ -4613,6 +4642,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 @@ -4731,29 +4763,54 @@ 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 && mSim.Running { - writeMeterRegisters(mSim, totalPower, phasePowers, fuseBlown) + if mSim != nil && mSim.Protocol == "modbus" && mSim.Server != nil && isRunning(mSim) { + 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 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 { - writeMeterRegisters(sim, totalPower, phasePowers, fuseBlown) + writeMeterRegisters(sim, totalPower, phasePowers, fuseBlown, q) } } } else { @@ -4770,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) @@ -4788,7 +4847,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) @@ -4802,7 +4861,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 { @@ -4818,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 } @@ -4829,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) @@ -4842,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 new file mode 100644 index 0000000..da5e51c --- /dev/null +++ b/cmd/simulator/run.go @@ -0,0 +1,899 @@ +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"` + // 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 { + 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"` + // 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. +// 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"` + 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 + 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"` // as wired; default 11.04 on three phases + Vendor string `json:"vendor,omitempty"` + Model string `json:"model,omitempty"` + PhaseSwitch bool `json:"phase_switch,omitempty"` + 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 +} + +// 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 +// 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"` + 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 + + // 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. +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) { + // 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 + } + 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 + // 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 +} + +// 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 { + 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" { + 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) + 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" + 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 { + 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) + 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 + } + 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) + latest := r.fault(ev.Device) + if ev.ForS > 0 { + id := ev.Device + 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": + var sim *Simulator + 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: + 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_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 new file mode 100644 index 0000000..dd33cde --- /dev/null +++ b/cmd/simulator/run_test.go @@ -0,0 +1,385 @@ +package main + +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 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 < 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", strconv.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 strconv.Itoa(n) } + +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") + } +} + +// 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") + } + }) +} + +// 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}, + 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) { + 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 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 c6eb456..070da58 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", "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" }, + "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..9c2c4e0 --- /dev/null +++ b/docs/HEADLESS_RUN.md @@ -0,0 +1,120 @@ +# 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 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. `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 + +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 | +| `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 | +|---|---| +| `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`, 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 | diff --git a/internal/mcp/server.go b/internal/mcp/server.go index ae6e9cc..38341cb 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, 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"), + "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/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 7684934..f23f6f8 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" @@ -261,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 @@ -268,3 +298,168 @@ 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. +// 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) + 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) + } +} + +// 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) + } +} + +// 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/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..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 @@ -124,6 +126,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 @@ -135,6 +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 + 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 } @@ -210,6 +219,78 @@ 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 +} + +// 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)) { + 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() @@ -252,6 +333,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() @@ -302,6 +387,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) @@ -331,7 +427,7 @@ func (s *Server) clientLoop() { select { case <-s.stopChan: return - case <-time.After(s.reconnectDelay): + case <-time.After(s.redialWait()): continue } } @@ -348,17 +444,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 +461,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 +491,53 @@ 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() + } +} + +// 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() + 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 +569,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 +579,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 +702,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 +727,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) @@ -1378,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 { @@ -1491,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) @@ -1600,7 +1747,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 +1756,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 +1770,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 +1935,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 +1982,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 +2005,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) } @@ -2377,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.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 @@ -2412,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 d1a7ea7..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 @@ -142,6 +151,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()