codetics-ocpp-service 0.1.0

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

//! Smart charging (OCPP 1.6 §3.13 / Smart Charging profile, OCPP 2.x block K).
//!
//! The CSMS keeps a registry of the charging profiles it installed (and of
//! those reported by stations), enforces the CSMS-side rules of the
//! specification before sending a profile, computes composite schedules, and
//! answers ISO 15118 charging needs with a TxProfile.

pub mod composite;
pub mod model;
pub mod v16;
pub mod v201;
pub mod v21;

use ocpp_core::{DateTime, OcppVersion};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};

use self::composite::{CompositePeriod, Params};
use self::model::{Kind, Profile, Purpose, RuleViolation, Unit};
use crate::calls;
use crate::db::{self, Ts, charging::ProfileRow, stations::Station};
use crate::error::AppError;
use crate::state::App;

fn rule_error(v: RuleViolation) -> AppError {
    AppError::BadRequest(format!("{} ({})", v.message, v.rule))
}

/// Normalizes a stored profile row.
pub fn normalize_row(row: &ProfileRow) -> Option<Profile> {
    let evse = row.evse_number as i64;
    match row.ocpp_version.as_str() {
        "1.6" => v16::decode(&row.profile)
            .ok()
            .map(|p| v16::normalize(evse, &p)),
        "2.0.1" => v201::decode(&row.profile)
            .ok()
            .map(|p| v201::normalize(evse, &p)),
        "2.1" => v21::decode(&row.profile).ok().map(|p| {
            let update = row.dyn_update.as_ref().and_then(|u| {
                ocpp_core::decode_value::<ocpp21::ChargingScheduleUpdate>(u.clone()).ok()
            });
            v21::normalize(
                evse,
                &p,
                update
                    .as_ref()
                    .zip(row.dyn_updated_at)
                    .map(|(u, t)| (u, t.0)),
            )
        }),
        _ => None,
    }
}

/// External limits (NotifyChargingLimit) as profiles of purpose
/// ChargingStationExternalConstraints (or LocalGeneration, 2.1).
fn external_profiles(limits: &[db::charging::ExternalLimit]) -> Vec<Profile> {
    let mut out = Vec::new();
    for (i, l) in limits.iter().enumerate() {
        let Some(Value::Array(schedules)) = &l.schedules else {
            continue;
        };
        // A 2.0.1 ChargingScheduleType is a subset of the 2.1 one.
        let Some(first) = schedules
            .first()
            .and_then(|s| ocpp_core::decode_value::<ocpp21::ChargingSchedule>(s.clone()).ok())
        else {
            continue;
        };
        let mut schedule = v21::schedule(&first);
        if schedule.start.is_none() {
            schedule.start = Some(l.received_at.0);
        }
        out.push(Profile {
            id: -(i as i64) - 1,
            evse: l.evse_number as i64,
            purpose: if l.is_local_generation == Some(true) {
                Purpose::LocalGeneration
            } else {
                Purpose::ExternalConstraints
            },
            stack_level: 0,
            kind: Kind::Absolute,
            recurrency: None,
            valid_from: None,
            valid_to: None,
            transaction: None,
            schedule,
            dyn_updated_at: None,
        });
    }
    out
}

/// All profiles currently applicable at the station, including external limits.
pub async fn station_profiles(app: &App, station_id: i64) -> Result<Vec<Profile>, AppError> {
    let rows = db::charging::installed(&app.db, station_id).await?;
    let mut profiles: Vec<Profile> = rows.iter().filter_map(normalize_row).collect();
    profiles.extend(external_profiles(
        &db::charging::external_limits(&app.db, station_id).await?,
    ));
    Ok(profiles)
}

