package ocpp import ( "context" "encoding/json" "errors" "fmt" "sync" "time" ) // CSMS is the Central System: it holds the live charge-point sessions keyed by // serial and answers the inbound OCPP calls a charger makes in own mode. type CSMS struct { mu sync.RWMutex sessions map[string]*Session logf func(string, ...any) } // NewCSMS builds an empty Central System. logf may be nil. func NewCSMS(logf func(string, ...any)) *CSMS { if logf == nil { logf = func(string, ...any) {} } return &CSMS{sessions: map[string]*Session{}, logf: logf} } // Accept registers a newly-connected charger. In own mode cp is the charger // connection and up must be nil; in proxy mode up is the (already-dialed) // upstream CSMS connection. Any prior session for the same serial is closed. The // returned Session is where control commands are issued. func (c *CSMS) Accept(serial, mode string, cp, up *Conn) (*Session, error) { if serial == "" { return nil, errors.New("ocpp: empty serial") } switch mode { case ModeOwn: if up != nil { _ = up.Close() up = nil } case ModeProxy: if up == nil { return nil, errors.New("ocpp: proxy mode requires an upstream connection") } default: return nil, fmt.Errorf("ocpp: cannot accept in mode %q", mode) } // Evict any existing session for this serial (a reconnect) before inserting. c.mu.Lock() old := c.sessions[serial] delete(c.sessions, serial) c.mu.Unlock() if old != nil { old.close(nil) } sess := newSession(serial, mode, cp, up, c.handleCall, c.deregister, c.logf) c.mu.Lock() c.sessions[serial] = sess c.mu.Unlock() sess.start() return sess, nil } // SessionFor returns the live session for a serial, if the charger is connected. func (c *CSMS) SessionFor(serial string) (*Session, bool) { c.mu.RLock() defer c.mu.RUnlock() s, ok := c.sessions[serial] return s, ok } // Statuses returns a snapshot of every connected charger. func (c *CSMS) Statuses() []Status { c.mu.RLock() defer c.mu.RUnlock() out := make([]Status, 0, len(c.sessions)) for _, s := range c.sessions { out = append(out, s.Snapshot()) } return out } // Shutdown closes every session. func (c *CSMS) Shutdown(context.Context) { c.mu.Lock() sessions := make([]*Session, 0, len(c.sessions)) for _, s := range c.sessions { sessions = append(sessions, s) } c.sessions = map[string]*Session{} c.mu.Unlock() for _, s := range sessions { s.close(nil) } } func (c *CSMS) deregister(s *Session) { c.mu.Lock() if c.sessions[s.serial] == s { delete(c.sessions, s.serial) } c.mu.Unlock() } // handleCall answers an inbound CALL from a charger in own mode. It implements // the CSMS side of the OCPP 1.6 core profile: enough for a charger to boot, // heartbeat, report status/meter values and open/close transactions against us. func (c *CSMS) handleCall(s *Session, action string, payload json.RawMessage) (any, string, string) { now := time.Now().UTC().Format(time.RFC3339) switch action { case "BootNotification": return map[string]any{"status": "Accepted", "currentTime": now, "interval": 300}, "", "" case "Heartbeat": return map[string]any{"currentTime": now}, "", "" case "StatusNotification", "MeterValues", "FirmwareStatusNotification", "DiagnosticsStatusNotification": return map[string]any{}, "", "" case "Authorize": return map[string]any{"idTagInfo": map[string]any{"status": "Accepted"}}, "", "" case "StartTransaction": return map[string]any{ "transactionId": s.assignTxn(), "idTagInfo": map[string]any{"status": "Accepted"}, }, "", "" case "StopTransaction": return map[string]any{"idTagInfo": map[string]any{"status": "Accepted"}}, "", "" case "DataTransfer": return map[string]any{"status": "Accepted"}, "", "" default: return nil, ErrNotImplemented, "action not supported by DriverVault CSMS" } } // assignTxn allocates a transaction id in own mode and records it in the status. func (s *Session) assignTxn() int { s.mu.Lock() s.nextTxn++ txn := s.nextTxn s.status.TransactionID = txn s.status.LastUpdated = time.Now() s.mu.Unlock() return txn }