ocpp

Type-safe OCPP (Open Charge Point Protocol) message models and OCPP-J codecs for Gleam, generated from the OCA JSON schemas with spec-derived documentation.

One package, every protocol version — each under its own namespace, so you import only the version(s) you speak:

Each version namespaces one module per spec entity (message/<action>, datatype/<type>, enum/<enum>), so the module is the namespace and no name carries a type prefix:

import ocpp/v2_0_1/enum/boot_reason
import ocpp/v2_0_1/message/boot_notification

// Build a request from its required fields (optionals default to None):
let request =
  boot_notification.new_request(
    charging_station: station,
    reason: boot_reason.PowerUp,
  )

// Encode / decode the exact OCPP-J JSON (schema constraints enforced):
let wire = boot_notification.request_to_json(request)
json.parse(body, boot_notification.response_decoder())

Dispatch by action string (ocpp/v2_0_1/dispatch) maps a wire frame's action to the right decoder and back; the frame layer lives in ocpp/transport/json/frame.

Worked example: a BootNotification round-trip

The protocol layer (ocpp/protocol/*) is sans-IO: the machines are pure functions over plain data, with no processes, sockets, or clocks. Every transition returns an Effect — an ordered list of outputs (frames to write, responses to deliver, violations to report) that your adapter interprets — and time enters only as the now argument (absolute monotonic milliseconds). The snippets below are compiled: they live verbatim in test/readme_examples_test.gleam and run with the suite.

Negotiating the version

OCPP-J carries the protocol version in the WebSocket handshake's Sec-WebSocket-Protocol header. The root ocpp module maps versions to their IANA-registered tokens and picks the winner by server preference (per RFC 6455 the server chooses; Error(Nil) means "complete the handshake without a subprotocol and close"):

pub fn subprotocol_negotiation_test() {
  // The station listed these in `Sec-WebSocket-Protocol`; this server
  // prefers 2.0.1. Echo the winner's token back in the handshake response.
  let assert Ok(version) =
    ocpp.negotiate(offered: ["ocpp2.1", "ocpp2.0.1"], supported: [
      ocpp.v2_0_1,
      ocpp.v1_6,
    ])
  ocpp.subprotocol(version) |> should.equal("ocpp2.0.1")
}

Charge-point side: boot over the station machine

ocpp/protocol/station wraps the generic RPC endpoint and drives the boot / heartbeat lifecycle. ocpp/protocol/v2_0_1 (imported as protocol here) binds it to the 2.0.1 codecs; ocpp/protocol/v1_6 and ocpp/protocol/v2_1 do the same for the other versions:

pub fn station_boot_round_trip_test() {
  // Bind the generic station machine to OCPP 2.0.1: the boot payload it
  // introduces itself with, a 30 s call timeout, a 5 s boot retry.
  let config =
    protocol.station_config(
      call_timeout_ms: 30_000,
      boot_request: boot_notification.new_request(
        charging_station: charging_station.new(
          model: "Wallbox One",
          vendor_name: "Acme",
        ),
        reason: boot_reason.PowerUp,
      ),
      boot_retry_ms: 5000,
    )

  // `init` starts the provisioning lifecycle: the machine immediately asks
  // the adapter to write the BootNotification CALL to the socket.
  let #(machine, fx) = station.init(config, 0)
  let assert [station.WireOut(call)] = effect.to_list(fx)
  let assert Ok(frame.Call(boot_id, "BootNotification", _)) = frame.decode(call)

  // The CSMS accepts with a 300 s heartbeat interval. (Forged by hand here;
  // a real CALLRESULT arrives on the WebSocket.)
  let assert Ok(csms_time) = datetime.parse_rfc3339("2026-01-01T00:00:00Z")
  let reply =
    frame.encode_call_result(
      boot_id,
      boot_notification.response_to_json(boot_notification.new_response(
        current_time: csms_time,
        interval: 300,
        status: registration_status.Accepted,
      )),
    )

  // Feeding the CALLRESULT registers the station. The machine consumes the
  // boot reply itself — no output for the application — and schedules the
  // first heartbeat one interval after acceptance.
  let #(machine, fx) =
    station.update(machine, station.WireIn(json.to_string(reply)), 100)
  effect.to_list(fx) |> should.equal([])
  station.registration(machine)
  |> should.equal(station.Registered(
    interval_ms: 300_000,
    heartbeat_due: 300_100,
  ))
  station.next_deadline(machine) |> should.equal(Some(300_100))
}

From here the adapter keeps one timer aimed at next_deadline and feeds station.Tick when it fires (heartbeats, boot retries, and call timeouts all run off it), station.WireIn for every inbound frame, and station.CallRequested/station.ReplyProvided for the application's own traffic.

CSMS side: answering inbound calls

ocpp/protocol/csms is the per-connection CSMS machine. Inbound CALLs surface as CallReceived(id, request) with the request already decoded to the version's dispatch.Request union; the application answers with the endpoint's typed reply, Result(res, #(rpc.ErrorCode, String)):

/// The CSMS application's request handler: pattern-match the typed request
/// union and return the endpoint's typed reply — `Ok(response)` goes out as
/// a CALLRESULT, `Error(#(code, description))` as a CALLERROR.
fn handle_request(
  request: dispatch.Request,
) -> Result(dispatch.Response, #(rpc.ErrorCode, String)) {
  case request {
    dispatch.BootNotificationRequest(_) -> {
      let assert Ok(now) = datetime.parse_rfc3339("2026-01-01T00:00:00Z")
      Ok(
        dispatch.BootNotificationResponse(boot_notification.new_response(
          current_time: now,
          interval: 300,
          status: registration_status.Accepted,
        )),
      )
    }
    _ -> Error(#(rpc.NotImplemented, "This CSMS only handles boots"))
  }
}

pub fn csms_inbound_call_handling_test() {
  // One machine per station connection. `RejectUnregistered` answers
  // non-boot CALLs from an unaccepted station with a CALLERROR
  // `SecurityError` (B02.FR.09) instead of delivering them.
  let #(machine, _fx) =
    csms.init(
      protocol.csms_config(
        call_timeout_ms: 30_000,
        enforcement: csms.RejectUnregistered,
      ),
      0,
    )

  // A BootNotification CALL arrives on the socket. (Forged here with the
  // dispatch codec: `request_to_json` recovers the action string and the
  // JSON payload from the typed union.)
  let #(action, payload) =
    dispatch.request_to_json(
      dispatch.BootNotificationRequest(boot_notification.new_request(
        charging_station: charging_station.new(
          model: "Wallbox One",
          vendor_name: "Acme",
        ),
        reason: boot_reason.PowerUp,
      )),
    )
  let call = json.to_string(frame.encode_call("19223201", action, payload))

  // The machine decodes the frame and hands the application a typed
  // request, keyed by the CALL's message id.
  let #(machine, fx) = csms.update(machine, csms.WireIn(call), 50)
  let assert [csms.CallReceived(id, request)] = effect.to_list(fx)

  // The application replies through its handler; the machine writes the
  // CALLRESULT and, because the reply accepted a boot request, now tracks
  // the peer station as registered.
  let #(machine, fx) =
    csms.update(machine, csms.ReplyProvided(id, handle_request(request)), 60)
  let assert [csms.WireOut(result)] = effect.to_list(fx)
  let assert Ok(frame.CallResult(_, _)) = frame.decode(result)
  csms.peer_registration(machine) |> should.equal(csms.PeerRegistered)
}

For a full station ↔ CSMS conversation — both machines wired back to back, covering heartbeats, CSMS-initiated calls, timeouts, and pre-acceptance queueing — see test/ocpp_protocol_conversation_test.gleam (and its 1.6 and 2.1 siblings).

Development

The per-version modules are generated — do not edit them by hand. The generator lives in dev/ (so it ships with neither the package nor its dependencies). To regenerate from the vendored schemas, from the repo root:

gleam dev                 # writes the staging tree under generated/
gleam format generated
for v in v1_6 v2_0_1 v2_1; do
  rsync -a --delete generated/ocpp/$v/ src/ocpp/$v/
done

Run the tests (codec round-trips, protocol-machine conversations, and the codegen suites together):

gleam test