Energy class for inventorying multiple energy models

flow/core/kernel/vehicle/base.py

@@ -363,6 +363,22 @@ def get_fuel_consumption(self, veh_id, error=-1001):
         """
         pass
 
+    @abstractmethod
+    def get_energy_model(self, veh_id, error=""):
+        """Return the energy model class object of the specified vehicle.
+
+        Parameters
+        ----------
+        veh_id : str or list of str
+            vehicle id, or list of vehicle ids
+        error : str
+            value that is returned if the vehicle is not found
+        Returns
+        -------
+        subclass of BaseEnergyModel
+        """
+        pass
+
     @abstractmethod
     def get_speed(self, veh_id, error=-1001):
         """Return the speed of the specified vehicle.

flow/core/kernel/vehicle/traci.py

@@ -292,6 +292,12 @@ def _add_departed(self, veh_id, veh_type):
         # specify the type
         self.__vehicles[veh_id]["type"] = veh_type
 
+        # specify energy model
+        energy_model = \
+            self.type_parameters[veh_type]["energy_model"]
+        self.__vehicles[veh_id]["energy_model"] = \
+            energy_model[0](veh_id, **energy_model[1])
+
         car_following_params = \
             self.type_parameters[veh_type]["car_following_params"]
 
@@ -549,6 +555,16 @@ def get_fuel_consumption(self, veh_id, error=-1001):
             return [self.get_fuel_consumption(vehID, error) for vehID in veh_id]
         return self.__sumo_obs.get(veh_id, {}).get(tc.VAR_FUELCONSUMPTION, error) * ml_to_gallons
 
+    def get_energy_model(self, veh_id, error=""):
+        """See parent class."""
+        if isinstance(veh_id, (list, np.ndarray)):
+            return [self.get_energy_model(vehID) for vehID in veh_id]
+        try:
+            return self.__vehicles.get(veh_id, {'energy_model': error})['energy_model']
+        except KeyError:
+            print("Energy model not specified for vehicle {}".format(veh_id))
+            raise
+
     def get_previous_speed(self, veh_id, error=-1001):
         """See parent class."""
         if isinstance(veh_id, (list, np.ndarray)):

flow/core/params.py

@@ -7,6 +7,10 @@
 from flow.controllers.car_following_models import SimCarFollowingController
 from flow.controllers.rlcontroller import RLController
 from flow.controllers.lane_change_controllers import SimLaneChangeController
+from flow.energy_models.toyota_energy import PriusEnergy
+from flow.energy_models.toyota_energy import TacomaEnergy
+from flow.energy_models.power_demand import PDMCombustionEngine
+from flow.energy_models.power_demand import PDMElectric
 
 
 SPEED_MODES = {
@@ -39,6 +43,9 @@
     "only_right_drive_safe": 576
 }
 
+ENERGY_MODELS = set([PriusEnergy, TacomaEnergy, PDMCombustionEngine, PDMElectric])
+DEFAULT_ENERGY_MODEL = PDMCombustionEngine
+
 # Traffic light defaults
 PROGRAM_ID = 1
 MAX_GAP = 3.0
