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_warning_ids_path_map;

mod tariff;
mod v211;
mod v221;

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

use chrono::{DateTime, Datelike, TimeDelta, Utc};
use chrono_tz::Tz;
use rust_decimal::Decimal;
use tracing::{debug, error, instrument, trace};

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

use tariff::Tariff;

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 time consumed in this period.
    pub duration_parking: 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_parking",
                &self.duration_parking.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 parking time after applying step-size
    pub billed_parking_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).
    pub total_parking_time: Total<Option<TimeDelta>>,

    /// Total duration of the charging session (including the duration of charging and not charging).
    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 cost related to parking of this transaction, including fixed price components, in the specified currency.
    pub total_parking_cost: Total<Option<Price>>,

    /// Total sum of all the cost related to a reservation of a Charge Point, including fixed price components, in the specified currency.
    pub total_reservation_cost: Total<Option<Price>>,

    /// Total sum of all the cost related to duration of charging during this transaction, in the specified currency.
    pub total_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_parking_time",
                &self.billed_parking_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_parking_time", &self.total_parking_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_parking_cost", &self.total_parking_cost)
            .field("total_reservation_cost", &self.total_reservation_cost)
            .field("total_time_cost", &self.total_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,

    /// The given dimension should have a volume
    DimensionShouldHaveVolume {
        dimension_name: &'static str,
    },

    /// 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>,
    },

    /// An internal error occurred.
    ///
    /// The details are logged using debug level.
    InternalError,

    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),

    /// An error occurred while deserializing a `CDR` or tariff.
    Parse(ParseError),

    /// 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::DimensionShouldHaveVolume { dimension_name } => {
                write!(f, "Dimension `{dimension_name}` should have volume")
            }
            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::InternalError => f.write_str("Internal error"),
            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::Parse(err) => {
                write!(f, "{err}")
            }
            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::DimensionShouldHaveVolume { dimension_name } => {
                warning::Id::from_string(format!("dimension_should_have_volume({dimension_name})"))
            }
            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::InternalError => warning::Id::from_static("internal_error"),
            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::Parse(ParseError { object: _, kind }) => kind.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<warning::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.
#[derive(Debug)]
pub struct Dimensions {
    /// Energy consumed.
    pub energy: Dimension<Kwh>,

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

    /// Duration of time charging.
    pub duration_charging: Dimension<TimeDelta>,

    /// Duration of time not charging.
    pub duration_parking: Dimension<TimeDelta>,
}

impl Dimensions {
    fn new(components: ComponentSet, consumed: &Consumed) -> Self {
        let ComponentSet {
            energy: energy_price,
            flat: flat_price,
            duration_charging: duration_charging_price,
            duration_parking: duration_parking_price,
        } = components;

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

        Self {
            energy: Dimension {
                price: energy_price,
                volume: *energy,
                billed_volume: *energy,
            },
            flat: Dimension {
                price: flat_price,
                volume: Some(()),
                billed_volume: Some(()),
            },
            duration_charging: Dimension {
                price: duration_charging_price,
                volume: *duration_charging,
                billed_volume: *duration_charging,
            },
            duration_parking: Dimension {
                price: duration_parking_price,
                volume: *duration_parking,
                billed_volume: *duration_parking,
            },
        }
    }
}

#[derive(Debug)]
/// A report for a single dimension during a single 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.
    /// It could be that no volume was provided during this period for this dimension in which case
    /// the `volume` is `None`.
    pub volume: Option<V>,

    /// This field contains the optional 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: Option<V>,
}

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

        let excl_vat = volume.cost(price_component.price);

        let incl_vat = match price_component.vat {
            VatApplicable::Applicable(vat) => Some(excl_vat.apply_vat(vat)),
            VatApplicable::Inapplicable => Some(excl_vat),
            VatApplicable::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_parking: Option<Component>,
}

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

        flat.is_some()
            && energy.is_some()
            && duration_parking.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 {
    /// The index of the tariff this `Component` lives in.
    pub tariff_element_index: usize,

    /// Price per unit (excl. VAT) for this dimension.
    pub 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.
    pub vat: VatApplicable,

    /// 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.
    pub step_size: u64,
}

impl Component {
    fn new(component: &crate::tariff::v221::PriceComponent, tariff_element_index: usize) -> Self {
        let crate::tariff::v221::PriceComponent {
            price,
            vat,
            step_size,
            dimension_type: _,
        } = component;

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

/// 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`
/// - `total_reservation_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])
    }
}