/// K01.FR.06 / K01.FR.39: no two profiles with the same stack level and purpose valid at the same time.
fn check_conflicts(new: &Profile, installed: &[Profile]) -> Result<(), RuleViolation> {
    for other in installed
        .iter()
        .filter(|o| o.id != new.id && o.purpose == new.purpose && o.stack_level == new.stack_level)
    {
        if new.purpose == Purpose::Tx {
            if other.transaction == new.transaction {
                return Err(RuleViolation::new(
                    "K01.FR.39",
                    format!(
                        "TxProfile {} already uses stackLevel {} for this transaction",
                        other.id, new.stack_level
                    ),
                ));
            }
        } else if other.evse == new.evse && new.validity_overlaps(other) {
            return Err(RuleViolation::new(
                "K01.FR.06",
                format!(
                    "profile {} already uses this stackLevel and purpose on EVSE {} with an overlapping validity",
                    other.id, new.evse
                ),
            ));
        }
    }
    Ok(())
}

#[derive(Debug, Serialize)]
pub struct InstallOutcome {
    pub status: String,
    pub response: Value,
    pub profile: Value,
}

/// Validates a profile against the specification rules, sends it with
/// SetChargingProfile and records the result.
pub async fn install(
    app: &App,
    station: &Station,
    version: OcppVersion,
    evse: i64,
    profile_json: Value,
) -> Result<InstallOutcome, AppError> {
    let active = if evse > 0 {
        db::transactions::active_on_evse(&app.db, station.id, evse).await?
    } else {
        None
    };
    let active_uid = active.as_ref().map(|(uid, _)| uid.as_str());
    let (profile, payload) = match version {
        OcppVersion::V16 => v16::prepare(
            evse,
            v16::decode(&profile_json).map_err(AppError::BadRequest)?,
            active_uid,
        ),
        OcppVersion::V201 => v201::prepare(
            evse,
            v201::decode(&profile_json).map_err(AppError::BadRequest)?,
            active_uid,
        ),
        OcppVersion::V21 => v21::prepare(
            evse,
            v21::decode(&profile_json).map_err(AppError::BadRequest)?,
            active_uid,
        ),
    }
    .map_err(rule_error)?;
    if version != OcppVersion::V16 {
        let installed: Vec<Profile> = db::charging::installed(&app.db, station.id)
            .await?
            .iter()
            .filter_map(normalize_row)
            .collect();
        check_conflicts(&profile, &installed)
            .map_err(|v| AppError::Conflict(format!("{} ({})", v.message, v.rule)))?;
    }
    let stored = payload
        .get("csChargingProfiles")
        .or_else(|| payload.get("chargingProfile"))
        .cloned()
        .unwrap_or(Value::Null);
    let row_id = db::charging::insert_pending(
        &app.db,
        station.id,
        &db::charging::NewProfile {
            ocpp_version: version.as_str(),
            profile_id: profile.id,
            evse_number: evse,
            purpose: stored["chargingProfilePurpose"]
                .as_str()
                .unwrap_or_default(),
            stack_level: profile.stack_level,
            kind: stored["chargingProfileKind"].as_str().unwrap_or_default(),
            transaction_uid: profile.transaction.as_deref(),
            valid_from: profile.valid_from.map(Ts),
            valid_to: profile.valid_to.map(Ts),
            source: "CSO",
            profile: &stored,
        },
    )
    .await?;
    let response = match calls::call(
        app,
        &station.ocpp_identity,
        version,
        "SetChargingProfile",
        payload,
    )
    .await
    {
        Ok(r) => r,
        Err(e) => {
            db::charging::discard_pending(&app.db, row_id).await?;
            return Err(e);
        }
    };
    let status = calls::status_of(&response).to_owned();
    if status == "Accepted" {
        // 1.6 stations replace a profile with the same connector/purpose/stack level (§3.13.2).
        db::charging::mark_installed(&app.db, station.id, row_id, version == OcppVersion::V16)
            .await?;
    } else {
        db::charging::mark_rejected(&app.db, row_id, response.get("statusInfo")).await?;
    }
    tracing::info!(station = %station.ocpp_identity, profile = profile.id, evse, %status, "SetChargingProfile");
    Ok(InstallOutcome {
        status,
        response,
        profile: stored,
    })
}

