Survey Assist Using RF

conf/analysis/trip_model.conf.json.sample

@@ -1,5 +1,5 @@
 {
-    "model_type": "greedy",
+    "model_type": "forest",
     "model_storage": "document_database",
     "minimum_trips": 14,
     "model_parameters": {
@@ -8,6 +8,24 @@
             "similarity_threshold_meters": 500,
             "apply_cutoff": false,
             "incremental_evaluation": false
+        },
+        "forest": {
+            "loc_feature" : "coordinates",
+            "radius": 100,
+            "size_thresh":1,
+            "purity_thresh":1.0,
+            "gamma":0.05,
+            "C":1,
+            "n_estimators":100,
+            "criterion":"gini",
+            "max_depth":null,
+            "min_samples_split":2,
+            "min_samples_leaf":1,
+            "max_features":"sqrt",
+            "bootstrap":true,
+            "random_state":42,
+            "use_start_clusters":false,
+            "use_trip_clusters":true            
         }
     }
 }

emission/analysis/modelling/trip_model/forest_classifier.py

@@ -0,0 +1,197 @@
+import joblib
+from typing import Dict, List, Optional, Tuple
+import sklearn.metrics.pairwise as smp 
+import emission.core.wrapper.confirmedtrip as ecwc
+import logging
+from io import BytesIO
+
+import json
+import emission.analysis.modelling.trip_model.trip_model as eamuu
+import emission.analysis.modelling.trip_model.config as eamtc
+import emission.storage.timeseries.builtin_timeseries as estb
+import emission.storage.decorations.trip_queries as esdtq
+import emission.analysis.modelling.trip_model.models as eamtm
+
+EARTH_RADIUS = 6371000
+
+class ForestClassifierModel(eamuu.TripModel):
+
+    def __init__(self,config=None):
+
+        if config is None:
+            config = eamtc.get_config_value_or_raise('model_parameters.forest')
+            logging.debug(f'ForestClassifier loaded model config from file')
+        else:
+            logging.debug(f'ForestClassifier using model config argument')
+
+        random_forest_expected_keys = [
+            'loc_feature',
+            'n_estimators',
+            'criterion',
+            'min_samples_split',
+            'min_samples_leaf',
+            'max_features',
+            'bootstrap',
+        ]     
+        ######### Not Tested #########
+        # The below code is used when we cluster the coordinates (loc_cluster parameter = True)
+        # before passing to Random Forest. Commenting this for now since it is not used. Not tested either.
+        ###############################
+
+        # cluster_expected_keys= [
+        #     'radius',
+        #     'size_thresh',  
+        #     'purity_thresh',
+        #     'gamma',
+        #     'C',
+        #     'use_start_clusters',
+        #     'use_trip_clusters',
+        # ]
+        #
+        # if config['loc_feature'] == 'cluster':
+        #     for k in cluster_expected_keys:
+        #         if config.get(k) is None:
+        #             msg = f"cluster trip model config missing expected key {k}"
+        #             raise KeyError(msg)
+        #######################################
+        for k in random_forest_expected_keys:
+            if config.get(k) is None:
+                msg = f"forest trip model config missing expected key {k}"
+                raise KeyError(msg)   
+        self.model=eamtm.ForestClassifier(**config)
+
+
+    def fit(self,trips: List[ecwc.Confirmedtrip]):
+        '''
+        trips : List of Entry type data 
+        '''
+        # check and raise exception if no data to fit
+        logging.debug(f'fit called with {len(trips)} trips')
+
+        unlabeled = list(filter(lambda t: len(t['data']['user_input']) == 0, trips))
+        if len(unlabeled) > 0:
+            msg = f'model.fit cannot be called with unlabeled trips, found {len(unlabeled)}'
+            raise Exception(msg)    
+
+        #Convert List of Entry to dataframe     
+        data_df = estb.BuiltinTimeSeries.to_data_df("analysis/confirmed_trip",trips)
+        labeled_trip_df = esdtq.filter_labeled_trips(data_df)
+        expanded_labeled_trip_df= esdtq.expand_userinputs(labeled_trip_df)
+        #fit models on dataframe
+        self.model.fit(expanded_labeled_trip_df)       
+
+
+    def predict(self, trip: List[float]) -> Tuple[List[Dict], int]:
+        '''
+        trip : A single trip whose mode, pupose and replaced mode are required
+        returns.
+        '''
+
+        #check if theres no trip to predict        
+        logging.debug(f"forest classifier predict called with {len(trip)} trips")
+        if len(trip) == 0:
+            msg = f'model.predict cannot be called with an empty trip'
+            raise Exception(msg)        
+        # CONVERT TRIP TO dataFrame        
+        test_df = estb.BuiltinTimeSeries.to_data_df("analysis/confirmed_trip",[trip])
+        predcitions_df= self.model.predict(test_df)
+
+        # the predictions_df currently holds the highest probable options
+        # individually in all three categories. the predictions_df are in the form 
+        #
+        # purpose_pred | purpose_proba | mode_pred | mode_proba | replaced_pred | replaced proba 
+        # dog-park     |   1.0         |  e-bike   | 0.99       | walk          | 1.1 
+        #
+        #
+        # However, to keep the trip model general, the forest model is expected to return  
+        #
+        #PREDICTIONS [  {'labels': {'mode_confirm': 'e-bike', 'replaced_mode': 'walk', 'purpose_confirm': 'dog-park'},
+        #                 'p':  ( Currently average of the 3 probabilities)}]
+        labels= {
+            'mode_confirm': predcitions_df['mode_pred'].iloc[0],
+            'replaced_mode' : predcitions_df['replaced_pred'].iloc[0],
+            'purpose_confirm' : predcitions_df['purpose_pred'].iloc[0]
+            }
+
+        avg_proba = predcitions_df[['purpose_proba','mode_proba','replaced_proba']].mean(axis=1).iloc[0]
+        predictions =[{
+            'labels' : labels,
+            'p' : avg_proba
+        }]
+        return predictions, len(predictions)
+
+    def to_dict(self):
+        """
+        Convert the model to a dictionary suitable for storage.
+        """
+        data={}
+        attr=[ 'purpose_predictor','mode_predictor','replaced_predictor','purpose_enc','mode_enc','train_df']
+
+        ######### Not Tested #########
+        # The below code is used when we cluster the coordinates (loc_cluster parameter = True)
+        # before passing to Random Forest. Commenting this for now since it is not used. Not tested either.
+        ###############################
+        # if self.model.loc_feature == 'cluster':
+        #     ## confirm this includes all the extra encoders/models
+        #     attr.extend([ 'cluster_enc','end_cluster_model','start_cluster_model','trip_grouper'])
+
+        for attribute_name in attr:
+            if not hasattr(self.model,attribute_name):
+                raise ValueError(f"Attribute {attribute_name} not found in the model")
+
+            buffer=BytesIO()
+            try:
+                joblib.dump(getattr(self.model,attribute_name),buffer)
+            except Exception as e:
+                raise RuntimeError(f"Error serializing { attribute_name}: {str(e)}")    
+            buffer.seek(0)
+            data[attribute_name]=buffer.getvalue()
+
+        return data
+
+    def from_dict(self,model: Dict):
+        """
+        Load the model from a dictionary.
+        """
+        attr=[ 'purpose_predictor','mode_predictor','replaced_predictor','purpose_enc','mode_enc','train_df']
+
+        ######### Not Tested #########
+        # The below code is used when we cluster the coordinates (loc_cluster parameter = True)
+        # before passing to Random Forest. Commenting this for now since it is not used. Not tested either.
+        ###############################
+        # if self.model.loc_feature == 'cluster':
+        #     ## TODO : confirm this includes all the extra encoders/models
+        #     attr.extend([ 'cluster_enc','end_cluster_model','start_cluster_model','trip_grouper'])
+        for attribute_name in attr:
+            if attribute_name not in model:
+                raise ValueError(f"Attribute {attribute_name} missing in the model")
+            try:
+                buffer = BytesIO(model[attribute_name])
+                setattr(self.model,attribute_name, joblib.load(buffer))
+            except Exception as e:
+                raise RuntimeError(f"Error deserializing { attribute_name}: {str(e)}")    
+                # If we do not wish to raise the exception after logging the error, comment the line above
+
+    def extract_features(self, trip: ecwc.Confirmedtrip) -> List[float]:
+        """
+        extract the relevant features for learning from a trip for this model instance
+
+        :param trip: the trip to extract features from
+        :type trip: Confirmedtrip
+        :return: a vector containing features to predict from
+        :rtype: List[float]
+        """
+        # ForestClassifier class in models.py file handles features extraction.
+        pass
+
+    def is_incremental(self) -> bool:
+        """
+        whether this model requires the complete user history to build (False),
+        or, if only the incremental data since last execution is required (True).
+
+        :return: if the model is incremental. the current timestamp will be recorded
+        in the analysis pipeline. the next call to this model will only include 
+        trip data for trips later than the recorded timestamp.
+        :rtype: bool
+        """
+        pass

