<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Taxi Fare Prediction Using Realtime Traffic Data
This will be the same as our previous taxi fare model, but with the addition of ‘trips_last_5min’ data as a feature. This is our proxy for traffic
Step1:  Load raw data
These are the same files used previously for taxi fare prediction, however an additional field trips_last_5min has been added
Step2:  Train and Evaluate input functions
Step3:  Feature Engineering
Step4:  Feature Engineering
Step5:  Gather list of feature columns
Note we're standard normalizing our traffic proxy. The mean and standard deviation were calculated on the full dataset in BigQuery, then hardcoded here.
Step6:  Serving Input Receiver function
Step7:  Train and Evaluate
<ASSISTANT_TASK:>
Python Code:
import tensorflow as tf
import numpy as np
import shutil
print(tf.__version__)
Explanation: Taxi Fare Prediction Using Realtime Traffic Data
This will be the same as our previous taxi fare model, but with the addition of ‘trips_last_5min’ data as a feature. This is our proxy for traffic
End of explanation
!gsutil cp gs://cloud-training-demos/taxifare/traffic/small/*.csv .
!ls -l *.csv
Explanation: Load raw data
These are the same files used previously for taxi fare prediction, however an additional field trips_last_5min has been added
End of explanation
CSV_COLUMN_NAMES = ["fare_amount","dayofweek","hourofday","pickuplon","pickuplat",\
                    "dropofflon","dropofflat","trips_last_5min"]
CSV_DEFAULTS = [[0.0],[1],[0],[-74.0],[40.0],[-74.0],[40.7],[0]]
def read_dataset(csv_path):
    def _parse_row(row):
        # Decode the CSV row into list of TF tensors
        fields = tf.decode_csv(records = row, record_defaults = CSV_DEFAULTS)
        # Pack the result into a dictionary
        features = dict(zip(CSV_COLUMN_NAMES, fields))
        
        # NEW: Add engineered features
        features = add_engineered_features(features)
        
        # Separate the label from the features
        label = features.pop("fare_amount") # remove label from features and store
        return features, label
    
    # Create a dataset containing the text lines.
    dataset = tf.data.Dataset.list_files(file_pattern = csv_path) # (i.e. data_file_*.csv)
    dataset = dataset.flat_map(map_func = lambda filename:tf.data.TextLineDataset(filenames = filename).skip(count = 1))
    # Parse each CSV row into correct (features,label) format for Estimator API
    dataset = dataset.map(map_func = _parse_row)
    
    return dataset
def train_input_fn(csv_path, batch_size = 128):
    #1. Convert CSV into tf.data.Dataset with (features,label) format
    dataset = read_dataset(csv_path)
      
    #2. Shuffle, repeat, and batch the examples.
    dataset = dataset.shuffle(buffer_size = 1000).repeat(count = None).batch(batch_size = batch_size)
   
    return dataset
def eval_input_fn(csv_path, batch_size = 128):
    #1. Convert CSV into tf.data.Dataset with (features,label) format
    dataset = read_dataset(csv_path)
    #2.Batch the examples.
    dataset = dataset.batch(batch_size = batch_size)
   
    return dataset
Explanation: Train and Evaluate input functions
End of explanation
# 1. One hot encode dayofweek and hourofday
fc_dayofweek = tf.feature_column.categorical_column_with_identity(key = "dayofweek", num_buckets = 7)
fc_hourofday = tf.feature_column.categorical_column_with_identity(key = "hourofday", num_buckets = 24)
# 2. Bucketize latitudes and longitudes
NBUCKETS = 16
latbuckets = np.linspace(start = 38.0, stop = 42.0, num = NBUCKETS).tolist()
lonbuckets = np.linspace(start = -76.0, stop = -72.0, num = NBUCKETS).tolist()
def bucketize_fc(key,boundaries):
    return tf.feature_column.bucketized_column(
        source_column = tf.feature_column.numeric_column(key = key), 
        boundaries = boundaries)
