"""Transport-chain emission calculation and booking.
This module is the heart of the transport-side emiflow pipeline:
* :class:`TransportChainCalculator` walks a chain's elements, looks up
each operation's emission intensity, and produces per-element and
per-chain results (transport activity, distance, emissions for the
*operation* and *energy provision* categories).
* :class:`EmissionIntensityCalculator` derives a new emission
intensity from raw inputs (factor, fuel quantity, mass, distance),
performing unit normalisation so callers can supply values in any
supported unit.
* :class:`TransportBookingService` posts a chain: it geocodes points,
computes emissions, denormalises everything into
:class:`EmissionBookingRecord` rows (one per chain + one per
:class:`TransportLCCValue`), and flips the chain status to
``POSTED``.
ISO 14083 terminology is used throughout: TCE = Transport Chain
Element, transport activity = mass × distance (``tkm``), intensity =
emission / activity.
"""
import functools
from dataclasses import dataclass, field
from decimal import Decimal
from typing import List
from django.utils.translation import gettext as _
from ferrosoft.apps.dataimport.services.database import ImportCallback
from ferrosoft.apps.emiflow.models import (
TransportChain,
TransportChainElement,
DistanceType,
TransportChainVisitor,
BookingError,
EmissionBookingRecord,
PredefinedLCC,
BookingStatus,
VehicleType,
TransportChainType,
EmissionIntensity,
IntensityCapableEntity,
OperationEpitype,
EmissionCategory,
LifecycleCategory,
)
from ferrosoft.apps.emiflow.services.visitor import GeocodingRoutingVisitor
from ferrosoft.apps.ferrobase.models import (
ObjectLookupCache,
MeasurementUnit,
UnitCursor,
UnitCategory,
)
from ferrosoft.apps.ferrobase.models.decimal import DecimalWithUnit
from ferrosoft.apps.ferrobase.models.registry import ModelRegistry
from ferrosoft.apps.ferrobase.services.conversion import normalize_unit
from ferrosoft.apps.ferrobase.tenant.utils import tenant_db_connection
from ferrosoft.decimal import zero
[docs]
class EmissionServiceError(RuntimeError):
"""Base class for emiflow emission-service errors."""
pass
[docs]
class IntensityCalculationError(EmissionServiceError):
"""Raised when emission intensity cannot be computed (bad units, zero activity)."""
pass
[docs]
@dataclass
class EmissionResult:
"""One emission value tagged by category and operation epitype."""
emission_category: EmissionCategory
operation_epitype: OperationEpitype
emission: DecimalWithUnit
[docs]
class EmissionResultMixin:
"""Memoised summation helpers for classes that own a list of :class:`EmissionResult`."""
[docs]
@functools.cached_property
def emission_total(self) -> DecimalWithUnit:
"""Sum of all emissions; ``0 gCO2e`` when the list is empty."""
if len(self.emissions) == 0:
return DecimalWithUnit.zero("gCO2e")
return DecimalWithUnit.create(
sum([e.emission.value for e in self.emissions], zero),
self.emissions[0].emission.unit,
)
[docs]
@functools.cached_property
def emission_operation(self):
"""Sum of emissions in the OPERATION category."""
return self.emission_of_category(EmissionCategory.OPERATION)
[docs]
def emission(
self, emission_category: EmissionCategory, operation_epitype: OperationEpitype
) -> DecimalWithUnit | None:
"""Return the single emission matching both category and epitype, or ``None``."""
return next(
filter(
lambda e: e.emission_category == emission_category
and e.operation_epitype == operation_epitype,
self.emissions,
),
None,
)
[docs]
def emission_of_category(
self, emission_category: EmissionCategory
) -> DecimalWithUnit:
"""Sum emissions in ``emission_category`` (zero when none match)."""
emissions: List[EmissionResult] = list(
filter(lambda e: e.emission_category == emission_category, self.emissions)
)
if len(emissions) == 0:
return DecimalWithUnit.zero("gCO2e")
return DecimalWithUnit(
sum([e.emission.value for e in emissions], zero),
emissions[0].emission.try_unit(),
)
[docs]
@dataclass
class TransportChainElementResult(EmissionResultMixin):
"""Calculation result for one :class:`TransportChainElement`."""
distance_type: DistanceType | None
element: TransportChainElement
transport_activity: DecimalWithUnit
emissions: List[EmissionResult] = field(default_factory=list)
[docs]
@dataclass
class TransportChainResult(EmissionResultMixin):
"""Aggregated calculation result for one :class:`TransportChain`.
Mirrors the ISO 14083 reporting layout: emissions are accumulated
across all elements, distance is summed, and transport activity is
kept grouped by :class:`DistanceType` so it can be reported per
strategy.
"""
chain: TransportChain
emissions: List[EmissionResult] = field(default_factory=list)
tce_results: List[TransportChainElementResult] = field(default_factory=list)
# ISO standard specifies that transport activity shall be reported
# grouped by distance type.
transport_activity: dict[DistanceType, DecimalWithUnit] = field(
default_factory=dict
)
distance: DecimalWithUnit = field(default_factory=DecimalWithUnit.zero)
weight: DecimalWithUnit = field(default_factory=DecimalWithUnit.zero)
[docs]
def add_activity_distance(
self, dtype: DistanceType, activity_distance: DecimalWithUnit
):
"""Accumulate ``activity_distance`` into the bucket for ``dtype``."""
if dtype in self.transport_activity:
self.transport_activity[dtype] += activity_distance
else:
self.transport_activity[dtype] = activity_distance
# @functools.cache
[docs]
def transport_activity_sum(self) -> DecimalWithUnit:
"""Total transport activity across every distance-type bucket."""
return sum(self.transport_activity.values(), DecimalWithUnit.zero())
# @functools.cache
[docs]
def emission_intensity(self, category: EmissionCategory) -> DecimalWithUnit:
"""Return the chain's emission intensity for ``category`` (emission / activity).
Returns a zero ``DecimalWithUnit`` when there is no transport
activity (a chain made entirely of hub elements).
"""
activity = self.transport_activity_sum()
if activity.value == zero:
return DecimalWithUnit.zero()
emission = self.emission_of_category(category)
return DecimalWithUnit.create(
emission.value / activity.value,
"%s/%s" % (emission.try_unit().symbol, activity.try_unit().symbol),
)
[docs]
@functools.cached_property
def emission_intensity_total(self):
"""Sum of operation and energy-provision intensities."""
return self.emission_intensity(
EmissionCategory.OPERATION
) + self.emission_intensity(EmissionCategory.ENERGY_PROVISION)
[docs]
@functools.cached_property
def emission_intensity_operation(self):
"""Operation-only emission intensity."""
return self.emission_intensity(EmissionCategory.OPERATION)
[docs]
@dataclass
class TransportChainSetResult:
"""Aggregated result over a set of transport chains (cross-chain reporting).
Supports ``set_result += chain_result`` so callers can fold a
chain result into the running totals.
"""
emission_total: DecimalWithUnit
emission_operation: DecimalWithUnit
transport_activity: dict[DistanceType, DecimalWithUnit]
emission_intensity_total: DecimalWithUnit
emission_intensity_operation: DecimalWithUnit
distance: DecimalWithUnit
weight: DecimalWithUnit
def __add__(self, other):
"""Return a new set result with ``other`` (a chain result) folded in."""
if isinstance(other, TransportChainResult):
activity = self.transport_activity
for other_dtype, other_value in other.transport_activity.items():
if other_dtype in activity:
activity[other_dtype] += other_value
else:
activity[other_dtype] = other_value
return TransportChainSetResult(
self.emission_total + other.emission_total,
self.emission_operation
+ other.emission_of_category(EmissionCategory.OPERATION),
activity,
self.emission_intensity_total + other.emission_intensity_total,
self.emission_intensity_operation + other.emission_intensity_operation,
self.distance + other.distance,
self.weight + other.weight,
)
else:
raise ValueError("Expecting TransportChainResult")
[docs]
@dataclass
class TransportActivityGroups:
"""Helper container that buckets transport activity by :class:`DistanceType`."""
groups: dict[DistanceType, DecimalWithUnit] = field(default_factory=dict)
[docs]
def add(self, dtype: DistanceType, activity_distance: DecimalWithUnit):
"""Accumulate ``activity_distance`` into the bucket for ``dtype``."""
if dtype in self.groups:
self.groups[dtype] += activity_distance
else:
self.groups[dtype] = activity_distance
[docs]
class TransportChainCalculator:
"""Compute per-element and per-chain emissions for one or more transport chains.
The calculator carries running totals across multiple chains so it
can be reused via :meth:`add_transport_chain` for set-level
reporting; for a single-chain calculation use :meth:`calculate_for`
on a freshly constructed calculator.
"""
def __init__(self):
self.tce_results: List[TransportChainElementResult] = []
self.tc_results: List[TransportChainResult] = []
self.set_result = TransportChainSetResult(
DecimalWithUnit.zero(),
DecimalWithUnit.zero(),
{},
DecimalWithUnit.zero(),
DecimalWithUnit.zero(),
DecimalWithUnit.zero(),
DecimalWithUnit.zero(),
)
[docs]
def calculate_for(self, chain: TransportChain) -> TransportChainResult:
"""Compute the :class:`TransportChainResult` for ``chain``.
For each chain element the operation's emission intensity is
looked up and combined with the transport activity (mass ×
distance for transport TCEs, mass alone for hub TCEs) and the
distance-adjustment factor to yield emissions in the
*operation* and *energy provision* categories.
Elements whose operation lacks an intensity contribute an
empty result so the chain can still be reported on, instead of
failing the whole calculation.
Raises:
RuntimeError: When the chain's summed freight weight has
no measurement unit attached.
"""
# Weight sum of consignments freight items serves as weight input to expressions.
weight = chain.sum_weight()
if weight.unit is None:
raise RuntimeError("Summed weight does not include measurement unit")
# Kilogram must be converted to metric ton, as transport activity is
# measured in "tkm", never in "kgkm".
weight = normalize_unit(UnitCategory.MASS, weight)
# Where results are summed into.
tc_result = TransportChainResult(chain, weight=weight)
# Do "manual" prefetching, because Django insists on using a "base" model manager
# for looking up the operation field (operation category) of a TCE.
intensities = EmissionIntensity.objects.filter(
owner_type=IntensityCapableEntity.TRANSPORT_OPERATION,
owner_id__in=chain.elements.all().values("operation"),
)
intensity_mapping = ObjectLookupCache(intensities, lambda obj: (obj.owner_id,))
for tce in chain.elements_ordered():
if isinstance(tce, TransportChainElement):
# Emission intensity is a very important parameter for calculation of emissions.
intensity: EmissionIntensity | None = intensity_mapping.get(
(tce.operation_id,)
)
if intensity is None:
# Garbage data
tce_result = TransportChainElementResult(
distance_type=None,
element=tce,
transport_activity=DecimalWithUnit.zero(),
)
self.tce_results.append(tce_result)
tc_result.tce_results.append(tce_result)
continue
match OperationEpitype(tce.epitype):
case OperationEpitype.TRANSPORT:
# For TRANSPORT TCE's a transport activity must be defined.
transport_activity = tce.activity_distance_parameters
distance = transport_activity.distance
# For TRANSPORT TCE's the activity distance is the product of weight and distance.
activity_distance = weight * distance
# Group activity distance based on distance type for reporting.
tc_result.add_activity_distance(
transport_activity.distance_type, activity_distance
)
tce_result = TransportChainElementResult(
distance_type=transport_activity.distance_type,
element=tce,
transport_activity=activity_distance,
)
# Calculate emissions for OPERATION and ENERGY_PROVISION categories.
# EmissionResultMixin takes care of summing.
tce_result.emissions.append(
EmissionResult(
EmissionCategory.OPERATION,
OperationEpitype.TRANSPORT,
self._emission_tce(
intensity.operation_intensity,
activity_distance,
transport_activity.daf,
),
)
)
tce_result.emissions.append(
EmissionResult(
EmissionCategory.ENERGY_PROVISION,
OperationEpitype.TRANSPORT,
self._emission_tce(
intensity.energy_provision_intensity,
activity_distance,
transport_activity.daf,
),
)
)
case OperationEpitype.HUB:
distance = DecimalWithUnit.zero()
# For HUB TCE's transport activity is simply the weight.
tce_result = TransportChainElementResult(
distance_type=None,
element=tce,
transport_activity=weight,
)
tce_result.emissions.append(
EmissionResult(
EmissionCategory.OPERATION,
OperationEpitype.HUB,
self._emission_tce(
intensity.operation_intensity,
weight,
Decimal(1),
),
)
)
tce_result.emissions.append(
EmissionResult(
EmissionCategory.ENERGY_PROVISION,
OperationEpitype.HUB,
self._emission_tce(
intensity.energy_provision_intensity,
weight,
Decimal(1),
),
)
)
case _:
raise ValueError("Unsupported operation epitype")
self.tce_results.append(tce_result)
tc_result.tce_results.append(tce_result)
tc_result.emissions += tce_result.emissions
tc_result.distance += distance
return tc_result
[docs]
def add_transport_chain(self, chain: TransportChain):
"""Calculate ``chain`` and fold its result into :attr:`set_result`."""
tc_result = self.calculate_for(chain)
self.tc_results.append(tc_result)
self.set_result += tc_result
@staticmethod
def _emission_tce(
emission_intensity: DecimalWithUnit,
transport_activity: DecimalWithUnit,
daf: Decimal,
) -> DecimalWithUnit:
"""Apply intensity × activity × DAF and cancel the matching units.
Args:
emission_intensity: Intensity in *emission / activity* units
(read from the operation's :class:`EmissionIntensity`).
transport_activity: Activity in the divisor of the intensity
unit; the units must match or :class:`ValueError` is
raised.
daf: Distance adjustment factor (1 for hub TCEs).
Returns:
DecimalWithUnit: Emission in the numerator unit of the
intensity.
"""
# Emission intensity has unit "A/B", while transport activity has "B".
# Multiplying them leaves just A, but B has to match up.
denom, div = emission_intensity.try_unit().fraction
if div != transport_activity.unit.symbol:
raise ValueError(
"Transport activity and emission intensity units are unaligned"
)
return DecimalWithUnit.create(
emission_intensity.value * transport_activity.value * daf, denom
)
[docs]
class TransportChainVisitorImportCallback(ImportCallback):
"""Import callback that runs a :class:`TransportChainVisitor` after import finishes.
Collects every :class:`TransportChain` instance the import emits;
once ``on_finish`` fires (all rows committed), each chain accepts
the wrapped visitor. Used to defer geocoding/routing until imported
rows are queryable.
"""
def __init__(self, visitor: TransportChainVisitor):
self._chains: List[TransportChain] = []
self._visitor = visitor
[docs]
def on_object(self, instance):
"""Buffer the instance when it's a :class:`TransportChain`."""
if isinstance(instance, TransportChain):
self._chains.append(instance)
[docs]
def on_finish(self):
"""Walk every buffered chain with the wrapped visitor."""
for chain in self._chains:
chain.accept(self._visitor)
[docs]
class EmissionIntensityCalculator:
"""Compute emission intensities from input variables.
Either pass an existing :class:`EmissionIntensity` row whose input
fields are populated (the calculator pulls the energy carrier's
density from the related factor), or call :meth:`calculate`
directly with an :class:`EmissionIntensityInputs` value.
"""
[docs]
def calculateFor(self, model: EmissionIntensity) -> List[DecimalWithUnit]:
"""Return the operation and energy-provision intensities for ``model``.
Returns a two-element list in *(energy-provision, operation)*
order — mirroring the order of factor fields on the model.
"""
if isinstance(model, EmissionIntensity):
fac_energy_provision = model.factor_energy_provision_value
fac_operation = model.factor_operation_value
# noinspection PyTypeChecker
return [
self.calculate(
EmissionIntensityInputs(
emission_factor=fac_energy_provision.factor,
quantity=model.quantity,
mass=model.mass,
distance=model.distance,
energy_carrier_density=(
None
if fac_energy_provision is None
else fac_energy_provision.energy_carrier.density
),
)
),
self.calculate(
EmissionIntensityInputs(
emission_factor=fac_operation.factor,
quantity=model.quantity,
mass=model.mass,
distance=model.distance,
energy_carrier_density=(
None
if fac_operation is None
else fac_operation.energy_carrier.density
),
)
),
]
else:
raise ValueError("Unsupported model")
[docs]
def calculate(self, inputs: EmissionIntensityInputs) -> DecimalWithUnit:
"""Calculate one emission intensity from the supplied inputs.
Steps: normalise distance (treat zero as a neutral multiplier
for hub operations), normalise the factor and verify its
divider matches the quantity unit, normalise the quantity
(volume → mass via density when needed), compute activity
distance (mass × distance), and finally
``emission / activity``.
Raises:
IntensityCalculationError: When factor and quantity units
are incompatible, or when activity distance resolves
to zero (which would divide by zero).
"""
distance = normalize_unit(UnitCategory.DISTANCE, inputs.distance)
if distance == DecimalWithUnit.zero():
# Distance of 0 must be turned into a neutral multiplier, as it is a
# normal occurrence for hub operations.
distance = DecimalWithUnit.create(1, None)
factor = normalize_unit(UnitCategory.EMISSION_FACTOR, inputs.emission_factor)
factor_denom, factor_divider = factor.try_unit().fraction
quantity = normalize_unit(
UnitCategory.NATIVE_CONSUMPTION,
inputs.quantity,
density=inputs.energy_carrier_density,
)
mass = normalize_unit(UnitCategory.MASS, inputs.mass)
if factor_divider != quantity.try_unit().symbol:
raise IntensityCalculationError(
"Emission factor and quantity units are unaligned"
)
# Calculate activity distance, which is mass * distance, yielding tkm value.
ad: DecimalWithUnit = mass * distance
if ad == zero:
raise IntensityCalculationError(
_("Activity distance must be greater than zero.")
)
# Calculate emissions using the factor.
emission = normalize_unit(
UnitCategory.EMISSION_VALUE,
DecimalWithUnit.create(quantity.value * factor.value, factor_denom),
)
# Intensity unit is emission unit divided by activity distance unit, e.g. gCO2e/tkm.
intensity_unit = ModelRegistry.Unit.get(
"%s/%s" % (emission.try_unit().symbol, ad.try_unit().symbol)
)
# Finally, calculate intensity.
return DecimalWithUnit.create(emission.value / ad.value, intensity_unit)
[docs]
class TransportBookingService:
"""Post a transport chain's calculated emissions to the booking journal.
Refuses to double-post (already-``POSTED`` chains are skipped to
keep the subscription tonnage counter accurate), enforces that
each vehicle type within the chain uses a single distance
strategy, geocodes geo-points, runs the
:class:`TransportChainCalculator`, and writes one
:class:`EmissionBookingRecord` per chain plus one per
:class:`TransportLCCValue`. Finally moves the chain to
``POSTED``.
"""
_activity_distance_group_query = """
SELECT STRING_AGG(DISTINCT tce.activity_distance_type::text, ';'), em.vehicle_type
FROM emiflow_transportchainelement tce
JOIN emiflow_transportoperation op on tce.operation_id = op.id
JOIN emiflow_emitter em on op.emitter_id = em.id
WHERE chain_id = %s
GROUP BY vehicle_type
"""
[docs]
def book(self, transport: TransportChain) -> list[EmissionBookingRecord]:
"""Calculate and persist booking records, then mark the chain ``POSTED``.
Returns the upserted records, or an empty list when the chain
is not in :class:`~BookingStatus.OPEN` status.
Raises:
BookingError: When chain elements within one vehicle type
mix distance strategies (an ISO 14083 reporting
violation).
"""
if not transport.is_status(BookingStatus.OPEN):
# No double-posting, to prevent over-reporting of tonnage.
return []
# Sanity check
self._ensure_uniform_activity_type_per_vehicle_type(transport)
# Calculations may return DecimalWithUnit.zero(), which doesn't have a
# unit. Therefore, default units must be selected.
default_units = MeasurementUnit.objects.get_lookup(
"gCO2e",
"gCO2e/tkm",
"tkm",
"km",
"t",
)
# For emissions calculation to produce numbers, address has to be geocoded
# and distance calculated.
transport.accept(GeocodingRoutingVisitor.get_instance())
# Get actual emission numbers.
calculator = TransportChainCalculator()
result = calculator.calculate_for(transport)
incoming = transport.consignor
outgoing = transport.consignee
# Collect material IDs and names for booking params.
material_ids = []
material_codes = []
material_names = []
for item in transport.freight_items.all():
material = item.material
material_ids.append(material.pk)
material_codes.append(material.code)
material_names.append(material.name)
# Load predefined LCC to write its possibly customized name into
# emission booking records.
lcc_id = PredefinedLCC.value_from(TransportChainType(transport.transport_type))
lcc = LifecycleCategory.objects.get(pk=lcc_id.value)
activity_distance = result.transport_activity_sum()
ad_unit = activity_distance.unit
intensity_operation_unit = result.emission_intensity_operation.unit
intensity_total_unit = result.emission_intensity_total.unit
distance_unit = result.distance.unit
weight_unit = result.weight.unit
emission_unit = result.emission_total.unit
records = [
EmissionBookingRecord(
booking_date=transport.booking_date,
lifecycle_category=lcc.pk,
lifecycle_category_name=lcc.name,
entity_id=transport.pk,
entity_reference=transport.reference,
simulation=transport.simulation,
emission_total_value=result.emission_total.value,
emission_operation_value=result.emission_operation.value,
emission_unit_id=(emission_unit or default_units["gCO2e"]).pk,
emission_unit_symbol=(emission_unit or default_units["gCO2e"]).symbol,
weight_value=result.weight.value,
weight_unit_id=(weight_unit or default_units["t"]).pk,
weight_unit_symbol=(weight_unit or default_units["t"]).symbol,
material_id=material_ids,
material_code=material_codes,
material_name=material_names,
incoming_partner_id=None if incoming is None else incoming.pk,
incoming_partner_name=None if incoming is None else incoming.name,
outgoing_partner_id=None if outgoing is None else outgoing.pk,
outgoing_partner_name=None if outgoing is None else outgoing.name,
extra={
"activity_distance": {
"unit_id": (ad_unit or default_units["tkm"]).pk,
"unit_symbol": (ad_unit or default_units["tkm"]).symbol,
"value": activity_distance.value,
},
"emission_intensity": {
"operation": {
"unit_id": (
intensity_operation_unit or default_units["gCO2e/tkm"]
).pk,
"unit_symbol": (
intensity_operation_unit or default_units["gCO2e/tkm"]
).symbol,
"value": result.emission_intensity_operation.value,
},
"total": {
"unit_id": (
intensity_total_unit or default_units["gCO2e/tkm"]
).pk,
"unit_symbol": (
intensity_total_unit or default_units["gCO2e/tkm"]
).symbol,
"value": result.emission_intensity_total.value,
},
},
"operations": [
{"name": tce_result.element.operation.name}
for tce_result in result.tce_results
],
"transport_distance": {
"unit_id": (distance_unit or default_units["km"]).pk,
"unit_symbol": (distance_unit or default_units["km"]).symbol,
"value": result.distance.value,
},
},
)
]
for lcc in transport.lcc_values.all():
total_emission = normalize_unit(
UnitCategory.EMISSION_VALUE, lcc.total_emission
)
operation_emission = normalize_unit(
UnitCategory.EMISSION_VALUE, lcc.operation_emission
)
emission_unit = total_emission.unit
records.append(
EmissionBookingRecord(
booking_date=transport.booking_date,
lifecycle_category=lcc.lifecycle_category.pk,
lifecycle_category_name=lcc.lifecycle_category.name,
entity_id=transport.pk,
entity_reference=transport.reference,
simulation=transport.simulation,
emission_total_value=total_emission.value,
emission_operation_value=operation_emission.value,
emission_unit_id=(emission_unit or default_units["gCO2e"]).pk,
emission_unit_symbol=(
emission_unit or default_units["gCO2e"]
).symbol,
weight_value=result.weight.value,
weight_unit_id=result.weight.try_unit().pk,
weight_unit_symbol=result.weight.try_unit().symbol,
material_id=material_ids,
material_code=material_codes,
material_name=material_names,
incoming_partner_id=None if incoming is None else incoming.pk,
incoming_partner_name=None if incoming is None else incoming.name,
outgoing_partner_id=None if outgoing is None else outgoing.pk,
outgoing_partner_name=None if outgoing is None else outgoing.name,
)
)
records = EmissionBookingRecord.objects.upsert(records)
transport.status = BookingStatus.POSTED
transport.save(update_fields=["status"])
return records
def _ensure_uniform_activity_type_per_vehicle_type(self, transport: TransportChain):
"""Reject chains where one vehicle type uses multiple distance strategies.
Raises:
BookingError: When chain elements grouped by vehicle type
produce more than one distinct
``activity_distance_type``.
"""
with tenant_db_connection().cursor() as cursor:
for result in cursor.execute(
self._activity_distance_group_query, [str(transport.pk)]
):
distance_type_ids, vehicle_type_id = result
if distance_type_ids is None or vehicle_type_id is None:
# Query returns NULL for hub operations, which can be
# skipped, as they don't include transport activity.
continue
if ";" in distance_type_ids:
vehicle_type = VehicleType.try_parse(vehicle_type_id)
raise BookingError(
_(
"Different activity distance types per vehicle type are not allowed. Please review transport chain elements with vehicle type %(name)s."
)
% {
"name": "" if vehicle_type is None else vehicle_type.label,
}
)