@@ -262,6 +269,7 @@ def add(self,
             num_vehicles=0,
             car_following_params=None,
             lane_change_params=None,
+            energy_model=DEFAULT_ENERGY_MODEL,
             color=None):
         """Add a sequence of vehicles to the list of vehicles in the network.
 
@@ -298,6 +306,12 @@ def add(self,
             # FIXME: depends on simulator
             lane_change_params = SumoLaneChangeParams()
 
+        if energy_model not in ENERGY_MODELS:
+            print('{} for vehicle {} is not a valid energy model. Defaulting to {}\n'.format(energy_model,
+                                                                                              veh_id,
+                                                                                              DEFAULT_ENERGY_MODEL))
+            energy_model = DEFAULT_ENERGY_MODEL
+
         type_params = {}
         type_params.update(car_following_params.controller_params)
         type_params.update(lane_change_params.controller_params)
@@ -311,7 +325,8 @@ def add(self,
              "routing_controller": routing_controller,
              "initial_speed": initial_speed,
              "car_following_params": car_following_params,
-             "lane_change_params": lane_change_params}
+             "lane_change_params": lane_change_params,
+             "energy_model": energy_model}
 
         if color:
             type_params['color'] = color
@@ -334,7 +349,9 @@ def add(self,
             "car_following_params":
                 car_following_params,
             "lane_change_params":
-                lane_change_params
+                lane_change_params,
+            "energy_model":
+                energy_model
         })
 
         # This is used to return the actual headways from the vehicles class.

flow/core/rewards.py

@@ -306,139 +306,40 @@ def punish_rl_lane_changes(env, penalty=1):
     return total_lane_change_penalty
 
 
-def energy_consumption(env, gain=.001):
-    """Calculate power consumption of a vehicle.
-
-    Assumes vehicle is an average sized vehicle.
-    The power calculated here is the lower bound of the actual power consumed
-    by a vehicle.
-    """
-    power = 0
-
-    M = 1200  # mass of average sized vehicle (kg)
-    g = 9.81  # gravitational acceleration (m/s^2)
-    Cr = 0.005  # rolling resistance coefficient
-    Ca = 0.3  # aerodynamic drag coefficient
-    rho = 1.225  # air density (kg/m^3)
-    A = 2.6  # vehicle cross sectional area (m^2)
-    for veh_id in env.k.vehicle.get_ids():
-        if veh_id not in env.k.vehicle.previous_speeds.keys():
-            continue
-        speed = env.k.vehicle.get_speed(veh_id)
-        prev_speed = env.k.vehicle.get_previous_speed(veh_id)
-
-        accel = abs(speed - prev_speed) / env.sim_step
-
-        power += M * speed * accel + M * g * Cr * speed + 0.5 * rho * A * Ca * speed ** 3
-
-    return -gain * power
-
-
-def veh_energy_consumption(env, veh_id, gain=.001):
-    """Calculate power consumption of a vehicle.
-
-    Assumes vehicle is an average sized vehicle.
-    The power calculated here is the lower bound of the actual power consumed
-    by a vehicle.
-    """
-    power = 0
-
-    M = 1200  # mass of average sized vehicle (kg)
-    g = 9.81  # gravitational acceleration (m/s^2)
-    Cr = 0.005  # rolling resistance coefficient
-    Ca = 0.3  # aerodynamic drag coefficient
-    rho = 1.225  # air density (kg/m^3)
-    A = 2.6  # vehicle cross sectional area (m^2)
-
-    if veh_id not in env.k.vehicle.previous_speeds:
-        return 0
-
-    speed = env.k.vehicle.get_speed(veh_id)
-    prev_speed = env.k.vehicle.get_previous_speed(veh_id)
-
-    accel = abs(speed - prev_speed) / env.sim_step
-
-    power += M * speed * accel + M * g * Cr * speed + 0.5 * rho * A * Ca * speed ** 3
-
-    return -gain * power
-
-
-def miles_per_megajoule(env, veh_ids=None, gain=.001):
-    """Calculate miles per mega-joule of either a particular vehicle or the total average of all the vehicles.
-
-    Assumes vehicle is an average sized vehicle.
-    The power calculated here is the lower bound of the actual power consumed
-    by a vehicle.
+def instantaneous_mpg(env, veh_ids=None, gain=.001):
+    """Calculate the instantaneous mpg for every simulation step specific to the vehicle type.
 
     Parameters
     ----------
     env : flow.envs.Env
         the environment variable, which contains information on the current
         state of the system.
-    veh_ids : [list]
-        list of veh_ids to compute the reward over
+    veh_ids : [list] or str
+        list of veh_ids or single veh_id to compute the reward over
     gain : float
         scaling factor for the reward
     """
-    mpj = 0
-    counter = 0
     if veh_ids is None:
         veh_ids = env.k.vehicle.get_ids()
     elif not isinstance(veh_ids, list):
         veh_ids = [veh_ids]
-    for veh_id in veh_ids:
-        speed = env.k.vehicle.get_speed(veh_id)
-        # convert to be positive since the function called is a penalty
-        power = -veh_energy_consumption(env, veh_id, gain=1.0)
-        if power > 0 and speed >= 0.1:
-            counter += 1
-            # meters / joule is (v * \delta t) / (power * \delta t)
-            mpj += speed / power
-    if counter > 0:
-        mpj /= counter
-
-    # convert from meters per joule to miles per joule
-    mpj /= 1609.0
-    # convert from miles per joule to miles per megajoule
-    mpj *= 10 ** 6
-
-    return mpj * gain
-
 
