EV Trip Efficiency

# Initialization code that runs before all other cells
# NB: This post started life as a marimo notebook.
# It was exported as a Jupyter notebook so it could be rendered
# for this blog post, using the Nikola static site generator.
import datetime
from pathlib import Path

import altair as alt
import duckdb
import marimo as mo
import polars as pl

Relation = duckdb.DuckDBPyRelation

here = Path.cwd()
data_dir = here.parent / "data"

leg_distance_csv_path = data_dir / "supercharger_distances.csv"


def load_leg_distances() -> Relation:
    """Get distances between supercharger locations."""
    # Because of my bad update logic, the leg distance CSV
    # may contain multiple instances of rows whose start_location
    # == end_location.  This clunky code is the only way I can
    # find to de-dup -- can't do it via Python API composition and SELECT DISTINCT.
    duckdb.sql(
        f"""
        CREATE TABLE IF NOT EXISTS leg_distances AS 
        FROM read_csv('{str(leg_distance_csv_path)}')
            GROUP BY ALL
            ORDER BY start_location, end_location
        """
    )
    return duckdb.table("leg_distances")

# Get the distances between fast chargers, as provided by Google Maps API:
charger_distances = load_leg_distances()

def load_tesla_charging_data() -> Relation:
    """
    Load charging data from (Tesla) CSV files into a DuckDB relation.

    The CSV files are as exported from the Tesla iOS app version 4.44.5:
    Hamburger menu -> Charging -> History -> (download icon, top right)
    """
    csv_wildcard = str(data_dir / "tesla" / "Tesla_Charging_History_*.csv")
    return duckdb.read_csv(
        csv_wildcard
    ).select(
        """
        CAST(ChargeStartDateTime AS TIMESTAMP) AS charge_date,
        SiteLocationName AS location,
        TRY_CAST (
            replace(
                replace(QuantityBase, ' kwh', ''),
                'N/A',
                ''
            ) AS REAL
        ) AS kwh,
        "Total Inc. VAT" AS cost,
        """
    ).order("charge_date ASC")

def leg_deltas(charging_data: Relation) -> Relation:
    """Compute leg delta info."""
    return charging_data.select(
        """
        charge_date AS charge_date,
        location AS location,
        kwh,
        LAG(charge_date)
            OVER (ORDER BY charge_date)
            AS prev_charge_date,
        LAG(location)
            OVER (ORDER BY charge_date)
            AS prev_location
        """
    ).order("charge_date")

def label_trip_bounds(deltas: Relation) -> Relation:
    return deltas.select(
        """
        (charge_date - prev_charge_date) AS time_gap,
        *
        """
    ).select(
        """
        (CASE WHEN (time_gap > INTERVAL 18 HOUR) OR (time_gap IS NULL)
         THEN 1 ELSE 0 END) AS is_trip_start,
        * EXCLUDE(time_gap)
        """
    )

def add_trip_ids(charging_data: Relation) -> Relation:
    """Associate every record in charging_data with a trip_id."""
    deltas = leg_deltas(charging_data)
    with_trip_markers = label_trip_bounds(deltas)

    return with_trip_markers.select(
        """
        SUM(is_trip_start) OVER (ORDER BY charge_date) AS trip_id,
        * EXCLUDE(is_trip_start)
        """
    ).order("trip_id, charge_date")

def add_leg_distances(with_trip_ids: Relation, distances: Relation) -> Relation:
    """Add leg distances to a leg-deltas relation."""
    trips = with_trip_ids.set_alias("trips")
    dists = distances.set_alias("dists")
    return trips.join(
        dists,
        """
        (trips.prev_location = dists.start_location
        AND trips.location = dists.end_location)
        """,
        how="left",
    ).select(
        """
        trip_id,
        charge_date,
        location, 
        IFNULL(miles, 0) AS miles, -- TODO remove this for consistency
        IF(prev_location IS NULL, 0.0, kwh) AS kwh,
        prev_location,
        CASE WHEN kwh IS NULL THEN 1 ELSE 0 END AS missing_kwh,
        CASE WHEN miles > 320 THEN 1 ELSE 0 END AS missing_stop
        """
    ).order("trip_id, charge_date")

