depot.api.private.consumption ============================= .. py:module:: depot.api.private.consumption Attributes ---------- .. autoapisummary:: depot.api.private.consumption._interpolator_cache depot.api.private.consumption._nn_interpolator_cache depot.api.private.consumption._nn_fallback_warned depot.api.private.consumption._LUT_DIM_NAMES depot.api.private.consumption._EARTH_RADIUS_M depot.api.private.consumption._VERTEX_GAP_THRESHOLD_M Classes ------- .. autoapisummary:: depot.api.private.consumption.ConsumptionResult depot.api.private.consumption.TripSegment depot.api.private.consumption.ConsumptionInformation Functions --------- .. autoapisummary:: depot.api.private.consumption.clear_interpolator_cache depot.api.private.consumption._classify_nan_query depot.api.private.consumption._get_or_build_interpolator depot.api.private.consumption._get_or_build_nearest_neighbor_interpolator depot.api.private.consumption._haversine_cumulative depot.api.private.consumption._route_geom_knots depot.api.private.consumption._station_z depot.api.private.consumption._assoc_z depot.api.private.consumption._build_trip_segments depot.api.private.consumption.extract_trip_information depot.api.private.consumption.initialize_vehicle depot.api.private.consumption.add_initial_standby_event depot.api.private.consumption.find_charger_occupancy depot.api.private.consumption.find_best_timeslot depot.api.private.consumption.attempt_opportunity_charging_event Module Contents --------------- .. py:data:: _interpolator_cache :type: Dict[int, Dict[str, Any]] .. py:data:: _nn_interpolator_cache :type: Dict[int, scipy.interpolate.NearestNDInterpolator] .. py:data:: _nn_fallback_warned :type: set .. py:function:: clear_interpolator_cache() Clear the module-level interpolator cache. .. py:data:: _LUT_DIM_NAMES :type: Tuple[str, str, str, str] :value: ('incline', 't_amb', 'level_of_loading', 'mean_speed_kmh') .. py:function:: _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. .. py:function:: _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'. .. py:function:: _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. .. py:class:: 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. :param 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%). :param 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``. :param 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. .. py:attribute:: delta_soc_total :type: float .. py:attribute:: timestamps :type: List[datetime.datetime] | None .. py:attribute:: delta_soc :type: List[float] | None .. py:class:: TripSegment A piece of a trip between two adjacent knots — either consecutive. :class:`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. .. py:attribute:: distance_m :type: float 2D ground distance in meters. .. py:attribute:: duration_s :type: float Duration of this segment in seconds. .. py:attribute:: mean_speed_kmh :type: float Mean speed in km/h. .. py:attribute:: incline :type: float Signed Δz / distance_m. 0.0 if either knot lacks an elevation. .. py:attribute:: level_of_loading :type: Optional[float] Vehicle loading as a fraction of max payload, or ``None`` when no LUT is in use. .. py:attribute:: t_amb :type: Optional[float] Ambient temperature in °C at the segment midpoint, or ``None`` when no temperature data is available. .. py:attribute:: end_time :type: datetime.datetime Absolute timestamp at the end of this segment. .. py:attribute:: consumption_kwh :type: Optional[float] :value: None Energy used on this segment in kWh; populated by :meth:`ConsumptionInformation.calculate`. .. py:class:: ConsumptionInformation Per-trip consumption inputs decomposed into route-aware segments. Either ``consumption_lut`` or ``flat_consumption_per_km`` must be set. Calling :meth:`calculate` populates ``segment.consumption_kwh`` for every segment; :meth:`generate_consumption_result` then turns that into a :class:`ConsumptionResult` with cumulative SoC. .. py:attribute:: trip_id :type: int .. py:attribute:: segments :type: List[TripSegment] .. py:attribute:: consumption_lut :type: Optional[eflips.model.ConsumptionLut] :value: None .. py:attribute:: flat_consumption_per_km :type: Optional[float] :value: None .. py:attribute:: line_name :type: Optional[str] :value: None Human-readable line name, used purely for diagnostic warnings. .. py:attribute:: route_name :type: Optional[str] :value: None Human-readable route name, used purely for diagnostic warnings. .. py:attribute:: trip_departure :type: Optional[datetime.datetime] :value: None Trip departure time, used purely for diagnostic warnings. .. py:attribute:: trip_arrival :type: Optional[datetime.datetime] :value: None Trip arrival time, used purely for diagnostic warnings. .. py:method:: calculate() Compute energy consumption for every segment. - When ``consumption_lut`` is set, the LUT is evaluated once for all segments via a vectorized :class:`RegularGridInterpolator` call. Any segment whose result is NaN is filled in via a :class:`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. .. py:method:: _warn_nn_fallback(points, nan_mask, cached, lut_id) Emit a deduplicated :class:`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. .. py:method:: generate_consumption_result(battery_capacity) Build a :class:`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. .. py:data:: _EARTH_RADIUS_M :value: 6371008.8 .. py:data:: _VERTEX_GAP_THRESHOLD_M :value: 1000.0 .. py:function:: _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 — :func:`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. .. py:function:: _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). .. py:function:: _station_z(station) Return the Z coordinate of a station's geom, or ``None`` if absent. .. py:function:: _assoc_z(assoc) Return the Z coordinate of an AssocRouteStation.location, or ``None`` if absent. .. py:function:: _build_trip_segments(trip, level_of_loading, t_amb) Walk a trip's stop times and route geometry to produce :class:`TripSegment` objects. Knots come from :class:`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 :class:`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. .. py:function:: extract_trip_information(trip_id, scenario, passenger_mass=68, passenger_count=17.6, *, temperatures = None, consumption_luts = None) Build a :class:`ConsumptionInformation` for a trip, decomposed into segments. Segment knots come from the trip's :class:`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 :meth:`ConsumptionInformation.calculate` called on it. .. py:function:: 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. :param rotation: A :class:`Rotation` instance for which a new ``Vehicle`` should be created. The new vehicle will inherit its type and scenario from this rotation. :param session: An active SQLAlchemy :class:`Session` used to persist the new vehicle to the database. The vehicle is added to the session but not committed here. :return: ``None``. Changes are made to the session but are not committed yet. .. py:function:: 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 :class:`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). :param vehicle: A :class:`Vehicle` instance for which to add a new standby event. Must have associated rotations and trips. :param session: An active SQLAlchemy :class:`Session` used to persist the new event to the database. The event is added to the session but not committed here. :return: ``None``. A new event is added to the session for the earliest trip, but changes are not yet committed. .. py:function:: 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. :param station: The :class:`Station` whose charger occupancy is to be analyzed. :param time_start: The start time for the occupancy timeseries (inclusive). :param time_end: The end time for the occupancy timeseries (exclusive). :param session: An active SQLAlchemy :class:`Session` used to query the database. :param 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. .. py:function:: find_best_timeslot(station, time_start, time_end, charging_duration, session, resolution = timedelta(seconds=1)) .. py:function:: attempt_opportunity_charging_event(previous_trip, next_trip, vehicle, charge_start_soc, terminus_deadtime, session)