codetics-ocpp-service 0.1.0

Codetics OCPP: Charging Station Management System (CSMS) service for OCPP 1.6J, 2.0.1 and 2.1, with TLS security profiles, smart charging, reservations and tariffs
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

// SPDX-License-Identifier: Apache-2.0
// Copyright 2026 Kevin Vavelin, Codetics (https://codetics.fr)

//! Sending a validated CALL to a connected station and mapping the outcome.

use std::time::Duration;

use ocpp_core::OcppVersion;
use serde_json::{Value, json};

use crate::error::AppError;
use crate::hub::{CallFailure, SendError};
use crate::ocpp::{v16, v21, v201};
use crate::state::App;

/// Validates `payload` against the version's message definition.
pub fn validate(version: OcppVersion, action: &str, payload: Value) -> Result<Value, AppError> {
    match version {
        OcppVersion::V16 => v16::validate_command(action, payload),
        OcppVersion::V201 => v201::validate_command(action, payload),
        OcppVersion::V21 => v21::validate_command(action, payload),
    }
    .map_err(|e| {
        AppError::BadRequest(format!(
            "invalid {action} for OCPP {version}: {} ({})",
            e.description, e.code
        ))
    })
}

/// Sends `action` to the station and returns its CALLRESULT payload.
pub async fn call(
    app: &App,
    identity: &str,
    version: OcppVersion,
    action: &str,
    payload: Value,
) -> Result<Value, AppError> {
    let payload = validate(version, action, payload)?;
    // Queueing behind other calls of the same connection, plus the call itself.
    let wait = app.cfg.call_timeout * 2 + Duration::from_secs(5);
    match app.hub.call(identity, action, payload, wait).await {
        Err(SendError::NotConnected) => Err(AppError::NotConnected { node: None }),
        Err(SendError::QueueFull) => Err(AppError::Unavailable(
            "too many commands queued for this station".into(),
        )),
        Ok(Ok(response)) => Ok(response),
        Ok(Err(CallFailure::Error(e))) => Err(AppError::Upstream(e)),
        Ok(Err(CallFailure::InvalidResponse { violation, raw })) => Err(AppError::Upstream(
            ocpp_core::RpcError::new(
                violation.kind.error_code(),
                format!("station response is invalid: {violation}"),
            )
            .with_details(json!({ "response": raw })),
        )),
        Ok(Err(CallFailure::Timeout)) => Err(AppError::GatewayTimeout(format!(
            "{action} not answered in time"
        ))),
        Ok(Err(CallFailure::Disconnected)) => {
            Err(AppError::Unavailable("station disconnected".into()))
        }
    }
}

/// OCPP version of the station's live connection on this node, or why there is none.
pub async fn connected_version(app: &App, identity: &str) -> Result<OcppVersion, AppError> {
    if let Some(h) = app.hub.get(identity) {
        return Ok(h.version);
    }
    let station = crate::db::stations::get_by_identity(&app.db, identity)
        .await?
        .ok_or_else(|| AppError::NotFound(format!("charging station '{identity}' not found")))?;
    Err(AppError::NotConnected {
        node: station
            .connected
            .then_some(station.connected_node)
            .flatten(),
    })
}

/// Response `status` field of a CALLRESULT.
pub fn status_of(response: &Value) -> &str {
    response.get("status").and_then(Value::as_str).unwrap_or("")
}