emission/analysis/modelling/trip_model/model_type.py

@@ -3,6 +3,7 @@
 import emission.analysis.modelling.trip_model.trip_model as eamuu
 import emission.analysis.modelling.similarity.od_similarity as eamso
 import emission.analysis.modelling.trip_model.greedy_similarity_binning as eamug
+import emission.analysis.modelling.trip_model.forest_classifier as eamuf
 
 
 SIMILARITY_THRESHOLD_METERS=500
@@ -11,6 +12,7 @@
 class ModelType(Enum):
     # ENUM_NAME_CAPS = 'SHORTHAND_NAME_CAPS'
     GREEDY_SIMILARITY_BINNING = 'GREEDY'
+    RANDOM_FOREST_CLASSIFIER = 'FOREST'
 
     def build(self, config=None) -> eamuu.TripModel:
         """
@@ -24,16 +26,16 @@ def build(self, config=None) -> eamuu.TripModel:
         :raises KeyError: if the requested model name does not exist
         """
         # Dict[ModelType, TripModel]
-        MODELS = {
-                ModelType.GREEDY_SIMILARITY_BINNING: eamug.GreedySimilarityBinning(config)
-        }
+        MODELS = {                
+                ModelType.GREEDY_SIMILARITY_BINNING: eamug.GreedySimilarityBinning,
+                ModelType.RANDOM_FOREST_CLASSIFIER: eamuf.ForestClassifierModel
+                }    
         model = MODELS.get(self)
         if model is None:
-            model_names = list(lambda e: e.name, MODELS.keys())
-            models = ",".join(model_names)
-            raise KeyError(f"ModelType {self.name} not found in factory, please add to build method")
-
-        return model
+            available_models = ', '.join([ e.name for e in ModelType])
+            raise KeyError(f"ModelType {self.name} not found in factory, Available models are {available_models}."\
+                           "Otherwise please add new model to build method")
+        return model(config)
 
     @classmethod
     def names(cls):

emission/analysis/modelling/trip_model/models.py

@@ -19,7 +19,7 @@
 from emission.analysis.modelling.trip_model.clustering import get_distance_matrix, single_cluster_purity
 import emission.analysis.modelling.trip_model.data_wrangling as eamtd
 import emission.storage.decorations.trip_queries as esdtq
-from emission.analysis.classification.inference.labels.inferrers import predict_cluster_confidence_discounting
+import emission.analysis.classification.inference.labels.inferrers as eacili
 import emission.core.wrapper.entry as ecwe
 import emission.analysis.modelling.trip_model.greedy_similarity_binning as eamtg
 import emission.core.common as ecc
@@ -738,7 +738,7 @@ def predict_proba(self, test_df):
         replaced_distribs = []
 
         for trip in test_trips:
-            trip_prediction = predict_cluster_confidence_discounting(trip)
+            trip_prediction = eacili.predict_cluster_confidence_discounting(trip)
 
             if len(trip_prediction) == 0:
                 # model could not find cluster for the trip
