ocpi_tariffs/

price.rs

//! Price a CDR using a tariff and compare the prices embedded in the CDR with the prices computed here.

#[cfg(test)]
pub mod test;

#[cfg(test)]
mod test_normalize_periods;

#[cfg(test)]
mod test_periods;

#[cfg(test)]
mod test_real_world;

#[cfg(test)]
mod test_validate_cdr;

#[cfg(test)]
mod test_current_and_power_restrictions;

#[cfg(test)]
mod test_reservation_restriction;

#[cfg(test)]
mod test_min_max_price;

#[cfg(test)]
mod test_warning_path_map;

mod tariff;
mod v211;
mod v221;

use std::{borrow::Cow, collections::BTreeMap, fmt, ops::Range};

use chrono::{DateTime, Datelike as _, TimeDelta, Utc};
use chrono_tz::Tz;
use rust_decimal::Decimal;
use tariff::Tariff;
use tracing::{debug, instrument, trace};

use crate::{
    country, currency, datetime,
    duration::{self, AsHms as _, Hms},
    enumeration, from_warning_all,
    json::{self, FromJson as _},
    money::{self, VatOrigin},
    number::{self, RoundDecimal as _},
    string,
    warning::{
        self, GatherDeferredWarnings as _, GatherWarnings as _, IntoCaveat as _,
        IntoCaveatDeferred as _, VerdictExt as _, WithElement as _,
    },
    Ampere, Caveat, Cost, DisplayOption, Kw, Kwh, Money, Price, SaturatingAdd as _,
    SaturatingSub as _, Version, Versioned as _,
};

pub type Verdict<T> = crate::Verdict<T, Warning>;
type VerdictDeferred<T> = warning::VerdictDeferred<T, Warning>;

/// A normalized/expanded form of a charging period to make the pricing calculation simpler.
///
/// The simplicity comes through avoiding having to look up the next period to figure out the end
/// of the current period.
#[derive(Debug)]
struct PeriodNormalized {
    /// The set of quantities consumed across the duration of the `Period`.
    consumed: Consumed,

    /// A snapshot of the values of various quantities at the start of the charge period.
    start_snapshot: TotalsSnapshot,

    /// A snapshot of the values of various quantities at the end of the charge period.
    end_snapshot: TotalsSnapshot,
}

/// The set of quantities consumed across the duration of the `Period`.
#[derive(Clone)]
#[cfg_attr(test, derive(Default))]
pub(crate) struct Consumed {
    /// The peak current during this period.
    pub current_max: Option<Ampere>,

    /// The lowest current during this period.
    pub current_min: Option<Ampere>,

    /// The charging time consumed in this period.
    pub duration_charging: Option<TimeDelta>,

    /// The parking/idle time consumed in this period.
    pub duration_idle: Option<TimeDelta>,

    /// The energy consumed in this period.
    pub energy: Option<Kwh>,

    /// The maximum power reached during this period.
    pub power_max: Option<Kw>,

    /// The minimum power reached during this period.
    pub power_min: Option<Kw>,
}

impl fmt::Debug for Consumed {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Consumed")
            .field("current_max", &self.current_max)
            .field("current_min", &self.current_min)
            .field(
                "duration_charging",
                &self.duration_charging.map(|dt| dt.as_hms()),
            )
            .field("duration_idle", &self.duration_idle.map(|dt| dt.as_hms()))
            .field("energy", &self.energy)
            .field("power_max", &self.power_max)
            .field("power_min", &self.power_min)
            .finish()
    }
}

/// A snapshot of the values of various quantities at the start and end of the charge period.
#[derive(Clone)]
struct TotalsSnapshot {
    /// The `DateTime` this snapshot of total quantities was taken.
    date_time: DateTime<Utc>,

    /// The total energy consumed during a charging period.
    energy: Kwh,

    /// The local timezone.
    local_timezone: Tz,

    /// The total charging duration during a charging period.
    duration_charging: TimeDelta,

    /// The total period duration during a charging period.
    duration_total: TimeDelta,
}

impl fmt::Debug for TotalsSnapshot {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("TotalsSnapshot")
            .field("date_time", &self.date_time)
            .field("energy", &self.energy)
            .field("local_timezone", &self.local_timezone)
            .field("duration_charging", &self.duration_charging.as_hms())
            .field("duration_total", &self.duration_total.as_hms())
            .finish()
    }
}

impl TotalsSnapshot {
    /// Create a snapshot where all quantities are zero.
    fn zero(date_time: DateTime<Utc>, local_timezone: Tz) -> Self {
        Self {
            date_time,
            energy: Kwh::zero(),
            local_timezone,
            duration_charging: TimeDelta::zero(),
            duration_total: TimeDelta::zero(),
        }
    }

    /// Create a new snapshot based on the current snapshot with consumed quantities added.
    fn next(&self, consumed: &Consumed, date_time: DateTime<Utc>) -> Self {
        let duration = date_time.signed_duration_since(self.date_time);

        let mut next = Self {
            date_time,
            energy: self.energy,
            local_timezone: self.local_timezone,
            duration_charging: self.duration_charging,
            duration_total: self.duration_total.saturating_add(duration),
        };

        if let Some(duration) = consumed.duration_charging {
            next.duration_charging = next.duration_charging.saturating_add(duration);
        }

        if let Some(energy) = consumed.energy {
            next.energy = next.energy.saturating_add(energy);
        }
        next
    }

    /// Return the local time of this snapshot.
    fn local_time(&self) -> chrono::NaiveTime {
        self.date_time.with_timezone(&self.local_timezone).time()
    }

    /// Return the local date of this snapshot.
    fn local_date(&self) -> chrono::NaiveDate {
        self.date_time
            .with_timezone(&self.local_timezone)
            .date_naive()
    }

    /// Return the local `Weekday` of this snapshot.
    fn local_weekday(&self) -> chrono::Weekday {
        self.date_time.with_timezone(&self.local_timezone).weekday()
    }
}

/// Structure containing the charge session priced according to the specified tariff.
/// The fields prefixed `total` correspond to CDR fields with the same name.
pub struct Report {
    /// Charge session details per period.
    pub periods: Vec<PeriodReport>,

    /// The index of the tariff that was used for pricing.
    pub tariff_used: TariffOrigin,

    /// A list of reports for each tariff found in the CDR or supplied to the [`cdr::price`](crate::cdr::price) function.
    ///
    /// The order of the `tariff::Report`s are the same as the order in which they are given.
    pub tariff_reports: Vec<TariffReport>,

