Coverage for hopwise/utils/case_study.py: 0%

1# @Time : 2020/12/25 

2# @Author : Yushuo Chen 

3# @Email : chenyushuo@ruc.edu.cn 

4 

5# UPDATE 

6# @Time : 2020/12/25 

7# @Author : Yushuo Chen 

8# @email : chenyushuo@ruc.edu.cn 

9 

10# UPDATE 

11# @Time : 2025/06 

12# @Author : Alessandro Soccol, Giacomo Medda 

13# @email : alessandro.soccol@unica.it, giacomo.medda@unica.it 

14 

15"""hopwise.utils.case_study 

16##################################### 

17""" 

18 

19import numpy as np 

20import pandas as pd 

21import torch 

22 

23from hopwise.data.interaction import Interaction 

24 

25 

26@torch.no_grad() 

27def full_sort_scores(uid_series, model, test_data, device=None): 

28 """Calculate the scores of all items for each user in uid_series. 

29 

30 Note: 

31 The score of [pad] and history items will be set into -inf. 

32 

33 Args: 

34 uid_series (numpy.ndarray or list): User id series. 

35 model (AbstractRecommender): Model to predict. 

36 test_data (FullSortEvalDataLoader): The test_data of model. 

37 device (torch.device, optional): The device which model will run on. Defaults to ``None``. 

38 Note: ``device=None`` is equivalent to ``device=torch.device('cpu')``. 

39 

40 Returns: 

41 torch.Tensor: the scores of all items for each user in uid_series. 

42 """ 

43 device = device or torch.device("cpu") 

44 uid_series = torch.tensor(uid_series) 

45 uid_field = test_data.dataset.uid_field 

46 dataset = test_data.dataset 

47 model.eval() 

48 

49 if not test_data.is_sequential: 

50 input_interaction = dataset.join(Interaction({uid_field: uid_series})) 

51 history_item = test_data.uid2history_item[list(uid_series)] 

52 history_row = torch.cat([torch.full_like(hist_iid, i) for i, hist_iid in enumerate(history_item)]) 

53 history_col = torch.cat(list(history_item)) 

54 history_index = history_row, history_col 

55 else: 

56 _, index = (dataset.inter_feat[uid_field] == uid_series[:, None]).nonzero(as_tuple=True) 

57 input_interaction = dataset[index] 

58 history_index = None 

59 

60 # Get scores of all items 

61 input_interaction = input_interaction.to(device) 

62 try: 

63 scores = model.full_sort_predict(input_interaction) 

64 except NotImplementedError: 

65 input_interaction = input_interaction.repeat_interleave(dataset.item_num) 

66 input_interaction.update(test_data.dataset.get_item_feature().to(device).repeat(len(uid_series))) 

67 scores = model.predict(input_interaction) 

68 

69 scores = scores.view(-1, dataset.item_num) 

70 scores[:, 0] = -np.inf # set scores of [pad] to -inf 

71 if history_index is not None: 

72 scores[history_index] = -np.inf # set scores of history items to -inf 

73 

74 return scores 

75 

76 

77@torch.no_grad() 

78def full_sort_explanations(uid_series, model, test_data, device=None): 

79 """Calculate the scores of all items for each user in uid_series. 

80 

81 Note: 

82 The score of [pad] and history items will be set into -inf. 

83 

84 Args: 

85 uid_series (numpy.ndarray or list): User id series. 

86 model (AbstractRecommender): Model to predict. 

87 test_data (FullSortEvalDataLoader): The test_data of model. 

88 device (torch.device, optional): The device which model will run on. Defaults to ``None``. 

89 Note: ``device=None`` is equivalent to ``device=torch.device('cpu')``. 

90 

91 Returns: 

92 torch.Tensor: the scores of all items for each user in uid_series. 

93 """ 

94 device = device or torch.device("cpu") 

95 uid_series = torch.tensor(uid_series) 

96 uid_field = test_data.dataset.uid_field 

97 dataset = test_data.dataset 

98 model.eval() 

99 

100 if not test_data.is_sequential: 

101 input_interaction = dataset.join(Interaction({uid_field: uid_series})) 

102 else: 

103 _, index = (dataset.inter_feat[uid_field] == uid_series[:, None]).nonzero(as_tuple=True) 

104 input_interaction = dataset[index] 

105 

106 # Get scores of all items 

107 input_interaction = input_interaction.to(device) 

108 

109 _, explanations = model.explain(input_interaction) 

110 

111 # return explanations as pandas dataframe 

112 return pd.DataFrame(explanations, columns=["user", "item", "score", "path"]) 

113 

114 

115def full_sort_topk(uid_series, model, test_data, k, device=None): 

116 """Calculate the top-k items' scores and ids for each user in uid_series. 

117 

118 Note: 

119 The score of [pad] and history items will be set into -inf. 

120 

121 Args: 

122 uid_series (numpy.ndarray): User id series. 

123 model (AbstractRecommender): Model to predict. 

124 test_data (FullSortEvalDataLoader): The test_data of model. 

125 k (int): The top-k items. 

126 device (torch.device, optional): The device which model will run on. Defaults to ``None``. 

127 Note: ``device=None`` is equivalent to ``device=torch.device('cpu')``. 

128 

129 Returns: 

130 tuple: 

131 - topk_scores (torch.Tensor): The scores of topk items. 

132 - topk_index (torch.Tensor): The index of topk items, which is also the internal ids of items. 

133 """ 

134 scores = full_sort_scores(uid_series, model, test_data, device) 

135 return torch.topk(scores, k)