def accumulate_trip_stats(with_leg_distances: Relation) -> Relation:
    """
    Calculate statistics for all of the trips in a table of trip legs.
    """
    return with_leg_distances.aggregate(
        """
        trip_id,
        FIRST(location ORDER BY charge_date) AS origin,
        LAST(location ORDER BY charge_date) AS destination,
        MIN(charge_date) AS start_date,
        MAX(charge_date) AS end_date,
        ROUND(SUM(miles)) AS miles,
        ROUND(SUM(kwh)) AS kwh,
        COUNT(*) AS num_waypoints,
        SUM(missing_kwh) AS missing_kwh,
        SUM(missing_stop) AS missing_stops
        """,
        group_expr="trip_id",
    ).select(
        """
        trip_id, origin, destination, miles, kwh,
        ROUND(miles / kwh, 1) AS efficiency,
        num_waypoints, missing_kwh + missing_stops AS num_anomalies,
        missing_kwh, missing_stops,
        start_date, end_date,
        (end_date - start_date) AS duration
        """
    ).order("trip_id, start_date")

def get_good_stats(charging_data: Relation, distances: Relation) -> Relation:
    return accumulate_trip_stats(
        add_leg_distances(add_trip_ids(charging_data), distances)
    ).filter("num_waypoints > 2 AND num_anomalies = 0")

def udf_exists(fn_name: str) -> bool:
    """Find out whether a user-defined function is already registered in duckdb.

    Parameters:
        fn_name (str): the name of the function

    Returns:  True or False, whether the function is registered
    """
    matches = duckdb.sql(
        f"""
         SELECT function_name FROM duckdb_functions()
         WHERE lower(function_name) = lower('{fn_name}')
           AND lower(function_type) = 'scalar'
        """
    ).fetchall()
    return len(matches) > 0

def annotate_with_tire_type(stats: Relation) -> Relation:
    """
    Add a tire_type column to Tesla stats.  This is based on emails and receipts
    from Discount Tire, telling when each set of tires was installed on the car.
    """

    # This could be done with another read_csv, but it gives the
    # opportunity to learn about Python-based user-defined functions
    # in duckdb.

    _date = datetime.datetime
    OEM = "OEM"
    NOKIAN = "Nokian"
    MICHELIN = "Michelin Defender"
    tire_changes = [
        (_date(2020, 9, 30), OEM),  # car purchase date
        (_date(2020, 12, 24), NOKIAN),
        (_date(2021, 3, 21), OEM),
        (_date(2023, 2, 14), NOKIAN),
        (_date(2023, 3, 11), MICHELIN),
        (_date(2024, 1, 2), NOKIAN),
        (_date(2024, 3, 27), MICHELIN),
        (_date(2025, 1, 14), NOKIAN),
        (_date(2025, 3, 17), MICHELIN),
    ]

    def tire_type_for_date(ref_date: datetime.datetime) -> str:
        # Brute-force, linear, backward search for the loss:
        for date, tire in reversed(tire_changes):
            if ref_date >= date:
                return tire
        return OEM

    fn_name = "tire_type_for_date"
    if not udf_exists(fn_name):
        duckdb.create_function(fn_name, tire_type_for_date)
    return stats.select("*, tire_type_for_date(start_date) AS tire_type")

def overall_efficiency(good_stats: Relation) -> Relation:
    """Aggregate efficiency info from good_stats."""
    return good_stats.aggregate(
        """
        round(mean(efficiency), 1) AS "Mean miles/kWh",
        round(stddev_pop(efficiency), 1) AS "σ^2(miles/kWh)",
        min(efficiency) AS "Min miles/kWh",
        max(efficiency) AS "Max miles/kWh",
        """
    )

good_tesla_stats = annotate_with_tire_type(
    get_good_stats(
        load_tesla_charging_data(),
        charger_distances
    )
).select(
    """
    origin, destination,
    miles, efficiency,
    tire_type,
    start_date, duration
    """
).order("start_date")

tesla_overall_efficiency = overall_efficiency(good_tesla_stats)

good_tesla_stats.show(max_width=128)

gts_df = pl.DataFrame(good_tesla_stats.df())[["efficiency", "tire_type"]]