#[derive(Debug, Default, Deserialize)]
pub struct ClearRequest {
    pub profile_id: Option<i64>,
    pub evse_id: Option<i64>,
    /// Version-specific purpose name (e.g. `ChargePointMaxProfile` in 1.6).
    pub purpose: Option<String>,
    pub stack_level: Option<i64>,
}

pub async fn clear(
    app: &App,
    station: &Station,
    version: OcppVersion,
    req: &ClearRequest,
) -> Result<Value, AppError> {
    let payload = if version == OcppVersion::V16 {
        json!({ "id": req.profile_id, "connectorId": req.evse_id, "chargingProfilePurpose": req.purpose, "stackLevel": req.stack_level })
    } else {
        if req.profile_id.is_none()
            && req.evse_id.is_none()
            && req.purpose.is_none()
            && req.stack_level.is_none()
        {
            return Err(AppError::BadRequest(
                "give profile_id, or at least one of evse_id, purpose, stack_level (K10.FR.02)"
                    .into(),
            ));
        }
        if matches!(
            req.purpose.as_deref(),
            Some("ChargingStationExternalConstraints" | "LocalGeneration")
        ) {
            return Err(AppError::BadRequest("the CSMS must not clear ChargingStationExternalConstraints or LocalGeneration profiles (K10.FR.06)".into()));
        }
        let criteria = (req.evse_id.is_some() || req.purpose.is_some() || req.stack_level.is_some())
            .then(|| json!({ "evseId": req.evse_id, "chargingProfilePurpose": req.purpose, "stackLevel": req.stack_level }));
        json!({ "chargingProfileId": req.profile_id, "chargingProfileCriteria": criteria })
    };
    let payload = strip_nulls(payload);
    let response = calls::call(
        app,
        &station.ocpp_identity,
        version,
        "ClearChargingProfile",
        payload,
    )
    .await?;
    if calls::status_of(&response) == "Accepted" {
        let criteria = db::charging::ClearCriteria {
            profile_id: req.profile_id,
            evse_number: req.evse_id,
            purpose: req.purpose.as_deref(),
            stack_level: req.stack_level,
        };
        let n = db::charging::clear(&app.db, station.id, &criteria).await?;
        tracing::info!(station = %station.ocpp_identity, cleared = n, "ClearChargingProfile");
    }
    Ok(response)
}

/// Removes `null` members recursively (OCPP-J does not allow null for absent fields).
pub fn strip_nulls(v: Value) -> Value {
    match v {
        Value::Object(m) => Value::Object(
            m.into_iter()
                .filter(|(_, v)| !v.is_null())
                .map(|(k, v)| (k, strip_nulls(v)))
                .collect(),
        ),
        Value::Array(a) => Value::Array(a.into_iter().map(strip_nulls).collect()),
        other => other,
    }
}

/// CSMS-side composite schedule of an EVSE.
pub async fn composite_for(
    app: &App,
    station: &Station,
    evse: i64,
    start: DateTime,
    duration: i64,
    unit: Unit,
) -> Result<Vec<CompositePeriod>, AppError> {
    let active = if evse > 0 {
        db::transactions::active_on_evse(&app.db, station.id, evse).await?
    } else {
        None
    };
    let profiles: Vec<Profile> = station_profiles(app, station.id)
        .await?
        .into_iter()
        // A TxProfile only applies to its own (active) transaction.
        .filter(|p| {
            p.purpose != Purpose::Tx
                || (p.transaction.is_some()
                    && p.transaction.as_deref() == active.as_ref().map(|a| a.0.as_str()))
        })
        .collect();
    let params = Params {
        start,
        duration,
        unit,
        voltage: app.cfg.nominal_voltage,
        default_phases: 3,
        tx_start: active.and_then(|a| a.1).map(|t| t.0),
    };
    Ok(composite::composite(&profiles, evse, &params))
}

