ocpp_service/ocpp/
v21.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 Kevin Vavelin, Codetics (https://codetics.fr)
3
4//! OCPP 2.1 CSMS behaviour.
5
6use ocpp_core::{ErrorCode, OcppVersion, RpcError};
7use ocpp21::*;
8use serde_json::Value;
9
10use super::{StationCtx, i32_of, internal};
11use crate::db::{self, Ts, stations::BootInfo, stations::StatusUpdate, transactions::TxEvent};
12
13pub fn decode_error(e: DecodeError) -> RpcError {
14    match e {
15        DecodeError::UnknownAction(a) => RpcError::not_implemented(&a),
16        DecodeError::WrongDirection(a) => RpcError::not_supported(a.as_str()),
17        DecodeError::Invalid(v) => v.into(),
18    }
19}
20
21fn json<T: serde::Serialize>(v: T) -> Result<Value, RpcError> {
22    serde_json::to_value(v).map_err(internal)
23}
24
25fn id_token_info(a: &db::tokens::AuthResult) -> IdTokenInfo {
26    let status = match a.status {
27        "Accepted" => AuthorizationStatusEnum::Accepted,
28        "Blocked" => AuthorizationStatusEnum::Blocked,
29        "Expired" => AuthorizationStatusEnum::Expired,
30        "Unknown" => AuthorizationStatusEnum::Unknown,
31        _ => AuthorizationStatusEnum::Invalid,
32    };
33    IdTokenInfo {
34        custom_data: None,
35        status,
36        cache_expiry_date_time: a.expires_at.map(|t| t.0),
37        charging_priority: None,
38        language1: None,
39        evse_id: None,
40        group_id_token: a.group_id.as_ref().map(|g| IdToken {
41            custom_data: None,
42            additional_info: None,
43            id_token: g.clone(),
44            r#type: "Central".to_owned(),
45        }),
46        language2: None,
47        personal_message: None,
48    }
49}
50
51/// Value of a measurand in base units (Wh, W, A), from the last meter value
52/// containing it: the phase-less value if present, otherwise the sum of L1–L3.
53fn measurand(values: &[MeterValue], m: MeasurandEnum) -> Option<f64> {
54    let scale = |s: &SampledValue| {
55        let (unit, mult) = s
56            .unit_of_measure
57            .as_ref()
58            .map(|u| (u.unit.as_deref().unwrap_or("Wh"), u.multiplier.unwrap_or(0)))
59            .unwrap_or(("Wh", 0));
60        let v = s.value * 10f64.powi(mult as i32);
61        if unit.starts_with('k') { v * 1000.0 } else { v }
62    };
63    for mv in values.iter().rev() {
64        // An absent measurand means Energy.Active.Import.Register.
65        let matching: Vec<&SampledValue> = mv
66            .sampled_value
67            .iter()
68            .filter(|s| {
69                s.measurand
70                    .unwrap_or(MeasurandEnum::EnergyActiveImportRegister)
71                    == m
72            })
73            .collect();
74        if let Some(total) = matching.iter().find(|s| s.phase.is_none()) {
75            return Some(scale(total));
76        }
77        let phases: Vec<f64> = matching
78            .iter()
79            .filter(|s| matches!(s.phase, Some(PhaseEnum::L1 | PhaseEnum::L2 | PhaseEnum::L3)))
80            .map(|s| scale(s))
81            .collect();
82        if !phases.is_empty() {
83            return Some(phases.iter().sum());
84        }
85    }
86    None
87}
88
89fn energy_wh(values: &[MeterValue]) -> Option<f64> {
90    measurand(values, MeasurandEnum::EnergyActiveImportRegister)
91}
92
93fn message(text: String) -> MessageContent {
94    let content: String = text.chars().take(512).collect();
95    MessageContent {
96        format: MessageFormatEnum::UTF8,
97        language: None,
98        content,
99        custom_data: None,
100    }
101}
102
103fn additional_info<'a>(token: Option<&'a IdToken>, kind: &str) -> Option<&'a str> {
104    token?
105        .additional_info
106        .as_ref()?
107        .iter()
108        .find(|a| a.r#type == kind)
109        .map(|a| a.additional_id_token.as_str())
110}
111
112/// Handles a CALL from a 2.1 Charging Station and returns the CALLRESULT payload.
113pub async fn handle(ctx: &mut StationCtx, action: &str, payload: &str) -> Result<Value, RpcError> {
114    let call = StationCall::decode(action, payload).map_err(decode_error)?;
115    match call {
116        StationCall::BootNotification(r) => {
117            let cs = &r.charging_station;
118            let info = BootInfo {
119                vendor: &cs.vendor_name,
120                model: &cs.model,
121                serial_number: cs.serial_number.as_deref(),
122                firmware_version: cs.firmware_version.as_deref(),
123                iccid: cs.modem.as_ref().and_then(|m| m.iccid.as_deref()),
124                imsi: cs.modem.as_ref().and_then(|m| m.imsi.as_deref()),
125                meter_type: None,
126                meter_serial_number: None,
127                reason: Some(r.reason.as_str()),
128            };
129            let status = match ctx.boot(&info).await? {
130                super::Registration::Pending => RegistrationStatusEnum::Pending,
131                super::Registration::Rejected => RegistrationStatusEnum::Rejected,
132                _ => RegistrationStatusEnum::Accepted,
133            };
134            json(BootNotificationResponse {
135                custom_data: None,
136                current_time: StationCtx::now(),
137                interval: ctx.heartbeat_interval(),
138                status,
139                status_info: None,
140            })
141        }
142        StationCall::Heartbeat(_) => {
143            ctx.heartbeat().await?;
144            json(HeartbeatResponse {
145                custom_data: None,
146                current_time: StationCtx::now(),
147            })
148        }
149        StationCall::StatusNotification(r) => {
150            ctx.status(&StatusUpdate {
151                evse_number: i32_of(r.evse_id),
152                connector_number: i32_of(r.connector_id),
153                status: r.connector_status.as_str(),
154                error_code: None,
155                info: None,
156                vendor_error_code: None,
157                at: Ts(r.timestamp),
158            })
159            .await?;
160            json(StatusNotificationResponse::default())
161        }
162        StationCall::Authorize(r) => {
163            let ty = r.id_token.r#type.as_str();
164            let a = ctx.authorize(&r.id_token.id_token, Some(ty)).await?;
165            let mut info = id_token_info(&a);
166            let mut tariff = None;
167            if a.status == "Accepted" {
168                // I01 (central cost): tariff text; I08 (local cost): driver-specific tariff.
169                let extras = crate::tariff::on_authorize(
170                    &ctx.app,
171                    ctx.station_id,
172                    OcppVersion::V21,
173                    &r.id_token.id_token,
174                    Some(ty),
175                )
176                .await
177                .map_err(internal)?;
178                info.personal_message = extras.personal_message.map(message);
179                tariff = extras.tariff;
180            }
181            json(AuthorizeResponse {
182                id_token_info: info,
183                certificate_status: None,
184                allowed_energy_transfer: None,
185                tariff,
186                custom_data: None,
187            })
188        }
189        StationCall::TransactionEvent(r) => {
190            let ti = &r.transaction_info;
191            let event = TxEvent {
192                version: "2.1",
193                uid: &ti.transaction_id,
194                event_type: r.event_type.as_str(),
195                at: Some(Ts(r.timestamp)),
196                seq_no: r.seq_no,
197                evse_number: r.evse.as_ref().map(|e| i32_of(e.id)),
198                connector_number: r.evse.as_ref().and_then(|e| e.connector_id).map(i32_of),
199                id_token: r.id_token.as_ref().map(|t| t.id_token.as_str()),
200                id_token_type: r.id_token.as_ref().map(|t| t.r#type.as_str()),
201                charging_state: ti.charging_state.map(|s| s.as_str()),
202                stop_reason: ti.stopped_reason.map(|s| s.as_str()),
203                reservation_id: r.reservation_id.map(i32_of),
204                remote_start_id: ti.remote_start_id.map(i32_of),
205                energy_wh: r.meter_value.as_deref().and_then(energy_wh),
206            };
207            let tx_id = db::transactions::apply_event(&ctx.app.db, ctx.station_id, &event)
208                .await
209                .map_err(internal)?;
210            if let Some(reservation) = r.reservation_id {
211                // H01.FR.15: the reservation is used by this transaction.
212                crate::reservation::on_used(&ctx.app, ctx.station_id, reservation, tx_id)
213                    .await
214                    .map_err(internal)?;
215            }
216            if r.event_type == TransactionEventEnum::Ended {
217                // A TxProfile ends with its transaction.
218                db::charging::clear_for_transaction(
219                    &ctx.app.db,
220                    ctx.station_id,
221                    &ti.transaction_id,
222                )
223                .await
224                .map_err(internal)?;
225            }
226            let evse = event.evse_number.unwrap_or(0);
227            for mv in r.meter_value.iter().flatten() {
228                let sampled = serde_json::to_value(&mv.sampled_value).map_err(internal)?;
229                ctx.meter_values(Some(tx_id), evse, Ts(mv.timestamp), &sampled)
230                    .await?;
231            }
232            tracing::debug!(station = %ctx.identity, transaction = %ti.transaction_id, event = r.event_type.as_str(), seq = r.seq_no, "TransactionEvent");
233            let id_token_info = match &r.id_token {
234                Some(t) => Some(id_token_info(
235                    &ctx.authorize(&t.id_token, Some(t.r#type.as_str())).await?,
236                )),
237                None => None,
238            };
239            let values = r.meter_value.as_deref().unwrap_or_default();
240            let cost_event = crate::tariff::TxEventCost {
241                event_type: r.event_type.as_str(),
242                at: r.timestamp,
243                seq_no: r.seq_no,
244                charging_state: ti.charging_state.map(|s| s.as_str()),
245                energy_wh: energy_wh(values),
246                power_w: measurand(values, MeasurandEnum::PowerActiveImport),
247                current_a: measurand(values, MeasurandEnum::CurrentImport),
248                tariff_id: ti.tariff_id.as_deref(),
249                cost_details: r
250                    .cost_details
251                    .as_ref()
252                    .map(serde_json::to_value)
253                    .transpose()
254                    .map_err(internal)?,
255                payment_brand: additional_info(r.id_token.as_ref(), "PaymentBrand"),
256                payment_recognition: additional_info(r.id_token.as_ref(), "PaymentRecognition"),
257            };
258            // I02/I03: running and final cost.
259            let total_cost =
260                crate::tariff::on_transaction_event(&ctx.app, ctx.station_id, tx_id, &cost_event)
261                    .await
262                    .map_err(internal)?;
263            json(TransactionEventResponse {
264                id_token_info,
265                total_cost,
266                ..Default::default()
267            })
268        }
269        StationCall::MeterValues(r) => {
270            for mv in &r.meter_value {
271                let sampled = serde_json::to_value(&mv.sampled_value).map_err(internal)?;
272                ctx.meter_values(None, i32_of(r.evse_id), Ts(mv.timestamp), &sampled)
273                    .await?;
274            }
275            json(MeterValuesResponse::default())
276        }
277        StationCall::DataTransfer(r) => {
278            ctx.log_event(action, payload).await;
279            tracing::info!(station = %ctx.identity, vendor_id = %r.vendor_id, message_id = ?r.message_id, "DataTransfer");
280            json(DataTransferResponse {
281                custom_data: None,
282                status: DataTransferStatusEnum::UnknownVendorId,
283                status_info: None,
284                data: None,
285            })
286        }
287        StationCall::FirmwareStatusNotification(r) => {
288            ctx.log_event(action, payload).await;
289            if r.status == FirmwareStatusEnum::Installed {
290                tracing::info!(station = %ctx.identity, "firmware installed");
291            }
292            json(FirmwareStatusNotificationResponse::default())
293        }
294        // No certificate authority is wired in: certificate requests are declined explicitly.
295        StationCall::SignCertificate(_) => {
296            ctx.log_event(action, payload).await;
297            json(SignCertificateResponse {
298                custom_data: None,
299                status: GenericStatusEnum::Rejected,
300                status_info: None,
301            })
302        }
303        StationCall::Get15118EVCertificate(_) => {
304            ctx.log_event(action, payload).await;
305            json(Get15118EVCertificateResponse {
306                custom_data: None,
307                status: Iso15118EVCertificateStatusEnum::Failed,
308                status_info: None,
309                exi_response: String::new(),
310                remaining_contracts: None,
311            })
312        }
313        StationCall::GetCertificateStatus(_) => json(GetCertificateStatusResponse {
314            custom_data: None,
315            status: GetCertificateStatusEnum::Failed,
316            status_info: None,
317            ocsp_result: None,
318        }),
319        StationCall::NotifyEVChargingNeeds(r) => {
320            ctx.log_event(action, payload).await;
321            let n = &r.charging_needs;
322            let needs = crate::smart::EvNeeds {
323                evse: r.evse_id,
324                bidirectional: {
325                    let mode = n.requested_energy_transfer.as_str();
326                    mode.contains("BPT") || mode.contains("DER")
327                },
328                ac_max_current: n
329                    .ac_charging_parameters
330                    .as_ref()
331                    .map(|ac| ac.ev_max_current),
332                dc_max_power: n
333                    .dc_charging_parameters
334                    .as_ref()
335                    .map(|d| {
336                        d.ev_max_power
337                            .unwrap_or(d.ev_max_current * d.ev_max_voltage)
338                    })
339                    .or_else(|| {
340                        n.v2x_charging_parameters
341                            .as_ref()
342                            .and_then(|x| x.max_charge_power)
343                    }),
344                departure: n.departure_time,
345                max_schedule_tuples: r.max_schedule_tuples,
346            };
347            let status = crate::smart::on_ev_charging_needs(
348                &ctx.app,
349                ctx.station_id,
350                OcppVersion::V21,
351                needs,
352            )
353            .await
354            .map_err(internal)?;
355            let status = match status {
356                "Accepted" => NotifyEVChargingNeedsStatusEnum::Accepted,
357                _ => NotifyEVChargingNeedsStatusEnum::Rejected,
358            };
359            json(NotifyEVChargingNeedsResponse {
360                custom_data: None,
361                status,
362                status_info: None,
363            })
364        }
365        StationCall::NotifyEVChargingSchedule(_) => {
366            ctx.log_event(action, payload).await;
367            json(NotifyEVChargingScheduleResponse {
368                custom_data: None,
369                status: GenericStatusEnum::Accepted,
370                status_info: None,
371            })
372        }
373        // Notifications: stored in the event log, acknowledged with an empty response.
374        StationCall::ClearedChargingLimit(r) => {
375            db::charging::delete_external_limits(
376                &ctx.app.db,
377                ctx.station_id,
378                r.charging_limit_source.as_str(),
379                r.evse_id,
380            )
381            .await
382            .map_err(internal)?;
383            ack(
384                ctx,
385                action,
386                payload,
387                ClearedChargingLimitResponse::default(),
388            )
389            .await
390        }
391        StationCall::LogStatusNotification(_) => {
392            ack(
393                ctx,
394                action,
395                payload,
396                LogStatusNotificationResponse::default(),
397            )
398            .await
399        }
400        StationCall::NotifyChargingLimit(r) => {
401            let schedules = r
402                .charging_schedule
403                .as_ref()
404                .map(serde_json::to_value)
405                .transpose()
406                .map_err(internal)?;
407            db::charging::upsert_external_limit(
408                &ctx.app.db,
409                ctx.station_id,
410                r.evse_id.unwrap_or(0),
411                r.charging_limit.charging_limit_source.as_str(),
412                r.charging_limit.is_grid_critical,
413                r.charging_limit.is_local_generation,
414                schedules.as_ref(),
415            )
416            .await
417            .map_err(internal)?;
418            ack(ctx, action, payload, NotifyChargingLimitResponse::default()).await
419        }
420        StationCall::NotifyCustomerInformation(_) => {
421            ack(
422                ctx,
423                action,
424                payload,
425                NotifyCustomerInformationResponse::default(),
426            )
427            .await
428        }
429        StationCall::NotifyDisplayMessages(_) => {
430            ack(
431                ctx,
432                action,
433                payload,
434                NotifyDisplayMessagesResponse::default(),
435            )
436            .await
437        }
438        StationCall::NotifyEvent(_) => {
439            ack(ctx, action, payload, NotifyEventResponse::default()).await
440        }
441        StationCall::NotifyMonitoringReport(_) => {
442            ack(
443                ctx,
444                action,
445                payload,
446                NotifyMonitoringReportResponse::default(),
447            )
448            .await
449        }
450        StationCall::NotifyReport(_) => {
451            ack(ctx, action, payload, NotifyReportResponse::default()).await
452        }
453        StationCall::PublishFirmwareStatusNotification(_) => {
454            ack(
455                ctx,
456                action,
457                payload,
458                PublishFirmwareStatusNotificationResponse::default(),
459            )
460            .await
461        }
462        StationCall::ReportChargingProfiles(r) => {
463            let profiles: Vec<Value> = r
464                .charging_profile
465                .iter()
466                .filter_map(|p| serde_json::to_value(p).ok())
467                .collect();
468            crate::smart::record_reported(
469                &ctx.app,
470                ctx.station_id,
471                OcppVersion::V21,
472                r.evse_id,
473                r.charging_limit_source.as_str(),
474                &profiles,
475            )
476            .await
477            .map_err(internal)?;
478            ack(
479                ctx,
480                action,
481                payload,
482                ReportChargingProfilesResponse::default(),
483            )
484            .await
485        }
486        StationCall::ReservationStatusUpdate(r) => {
487            crate::reservation::on_status_update(
488                &ctx.app,
489                ctx.station_id,
490                r.reservation_id,
491                r.reservation_update_status.as_str(),
492            )
493            .await
494            .map_err(internal)?;
495            ack(
496                ctx,
497                action,
498                payload,
499                ReservationStatusUpdateResponse::default(),
500            )
501            .await
502        }
503        StationCall::SecurityEventNotification(_) => {
504            ack(
505                ctx,
506                action,
507                payload,
508                SecurityEventNotificationResponse::default(),
509            )
510            .await
511        }
512        // OCPP 2.1 additions.
513        StationCall::BatterySwap(_) => {
514            ack(ctx, action, payload, BatterySwapResponse::default()).await
515        }
516        StationCall::NotifyDERAlarm(_) => {
517            ack(ctx, action, payload, NotifyDERAlarmResponse::default()).await
518        }
519        StationCall::NotifyDERStartStop(_) => {
520            ack(ctx, action, payload, NotifyDERStartStopResponse::default()).await
521        }
522        StationCall::NotifyPriorityCharging(_) => {
523            ack(
524                ctx,
525                action,
526                payload,
527                NotifyPriorityChargingResponse::default(),
528            )
529            .await
530        }
531        StationCall::NotifySettlement(_) => {
532            ack(ctx, action, payload, NotifySettlementResponse::default()).await
533        }
534        StationCall::ReportDERControl(_) => {
535            ack(ctx, action, payload, ReportDERControlResponse::default()).await
536        }
537        StationCall::OpenPeriodicEventStream(_) => {
538            ctx.log_event(action, payload).await;
539            json(OpenPeriodicEventStreamResponse {
540                status: GenericStatusEnum::Accepted,
541                status_info: None,
542                custom_data: None,
543            })
544        }
545        StationCall::ClosePeriodicEventStream(_) => {
546            ack(
547                ctx,
548                action,
549                payload,
550                ClosePeriodicEventStreamResponse::default(),
551            )
552            .await
553        }
554        StationCall::PullDynamicScheduleUpdate(r) => {
555            // K28.FR.07 / K28.FR.12
556            let update =
557                crate::smart::pull_dynamic_update(&ctx.app, ctx.station_id, r.charging_profile_id)
558                    .await
559                    .map_err(internal)?;
560            let update =
561                update.and_then(|u| ocpp_core::decode_value::<ChargingScheduleUpdate>(u).ok());
562            json(match update {
563                Some(u) => PullDynamicScheduleUpdateResponse {
564                    schedule_update: Some(u),
565                    status: ChargingProfileStatusEnum::Accepted,
566                    status_info: None,
567                    custom_data: None,
568                },
569                None => PullDynamicScheduleUpdateResponse {
570                    schedule_update: None,
571                    status: ChargingProfileStatusEnum::Rejected,
572                    status_info: Some(StatusInfo {
573                        reason_code: "InvalidProfile".into(),
574                        additional_info: None,
575                        custom_data: None,
576                    }),
577                    custom_data: None,
578                },
579            })
580        }
581        StationCall::VatNumberValidation(r) => {
582            ctx.log_event(action, payload).await;
583            // No VAT registry is connected: report the number as not validated.
584            json(VatNumberValidationResponse {
585                company: None,
586                status_info: None,
587                vat_number: r.vat_number,
588                evse_id: r.evse_id,
589                status: GenericStatusEnum::Rejected,
590                custom_data: None,
591            })
592        }
593        // No certificate authority / OCSP responder is connected.
594        StationCall::GetCertificateChainStatus(_) => Err(RpcError::not_supported(action)),
595    }
596}
597
598/// Handles an unconfirmed SEND message (OCPP 2.1 message type 6). No reply is sent.
599pub async fn handle_send(ctx: &mut StationCtx, action: &str, payload: &str) {
600    match StationSend::decode(action, payload) {
601        Ok(StationSend::NotifyPeriodicEventStream(s)) => {
602            // Values are offsets from `basetime`; stored as received.
603            let at = Ts(s.basetime);
604            let data =
605                serde_json::json!({ "streamId": s.id, "pending": s.pending, "data": s.data });
606            if let Err(e) = ctx.meter_values(None, 0, at, &data).await {
607                tracing::warn!(station = %ctx.identity, error = %e, "failed to store periodic event stream data");
608            }
609        }
610        Err(e) => {
611            tracing::warn!(station = %ctx.identity, action, error = %e, "invalid SEND message ignored")
612        }
613    }
614}
615
616async fn ack<T: serde::Serialize>(
617    ctx: &StationCtx,
618    action: &str,
619    payload: &str,
620    response: T,
621) -> Result<Value, RpcError> {
622    ctx.log_event(action, payload).await;
623    json(response)
624}
625
626/// Validates a command payload sent through the API before it is queued.
627pub fn validate_command(action: &str, payload: Value) -> Result<Value, RpcError> {
628    CsmsCall::from_value(action, payload)
629        .map(|c| c.to_payload())
630        .map_err(|e| match e {
631            DecodeError::Invalid(v) => RpcError::new(v.kind.error_code(), v.message),
632            other => RpcError::new(ErrorCode::NotSupported, other.to_string()),
633        })
634}
635
636/// Validates a CALLRESULT received for one of our CALLs.
637pub fn validate_response(action: &str, payload: &str) -> Result<Value, ocpp_core::Violation> {
638    let a = Action::from_name(action).ok_or_else(|| {
639        ocpp_core::Violation::new(
640            ocpp_core::ViolationKind::Format,
641            format!("unknown action {action}"),
642        )
643    })?;
644    ocpp21::validate_response(a, payload)
645}
ocpp_service/ocpp/v21.rs

ocpp_service/ocpp/
v21.rs
