depot.api.private.consumption

Attributes

_interpolator_cache

_nn_interpolator_cache

_nn_fallback_warned

_LUT_DIM_NAMES

_EARTH_RADIUS_M

_VERTEX_GAP_THRESHOLD_M

Classes

ConsumptionResult

A dataclass that stores the results of a charging simulation for a single trip.

TripSegment

A piece of a trip between two adjacent knots — either consecutive.

ConsumptionInformation

Per-trip consumption inputs decomposed into route-aware segments.

Functions

clear_interpolator_cache()

Clear the module-level interpolator cache.

_classify_nan_query(point, cached)

Classify a 4D query that produced NaN under the regular-grid interpolator.

_get_or_build_interpolator(consumption_lut)

Build or retrieve a cached RegularGridInterpolator for a ConsumptionLut.

_get_or_build_nearest_neighbor_interpolator(cached, lut_id)

Build or retrieve a cached NearestNDInterpolator for a ConsumptionLut.

_haversine_cumulative(coords)

Cumulative haversine distance in meters along a polyline.

_route_geom_knots(route)

Return the route's geometry as an (N, 4) array of [elapsed_distance_m, lon, lat, z].

_station_z(station)

Return the Z coordinate of a station's geom, or None if absent.

_assoc_z(assoc)

Return the Z coordinate of an AssocRouteStation.location, or None if absent.

_build_trip_segments(trip, level_of_loading, t_amb)

Walk a trip's stop times and route geometry to produce TripSegment objects.

extract_trip_information(trip_id, scenario[, ...])

Build a ConsumptionInformation for a trip, decomposed into segments.

initialize_vehicle(rotation, session)

Create and add a new Vehicle object in the database for the given rotation.

add_initial_standby_event(vehicle, session)

Create and add a standby event immediately before the earliest trip of the given vehicle.

find_charger_occupancy(station, time_start, time_end, ...)

Build a timeseries of charger occupancy at a station between two points in time.

find_best_timeslot(station, time_start, time_end, ...)

attempt_opportunity_charging_event(previous_trip, ...)

Module Contents

depot.api.private.consumption._interpolator_cache: Dict[int, Dict[str, Any]]
depot.api.private.consumption._nn_interpolator_cache: Dict[int, scipy.interpolate.NearestNDInterpolator]
depot.api.private.consumption._nn_fallback_warned: set
depot.api.private.consumption.clear_interpolator_cache()

Clear the module-level interpolator cache.

Return type:

None

depot.api.private.consumption._LUT_DIM_NAMES: Tuple[str, str, str, str] = ('incline', 't_amb', 'level_of_loading', 'mean_speed_kmh')
depot.api.private.consumption._classify_nan_query(point, cached)

Classify a 4D query that produced NaN under the regular-grid interpolator.

Returns ("out_of_range", (dim_names...)) if any axis value is outside its scale, else ("ragged", ()) — meaning the query lies inside the bounding box but a neighbouring grid cell is unpopulated.

Parameters:

  • point (numpy.ndarray)

  • cached (Dict[str, Any])

Return type:

Tuple[str, Tuple[str, Ellipsis]]

depot.api.private.consumption._get_or_build_interpolator(consumption_lut)

Build or retrieve a cached RegularGridInterpolator for a ConsumptionLut.

Returns a dict with keys: ‘interpolator’, ‘consumption_array’, ‘incline_scale’, ‘temperature_scale’, ‘level_of_loading_scale’, ‘speed_scale’.

Parameters:

consumption_lut (eflips.model.ConsumptionLut)

Return type:

Dict[str, Any]

depot.api.private.consumption._get_or_build_nearest_neighbor_interpolator(cached, lut_id)

Build or retrieve a cached NearestNDInterpolator for a ConsumptionLut.

This is used as a fallback when the RegularGridInterpolator returns NaN.

Parameters:

  • cached (Dict[str, Any])

  • lut_id (int)

Return type:

scipy.interpolate.NearestNDInterpolator

class depot.api.private.consumption.ConsumptionResult

A dataclass that stores the results of a charging simulation for a single trip.

This class holds both the total change in battery State of Charge (SoC) over the trip as well as an optional timeseries of timestamps and incremental SoC changes. When an entry exists for a given trip in consumption_result, the simulation will use these precomputed values instead of recalculating the SoC changes from the vehicle distance and consumption.

Parameters:

  • delta_soc_total – The total change in the vehicle’s State of Charge over the trip, typically negative if the vehicle is consuming energy (e.g., -0.15 means the SoC dropped by 15%).

  • timestamps – A list of timestamps (e.g., arrival times at stops) that mark the times associated with the SoC changes. The number of timestamps must match the number of entries in delta_soc.

  • delta_soc – A list of cumulative SoC changes corresponding to the timestamps. For example, if delta_soc[i] = -0.02, it means the SoC decreased by 2% between from the start of the trip to timestamps[i]. This list should typically be a monotonic decreasing sequence.