/// K28.FR.07 / K28.FR.12: values for a PullDynamicScheduleUpdateRequest.
pub async fn pull_dynamic_update(
    app: &App,
    station_id: i64,
    profile_id: i64,
) -> Result<Option<Value>, AppError> {
    let Some(row) = db::charging::find_installed(&app.db, station_id, profile_id).await? else {
        return Ok(None);
    };
    if row.kind != "Dynamic" {
        return Ok(None);
    }
    if let Some(u) = row.dyn_update {
        return Ok(Some(u));
    }
    // No update yet: the values of the profile's single period.
    let period = &row.profile["chargingSchedule"][0]["chargingSchedulePeriod"][0];
    let fields = [
        "limit",
        "limit_L2",
        "limit_L3",
        "dischargeLimit",
        "dischargeLimit_L2",
        "dischargeLimit_L3",
        "setpoint",
        "setpoint_L2",
        "setpoint_L3",
        "setpointReactive",
        "setpointReactive_L2",
        "setpointReactive_L3",
    ];
    let update: serde_json::Map<String, Value> = fields
        .iter()
        .filter_map(|f| period.get(*f).map(|v| ((*f).to_owned(), v.clone())))
        .collect();
    Ok(Some(Value::Object(update)))
}

/// Stores a dynamic schedule update (2.1 K28) and, with `push`, sends UpdateDynamicSchedule.
pub async fn dynamic_update(
    app: &App,
    station: &Station,
    version: OcppVersion,
    profile_id: i64,
    update: Value,
    push: bool,
) -> Result<Value, AppError> {
    if version != OcppVersion::V21 {
        return Err(AppError::BadRequest(
            "dynamic charging profiles exist in OCPP 2.1 only".into(),
        ));
    }
    let row = db::charging::find_installed(&app.db, station.id, profile_id)
        .await?
        .ok_or_else(|| AppError::NotFound(format!("no installed charging profile {profile_id}")))?;
    if row.kind != "Dynamic" {
        return Err(AppError::BadRequest(
            "the charging profile is not Dynamic (K28.FR.11)".into(),
        ));
    }
    let update: ocpp21::ChargingScheduleUpdate =
        ocpp_core::decode_value(update).map_err(|e| AppError::BadRequest(e.to_string()))?;
    let update = serde_json::to_value(update).map_err(|e| AppError::Internal(e.to_string()))?;
    if !push {
        db::charging::set_dyn_update(&app.db, row.id, &update).await?;
        return Ok(json!({ "stored": true }));
    }
    let response = calls::call(
        app,
        &station.ocpp_identity,
        version,
        "UpdateDynamicSchedule",
        json!({ "chargingProfileId": profile_id, "scheduleUpdate": update }),
    )
    .await?;
    if calls::status_of(&response) == "Accepted" {
        db::charging::set_dyn_update(&app.db, row.id, &update).await?;
    }
    Ok(response)
}

/// ISO 15118 charging needs reported by the station (NotifyEVChargingNeeds), version-neutral.
#[derive(Debug, Clone)]
pub struct EvNeeds {
    pub evse: i64,
    /// The EV requests bidirectional power transfer (2.1 *_BPT / DER modes).
    pub bidirectional: bool,
    /// AC: maximum current per phase (A).
    pub ac_max_current: Option<f64>,
    /// DC: maximum power (W).
    pub dc_max_power: Option<f64>,
    pub departure: Option<DateTime>,
    pub max_schedule_tuples: Option<i64>,
}