#[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(Warning::NoValidTariff, cdr_elem.as_element());
    };

    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::try_from)
        .collect::<Result<Vec<_>, _>>()
        .map_err(|err| warning::ErrorSet::with_warn(Warning::Parse(err), cdr_elem.as_element()))?;

    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);

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

    Ok(report.into_caveat(warnings))
}

/// Price a list of normalized [`Period`]s using a [`Versioned`](crate::tariff::Versioned) 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, warnings) = period_totals.calculate_billed()?.into_parts();
    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 {
    fn new(period: &PeriodNormalized, dimensions: Dimensions) -> Self {
        Self {
            start_date_time: period.start_snapshot.date_time,
            end_date_time: period.end_snapshot.date_time,
            dimensions,
        }
    }

    /// The total cost of all dimensions in this period.
    pub fn cost(&self) -> Option<Price> {
        [
            self.dimensions.duration_charging.cost(),
            self.dimensions.duration_parking.cost(),
            self.dimensions.flat.cost(),
            self.dimensions.energy.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()),
                )
            }
        })
    }
}

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

    /// The computed step size.
    step_size: StepSize,

    /// 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 parking time used during a session.
    ///
    /// Some if there was idle time during the session.
    pub duration_parking: Option<TimeDelta>,
}

impl PeriodTotals {
    /// Calculate the billed dimensions by applying the step-size to each dimension.
    ///
    /// Applying the step size can mutate the dimension values contained in the `Period`.
    fn calculate_billed(self) -> VerdictDeferred<(Billable, Vec<PeriodReport>, Totals)> {
        let mut warnings = warning::SetDeferred::new();
        let Self {
            mut periods,
            step_size,
            totals,
        } = self;
        let charging_time = totals
            .duration_charging
            .map(|dt| step_size.apply_time(&mut periods, dt))
            .transpose()?
            .gather_deferred_warnings_into(&mut warnings);
        let energy = totals
            .energy
            .map(|kwh| step_size.apply_energy(&mut periods, kwh))
            .transpose()?
            .gather_deferred_warnings_into(&mut warnings);
        let parking_time = totals
            .duration_parking
            .map(|dt| step_size.apply_parking_time(&mut periods, dt))
            .transpose()?
            .gather_deferred_warnings_into(&mut warnings);
        let billed = Billable {
            charging_time,
            energy,
            parking_time,
        };
        Ok((billed, 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.
    charging_time: Option<TimeDelta>,

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

    /// The billable parking time.
    parking_time: Option<TimeDelta>,
}

/// 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 step_size = StepSize::new();
    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;
                }
            }

            step_size.update(index, &component_set, period);

            trace!(period_index = index, "Step size updated\n{step_size:#?}");

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

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

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

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

            if let Some(dt) = dimensions.duration_parking.volume {
                let acc = totals.duration_parking.get_or_insert_default();
                *acc = acc.saturating_add(dt);
            }

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

            PeriodReport::new(period, dimensions)
        })
        .collect::<Vec<_>>();

    PeriodTotals {
        periods,
        step_size,
        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 parking time used during a session.
    pub duration_parking: 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_parking,
        } = self;
        debug!(
            energy = %DisplayOption(*energy),
            fixed = %DisplayOption(*fixed),
            duration_charging = %DisplayOption(*duration_charging),
            duration_parking = %DisplayOption(*duration_parking),
            "Calculating total costs."
        );
        [energy, fixed, duration_charging, duration_parking]
            .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.cost();
        let fixed_cost = dimensions.flat.cost();
        let duration_charging_cost = dimensions.duration_charging.cost();
        let duration_parking_cost = dimensions.duration_parking.cost();

        trace!(?total_costs.energy, ?energy_cost, "Energy cost");
        trace!(?total_costs.duration_charging, ?duration_charging_cost, "Charging cost");
        trace!(?total_costs.duration_parking, ?duration_parking_cost, "Parking 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_parking = match (total_costs.duration_parking, duration_parking_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_parking_time: billable.parking_time,
        billed_energy: billable.energy.round_to_ocpi_scale(),
        billed_charging_time: billable.charging_time,
        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_time_cost: Total {
            cdr: cdr_totals.time_cost.round_to_ocpi_scale(),
            calculated: total_costs.duration_charging.round_to_ocpi_scale(),
        },
        total_time: Total {
            cdr: cdr_totals.time,
            calculated: total_time,
        },
        total_parking_cost: Total {
            cdr: cdr_totals.parking_cost.round_to_ocpi_scale(),
            calculated: total_costs.duration_parking.round_to_ocpi_scale(),
        },
        total_parking_time: Total {
            cdr: cdr_totals.parking_time,
            calculated: totals.duration_parking,
        },
        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(),
        },
        total_reservation_cost: Total {
            cdr: cdr_totals.reservation_cost.round_to_ocpi_scale(),
            calculated: None,
        },
    };

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

    report
}