delta_soc_total: float
timestamps: List[datetime.datetime] | None
delta_soc: List[float] | None
class depot.api.private.consumption.TripSegment

A piece of a trip between two adjacent knots — either consecutive.

StopTime boundaries, or synthetic knots inserted from Route.geom vertices to capture intermediate elevation changes.

Each segment is the unit at which the consumption LUT is evaluated.

distance_m: float

2D ground distance in meters.

duration_s: float

Duration of this segment in seconds.

mean_speed_kmh: float

Mean speed in km/h.

incline: float

Signed Δz / distance_m.

0.0 if either knot lacks an elevation.

level_of_loading: float | None

Vehicle loading as a fraction of max payload, or None when no LUT is in use.

t_amb: float | None

Ambient temperature in °C at the segment midpoint, or None when no temperature data is available.

end_time: datetime.datetime

Absolute timestamp at the end of this segment.

consumption_kwh: float | None = None

Energy used on this segment in kWh; populated by ConsumptionInformation.calculate().

class depot.api.private.consumption.ConsumptionInformation

Per-trip consumption inputs decomposed into route-aware segments.

Either consumption_lut or flat_consumption_per_km must be set. Calling calculate() populates segment.consumption_kwh for every segment; generate_consumption_result() then turns that into a ConsumptionResult with cumulative SoC.

trip_id: int
segments: List[TripSegment]
consumption_lut: eflips.model.ConsumptionLut | None = None
flat_consumption_per_km: float | None = None
line_name: str | None = None

Human-readable line name, used purely for diagnostic warnings.

route_name: str | None = None

Human-readable route name, used purely for diagnostic warnings.

trip_departure: datetime.datetime | None = None

Trip departure time, used purely for diagnostic warnings.

trip_arrival: datetime.datetime | None = None

Trip arrival time, used purely for diagnostic warnings.

calculate()

Compute energy consumption for every segment.

  • When consumption_lut is set, the LUT is evaluated once for all segments via a vectorized RegularGridInterpolator call. Any segment whose result is NaN is filled in via a NearestNDInterpolator fallback (a single warning is emitted).

  • When only flat_consumption_per_km is set, each segment’s energy is just flat_consumption_per_km * distance_m / 1000.

The LUT reference is dropped after evaluation to avoid pinning the whole table in memory.

Return type:

None

_warn_nn_fallback(points, nan_mask, cached, lut_id)

Emit a deduplicated ConsistencyWarning for every distinct nearest-neighbor fallback reason encountered on this trip.

Each NaN query is classified as either out_of_range (at least one LUT axis outside its scale; the offending axes are listed) or ragged (all four axes in-range, but a neighbouring grid cell is unpopulated). Warnings are deduplicated by (lut_id, kind, dim_tuple) across the whole process.

Parameters:

  • points (numpy.ndarray)

  • nan_mask (numpy.ndarray)

  • cached (Dict[str, Any])

  • lut_id (int)

Return type:

None

generate_consumption_result(battery_capacity)

Build a ConsumptionResult from per-segment consumption_kwh values.

timestamps matches [s.end_time for s in segments] and delta_soc is the cumulative SoC drop at the end of each segment.

Parameters:

battery_capacity (float)

Return type:

ConsumptionResult

depot.api.private.consumption._EARTH_RADIUS_M = 6371008.8
depot.api.private.consumption._VERTEX_GAP_THRESHOLD_M = 1000.0
depot.api.private.consumption._haversine_cumulative(coords)

Cumulative haversine distance in meters along a polyline.

coords is an (N, 2+) array of (lon, lat[, z]) in degrees. Z is ignored — Route.calculate_length() (PostGIS ST_Length(..., true)) also ignores Z, so the route’s stored distance is purely 2D and we want to match that for elapsed_distance bookkeeping.

Returns an (N,) array, with index 0 always 0.

Parameters:

coords (numpy.ndarray)

Return type:

numpy.ndarray

depot.api.private.consumption._route_geom_knots(route)

Return the route’s geometry as an (N, 4) array of [elapsed_distance_m, lon, lat, z].

The cumulative haversine distance is rescaled so that the last vertex sits at route.distance. Returns None when the route has no geometry or fewer than three vertices (only the endpoints — no useful intermediate Z to add).

Parameters:

route (eflips.model.Route)

Return type:

Optional[numpy.ndarray]

depot.api.private.consumption._station_z(station)

Return the Z coordinate of a station’s geom, or None if absent.

Parameters:

station (eflips.model.Station)

Return type:

Optional[float]

depot.api.private.consumption._assoc_z(assoc)

Return the Z coordinate of an AssocRouteStation.location, or None if absent.

Parameters:

assoc (Optional[eflips.model.AssocRouteStation])

Return type:

Optional[float]

depot.api.private.consumption._build_trip_segments(trip, level_of_loading, t_amb)

Walk a trip’s stop times and route geometry to produce TripSegment objects.