fc_bucketized_plat = bucketize_fc("pickuplon",lonbuckets)
fc_bucketized_plon = bucketize_fc("pickuplat",latbuckets)
fc_bucketized_dlat = bucketize_fc("dropofflon",lonbuckets)
fc_bucketized_dlon = bucketize_fc("dropofflat",latbuckets)
# 3. Cross features to get combination of day and hour
fc_crossed_day_hr = tf.feature_column.crossed_column(keys = [fc_dayofweek, fc_hourofday], hash_bucket_size = 24 * 7)
Explanation: Feature Engineering: feature columns
End of explanation
def add_engineered_features(features):
    features["dayofweek"] = features["dayofweek"] - 1 # subtract one since our days of week are 1-7 instead of 0-6
    
    features["latdiff"] = features["pickuplat"] - features["dropofflat"] # East/West
    features["londiff"] = features["pickuplon"] - features["dropofflon"] # North/South
    features["euclidean_dist"] = tf.sqrt(x = features["latdiff"]**2 + features["londiff"]**2)
    return features
Explanation: Feature Engineering: input function
End of explanation
feature_cols = [
    #1. Engineered using tf.feature_column module
    tf.feature_column.indicator_column(categorical_column = fc_crossed_day_hr),
    fc_bucketized_plat,
    fc_bucketized_plon,
    fc_bucketized_dlat,
    fc_bucketized_dlon,
    #2. Engineered in input functions
    tf.feature_column.numeric_column(key = "latdiff"),
    tf.feature_column.numeric_column(key = "londiff"),
    tf.feature_column.numeric_column(key = "euclidean_dist"),
    #3. Traffic proxy
    tf.feature_column.numeric_column(key = "trips_last_5min",
                                     normalizer_fn=lambda x: (x - 2070) / 616)
]
Explanation: Gather list of feature columns
Note we're standard normalizing our traffic proxy. The mean and standard deviation were calculated on the full dataset in BigQuery, then hardcoded here.
End of explanation
def serving_input_receiver_fn():
    receiver_tensors = {
        'dayofweek' : tf.placeholder(dtype = tf.int32, shape = [None]), # shape is vector to allow batch of requests
        'hourofday' : tf.placeholder(dtype = tf.int32, shape = [None]),
        'pickuplon' : tf.placeholder(dtype = tf.float32, shape = [None]), 
        'pickuplat' : tf.placeholder(dtype = tf.float32, shape = [None]),
        'dropofflat' : tf.placeholder(dtype = tf.float32, shape = [None]),
        'dropofflon' : tf.placeholder(dtype = tf.float32, shape = [None]),
        'trips_last_5min' : tf.placeholder(dtype = tf.float32, shape = [None]),
    }
    
    features = add_engineered_features(receiver_tensors) # 'features' is what is passed on to the model
    
    return tf.estimator.export.ServingInputReceiver(features = features, receiver_tensors = receiver_tensors)
Explanation: Serving Input Receiver function
End of explanation
%%time
OUTDIR = "taxi_trained"
shutil.rmtree(path = OUTDIR, ignore_errors = True) # start fresh each time
tf.summary.FileWriterCache.clear() # ensure filewriter cache is clear for TensorBoard events file
tf.logging.set_verbosity(v = tf.logging.INFO) # so loss is printed during training
model = tf.estimator.DNNRegressor(
    hidden_units = [10,10], # specify neural architecture
    feature_columns = feature_cols, 
    model_dir = OUTDIR,
    config = tf.estimator.RunConfig(
        tf_random_seed = 1, # for reproducibility
        save_checkpoints_steps = 200 # checkpoint every N steps
    ) 
)
# Add custom evaluation metric
def my_rmse(labels, predictions):
    pred_values = tf.squeeze(input = predictions["predictions"], axis = -1)
    return {"rmse": tf.metrics.root_mean_squared_error(labels = labels, predictions = pred_values)}
model = tf.contrib.estimator.add_metrics(estimator = model, metric_fn = my_rmse) 
    
train_spec = tf.estimator.TrainSpec(
    input_fn = lambda: train_input_fn("./taxi-train.csv"),
    max_steps = 5000)
exporter = tf.estimator.FinalExporter(name = "exporter", serving_input_receiver_fn = serving_input_receiver_fn) # export SavedModel once at the end of training
# Note: alternatively use tf.estimator.BestExporter to export at every checkpoint that has lower loss than the previous checkpoint
eval_spec = tf.estimator.EvalSpec(
    input_fn = lambda: eval_input_fn("./taxi-valid.csv"),
    steps = None,
    start_delay_secs = 1, # wait at least N seconds before first evaluation (default 120)
    throttle_secs = 1, # wait at least N seconds before each subsequent evaluation (default 600)
    exporters = exporter) # export SavedModel once at the end of training
tf.estimator.train_and_evaluate(estimator = model, train_spec = train_spec, eval_spec = eval_spec)
Explanation: Train and Evaluate
End of explanation
<END_TASK>