    /// Time-zone that was either specified or detected.
    pub timezone: String,

    /* Billed Quantities */
    /// The total charging time after applying step-size.
    pub billed_charging_time: Option<TimeDelta>,

    /// The total energy after applying step-size.
    pub billed_energy: Option<Kwh>,

    /// The total idle time after applying step-size.
    pub billed_idle_time: Option<TimeDelta>,

    /* Totals */
    /// Total duration of the charging session (excluding not charging), in hours.
    ///
    /// This is a total that has no direct source field in the `CDR` as it is calculated in the
    /// [`cdr::price`](crate::cdr::price) function.
    pub total_charging_time: Option<TimeDelta>,

    /// Total energy charged, in kWh.
    pub total_energy: Total<Kwh, Option<Kwh>>,

    /// Total duration of the charging session where the EV was not charging (no energy was transferred between EVSE and EV).
    ///
    /// See: `total_parking_time` field in <https://github.com/ocpi/ocpi/blob/release-2.2.1-bugfixes/mod_cdrs.asciidoc#131-cdr-object>.
    /// Note: We use `total_idle_time` as it's clearer than `total_parking_time`.
    /// Some people interpret `parking` to mean the total time that the vehicle is standing by the charge point.
    /// The OCPI spec defines `parking` as the time spent not charging.
    pub total_idle_time: Total<Option<TimeDelta>>,

    /// Total duration of the charging session (including the duration of charging and idle phases).
    pub total_time: Total<TimeDelta>,

    /* Costs */
    /// Total sum of all the costs of this transaction in the specified currency.
    pub total_cost: Total<Price, Option<Price>>,

    /// Total sum of all the cost of all the energy used, in the specified currency.
    pub total_energy_cost: Total<Option<Price>>,

    /// Total sum of all the fixed costs in the specified currency, except fixed price components of `parking` and `reservation`.
    /// The cost not depending on amount of time/energy used etc. Can contain costs like a start tariff.
    pub total_fixed_cost: Total<Option<Price>>,

    /// Total sum of all the costs related to idleness during this transaction. This includes fixed price components, in the specified currency.
    ///
    /// See: `total_parking_cost` field in <https://github.com/ocpi/ocpi/blob/release-2.2.1-bugfixes/mod_cdrs.asciidoc#131-cdr-object>.
    /// Note: We use `total_idle_cost` as it's clearer than `total_parking_cost`.
    /// Some people interpret `parking` to mean the total time that the vehicle is standing by the charge point.
    /// The OCPI spec defines `parking` as the time spent not charging.
    pub total_idle_cost: Total<Option<Price>>,

    /// Total sum of all the cost related to duration of charging during this transaction, in the specified currency.
    ///
    /// See: `total_time_cost` field in <https://github.com/ocpi/ocpi/blob/release-2.2.1-bugfixes/mod_cdrs.asciidoc#131-cdr-object>.
    /// Note: We use `total_charging_time_cost` as it's clearer than `total_time_cost`.
    pub total_charging_time_cost: Total<Option<Price>>,
}

impl fmt::Debug for Report {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Report")
            .field("periods", &self.periods)
            .field("tariff_used", &self.tariff_used)
            .field("tariff_reports", &self.tariff_reports)
            .field("timezone", &self.timezone)
            .field(
                "billed_charging_time",
                &self.billed_charging_time.map(|dt| dt.as_hms()),
            )
            .field("billed_energy", &self.billed_energy)
            .field(
                "billed_idle_time",
                &self.billed_idle_time.map(|dt| dt.as_hms()),
            )
            .field(
                "total_charging_time",
                &self.total_charging_time.map(|dt| dt.as_hms()),
            )
            .field("total_energy", &self.total_energy)
            .field("total_idle_time", &self.total_idle_time)
            .field("total_time", &self.total_time)
            .field("total_cost", &self.total_cost)
            .field("total_energy_cost", &self.total_energy_cost)
            .field("total_fixed_cost", &self.total_fixed_cost)
            .field("total_idle_cost", &self.total_idle_cost)
            .field("total_charging_time_cost", &self.total_charging_time_cost)
            .finish()
    }
}

/// The warnings that happen when pricing a CDR.
#[derive(Debug)]
pub enum Warning {
    Country(country::Warning),
    Currency(currency::Warning),
    DateTime(datetime::Warning),
    Decode(json::decode::Warning),
    Duration(duration::Warning),
    Enum(enumeration::Warning),

    /// The `$.country` field should be an alpha-2 country code.
    ///
    /// The alpha-3 code can be converted into an alpha-3 but the caller should be warned.
    CountryShouldBeAlpha2,

    /// A field in the tariff doesn't have the expected type.
    FieldInvalidType {
        /// The type that the given field should have according to the schema.
        expected_type: json::ValueKind,
    },

    /// A field in the tariff doesn't have the expected value.
    FieldInvalidValue {
        /// The value encountered.
        value: String,

        /// A message about what values are expected for this field.
        message: Cow<'static, str>,
    },

    /// The given field is required.
    FieldRequired {
        field_name: Cow<'static, str>,
    },

    Money(money::Warning),

    /// The CDR has no charging periods.
    NoPeriods,

    /// No valid tariff has been found in the list of provided tariffs.
    /// The tariff list can be sourced from either the tariffs contained in the CDR or from a list
    /// provided by the caller.
    ///
    /// A valid tariff must have a start date-time before the start of the session and an end
    /// date-time after the start of the session.
    ///
    /// If the CDR does not contain any tariffs consider providing a them using [`TariffSource`]
    /// when calling [`cdr::price`](crate::cdr::price).
    NoValidTariff,

    Number(number::Warning),

    /// The `start_date_time` of at least one of the `charging_periods` is outside of the
    /// CDR's `start_date_time`-`end_date_time` range.
    PeriodsOutsideStartEndDateTime {
        cdr_range: Range<DateTime<Utc>>,
        period_range: PeriodRange,
    },

    String(string::Warning),

    /// Converting the `tariff::Versioned` into a structured `tariff::v221::Tariff` caused an
    /// unrecoverable error.
    Tariff(crate::tariff::Warning),
}

impl Warning {
    /// Create a new `Warning::FieldInvalidValue` where the field is built from the given `json::Element`.
    fn field_invalid_value(
        value: impl Into<String>,
        message: impl Into<Cow<'static, str>>,
    ) -> Self {
        Warning::FieldInvalidValue {
            value: value.into(),
            message: message.into(),
        }
    }
}

