Spaces:
Runtime error
Runtime error
File size: 4,548 Bytes
69a2a60 b6b5151 69a2a60 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 |
from datetime import datetime
import requests
import os
import joblib
import pandas as pd
import json
def decode_features(df, feature_view):
"""Decodes features in the input DataFrame using corresponding Hopsworks Feature Store transformation functions"""
df_res = df.copy()
import inspect
td_transformation_functions = feature_view._batch_scoring_server._transformation_functions
res = {}
for feature_name in td_transformation_functions:
if feature_name in df_res.columns:
td_transformation_function = td_transformation_functions[feature_name]
sig, foobar_locals = inspect.signature(td_transformation_function.transformation_fn), locals()
param_dict = dict([(param.name, param.default) for param in sig.parameters.values() if param.default != inspect._empty])
if td_transformation_function.name == "min_max_scaler":
df_res[feature_name] = df_res[feature_name].map(
lambda x: x * (param_dict["max_value"] - param_dict["min_value"]) + param_dict["min_value"])
elif td_transformation_function.name == "standard_scaler":
df_res[feature_name] = df_res[feature_name].map(
lambda x: x * param_dict['std_dev'] + param_dict["mean"])
elif td_transformation_function.name == "label_encoder":
dictionary = param_dict['value_to_index']
dictionary_ = {v: k for k, v in dictionary.items()}
df_res[feature_name] = df_res[feature_name].map(
lambda x: dictionary_[x])
return df_res
def get_model1(project, model_name, evaluation_metric, sort_metrics_by):
"""Retrieve desired model or download it from the Hopsworks Model Registry.
In second case, it will be physically downloaded to this directory"""
TARGET_FILE = "model_tempmax.pkl"
list_of_files = [os.path.join(dirpath,filename) for dirpath, _, filenames \
in os.walk('.') for filename in filenames if filename == TARGET_FILE]
if list_of_files:
model_path = list_of_files[0]
model = joblib.load(model_path)
else:
if not os.path.exists(TARGET_FILE):
mr = project.get_model_registry()
# get best model based on custom metrics
model = mr.get_best_model(model_name,
evaluation_metric,
sort_metrics_by)
model_dir = model.download()
model = joblib.load(model_dir + "/model_tempmax.pkl")
return model
def get_model2(project, model_name, evaluation_metric, sort_metrics_by):
"""Retrieve desired model or download it from the Hopsworks Model Registry.
In second case, it will be physically downloaded to this directory"""
TARGET_FILE = "model_tempmin.pkl"
list_of_files = [os.path.join(dirpath,filename) for dirpath, _, filenames \
in os.walk('.') for filename in filenames if filename == TARGET_FILE]
if list_of_files:
model_path = list_of_files[0]
model = joblib.load(model_path)
else:
if not os.path.exists(TARGET_FILE):
mr = project.get_model_registry()
# get best model based on custom metrics
model = mr.get_best_model(model_name,
evaluation_metric,
sort_metrics_by)
model_dir = model.download()
model = joblib.load(model_dir + "/model_tempmin.pkl")
return model
def get_model(project, model_name, evaluation_metric, sort_metrics_by):
"""Retrieve desired model or download it from the Hopsworks Model Registry.
In second case, it will be physically downloaded to this directory"""
TARGET_FILE = "model_temp.pkl"
list_of_files = [os.path.join(dirpath,filename) for dirpath, _, filenames \
in os.walk('.') for filename in filenames if filename == TARGET_FILE]
if list_of_files:
model_path = list_of_files[0]
model = joblib.load(model_path)
else:
if not os.path.exists(TARGET_FILE):
mr = project.get_model_registry()
# get best model based on custom metrics
model = mr.get_best_model(model_name,
evaluation_metric,
sort_metrics_by)
model_dir = model.download()
model = joblib.load(model_dir + "/model_temp.pkl")
return model
|