#[derive(Debug)]
struct StepSize {
    charging_time: Option<(usize, Component)>,
    parking_time: Option<(usize, Component)>,
    energy: Option<(usize, Component)>,
}

/// 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()))
}

impl StepSize {
    fn new() -> Self {
        Self {
            charging_time: None,
            parking_time: None,
            energy: None,
        }
    }

    fn update(&mut self, index: usize, components: &ComponentSet, period: &PeriodNormalized) {
        if period.consumed.energy.is_some() {
            if let Some(energy) = components.energy.clone() {
                self.energy = Some((index, energy));
            }
        }

        if period.consumed.duration_charging.is_some() {
            if let Some(time) = components.duration_charging.clone() {
                self.charging_time = Some((index, time));
            }
        }

        if period.consumed.duration_parking.is_some() {
            if let Some(parking) = components.duration_parking.clone() {
                self.parking_time = Some((index, parking));
            }
        }
    }

    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 apply_time(
        &self,
        periods: &mut [PeriodReport],
        total: TimeDelta,
    ) -> VerdictDeferred<TimeDelta> {
        let (Some((time_index, price)), None) = (&self.charging_time, &self.parking_time) else {
            return Ok(total.into_caveat_deferred(warning::SetDeferred::new()));
        };

        let Some(period) = periods.get_mut(*time_index) else {
            error!(time_index, "Invalid period index");
            return Err(warning::ErrorSetDeferred::with_warn(Warning::InternalError));
        };
        let Some(volume) = period.dimensions.duration_charging.billed_volume.as_mut() else {
            return Err(warning::ErrorSetDeferred::with_warn(
                Warning::DimensionShouldHaveVolume {
                    dimension_name: "time",
                },
            ));
        };

        Self::duration_step_size(total, volume, price.step_size)
    }

    fn apply_parking_time(
        &self,
        periods: &mut [PeriodReport],
        total: TimeDelta,
    ) -> VerdictDeferred<TimeDelta> {
        let warnings = warning::SetDeferred::new();
        let Some((parking_index, price)) = &self.parking_time else {
            return Ok(total.into_caveat_deferred(warnings));
        };

        let Some(period) = periods.get_mut(*parking_index) else {
            error!(parking_index, "Invalid period index");
            return warnings.bail(Warning::InternalError);
        };
        let Some(volume) = period.dimensions.duration_parking.billed_volume.as_mut() else {
            return warnings.bail(Warning::DimensionShouldHaveVolume {
                dimension_name: "parking_time",
            });
        };

        Self::duration_step_size(total, volume, price.step_size)
    }

    fn apply_energy(
        &self,
        periods: &mut [PeriodReport],
        total_volume: Kwh,
    ) -> VerdictDeferred<Kwh> {
        let warnings = warning::SetDeferred::new();
        let Some((energy_index, price)) = &self.energy else {
            return Ok(total_volume.into_caveat_deferred(warnings));
        };

        if price.step_size == 0 {
            return Ok(total_volume.into_caveat_deferred(warnings));
        }

        let Some(period) = periods.get_mut(*energy_index) else {
            error!(energy_index, "Invalid period index");
            return warnings.bail(Warning::InternalError);
        };
        let step_size = Decimal::from(price.step_size);

        let Some(period_billed_volume) = period.dimensions.energy.billed_volume.as_mut() else {
            return warnings.bail(Warning::DimensionShouldHaveVolume {
                dimension_name: "energy",
            });
        };

        let Some(watt_hours) = total_volume.watt_hours().checked_div(step_size) else {
            return warnings.bail(duration::Warning::Overflow.into());
        };

        let total_billed_volume = Kwh::from_watt_hours(watt_hours.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.into_caveat_deferred(warnings))
    }
}

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()),
    }
}