impl fmt::Display for Warning {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Country(warning) => write!(f, "{warning}"),
            Self::CountryShouldBeAlpha2 => {
                f.write_str("The `$.country` field should be an alpha-2 country code.")
            }
            Self::Currency(warning) => write!(f, "{warning}"),
            Self::DateTime(warning) => write!(f, "{warning}"),
            Self::Decode(warning) => write!(f, "{warning}"),
            Self::Duration(warning) => write!(f, "{warning}"),
            Self::Enum(warning) => write!(f, "{warning}"),
            Self::FieldInvalidType { expected_type } => {
                write!(f, "Field has invalid type. Expected type `{expected_type}`")
            }
            Self::FieldInvalidValue { value, message } => {
                write!(f, "Field has invalid value `{value}`: {message}")
            }
            Self::FieldRequired { field_name } => {
                write!(f, "Field is required: `{field_name}`")
            }
            Self::Money(warning) => write!(f, "{warning}"),
            Self::NoPeriods => f.write_str("The CDR has no charging periods"),
            Self::NoValidTariff => {
                f.write_str("No valid tariff has been found in the list of provided tariffs")
            }
            Self::Number(warning) => write!(f, "{warning}"),
            Self::PeriodsOutsideStartEndDateTime {
                cdr_range: Range { start, end },
                period_range,
            } => {
                write!(
                    f,
                    "The CDR's charging period time range is not contained within the `start_date_time` \
                    and `end_date_time`; cdr: [start: {start}, end: {end}], period: {period_range}",
                )
            }
            Self::String(warning) => write!(f, "{warning}"),
            Self::Tariff(warnings) => {
                write!(f, "Tariff warnings: {warnings:?}")
            }
        }
    }
}

impl crate::Warning for Warning {
    fn id(&self) -> warning::Id {
        match self {
            Self::Country(warning) => warning.id(),
            Self::CountryShouldBeAlpha2 => warning::Id::from_static("country_should_be_alpha_2"),
            Self::Currency(warning) => warning.id(),
            Self::DateTime(warning) => warning.id(),
            Self::Decode(warning) => warning.id(),
            Self::Duration(warning) => warning.id(),
            Self::Enum(warning) => warning.id(),
            Self::FieldInvalidType { expected_type } => {
                warning::Id::from_string(format!("field_invalid_type({expected_type})"))
            }
            Self::FieldInvalidValue { value, .. } => {
                warning::Id::from_string(format!("field_invalid_value({value})"))
            }
            Self::FieldRequired { field_name } => {
                warning::Id::from_string(format!("field_required({field_name})"))
            }
            Self::Money(warning) => warning.id(),
            Self::NoPeriods => warning::Id::from_static("no_periods"),
            Self::NoValidTariff => warning::Id::from_static("no_valid_tariff"),
            Self::Number(warning) => warning.id(),
            Self::PeriodsOutsideStartEndDateTime { .. } => {
                warning::Id::from_static("periods_outside_start_end_date_time")
            }
            Self::String(warning) => warning.id(),
            Self::Tariff(warning) => warning.id(),
        }
    }
}

from_warning_all!(
    country::Warning => Warning::Country,
    currency::Warning => Warning::Currency,
    datetime::Warning => Warning::DateTime,
    duration::Warning => Warning::Duration,
    enumeration::Warning => Warning::Enum,
    json::decode::Warning => Warning::Decode,
    money::Warning => Warning::Money,
    number::Warning => Warning::Number,
    string::Warning => Warning::String,
    crate::tariff::Warning => Warning::Tariff
);

/// A report of parsing and using the referenced tariff to price a CDR.
#[derive(Debug)]
pub struct TariffReport {
    /// The id of the tariff.
    pub origin: TariffOrigin,

    /// Warnings from parsing a tariff.
    ///
    /// Each entry in the map is an element path and a list of associated warnings.
    pub warnings: BTreeMap<json::Path, Vec<crate::tariff::Warning>>,
}

/// The origin data for a tariff.
#[derive(Clone, Debug)]
pub struct TariffOrigin {
    /// The index of the tariff in the CDR JSON or in the list of override tariffs.
    pub index: usize,

    /// The value of the `id` field in the tariff JSON.
    pub id: String,

    // The currency code of the tariff.
    pub currency: currency::Code,
}

/// A CDR charge period in a normalized form ready for pricing.
#[derive(Debug)]
pub(crate) struct Period {
    /// The start time of this period.
    pub start_date_time: DateTime<Utc>,

    /// The quantities consumed during this period.
    pub consumed: Consumed,
}

/// A structure containing a report for each dimension of a CDRs charging [`Period`].
#[derive(Debug)]
pub struct Dimensions {
    /// Energy consumed. `None` if the CDR period had no energy dimension.
    pub energy: Option<Dimension<Kwh>>,

    /// Flat fee without unit for `step_size`.
    pub flat: Dimension<()>,

    /// Duration of time charging. `None` if the CDR period had no time dimension.
    pub duration_charging: Option<Dimension<TimeDelta>>,

    /// Duration of time not charging. `None` if the CDR period had no parking-time dimension.
    pub duration_idle: Option<Dimension<TimeDelta>>,
}

impl Dimensions {
    /// Create a new `Dimensions` object.
    fn new(components: ComponentSet, consumed: &Consumed) -> Self {
        let ComponentSet {
            energy: energy_price,
            flat: flat_price,
            duration_charging: duration_charging_price,
            duration_idle: duration_idle_price,
        } = components;

        let Consumed {
            duration_charging,
            duration_idle,
            energy,
            current_max: _,
            current_min: _,
            power_max: _,
            power_min: _,
        } = consumed;

        Self {
            energy: (*energy).map(|e| Dimension {
                price: energy_price,
                volume: e,
                billed_volume: e,
            }),
            flat: Dimension {
                price: flat_price,
                volume: (),
                billed_volume: (),
            },
            duration_charging: (*duration_charging).map(|dc| Dimension {
                price: duration_charging_price,
                volume: dc,
                billed_volume: dc,
            }),
            duration_idle: (*duration_idle).map(|di| Dimension {
                price: duration_idle_price,
                volume: di,
                billed_volume: di,
            }),
        }
    }
}

#[derive(Debug)]
/// A report for a single dimension during a single charging [`Period`].
pub struct Dimension<V> {
    /// The price component that was active during this period for this dimension.
    /// It could be that no price component was active during this period for this dimension in
    /// which case `price` is `None`.
    pub price: Option<Component>,

    /// The volume of this dimension during this period, as received in the provided charge detail record.
    pub volume: V,