Knots come from StopTime rows when present (otherwise from the trip’s departure/arrival), with synthetic knots inserted from Route.geom vertices whenever two consecutive stop-time knots are more than _VERTEX_GAP_THRESHOLD_M apart. A single ConsistencyWarning is emitted per trip when at least one knot has no Z. Ambient temperature is constant across all segments — sample it once at the trip midpoint upstream.

Parameters:

  • trip (eflips.model.Trip)

  • level_of_loading (Optional[float])

  • t_amb (Optional[float])

Return type:

List[TripSegment]

depot.api.private.consumption.extract_trip_information(trip_id, scenario, passenger_mass=68, passenger_count=17.6, *, temperatures=None, consumption_luts=None)

Build a ConsumptionInformation for a trip, decomposed into segments.

Segment knots come from the trip’s StopTime rows; route-vertex knots are inserted between stops more than 1 km apart so that intermediate elevation changes are captured. The returned object has already had ConsumptionInformation.calculate() called on it.

Parameters:

  • trip_id (int)

  • scenario (eflips.model.Scenario)

  • temperatures (Optional[eflips.model.Temperatures])

  • consumption_luts (Optional[Dict[int, eflips.model.ConsumptionLut]])

Return type:

ConsumptionInformation

depot.api.private.consumption.initialize_vehicle(rotation, session)

Create and add a new Vehicle object in the database for the given rotation.

This function:

  1. Creates a new Vehicle instance using the provided rotation’s vehicle type and scenario ID.

  2. Names it based on the rotation’s ID.

  3. Adds the vehicle to the specified SQLAlchemy session.

  4. Assigns the new vehicle to the rotation’s vehicle attribute.

Parameters:

  • rotation (eflips.model.Rotation) – A Rotation instance for which a new Vehicle should be created. The new vehicle will inherit its type and scenario from this rotation.

  • session (sqlalchemy.orm.session.Session) – An active SQLAlchemy Session used to persist the new vehicle to the database. The vehicle is added to the session but not committed here.

Returns:

None. Changes are made to the session but are not committed yet.

depot.api.private.consumption.add_initial_standby_event(vehicle, session)

Create and add a standby event immediately before the earliest trip of the given vehicle.

This function:

  1. Gathers all rotations assigned to the vehicle, sorted by their first trip’s departure time.

  2. Identifies the earliest trip across those rotations.

  3. Fetches an appropriate Area record from the database based on the vehicle’s scenario and vehicle type (for depot and subloc capacity).

  4. Constructs a dummy standby event starting one second before the earliest trip’s departure time, ending at the trip’s departure time, with 100% SoC.

  5. Adds the event to the session without committing (the caller is responsible for commits).

Parameters:

  • vehicle (eflips.model.Vehicle) – A Vehicle instance for which to add a new standby event. Must have associated rotations and trips.

  • session (sqlalchemy.orm.session.Session) – An active SQLAlchemy Session used to persist the new event to the database. The event is added to the session but not committed here.

Returns:

None. A new event is added to the session for the earliest trip, but changes are not yet committed.

Return type:

None

depot.api.private.consumption.find_charger_occupancy(station, time_start, time_end, session, resolution=timedelta(seconds=1))

Build a timeseries of charger occupancy at a station between two points in time.

For each discrete timestep between time_start and time_end (at the given resolution), this function calculates how many charging events (from the database) overlap with that time, thus producing a count of the active chargers at each timestep.

Parameters:

  • station (eflips.model.Station) – The Station whose charger occupancy is to be analyzed.

  • time_start (datetime.datetime) – The start time for the occupancy timeseries (inclusive).

  • time_end (datetime.datetime) – The end time for the occupancy timeseries (exclusive).

  • session (sqlalchemy.orm.session.Session) – An active SQLAlchemy Session used to query the database.

  • resolution – The timestep interval used to build the timeseries (default is 1 second). Note that using a very fine resolution over a large time range can produce large arrays.

Returns:

A tuple of two numpy arrays:

  1. times: The array of discrete timesteps (shape: (n,)).

  2. occupancy: The array of integer occupancy values for each timestep (shape: (n,)), indicating how many charging events are active.

Return type:

Tuple[numpy.ndarray, numpy.ndarray]

depot.api.private.consumption.find_best_timeslot(station, time_start, time_end, charging_duration, session, resolution=timedelta(seconds=1))
Parameters:

  • station (eflips.model.Station)

  • time_start (datetime.datetime)

  • time_end (datetime.datetime)

  • charging_duration (datetime.timedelta)

  • session (sqlalchemy.orm.session.Session)

  • resolution (datetime.timedelta)

Return type:

datetime.datetime

depot.api.private.consumption.attempt_opportunity_charging_event(previous_trip, next_trip, vehicle, charge_start_soc, terminus_deadtime, session)
Parameters:

  • previous_trip (eflips.model.Trip)

  • next_trip (eflips.model.Trip)

  • vehicle (eflips.model.Vehicle)

  • charge_start_soc (float)

  • terminus_deadtime (datetime.timedelta)

  • session (sqlalchemy.orm.session.Session)

Return type:

float