/// K15/K17/K18/K19: decides the NotifyEVChargingNeeds status and, when
/// accepted, sends a TxProfile for the transaction in the background (the
/// station expects it within 60 s).
pub async fn on_ev_charging_needs(
    app: &App,
    station_id: i64,
    version: OcppVersion,
    needs: EvNeeds,
) -> Result<&'static str, AppError> {
    if !app.cfg.ev_charging_needs {
        return Ok("Rejected");
    }
    if needs.bidirectional && !app.cfg.allow_bidirectional {
        // K18.FR.02 / K19.FR.02: requested energy transfer not accepted.
        return Ok("Rejected");
    }
    if needs.ac_max_current.is_none() && needs.dc_max_power.is_none() {
        return Ok("Rejected");
    }
    let Some((uid, _)) = db::transactions::active_on_evse(&app.db, station_id, needs.evse).await?
    else {
        return Ok("Rejected");
    };
    let app = app.clone();
    tokio::spawn(async move {
        if let Err(e) = send_ev_schedule(&app, station_id, version, &uid, &needs).await {
            tracing::warn!(station_id, transaction = %uid, error = %e, "could not send a TxProfile for the EV charging needs");
        }
    });
    Ok("Accepted")
}

async fn send_ev_schedule(
    app: &App,
    station_id: i64,
    version: OcppVersion,
    uid: &str,
    needs: &EvNeeds,
) -> Result<(), AppError> {
    let station = db::stations::get(&app.db, station_id)
        .await?
        .ok_or_else(|| AppError::NotFound("station".into()))?;
    let now = DateTime::now();
    let horizon = needs.departure.map_or(86_400, |d| {
        (d.unix_seconds() - now.unix_seconds()).clamp(3600, 86_400)
    });
    let (unit, ev_max) = match (needs.ac_max_current, needs.dc_max_power) {
        (Some(a), _) => (Unit::A, a),
        (None, Some(w)) => (Unit::W, w),
        _ => return Ok(()),
    };
    // Station-wide and default limits, without the TxProfiles of this transaction (they get replaced).
    let installed = db::charging::installed(&app.db, station.id).await?;
    let own: Vec<&ProfileRow> = installed
        .iter()
        .filter(|r| r.purpose == "TxProfile" && r.transaction_uid.as_deref() == Some(uid))
        .collect();
    let profiles: Vec<Profile> = station_profiles(app, station.id)
        .await?
        .into_iter()
        .filter(|p| p.purpose != Purpose::Tx)
        .collect();
    let params = Params {
        start: now,
        duration: horizon,
        unit,
        voltage: app.cfg.nominal_voltage,
        default_phases: 3,
        tx_start: None,
    };
    let mut periods: Vec<(i64, f64)> = composite::composite(&profiles, needs.evse, &params)
        .into_iter()
        .map(|p| (p.start_period, p.limit.map_or(ev_max, |l| l.min(ev_max))))
        .collect();
    periods.dedup_by(|b, a| a.1 == b.1);
    // Respect maxScheduleTuples: the last kept period takes the lowest remaining limit.
    if let Some(max) = needs
        .max_schedule_tuples
        .filter(|m| *m > 0)
        .map(|m| m as usize)
        && periods.len() > max
    {
        let tail_min = periods[max - 1..]
            .iter()
            .map(|p| p.1)
            .fold(f64::INFINITY, f64::min);
        periods.truncate(max);
        periods[max - 1].1 = tail_min;
    }
    // Reuse the profile id of a previous EV-needs TxProfile of this transaction, otherwise allocate one.
    let (id, stack) = match own.iter().find(|r| r.profile_id >= 100_000) {
        Some(r) => (r.profile_id as i64, r.stack_level as i64),
        None => (
            db::charging::next_profile_id(&app.db, station.id).await?,
            own.iter()
                .map(|r| r.stack_level as i64 + 1)
                .max()
                .unwrap_or(0),
        ),
    };
    let profile = json!({
        "id": id,
        "stackLevel": stack,
        "chargingProfilePurpose": "TxProfile",
        "chargingProfileKind": "Absolute",
        "transactionId": uid,
        "chargingSchedule": [{
            "id": 1,
            "startSchedule": now,
            "duration": horizon,
            "chargingRateUnit": if unit == Unit::A { "A" } else { "W" },
            "chargingSchedulePeriod": periods.iter().map(|(s, l)| json!({ "startPeriod": s, "limit": (l * 10.0).round() / 10.0 })).collect::<Vec<_>>(),
        }],
    });
    let outcome = install(app, &station, version, needs.evse, profile).await?;
    tracing::info!(station = %station.ocpp_identity, transaction = uid, status = %outcome.status, "TxProfile sent for EV charging needs");
    Ok(())
}