    /// The value of `volume` after a potential step size was applied.
    /// Step size is applied over the total volume during the whole session of a dimension. But the
    /// resulting additional volume should be billed according to the price component in this
    /// period.
    ///
    /// If no step-size was applied for this period, the volume is exactly equal to the `volume`
    /// field.
    pub billed_volume: V,
}

impl<V: Cost> Dimension<V> {
    /// The total cost of this dimension during a period.
    pub fn cost(&self) -> Option<Price> {
        let Some(price_component) = &self.price else {
            return None;
        };

        let excl_vat = self.billed_volume.cost(price_component.price);

        let incl_vat = match price_component.vat {
            VatOrigin::Provided(vat) => Some(excl_vat.apply_vat(vat)),
            VatOrigin::NotProvided => Some(excl_vat),
            VatOrigin::Unknown => None,
        };

        Some(Price { excl_vat, incl_vat })
    }
}

/// A set of price `Component`s, one for each dimension.
///
/// See: <https://github.com/ocpi/ocpi/blob/release-2.2.1-bugfixes/mod_tariffs.asciidoc#142-pricecomponent-class>.
/// See: <https://github.com/ocpi/ocpi/blob/release-2.2.1-bugfixes/mod_tariffs.asciidoc#145-tariffdimensiontype-enum>.
#[derive(Debug)]
pub struct ComponentSet {
    /// Energy consumed.
    pub energy: Option<Component>,

    /// Flat fee without unit for `step_size`.
    pub flat: Option<Component>,

    /// Duration of time charging.
    pub duration_charging: Option<Component>,

    /// Duration of time not charging.
    pub duration_idle: Option<Component>,
}

impl ComponentSet {
    /// Returns true if all components are `Some`.
    fn has_all_components(&self) -> bool {
        let Self {
            energy,
            flat,
            duration_charging,
            duration_idle,
        } = self;

        flat.is_some() && energy.is_some() && duration_idle.is_some() && duration_charging.is_some()
    }
}

/// A Price Component describes how a certain amount of a certain dimension being consumed
/// translates into an amount of money owed.
///
/// See: <https://github.com/ocpi/ocpi/blob/release-2.2.1-bugfixes/mod_tariffs.asciidoc#142-pricecomponent-class>.
#[derive(Clone, Debug)]
pub struct Component {
    /// Price per unit (excl. VAT) for this dimension.
    price: Money,

    /// Applicable VAT percentage for this tariff dimension. If omitted, no VAT is applicable.
    /// Not providing a VAT is different from 0% VAT, which would be a value of 0.0 here.
    vat: VatOrigin,

    /// Minimum amount to be billed. That is, the dimension will be billed in this `step_size` blocks.
    /// Consumed amounts are rounded up to the smallest multiple of `step_size` that is greater than
    /// the consumed amount.
    ///
    /// For example: if type is TIME and `step_size` has a value of 300, then time will be billed in
    /// blocks of 5 minutes. If 6 minutes were consumed, 10 minutes (2 blocks of `step_size`) will
    /// be billed.
    step_size: u64,
}

impl Component {
    /// Create a new `Component` object.
    fn new(component: &crate::tariff::v221::PriceComponent) -> Self {
        let crate::tariff::v221::PriceComponent {
            price,
            vat,
            step_size,
            dimension_type: _,
        } = component;

        Self {
            price: *price,
            vat: *vat,
            step_size: *step_size,
        }
    }

    /// Return the price of the `Component`.
    pub fn price(&self) -> Money {
        self.price
    }
}

/// A related source and calculated pair of total amounts.
///
/// This is used to express the source and calculated amounts for the total fields of a `CDR`.
///
/// - `total_cost`
/// - `total_fixed_cost`
/// - `total_energy`
/// - `total_energy_cost`
/// - `total_time`
/// - `total_time_cost`
/// - `total_parking_time`
/// - `total_parking_cost`
#[derive(Debug)]
pub struct Total<TCdr, TCalc = TCdr> {
    /// The source value from the `CDR`.
    pub cdr: TCdr,

    /// The value calculated by the [`cdr::price`](crate::cdr::price) function.
    pub calculated: TCalc,
}

/// The range of time the CDR periods span.
#[derive(Debug)]
pub enum PeriodRange {
    /// There are many periods in the CDR and so the range is from the `start_date_time` of the first to
    /// the `start_date_time` of the last.
    Many(Range<DateTime<Utc>>),

    /// There is one period in the CDR and so one `start_date_time`.
    Single(DateTime<Utc>),
}

impl fmt::Display for PeriodRange {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            PeriodRange::Many(Range { start, end }) => write!(f, "[start: {start}, end: {end}]"),
            PeriodRange::Single(date_time) => write!(f, "{date_time}"),
        }
    }
}

/// Where should the tariffs come from when pricing a `CDR`.
///
/// Used with [`cdr::price`](crate::cdr::price).
#[derive(Debug)]
pub enum TariffSource<'buf> {
    /// Use the tariffs from the `CDR`.
    UseCdr,

    /// Ignore the tariffs from the `CDR` and use these instead.
    Override(Vec<crate::tariff::Versioned<'buf>>),
}

impl<'buf> TariffSource<'buf> {
    /// Convenience method to provide a single override tariff.
    pub fn single(tariff: crate::tariff::Versioned<'buf>) -> Self {
        Self::Override(vec![tariff])
    }
}

/// Price a CDR.
///
/// See: [`crate::cdr::price`].
#[instrument(skip_all)]
pub(super) fn cdr(
    cdr_elem: &crate::cdr::Versioned<'_>,
    tariff_source: TariffSource<'_>,
    timezone: Tz,
) -> Verdict<Report> {
    let source_version = cdr_elem.version();
    let cdr = parse_cdr(cdr_elem)?;

    match tariff_source {
        TariffSource::UseCdr => {
            let (v221::cdr::WithTariffs { cdr, tariffs }, warnings) = cdr.into_parts();
            debug!("Using tariffs from CDR");
            let tariffs = tariffs
                .iter()
                .map(|elem| {
                    // Parse the nested tariffs based on the original version of the CDR.
                    match source_version {
                        Version::V221 => crate::tariff::v221::Tariff::from_json(elem),
                        Version::V211 => {
                            let tariff = crate::tariff::v211::Tariff::from_json(elem);
                            // `ocpi-tariffs` considers the `v221` tariff to be the "normalized" version.
                            // A `v211` tariff is converted to a `v221` tariff for various operations.
                            tariff.map_caveat(crate::tariff::v221::Tariff::from)
                        }
                    }
                })
                .collect::<Result<Vec<_>, _>>()?;

            let cdr = cdr.into_caveat(warnings);

            Ok(price_v221_cdr_with_tariffs(
                cdr_elem, cdr, tariffs, timezone,
            )?)
        }
        TariffSource::Override(tariffs) => {
            let cdr = cdr.map(v221::cdr::WithTariffs::discard_tariffs);

            debug!("Using override tariffs");
            let tariffs = tariffs
                .iter()
                .map(tariff::parse)
                .collect::<Result<Vec<_>, _>>()?;

            Ok(price_v221_cdr_with_tariffs(
                cdr_elem, cdr, tariffs, timezone,
            )?)
        }
    }
}

