# @Time : 2020/12/25
# @Author : Yushuo Chen
# @Email : chenyushuo@ruc.edu.cn
# UPDATE
# @Time : 2020/12/25
# @Author : Yushuo Chen
# @email : chenyushuo@ruc.edu.cn
"""
recbole.utils.case_study
#####################################
"""
import numpy as np
import torch
from recbole.data.interaction import Interaction
[docs]@torch.no_grad()
def full_sort_scores(uid_series, model, test_data, device=None):
"""Calculate the scores of all items for each user in uid_series.
Note:
The score of [pad] and history items will be set into -inf.
Args:
uid_series (numpy.ndarray or list): User id series.
model (AbstractRecommender): Model to predict.
test_data (FullSortEvalDataLoader): The test_data of model.
device (torch.device, optional): The device which model will run on. Defaults to ``None``.
Note: ``device=None`` is equivalent to ``device=torch.device('cpu')``.
Returns:
torch.Tensor: the scores of all items for each user in uid_series.
"""
device = device or torch.device("cpu")
uid_series = torch.tensor(uid_series)
uid_field = test_data.dataset.uid_field
dataset = test_data.dataset
model.eval()
if not test_data.is_sequential:
input_interaction = dataset.join(Interaction({uid_field: uid_series}))
history_item = test_data.uid2history_item[list(uid_series)]
history_row = torch.cat(
[torch.full_like(hist_iid, i) for i, hist_iid in enumerate(history_item)]
)
history_col = torch.cat(list(history_item))
history_index = history_row, history_col
else:
_, index = (dataset.inter_feat[uid_field] == uid_series[:, None]).nonzero(
as_tuple=True
)
input_interaction = dataset[index]
history_index = None
# Get scores of all items
input_interaction = input_interaction.to(device)
try:
scores = model.full_sort_predict(input_interaction)
except NotImplementedError:
input_interaction = input_interaction.repeat_interleave(dataset.item_num)
input_interaction.update(
test_data.dataset.get_item_feature().to(device).repeat(len(uid_series))
)
scores = model.predict(input_interaction)
scores = scores.view(-1, dataset.item_num)
scores[:, 0] = -np.inf # set scores of [pad] to -inf
if history_index is not None:
scores[history_index] = -np.inf # set scores of history items to -inf
return scores
[docs]def full_sort_topk(uid_series, model, test_data, k, device=None):
"""Calculate the top-k items' scores and ids for each user in uid_series.
Note:
The score of [pad] and history items will be set into -inf.
Args:
uid_series (numpy.ndarray): User id series.
model (AbstractRecommender): Model to predict.
test_data (FullSortEvalDataLoader): The test_data of model.
k (int): The top-k items.
device (torch.device, optional): The device which model will run on. Defaults to ``None``.
Note: ``device=None`` is equivalent to ``device=torch.device('cpu')``.
Returns:
tuple:
- topk_scores (torch.Tensor): The scores of topk items.
- topk_index (torch.Tensor): The index of topk items, which is also the internal ids of items.
"""
scores = full_sort_scores(uid_series, model, test_data, device)
return torch.topk(scores, k)