/// Stores profiles reported with ReportChargingProfiles (2.x K09).
pub async fn record_reported(
    app: &App,
    station_id: i64,
    version: OcppVersion,
    evse: i64,
    source: &str,
    profiles: &[Value],
) -> Result<(), AppError> {
    for p in profiles {
        let normalized = match version {
            OcppVersion::V201 => v201::decode(p).map(|x| v201::normalize(evse, &x)),
            OcppVersion::V21 => v21::decode(p).map(|x| v21::normalize(evse, &x, None)),
            OcppVersion::V16 => continue,
        };
        let Ok(n) = normalized else { continue };
        db::charging::upsert_reported(
            &app.db,
            station_id,
            &db::charging::NewProfile {
                ocpp_version: version.as_str(),
                profile_id: n.id,
                evse_number: evse,
                purpose: p["chargingProfilePurpose"].as_str().unwrap_or_default(),
                stack_level: n.stack_level,
                kind: p["chargingProfileKind"].as_str().unwrap_or_default(),
                transaction_uid: n.transaction.as_deref(),
                valid_from: n.valid_from.map(Ts),
                valid_to: n.valid_to.map(Ts),
                source,
                profile: p,
            },
        )
        .await?;
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::smart::model::{Period, Schedule};

    fn p(id: i64, evse: i64, purpose: Purpose, stack: i64, tx: Option<&str>) -> Profile {
        Profile {
            id,
            evse,
            purpose,
            stack_level: stack,
            kind: Kind::Absolute,
            recurrency: None,
            valid_from: None,
            valid_to: None,
            transaction: tx.map(str::to_owned),
            schedule: Schedule {
                start: Some(DateTime::UNIX_EPOCH),
                duration: None,
                unit: Unit::A,
                periods: vec![Period {
                    start: 0,
                    limit: Some(16.0),
                    phases: None,
                }],
            },
            dyn_updated_at: None,
        }
    }

    #[test]
    fn conflict_rules() {
        let installed = vec![
            p(1, 1, Purpose::TxDefault, 0, None),
            p(2, 1, Purpose::Tx, 1, Some("T1")),
        ];
        assert!(check_conflicts(&p(3, 1, Purpose::TxDefault, 0, None), &installed).is_err());
        assert!(
            check_conflicts(&p(1, 1, Purpose::TxDefault, 0, None), &installed).is_ok(),
            "same id replaces"
        );
        assert!(
            check_conflicts(&p(3, 2, Purpose::TxDefault, 0, None), &installed).is_ok(),
            "other EVSE"
        );
        assert!(
            check_conflicts(&p(3, 1, Purpose::TxDefault, 1, None), &installed).is_ok(),
            "other stack level"
        );
        assert!(check_conflicts(&p(4, 1, Purpose::Tx, 1, Some("T1")), &installed).is_err());
        assert!(check_conflicts(&p(4, 1, Purpose::Tx, 1, Some("T2")), &installed).is_ok());
        let mut later = p(5, 1, Purpose::TxDefault, 0, None);
        later.valid_from = Some(DateTime::from_unix_seconds(100));
        let mut earlier = p(1, 1, Purpose::TxDefault, 0, None);
        earlier.valid_to = Some(DateTime::from_unix_seconds(100));
        assert!(
            check_conflicts(&later, &[earlier]).is_ok(),
            "non-overlapping validity"
        );
    }

    #[test]
    fn strip_nulls_removes_absent_fields() {
        assert_eq!(
            strip_nulls(json!({ "a": null, "b": { "c": null, "d": 1 } })),
            json!({ "b": { "d": 1 } })
        );
    }
}