/// Price a single charge-session using a tariff selected from a list.
///
/// Returns a report containing the totals, subtotals, and a breakdown of the calculation.
/// Price a single charge-session using a single tariff.
///
/// Returns a report containing the totals, subtotals, and a breakdown of the calculation.
fn price_v221_cdr_with_tariffs(
    cdr_elem: &crate::cdr::Versioned<'_>,
    cdr: Caveat<v221::Cdr, Warning>,
    tariffs: Vec<Caveat<crate::tariff::v221::Tariff<'_>, crate::tariff::Warning>>,
    timezone: Tz,
) -> Verdict<Report> {
    debug!(?timezone, version = ?cdr_elem.version(), "Pricing CDR");
    let (cdr, mut warnings) = cdr.into_parts();
    let v221::Cdr {
        start_date_time,
        end_date_time,
        charging_periods,
        totals: cdr_totals,
    } = cdr;

    // Convert each versioned tariff JSON to a structured tariff.
    //
    // This generates a list of `TariffReport`s that are returned to the caller in the `Report`.
    // One of the structured tariffs is selected for use in the `price_periods` function.
    let (tariff_reports, tariffs): (Vec<_>, Vec<_>) = tariffs
        .into_iter()
        .enumerate()
        .map(|(index, tariff)| {
            let (tariff, warnings) = tariff.into_parts();
            (
                TariffReport {
                    origin: TariffOrigin {
                        index,
                        id: tariff.id.to_string(),
                        currency: tariff.currency,
                    },
                    warnings: warnings.into_path_map(),
                },
                tariff,
            )
        })
        .unzip();

    debug!(tariffs = ?tariffs.iter().map(|t| t.id).collect::<Vec<_>>(), "Found tariffs(by id) in CDR");

    let tariffs_normalized = tariff::normalize_all(&tariffs);
    let Some((tariff_index, tariff)) =
        tariff::find_first_active(tariffs_normalized, start_date_time)
    else {
        return warnings.bail(cdr_elem.as_element(), Warning::NoValidTariff);
    };

    debug!(tariff_index, id = ?tariff.id(), "Found active tariff");
    debug!(%timezone, "Found timezone");

    // Convert the CDRs periods to the API input period.
    let periods = charging_periods.into_iter().map(Period::from).collect();

    let periods = normalize_periods(periods, end_date_time, timezone);
    let price_cdr_report = price_periods(&periods, &tariff)
        .with_element(cdr_elem.as_element())?
        .gather_warnings_into(&mut warnings);

    if tariff.has_reservation_elements() {
        warnings.insert(
            cdr_elem.as_element(),
            Warning::Tariff(crate::tariff::Warning::ReservationElementSkipped),
        );
    }

    let mut report = generate_report(
        &cdr_totals,
        timezone,
        tariff_reports,
        price_cdr_report,
        TariffOrigin {
            index: tariff_index,
            id: tariff.id().to_owned(),
            currency: tariff.currency(),
        },
    );

    if let Some(total_cost) = report.total_cost.calculated.as_mut() {
        if let Some(min_price) = tariff.min_price() {
            if *total_cost < min_price {
                *total_cost = min_price;
                warnings.insert(
                    cdr_elem.as_element(),
                    crate::tariff::Warning::TotalCostClampedToMin.into(),
                );
            }
        }

        if let Some(max_price) = tariff.max_price() {
            if *total_cost > max_price {
                *total_cost = max_price;
                warnings.insert(
                    cdr_elem.as_element(),
                    crate::tariff::Warning::TotalCostClampedToMax.into(),
                );
            }
        }
    }

    Ok(report.into_caveat(warnings))
}

/// Price a list of normalized [`Period`]s using a [`VersionedJson`](crate::tariff::VersionedJson) tariff.
pub(crate) fn periods(
    end_date_time: DateTime<Utc>,
    timezone: Tz,
    tariff_elem: &crate::tariff::v221::Tariff<'_>,
    mut periods: Vec<Period>,
) -> VerdictDeferred<PeriodsReport> {
    // Make sure the periods are sorted by time as the start date of one period determines the end
    // date of the previous period.
    periods.sort_by_key(|p| p.start_date_time);
    let tariff = Tariff::from_v221(tariff_elem);
    let periods = normalize_periods(periods, end_date_time, timezone);
    price_periods(&periods, &tariff)
}

fn normalize_periods(
    periods: Vec<Period>,
    end_date_time: DateTime<Utc>,
    local_timezone: Tz,
) -> Vec<PeriodNormalized> {
    debug!("Normalizing CDR periods");

    // Each new period is linked to the previous periods data.
    let mut previous_end_snapshot = Option::<TotalsSnapshot>::None;

    // The end-date of the first period is the start-date of the second and so on.
    let end_dates = {
        let mut end_dates = periods
            .iter()
            .skip(1)
            .map(|p| p.start_date_time)
            .collect::<Vec<_>>();

        // The last end-date is the end-date of the CDR.
        end_dates.push(end_date_time);
        end_dates
    };

    let periods = periods
        .into_iter()
        .zip(end_dates)
        .enumerate()
        .map(|(index, (period, end_date_time))| {
            trace!(index, "processing\n{period:#?}");
            let Period {
                start_date_time,
                consumed,
            } = period;

            let period = if let Some(prev_end_snapshot) = previous_end_snapshot.take() {
                let start_snapshot = prev_end_snapshot;
                let end_snapshot = start_snapshot.next(&consumed, end_date_time);

                let period = PeriodNormalized {
                    consumed,
                    start_snapshot,
                    end_snapshot,
                };
                trace!("Adding new period based on the last added\n{period:#?}");
                period
            } else {
                let start_snapshot = TotalsSnapshot::zero(start_date_time, local_timezone);
                let end_snapshot = start_snapshot.next(&consumed, end_date_time);

                let period = PeriodNormalized {
                    consumed,
                    start_snapshot,
                    end_snapshot,
                };
                trace!("Adding new period\n{period:#?}");
                period
            };

            previous_end_snapshot.replace(period.end_snapshot.clone());
            period
        })
        .collect::<Vec<_>>();

    periods
}

