EV Compatibility Score - Charging at Anchor

Compatibility of internal combustion engine vehicle driving with electric vehicles based on simulated charging at an anchor location

As in: https://dimo.zone/news/dimo-insights-2, the frequency of charging away from a primary anchor location (where charging would likely occur) that an ICE vehicle owner's driving style.

The method primary_anchor (given in the Code Appendix) finds the location where the most trips have started and assumes that to be the location where the vehicle would be charging (and nowhere else). This process will essentially test if each instance of being away from the primary anchor location would have required charging away from home, had the vehicle been solely battery powered.

Using the code below, we can create a DataFrame with all instances where the vehicle was away from its primary anchor location for at least (note that when the vehicle stops at the primary anchor location for less than 30 minutes, it is assumed that the vehicle could not have charged).

anchor = primary_anchor(gdf)

gdf.loc[:,'lat_rounded'] = gdf.loc[:,'latitude'].round(decimals=2)
gdf.loc[:,'lng_rounded'] = gdf.loc[:,'longitude'].round(decimals=2)
gdf['rounded_point'] = gdf['lng_rounded'].astype(str) + '_' + gdf['lat_rounded'].astype(str)

gdf.loc[:,'lat_rounded'] = gdf.loc[:,'latitude'].round(decimals=3)
gdf.loc[:,'lng_rounded'] = gdf.loc[:,'longitude'].round(decimals=3)
gdf['geometry'] = gpd.GeoSeries.from_xy(x=gdf['lng_rounded'],y=gdf['lat_rounded'])
gdf['haversine_dist_shift'] = gdf['geometry'].to_crs('EPSG:6340').distance(gdf['geometry'].shift(1).to_crs('EPSG:6340')).fillna(0)


gdf['derived_speed'] = gdf['haversine_dist_shift']/gdf['timestamp_diff_second']
gdf['near_anchor'] = gdf['rounded_point'] == anchor
gdf['near_anchor_and_stopped'] = (gdf['rounded_point'] == anchor) &(gdf['derived_speed'] == 0 )

gdf['near_anchor_diff'] = gdf['near_anchor_and_stopped'].diff().fillna(method = 'backfill')
gdf['near_anchor_session_id'] = gdf['near_anchor_diff'].cumsum()


for g in gdf.groupby('near_anchor_session_id'):
    # If this is identified as an instance that the vehicle stopped at anchor:
    #print()
    if g[1]['near_anchor_and_stopped'].mean() == 1:
        # Make sure that the vehicle is stopped for at least 30 minutes. 
        # If stopped for less than 30 minutes, this is not "truly" a stop at anchor 
        # that one might charge, so push this instance into the next near_anchor_session_id
        if (g[1]['timestamp_diff_second'].sum())/60 <= 30:
            gdf.loc[g[1].index, 'near_anchor_session_id'] = g[1].iloc[0]['near_anchor_session_id'] + 1
            gdf.loc[g[1].index, 'near_anchor_and_stopped'] = False

gdf['near_anchor_diff'] = gdf['near_anchor_and_stopped'].diff().fillna(gdf['near_anchor_and_stopped'].iloc[0])
gdf['near_anchor_session_id'] = gdf['near_anchor_diff'].cumsum()



groups = gdf.groupby('near_anchor_session_id').agg(
            {'timestamp':[min,max],                  
            'haversine_dist_shift':np.sum,                            
            'timestamp_diff_second':np.sum,                         
            'near_anchor_and_stopped':np.mean})

groups.columns = [i + '_' + j for i,j in groups.columns.to_list()]
groups['haversine_dist_shift_sum_km'] = groups['haversine_dist_shift_sum']/1000
groups['range_change'] = -(groups['haversine_dist_shift_sum_km'])

groups['total_time_hour'] = groups['timestamp_diff_second_sum']/(60*60)

groups['added_range_L2'] = groups['near_anchor_and_stopped_mean']*(50*groups['total_time_hour'] + groups['range_change'])
groups['added_range_L1'] = groups['near_anchor_and_stopped_mean']*(8*groups['total_time_hour'] + groups['range_change'])

groups['starting_range_L2'] = 400
groups['ending_range_L2'] = 0

groups['starting_range_L1'] = 400
groups['ending_range_L1'] = 0

starting_range = 400
groups['insufficient_range_L2'] = False
for i in groups.index:
    if i % 2 == 0:
        groups.loc[i,'starting_range_L2'] = starting_range
        starting_range += groups.loc[i,'range_change']
        if starting_range <= 0:
            groups.loc[i,'insufficient_range_L2'] = True
            starting_range = 0
            
        groups.loc[i,'ending_range_L2'] = starting_range
    else:
        groups.loc[i,'starting_range_L2'] = starting_range
        starting_range = min( starting_range + groups.loc[i,'added_range_L2'] + groups.loc[i,'range_change'] ,400)
        groups.loc[i,'ending_range_L2'] = starting_range

starting_range = 400
groups['insufficient_range_L1'] = False
for i in groups.index:
    if i % 2 == 0:
        groups.loc[i,'starting_range_L1'] = starting_range
        starting_range += groups.loc[i,'range_change']
        if starting_range <= 0:
            groups.loc[i,'insufficient_range_L1'] = True
            starting_range = 0
            
        groups.loc[i,'ending_range_L1'] = starting_range
    else:
        groups.loc[i,'starting_range_L1'] = starting_range
        starting_range = min( starting_range + groups.loc[i,'added_range_L1'] + groups.loc[i,'range_change'] ,400)
        groups.loc[i,'ending_range_L1'] = starting_range

Code Appendix

def primary_anchor(gdf):    
    gdf.loc[:,'lat_rounded_2'] = gdf.loc[:,'latitude'].round(decimals=2)
    gdf.loc[:,'lng_rounded_2'] = gdf.loc[:,'longitude'].round(decimals=2)
    gdf['rounded_point'] = gdf['lng_rounded_2'].astype(str) + '_' + gdf['lat_rounded_2'].astype(str)
    gdf['rounded_pt_equal'] = gdf['rounded_point'] == gdf['rounded_point'].shift().fillna(0)

    
    # Create a trip identifier
    temp = gdf.loc[gdf['rounded_pt_equal'] == 1]
    temp.loc[:,'index_values'] = temp.index
    temp.loc[:,'new_stop'] = (temp['index_values'].diff() - 1).fillna(0)
    temp.loc[:,'trip_id'] = temp.loc[:,'new_stop'].cumsum()

    gb = temp.groupby('trip_id')

    trip_times = pd.DataFrame((gb['timestamp'].max() - gb['timestamp'].min())/np.timedelta64(1, 'm'))
    trip_times['first_index_value'] = gb['index_values'].first()

    try:
        # Find the most common gps location where trips 
        # start and end and call it anchor
        anchor = temp.merge(trip_times.loc[trip_times['timestamp'] > 60], left_on = 'trip_id', right_index = True).groupby('trip_id')['rounded_point'].first().value_counts().index[0]
    except:
        anchor = np.nan


    return anchor