-def miles_per_gallon(env, veh_ids=None, gain=.001):
-    """Calculate mpg of either a particular vehicle or the total average of all the vehicles.
-
-    Assumes vehicle is an average sized vehicle.
-    The power calculated here is the lower bound of the actual power consumed
-    by a vehicle.
-
-    Parameters
-    ----------
-    env : flow.envs.Env
-        the environment variable, which contains information on the current
-        state of the system.
-    veh_ids : [list]
-        list of veh_ids to compute the reward over
-    gain : float
-        scaling factor for the reward
-    """
-    mpg = 0
-    counter = 0
-    if veh_ids is None:
-        veh_ids = env.k.vehicle.get_ids()
-    elif not isinstance(veh_ids, list):
-        veh_ids = [veh_ids]
+    cumulative_gallons = 0
+    cumulative_distance = 0
     for veh_id in veh_ids:
-        speed = env.k.vehicle.get_speed(veh_id)
-        gallons_per_s = env.k.vehicle.get_fuel_consumption(veh_id)
-        if gallons_per_s > 0 and speed >= 0.0:
-            counter += 1
-            # meters / gallon is (v * \delta t) / (gallons_per_s * \delta t)
-            mpg += speed / gallons_per_s
-    if counter > 0:
-        mpg /= counter
-
-    # convert from meters per gallon to miles per gallon
-    mpg /= 1609.0
+        energy_model = env.k.vehicle.get_energy_model(veh_id)
+        if energy_model != "":
+            speed = env.k.vehicle.get_speed(veh_id)
+            accel = env.k.vehicle.get_accel_no_noise_with_failsafe(veh_id)
+            grade = env.k.vehicle.get_road_grade(veh_id)
+            gallons_per_hr = energy_model.get_instantaneous_fuel_consumption(accel, speed, grade)
+            if gallons_per_hr > 0 and speed >= 0.0:
+                cumulative_gallons += gallons_per_hr
+                cumulative_distance += speed
+
+    cumulative_gallons /= 3600.0
+    cumulative_distance /= 1609.0
+    # miles / gallon is (distance_dot * \delta t) / (gallons_dot * \delta t)
+    mpg = cumulative_distance / cumulative_gallons
 
     return mpg * gain

flow/energy_models/base_energy.py

@@ -0,0 +1,59 @@
+"""Script containing the base vehicle energy class."""
+from abc import ABCMeta, abstractmethod
+
+
+class BaseEnergyModel(metaclass=ABCMeta):
+    """Base energy model class.
+
+    Calculate the instantaneous power consumption of a vehicle in
+    the network.  It returns the power in Watts regardless of the
+    vehicle type: whether EV or Combustion Engine, Toyota Prius or Tacoma
+    or non-Toyota vehicles. Non-Toyota vehicles are set by default
+    to be an averaged-size vehicle.
+    """
+
+    def __init__(self, kernel):
+        self.k = kernel
+
+        # 15 kilowatts = 1 gallon/hour conversion factor
+        self.conversion = 15e3
+
+    @abstractmethod
+    def get_instantaneous_power(self, accel, speed, grade):
+        """Calculate the instantaneous power consumption of a vehicle.
+
+        Must be implemented by child classes.
+
+        Parameters
+        ----------
+        accel : float
+            Instantaneous acceleration of the vehicle
+        speed : float
+            Instantaneous speed of the vehicle
+        grade : float
+            Instantaneous road grade of the vehicle
+        Returns
+        -------
+        float
+        """
+        pass
+
+    def get_instantaneous_fuel_consumption(self, accel, speed, grade):
+        """Calculate the instantaneous fuel consumption of a vehicle.
+
+        Fuel consumption is reported in gallons per hour, with the conversion
+        rate of 15kW = 1 gallon/hour.
+
+        Parameters
+        ----------
+        accel : float
+            Instantaneous acceleration of the vehicle
+        speed : float
+            Instantaneous speed of the vehicle
+        grade : float
+            Instantaneous road grade of the vehicle
+        Returns
+        -------
+        float
+        """
+        return self.get_instantaneous_power(accel, speed, grade) * self.conversion