/// Price the given set of CDR periods using a normalized `Tariff`.
fn price_periods(periods: &[PeriodNormalized], tariff: &Tariff) -> VerdictDeferred<PeriodsReport> {
    debug!(count = periods.len(), "Pricing CDR periods");

    if tracing::enabled!(tracing::Level::TRACE) {
        trace!("# CDR period list:");
        for period in periods {
            trace!("{period:#?}");
        }
    }

    let period_totals = period_totals(periods, tariff);
    let (billed, mut warnings) = period_totals.calculate_billed()?.into_parts();

    if tariff.has_reservation_elements() {
        warnings.insert(Warning::Tariff(
            crate::tariff::Warning::ReservationElementSkipped,
        ));
    }

    let (billable, periods, totals) = billed;
    let total_costs = total_costs(&periods, tariff);
    let report = PeriodsReport {
        billable,
        periods,
        totals,
        total_costs,
    };

    Ok(report.into_caveat_deferred(warnings))
}

/// The internal report generated from the [`periods`] fn.
pub(crate) struct PeriodsReport {
    /// The billable dimensions calculated by applying the step-size to each dimension.
    pub billable: Billable,

    /// A list of reports for each charging period that occurred during a session.
    pub periods: Vec<PeriodReport>,

    /// The totals for each dimension.
    pub totals: Totals,

    /// The total costs for each dimension.
    pub total_costs: TotalCosts,
}

/// A report for a single charging period that occurred during a session.
///
/// A charging period is a period of time that has relevance for the total costs of a CDR.
/// During a charging session, different parameters change all the time, like the amount of energy used,
/// or the time of day. These changes can result in another [`PriceComponent`](https://github.com/ocpi/ocpi/blob/release-2.2.1-bugfixes/mod_tariffs.asciidoc#142-pricecomponent-class) of the Tariff becoming active.
#[derive(Debug)]
pub struct PeriodReport {
    /// The start time of this period.
    pub start_date_time: DateTime<Utc>,

    /// The end time of this period.
    pub end_date_time: DateTime<Utc>,

    /// A structure that contains results per dimension.
    pub dimensions: Dimensions,
}

impl PeriodReport {
    /// The total cost of all dimensions in this period.
    pub fn cost(&self) -> Option<Price> {
        [
            self.dimensions
                .duration_charging
                .as_ref()
                .and_then(Dimension::cost),
            self.dimensions
                .duration_idle
                .as_ref()
                .and_then(Dimension::cost),
            self.dimensions.flat.cost(),
            self.dimensions.energy.as_ref().and_then(Dimension::cost),
        ]
        .into_iter()
        .fold(None, |accum, next| {
            if accum.is_none() && next.is_none() {
                None
            } else {
                Some(
                    accum
                        .unwrap_or_default()
                        .saturating_add(next.unwrap_or_default()),
                )
            }
        })
    }
}

/// A [`PeriodReport`] under construction during the step-size calculation.
///
/// After step sizes are applied the step-size fields are dropped and this converts into
/// a [`PeriodReport`].
#[derive(Debug)]
struct PeriodReportScratch {
    start_date_time: DateTime<Utc>,
    end_date_time: DateTime<Utc>,
    dimensions: Dimensions,
    step_size_duration_charging: Option<Component>,
    step_size_duration_idle: Option<Component>,
    step_size_energy: Option<Component>,
}

impl From<PeriodReportScratch> for PeriodReport {
    fn from(scratch: PeriodReportScratch) -> Self {
        Self {
            start_date_time: scratch.start_date_time,
            end_date_time: scratch.end_date_time,
            dimensions: scratch.dimensions,
        }
    }
}

/// The result of normalizing the CDR charging periods.
#[derive(Debug)]
struct PeriodTotals {
    /// The list of periods under construction.
    periods: Vec<PeriodReportScratch>,

    /// The totals for each dimension.
    totals: Totals,
}

/// The totals for each dimension.
#[derive(Debug, Default)]
pub(crate) struct Totals {
    /// The total energy used during a session.
    pub energy: Option<Kwh>,

    /// The total charging time used during a session.
    ///
    /// Some if the charging happened during the session.
    pub duration_charging: Option<TimeDelta>,

    /// The total idle time during a session.
    ///
    /// Some if there was idle time during the session.
    pub duration_idle: Option<TimeDelta>,
}

impl PeriodTotals {
    /// Apply step sizes and convert scratch periods into final [`PeriodReport`]s.
    fn calculate_billed(self) -> VerdictDeferred<(Billable, Vec<PeriodReport>, Totals)> {
        let mut warnings = warning::SetDeferred::new();
        let Self {
            mut periods,
            totals,
        } = self;

        let billable =
            apply_step_sizes(&mut periods, &totals)?.gather_deferred_warnings_into(&mut warnings);

        let periods = periods.into_iter().map(PeriodReport::from).collect();

        Ok((billable, periods, totals).into_caveat_deferred(warnings))
    }
}

/// The billable dimensions calculated by applying the step-size to each dimension.
#[derive(Debug)]
pub(crate) struct Billable {
    /// The billable charging time.
    duration_charging: Option<TimeDelta>,

    /// The billable idle time.
    duration_idle: Option<TimeDelta>,

    /// The billable energy use.
    energy: Option<Kwh>,
}