def efficiency_hist(df):
    bins = alt.Bin(extent=[0, 5], step=0.25)
    return (
        alt.Chart(df)
        .mark_bar()
        .encode(
            x=alt.X("efficiency:Q", bin=bins),
            y=alt.Y("count()", scale=alt.Scale(domain=[0, 5])),
            tooltip=[
                alt.Tooltip("efficiency:Q", bin=bins, title="Bin"),
                alt.Tooltip("count()", title="Count"),
            ],
        )
        .properties(width=300, height=150)
    )

# I don't know why, but "SELECT DISTINCT" in duckdb always
# returns duplicates.  Hence the use of `set`.
tire_types = set(
    [row[0] for row in good_tesla_stats.select("tire_type").fetchall()]
)

charts = [
    efficiency_hist(gts_df).properties(title="Overall")
]
for tire_type in tire_types:
    charts.append(
        efficiency_hist(
            gts_df.filter(pl.col("tire_type") == tire_type)
        ).properties(title=tire_type)
    )
alt.vconcat(*charts)

tesla_overall_efficiency

tesla_eff_stats_by_tire_type = """
tire_type AS "Tire Type",
round(mean(efficiency), 1) AS "Mean miles/kWh",
round(stddev_pop(efficiency), 1) AS "σ^2(miles/kWh)"
"""

good_tesla_stats.aggregate(tesla_eff_stats_by_tire_type).order(
    '"Mean miles/kWh" DESC'
)

def parse_date_str(ds: str) -> datetime.datetime:
    """
    Timestamps in Electrify America CSV files have their own format.
    Try to handle those formats.
    """
    valid_formats = ["%Y/%m/%d %H:%M:%S %Z", "%m/%d/%y, %I:%M:%S %p %Z"]
    for format in valid_formats:
        try:
            return datetime.datetime.strptime(ds, format)
        except Exception:
            pass
    else:
        raise ValueError(f"Could not parse time value '{ds}'")

def read_vw_charging_data(csv_path: Path) -> pl.DataFrame:
    """
    Read VW (Electrify America-ish) charging data from a single file.
    """
    return pl.read_csv(
        csv_path, use_pyarrow=True, schema_overrides={" TotalCost": pl.String}
    ).with_columns(
        [
            pl.col("Date")
            .map_elements(lambda x: parse_date_str(x), return_dtype=pl.Datetime)
            .alias("charge_date"),
            pl.col(" LocationAddress").alias("location"),
            pl.col(" TotalEnergyDelivered")
            .str.replace(" kWh", "")
            .cast(pl.Float64)
            .alias("kwh"),
            pl.col(" TotalCost")
            .str.replace("$", "", literal=True)
            .cast(pl.Float64)
            .alias("cost"),
        ]
    )[["charge_date", "location", "kwh", "cost"]]

def load_vw_charging_data() -> Relation:
    """
    Load charging data from Electrify America and other vendor CSV files into a
    DuckDB relation.

    The CSV files are as exported from the
    Electrify America iOS app version 7.23.0:
    Account -> Charge History -> (Share icon, top right)

    Some other charging providers, e.g., EVgo, don't provide for exporting/
    sharing data.  Data has been extracted from those apps manually.
    """
    vw_data_dir = data_dir / "volkswagen_id_4"
    csv_wild = "*.csv"
    vw_data_paths = vw_data_dir.glob(csv_wild)
    vw_dfs = [read_vw_charging_data(pathname) for pathname in vw_data_paths]
    df = pl.concat(vw_dfs)
    return duckdb.from_arrow(df).order("charge_date")

good_id4_stats = get_good_stats(
    load_vw_charging_data(),
    charger_distances
).select(
    """
    origin, destination,
    miles, efficiency,
    start_date, duration
    """
).order("start_date")

id4_overall_efficiency = overall_efficiency(good_id4_stats).set_alias("id4_overall_efficiency")

good_id4_stats.show(max_width=128)

id4_overall_efficiency

efficiency_hist(good_id4_stats).properties(title="VW ID.4")

duckdb.sql(
    """
    SELECT 'Tesla Model 3 LR AWD' AS Vehicle, * from tesla_overall_efficiency
    UNION ALL
    SELECT 'Volkswagen ID.4 Pro AWD' AS Vehicle, * from id4_overall_efficiency
    """
)