@@ -1514,7 +1514,7 @@ def __init__(
         self.C = C
         self.n_estimators = n_estimators
         self.criterion = criterion
-        self.max_depth = max_depth
+        self.max_depth = max_depth if max_depth!= 'null' else None
         self.min_samples_split = min_samples_split
         self.min_samples_leaf = min_samples_leaf
         self.max_features = max_features
@@ -1524,36 +1524,42 @@ def __init__(
         self.use_start_clusters = use_start_clusters
         self.use_trip_clusters = use_trip_clusters
 
-        if self.loc_feature == 'cluster':
-            # clustering algorithm to generate end clusters
-            self.end_cluster_model = DBSCANSVMCluster(
-                loc_type='end',
-                radius=self.radius,
-                size_thresh=self.size_thresh,
-                purity_thresh=self.purity_thresh,
-                gamma=self.gamma,
-                C=self.C)
-
-            if self.use_start_clusters or self.use_trip_clusters:
-                # clustering algorithm to generate start clusters
-                self.start_cluster_model = DBSCANSVMCluster(
-                    loc_type='start',
-                    radius=self.radius,
-                    size_thresh=self.size_thresh,
-                    purity_thresh=self.purity_thresh,
-                    gamma=self.gamma,
-                    C=self.C)
-
-                if self.use_trip_clusters:
-                    # helper class to generate trip-level clusters
-                    self.trip_grouper = TripGrouper(
-                        start_cluster_col='start_cluster_idx',
-                        end_cluster_col='end_cluster_idx')
-
-            # wrapper class to generate one-hot encodings for cluster indices
-            self.cluster_enc = OneHotWrapper(sparse=False,
-                                             handle_unknown='ignore')
+        ######### Not Tested #########
+        # The below code is used when we cluster the coordinates (loc_cluster parameter = True)
+        # before passing to Random Forest. Commenting this for now since it is not tested.
+        ###############################
+        # if self.loc_feature == 'cluster':
+        #     # clustering algorithm to generate end clusters
+        #     self.end_cluster_model = DBSCANSVMCluster(
+        #         loc_type='end',
+        #         radius=self.radius,
+        #         size_thresh=self.size_thresh,
+        #         purity_thresh=self.purity_thresh,
+        #         gamma=self.gamma,
+        #         C=self.C)
+
+        #     if self.use_start_clusters or self.use_trip_clusters:
+        #         # clustering algorithm to generate start clusters
+        #         self.start_cluster_model = DBSCANSVMCluster(
+        #             loc_type='start',
+        #             radius=self.radius,
+        #             size_thresh=self.size_thresh,
+        #             purity_thresh=self.purity_thresh,
+        #             gamma=self.gamma,
+        #             C=self.C)
+
+        #         if self.use_trip_clusters:
+        #             # helper class to generate trip-level clusters
+        #             self.trip_grouper = TripGrouper(
+        #                 start_cluster_col='start_cluster_idx',
+        #                 end_cluster_col='end_cluster_idx')
+
+        #     # wrapper class to generate one-hot encodings for cluster indices
+        #     self.cluster_enc = OneHotWrapper(sparse=False,
+        #                                      handle_unknown='ignore')
+        #############################################################################
 
+
         # wrapper class to generate one-hot encodings for purposes and modes
         self.purpose_enc = OneHotWrapper(impute_missing=True,
                                          sparse=False,

emission/analysis/modelling/trip_model/run_model.py

@@ -56,9 +56,7 @@ def update_trip_model(
         time_query = time_query_from_pipeline if model.is_incremental else None
         logging.debug(f'model type {model_type.name} is incremental? {model.is_incremental}')
         logging.debug(f'time query for training data collection: {time_query}')
-
         trips = _get_training_data(user_id, time_query)
-
         # don't start training for a user that doesn't have at least $trips many trips
         # (assume if a stored model exists for the user, that they met this requirement previously)
         if len(trips) == 0:
@@ -74,7 +72,8 @@ def update_trip_model(
             epq.mark_trip_model_failed(user_id)
         else:
 
-            # train and store the model
+            # train and store the model. pass only List of event and only convert 
+            # to dataframe type data whereever required.
             model.fit(trips)
             model_data_next = model.to_dict()