/// Map the `session::ChargePeriod`s to a normalized `Period` and calculate the step size and
/// totals for each dimension.
fn period_totals(periods: &[PeriodNormalized], tariff: &Tariff) -> PeriodTotals {
    let mut has_flat_fee = false;
    let mut totals = Totals::default();

    debug!(
        tariff_id = tariff.id(),
        period_count = periods.len(),
        "Accumulating dimension totals for each period"
    );

    let periods = periods
        .iter()
        .enumerate()
        .map(|(index, period)| {
            let mut component_set = tariff.active_components(period);
            trace!(
                index,
                "Creating charge period with Dimension\n{period:#?}\n{component_set:#?}"
            );

            if component_set.flat.is_some() {
                if has_flat_fee {
                    component_set.flat = None;
                } else {
                    has_flat_fee = true;
                }
            }

            // Extract step-size components before consuming the component_set.
            let step_size_duration_charging = if period.consumed.duration_charging.is_some() {
                component_set.duration_charging.clone()
            } else {
                None
            };
            let step_size_duration_idle = if period.consumed.duration_idle.is_some() {
                component_set.duration_idle.clone()
            } else {
                None
            };
            let step_size_energy = if period.consumed.energy.is_some() {
                component_set.energy.clone()
            } else {
                None
            };

            let dimensions = Dimensions::new(component_set, &period.consumed);

            trace!(period_index = index, "Dimensions created\n{dimensions:#?}");

            if let Some(dim) = &dimensions.duration_charging {
                let acc = totals.duration_charging.get_or_insert_default();
                *acc = acc.saturating_add(dim.volume);
            }

            if let Some(dim) = &dimensions.energy {
                let acc = totals.energy.get_or_insert_default();
                *acc = acc.saturating_add(dim.volume);
            }

            if let Some(dim) = &dimensions.duration_idle {
                let acc = totals.duration_idle.get_or_insert_default();
                *acc = acc.saturating_add(dim.volume);
            }

            trace!(period_index = index, ?totals, "Update totals");

            PeriodReportScratch {
                start_date_time: period.start_snapshot.date_time,
                end_date_time: period.end_snapshot.date_time,
                dimensions,
                step_size_duration_charging,
                step_size_duration_idle,
                step_size_energy,
            }
        })
        .collect::<Vec<_>>();

    PeriodTotals { periods, totals }
}

/// The total costs for each dimension.
#[derive(Debug, Default)]
pub(crate) struct TotalCosts {
    /// The [`Price`] for all energy used during a session.
    pub energy: Option<Price>,

    /// The [`Price`] for all flat rates applied during a session.
    pub fixed: Option<Price>,

    /// The [`Price`] for all charging time used during a session.
    pub duration_charging: Option<Price>,

    /// The [`Price`] for all idle time used during a session.
    pub duration_idle: Option<Price>,
}

impl TotalCosts {
    /// Summate each dimension total into a single total.
    ///
    /// Return `None` if there are no cost dimensions otherwise return `Some`.
    pub(crate) fn total(&self) -> Option<Price> {
        let Self {
            energy,
            fixed,
            duration_charging,
            duration_idle,
        } = self;
        debug!(
            energy = %DisplayOption(*energy),
            fixed = %DisplayOption(*fixed),
            duration_charging = %DisplayOption(*duration_charging),
            duration_idle = %DisplayOption(*duration_idle),
            "Calculating total costs."
        );
        [energy, fixed, duration_charging, duration_idle]
            .into_iter()
            .fold(None, |accum: Option<Price>, next| match (accum, next) {
                (None, None) => None,
                _ => Some(
                    accum
                        .unwrap_or_default()
                        .saturating_add(next.unwrap_or_default()),
                ),
            })
    }
}

/// Accumulate total costs per dimension across all periods.
fn total_costs(periods: &[PeriodReport], tariff: &Tariff) -> TotalCosts {
    let mut total_costs = TotalCosts::default();

    debug!(
        tariff_id = tariff.id(),
        period_count = periods.len(),
        "Accumulating dimension costs for each period"
    );
    for (index, period) in periods.iter().enumerate() {
        let dimensions = &period.dimensions;

        trace!(period_index = index, "Processing period");

        let energy_cost = dimensions.energy.as_ref().and_then(Dimension::cost);
        let fixed_cost = dimensions.flat.cost();
        let duration_charging_cost = dimensions
            .duration_charging
            .as_ref()
            .and_then(Dimension::cost);
        let duration_idle_cost = dimensions.duration_idle.as_ref().and_then(Dimension::cost);

        trace!(?total_costs.energy, ?energy_cost, "Energy cost");
        trace!(?total_costs.duration_charging, ?duration_charging_cost, "Charging cost");
        trace!(?total_costs.duration_idle, ?duration_idle_cost, "Idle cost");
        trace!(?total_costs.fixed, ?fixed_cost, "Fixed cost");

        total_costs.energy = match (total_costs.energy, energy_cost) {
            (None, None) => None,
            (total, period) => Some(
                total
                    .unwrap_or_default()
                    .saturating_add(period.unwrap_or_default()),
            ),
        };

        total_costs.duration_charging =
            match (total_costs.duration_charging, duration_charging_cost) {
                (None, None) => None,
                (total, period) => Some(
                    total
                        .unwrap_or_default()
                        .saturating_add(period.unwrap_or_default()),
                ),
            };

        total_costs.duration_idle = match (total_costs.duration_idle, duration_idle_cost) {
            (None, None) => None,
            (total, period) => Some(
                total
                    .unwrap_or_default()
                    .saturating_add(period.unwrap_or_default()),
            ),
        };

        total_costs.fixed = match (total_costs.fixed, fixed_cost) {
            (None, None) => None,
            (total, period) => Some(
                total
                    .unwrap_or_default()
                    .saturating_add(period.unwrap_or_default()),
            ),
        };

        trace!(period_index = index, ?total_costs, "Update totals");
    }

    total_costs
}

fn generate_report(
    cdr_totals: &v221::cdr::Totals,
    timezone: Tz,
    tariff_reports: Vec<TariffReport>,
    price_periods_report: PeriodsReport,
    tariff_used: TariffOrigin,
) -> Report {
    let PeriodsReport {
        billable,
        periods,
        totals,
        total_costs,
    } = price_periods_report;
    trace!("Update billed totals {billable:#?}");

    let total_cost = total_costs.total();

    debug!(total_cost = %DisplayOption(total_cost.as_ref()));

    let total_time = {
        debug!(
            period_start = %DisplayOption(periods.first().map(|p| p.start_date_time)),
            period_end = %DisplayOption(periods.last().map(|p| p.end_date_time)),
            "Calculating `total_time`"
        );

        periods
            .first()
            .zip(periods.last())
            .map(|(first, last)| {
                last.end_date_time
                    .signed_duration_since(first.start_date_time)
            })
            .unwrap_or_default()
    };
    debug!(total_time = %Hms(total_time));

    let report = Report {
        periods,
        tariff_used,
        timezone: timezone.to_string(),
        billed_idle_time: billable.duration_idle,
        billed_energy: billable.energy.round_to_ocpi_scale(),
        billed_charging_time: billable.duration_charging,
        tariff_reports,
        total_charging_time: totals.duration_charging,
        total_cost: Total {
            cdr: cdr_totals.cost.round_to_ocpi_scale(),
            calculated: total_cost.round_to_ocpi_scale(),
        },
        total_charging_time_cost: Total {
            cdr: cdr_totals.duration_charging_cost.round_to_ocpi_scale(),
            calculated: total_costs.duration_charging.round_to_ocpi_scale(),
        },
        total_time: Total {
            cdr: cdr_totals.duration_charging,
            calculated: total_time,
        },
        total_idle_cost: Total {
            cdr: cdr_totals.duration_idle_cost.round_to_ocpi_scale(),
            calculated: total_costs.duration_idle.round_to_ocpi_scale(),
        },
        total_idle_time: Total {
            cdr: cdr_totals.duration_idle,
            calculated: totals.duration_idle,
        },
        total_energy_cost: Total {
            cdr: cdr_totals.energy_cost.round_to_ocpi_scale(),
            calculated: total_costs.energy.round_to_ocpi_scale(),
        },
        total_energy: Total {
            cdr: cdr_totals.energy.round_to_ocpi_scale(),
            calculated: totals.energy.round_to_ocpi_scale(),
        },
        total_fixed_cost: Total {
            cdr: cdr_totals.fixed_cost.round_to_ocpi_scale(),
            calculated: total_costs.fixed.round_to_ocpi_scale(),
        },
    };

    trace!("{report:#?}");

    report
}

