ocpp

# ocpp

Type-safe [OCPP](https://openchargealliance.org/) (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:

| Namespace | Contents |
|---|---|
| `ocpp`, `ocpp/rpc`, `ocpp/transport/json/*` | `Version` and WebSocket subprotocol negotiation, RPC error codes, OCPP-J frame layer, `DateTime`/`CustomData`/`JsonValue`, codec helpers |
| `ocpp/protocol/*` | Sans-IO protocol machines: the OCPP-J RPC endpoint, the charging-station and CSMS roles, and their per-version bindings |
| `ocpp/v1_6/*` | OCPP 1.6 (39 actions, Security Whitepaper extension included) |
| `ocpp/v2_0_1/*` | OCPP 2.0.1 (64 actions) |
| `ocpp/v2_1/*` | OCPP 2.1 (91 actions, including the send-only `NotifyPeriodicEventStream`) |

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:

```gleam
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"):

```gleam
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:

```gleam
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))`:

```gleam
/// 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:

```sh
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):

```sh
gleam test
```