/// Apply step sizes by iterating in reverse over the scratch periods to find the last period
/// that had an active component for each dimension, then mutating its `billed_volume`.
fn apply_step_sizes(
    periods: &mut [PeriodReportScratch],
    totals: &Totals,
) -> VerdictDeferred<Billable> {
    let mut warnings = warning::SetDeferred::new();

    let has_idle_step_size = periods.iter().any(|p| p.step_size_duration_idle.is_some());

    let duration_charging = if let Some(total) = totals.duration_charging {
        let mut result = Some(total);
        for period in periods.iter_mut().rev() {
            let Some(step) = period
                .step_size_duration_charging
                .as_ref()
                .map(|c| c.step_size)
            else {
                continue;
            };
            if has_idle_step_size {
                result = Some(total);
            } else if let Some(dim) = period.dimensions.duration_charging.as_mut() {
                let dt = duration_step_size(total, &mut dim.billed_volume, step)?
                    .gather_deferred_warnings_into(&mut warnings);
                result = Some(dt);
            }
            break;
        }
        result
    } else {
        None
    };

    let duration_idle = if let Some(total) = totals.duration_idle {
        let mut result = Some(total);
        for period in periods.iter_mut().rev() {
            let Some(step) = period.step_size_duration_idle.as_ref().map(|c| c.step_size) else {
                continue;
            };
            if let Some(dim) = period.dimensions.duration_idle.as_mut() {
                let dt = duration_step_size(total, &mut dim.billed_volume, step)?
                    .gather_deferred_warnings_into(&mut warnings);
                result = Some(dt);
            }
            break;
        }
        result
    } else {
        None
    };

    let energy = if let Some(total) = totals.energy {
        let mut result = Some(total);
        for period in periods.iter_mut().rev() {
            let Some(step) = period.step_size_energy.as_ref().map(|c| c.step_size) else {
                continue;
            };
            if step == 0 {
                result = Some(total);
            } else {
                let step_dec = Decimal::from(step);
                if let Some(dim) = period.dimensions.energy.as_mut() {
                    let Some(watt_hours) = total.watt_hours().checked_div(step_dec) else {
                        return warnings.bail(duration::Warning::Overflow.into());
                    };
                    let total_billed_volume =
                        Kwh::from_watt_hours(watt_hours.ceil().saturating_mul(step_dec));
                    let period_delta_volume = total_billed_volume.saturating_sub(total);
                    dim.billed_volume = dim.billed_volume.saturating_add(period_delta_volume);
                    result = Some(total_billed_volume);
                }
            }
            break;
        }
        result
    } else {
        None
    };

    Ok(Billable {
        duration_charging,
        duration_idle,
        energy,
    }
    .into_caveat_deferred(warnings))
}

/// Return the duration as a `Decimal` amount of seconds.
fn delta_as_seconds_dec(delta: TimeDelta) -> Decimal {
    Decimal::from(delta.num_milliseconds())
        .checked_div(Decimal::from(duration::MILLIS_IN_SEC))
        .expect("Can't overflow; See test `as_seconds_dec_should_not_overflow`")
}

/// Create a `HoursDecimal` from a `Decimal` amount of seconds.
fn delta_from_seconds_dec(seconds: Decimal) -> VerdictDeferred<TimeDelta> {
    let millis = seconds.saturating_mul(Decimal::from(duration::MILLIS_IN_SEC));
    let Ok(millis) = i64::try_from(millis) else {
        return Err(warning::ErrorSetDeferred::with_warn(
            duration::Warning::Overflow.into(),
        ));
    };
    let Some(delta) = TimeDelta::try_milliseconds(millis) else {
        return Err(warning::ErrorSetDeferred::with_warn(
            duration::Warning::Overflow.into(),
        ));
    };
    Ok(delta.into_caveat_deferred(warning::SetDeferred::new()))
}

/// Apply a duration based step size for either `time` or `idle_time`.
fn duration_step_size(
    total_volume: TimeDelta,
    period_billed_volume: &mut TimeDelta,
    step_size: u64,
) -> VerdictDeferred<TimeDelta> {
    if step_size == 0 {
        return Ok(total_volume.into_caveat_deferred(warning::SetDeferred::new()));
    }

    let total_seconds = delta_as_seconds_dec(total_volume);
    let step_size = Decimal::from(step_size);

    let Some(x) = total_seconds.checked_div(step_size) else {
        return Err(warning::ErrorSetDeferred::with_warn(
            duration::Warning::Overflow.into(),
        ));
    };
    let total_billed_volume = delta_from_seconds_dec(x.ceil().saturating_mul(step_size))?;

    let period_delta_volume = total_billed_volume.saturating_sub(total_volume);
    *period_billed_volume = period_billed_volume.saturating_add(period_delta_volume);

    Ok(total_billed_volume)
}

fn parse_cdr<'buf>(cdr: &crate::cdr::Versioned<'buf>) -> Verdict<v221::cdr::WithTariffs<'buf>> {
    match cdr.version() {
        Version::V211 => {
            let cdr = v211::cdr::WithTariffs::from_json(cdr.as_element())?;
            Ok(cdr.map(v221::cdr::WithTariffs::from))
        }
        Version::V221 => v221::cdr::WithTariffs::from_json(cdr.as_element()),
    }
}