Skip to content

filters

logger = logging.getLogger(__name__) module-attribute

T = TypeVar('T', bound='InteractionMatrix') module-attribute

SelectionIDMixin

Source code in src/recnexteval/matrix/filters.py 
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
class SelectionIDMixin:
    @overload
    def users_in(self: T, U: set[int]) -> T: ...
    @overload
    def users_in(self: T, U: set[int], inplace: Literal[False]) -> T: ...
    @overload
    def users_in(self: T, U: set[int], inplace: Literal[True]) -> None: ...
    def users_in(self: T, U: set[int], inplace: bool = False) -> None | T:
        logger.debug("Performing users_in comparison")
        mask = self._df[self.USER_IX].isin(U)
        return self._apply_mask(mask, inplace=inplace)

    @overload
    def users_not_in(self: T, U: set[int]) -> T: ...
    @overload
    def users_not_in(self: T, U: set[int], inplace: Literal[False]) -> T: ...
    @overload
    def users_not_in(self: T, U: set[int], inplace: Literal[True]) -> None: ...
    def users_not_in(self: T, U: set[int], inplace: bool = False) -> None | T:
        logger.debug("Performing users_not_in comparison")
        mask = ~self._df[self.USER_IX].isin(U)
        return self._apply_mask(mask, inplace=inplace)

    @overload
    def items_in(self: T, id_set: set[int]) -> T: ...
    @overload
    def items_in(self: T, id_set: set[int], inplace: Literal[False]) -> T: ...
    @overload
    def items_in(self: T, id_set: set[int], inplace: Literal[True]) -> None: ...
    def items_in(self: T, id_set: set[int], inplace: bool = False) -> None | T:
        logger.debug("Performing items_in comparison")
        mask = self._df[self.ITEM_IX].isin(id_set)
        return self._apply_mask(mask, inplace=inplace)

    @overload
    def items_not_in(self: T, id_set: set[int]) -> T: ...
    @overload
    def items_not_in(self: T, id_set: set[int], inplace: Literal[False]) -> T: ...
    @overload
    def items_not_in(self: T, id_set: set[int], inplace: Literal[True]) -> None: ...
    def items_not_in(self: T, id_set: set[int], inplace: bool = False) -> None | T:
        logger.debug("Performing items_not_in comparison")
        mask = ~self._df[self.ITEM_IX].isin(id_set)
        return self._apply_mask(mask, inplace=inplace)

users_in(U, inplace=False) 

users_in(U: set[int]) -> T

users_in(U: set[int], inplace: Literal[False]) -> T

users_in(U: set[int], inplace: Literal[True]) -> None
Source code in src/recnexteval/matrix/filters.py 
23
24
25
26
def users_in(self: T, U: set[int], inplace: bool = False) -> None | T:
    logger.debug("Performing users_in comparison")
    mask = self._df[self.USER_IX].isin(U)
    return self._apply_mask(mask, inplace=inplace)

users_not_in(U, inplace=False) 

users_not_in(U: set[int]) -> T

users_not_in(U: set[int], inplace: Literal[False]) -> T

users_not_in(U: set[int], inplace: Literal[True]) -> None
Source code in src/recnexteval/matrix/filters.py 
34
35
36
37
def users_not_in(self: T, U: set[int], inplace: bool = False) -> None | T:
    logger.debug("Performing users_not_in comparison")
    mask = ~self._df[self.USER_IX].isin(U)
    return self._apply_mask(mask, inplace=inplace)

items_in(id_set, inplace=False) 

items_in(id_set: set[int]) -> T

items_in(id_set: set[int], inplace: Literal[False]) -> T

items_in(id_set: set[int], inplace: Literal[True]) -> None
Source code in src/recnexteval/matrix/filters.py 
45
46
47
48
def items_in(self: T, id_set: set[int], inplace: bool = False) -> None | T:
    logger.debug("Performing items_in comparison")
    mask = self._df[self.ITEM_IX].isin(id_set)
    return self._apply_mask(mask, inplace=inplace)

items_not_in(id_set, inplace=False) 

items_not_in(id_set: set[int]) -> T

items_not_in(
    id_set: set[int], inplace: Literal[False]
) -> T

items_not_in(
    id_set: set[int], inplace: Literal[True]
) -> None
Source code in src/recnexteval/matrix/filters.py 
56
57
58
59
def items_not_in(self: T, id_set: set[int], inplace: bool = False) -> None | T:
    logger.debug("Performing items_not_in comparison")
    mask = ~self._df[self.ITEM_IX].isin(id_set)
    return self._apply_mask(mask, inplace=inplace)

SelectionTimestampMixin

Source code in src/recnexteval/matrix/filters.py 
 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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
class SelectionTimestampMixin:
    @overload
    def timestamps_gt(self: T, timestamp: float) -> T: ...
    @overload
    def timestamps_gt(self: T, timestamp: float, inplace: Literal[True]) -> None: ...
    def timestamps_gt(self: T, timestamp: float, inplace: bool = False) -> None | T:
        """Select interactions after a given timestamp.

        :param timestamp: The timestamp with which
            the interactions timestamp is compared.
        :type timestamp: float
        :param inplace: Apply the selection in place if True, defaults to False
        :type inplace: bool, optional
        :return: None if `inplace`, otherwise returns a new InteractionMatrix object
        :rtype: Union[InteractionMatrix, None]
        """
        return self._timestamps_cmp(operator.gt, timestamp, inplace)

    @overload
    def timestamps_gte(self: T, timestamp: float) -> T: ...
    @overload
    def timestamps_gte(self: T, timestamp: float, inplace: Literal[True]) -> None: ...
    def timestamps_gte(self: T, timestamp: float, inplace: bool = False) -> None | T:
        """Select interactions after and including a given timestamp.

        :param timestamp: The timestamp with which
            the interactions timestamp is compared.
        :type timestamp: float
        :param inplace: Apply the selection in place if True, defaults to False
        :type inplace: bool, optional
        :return: None if `inplace`, otherwise returns a new InteractionMatrix object
        :rtype: Union[InteractionMatrix, None]
        """
        return self._timestamps_cmp(operator.ge, timestamp, inplace)

    @overload
    def timestamps_lt(self: T, timestamp: float) -> T: ...
    @overload
    def timestamps_lt(self: T, timestamp: float, inplace: Literal[True]) -> None: ...
    def timestamps_lt(self: T, timestamp: float, inplace: bool = False) -> None | T:
        """Select interactions up to a given timestamp.

        :param timestamp: The timestamp with which
            the interactions timestamp is compared.
        :type timestamp: float
        :param inplace: Apply the selection in place if True, defaults to False
        :type inplace: bool, optional
        :return: None if `inplace`, otherwise returns a new InteractionMatrix object
        :rtype: Union[InteractionMatrix, None]
        """
        return self._timestamps_cmp(operator.lt, timestamp, inplace)

    @overload
    def timestamps_lte(self: T, timestamp: float) -> T: ...
    @overload
    def timestamps_lte(self: T, timestamp: float, inplace: Literal[True]) -> None: ...
    def timestamps_lte(self: T, timestamp: float, inplace: bool = False) -> None | T:
        """Select interactions up to and including a given timestamp.

        :param timestamp: The timestamp with which
            the interactions timestamp is compared.
        :type timestamp: float
        :param inplace: Apply the selection in place if True, defaults to False
        :type inplace: bool, optional
        :return: None if `inplace`, otherwise returns a new InteractionMatrix object
        :rtype: Union[InteractionMatrix, None]
        """
        return self._timestamps_cmp(operator.le, timestamp, inplace)

    def _timestamps_cmp(self: T, op, timestamp: float, inplace: bool = False) -> None | T:
        # import here to avoid circular imports at module import time
        from .exception import TimestampAttributeMissingError

        if not self.has_timestamps:
            raise TimestampAttributeMissingError()

        logger.debug(f"Performing {op.__name__}(t, {timestamp})")

        mask = op(self._df[self.TIMESTAMP_IX], timestamp)
        return self._apply_mask(mask, inplace=inplace)

    def get_users_n_last_interaction(
        self: T, n_seq_data: int = 1, t_upper: None | int = None, user_in: None | set[int] = None, inplace: bool = False
    ) -> T:
        return self._get_last_n_interactions(
            by=ItemUserBasedEnum.USER, n_seq_data=n_seq_data, t_upper=t_upper, id_in=user_in, inplace=inplace
        )

    def get_items_n_last_interaction(
        self: T, n_seq_data: int = 1, t_upper: None | int = None, item_in: None | set[int] = None, inplace: bool = False
    ) -> T:
        return self._get_last_n_interactions(by=ItemUserBasedEnum.ITEM, n_seq_data=n_seq_data, t_upper=t_upper, id_in=item_in, inplace=inplace)

    def get_users_n_first_interaction(
        self: T, n_seq_data: int = 1, t_lower: None | int = None, inplace: bool = False
    ) -> T:
        return self._get_first_n_interactions(ItemUserBasedEnum.USER, n_seq_data, t_lower, inplace)

    def get_items_n_first_interaction(
        self: T, n_seq_data: int = 1, t_lower: None | int = None, inplace: bool = False
    ) -> T:
        return self._get_first_n_interactions(ItemUserBasedEnum.ITEM, n_seq_data, t_lower, inplace)

timestamps_gt(timestamp, inplace=False) 

timestamps_gt(timestamp: float) -> T

timestamps_gt(
    timestamp: float, inplace: Literal[True]
) -> None

Select interactions after a given timestamp.

:param timestamp: The timestamp with which the interactions timestamp is compared. :type timestamp: float :param inplace: Apply the selection in place if True, defaults to False :type inplace: bool, optional :return: None if inplace, otherwise returns a new InteractionMatrix object :rtype: Union[InteractionMatrix, None]

Source code in src/recnexteval/matrix/filters.py 
67
68
69
70
71
72
73
74
75
76
77
78
def timestamps_gt(self: T, timestamp: float, inplace: bool = False) -> None | T:
    """Select interactions after a given timestamp.

    :param timestamp: The timestamp with which
        the interactions timestamp is compared.
    :type timestamp: float
    :param inplace: Apply the selection in place if True, defaults to False
    :type inplace: bool, optional
    :return: None if `inplace`, otherwise returns a new InteractionMatrix object
    :rtype: Union[InteractionMatrix, None]
    """
    return self._timestamps_cmp(operator.gt, timestamp, inplace)

timestamps_gte(timestamp, inplace=False) 

timestamps_gte(timestamp: float) -> T

timestamps_gte(
    timestamp: float, inplace: Literal[True]
) -> None

Select interactions after and including a given timestamp.

:param timestamp: The timestamp with which the interactions timestamp is compared. :type timestamp: float :param inplace: Apply the selection in place if True, defaults to False :type inplace: bool, optional :return: None if inplace, otherwise returns a new InteractionMatrix object :rtype: Union[InteractionMatrix, None]

Source code in src/recnexteval/matrix/filters.py 
84
85
86
87
88
89
90
91
92
93
94
95
def timestamps_gte(self: T, timestamp: float, inplace: bool = False) -> None | T:
    """Select interactions after and including a given timestamp.

    :param timestamp: The timestamp with which
        the interactions timestamp is compared.
    :type timestamp: float
    :param inplace: Apply the selection in place if True, defaults to False
    :type inplace: bool, optional
    :return: None if `inplace`, otherwise returns a new InteractionMatrix object
    :rtype: Union[InteractionMatrix, None]
    """
    return self._timestamps_cmp(operator.ge, timestamp, inplace)

timestamps_lt(timestamp, inplace=False) 

timestamps_lt(timestamp: float) -> T

timestamps_lt(
    timestamp: float, inplace: Literal[True]
) -> None

Select interactions up to a given timestamp.

:param timestamp: The timestamp with which the interactions timestamp is compared. :type timestamp: float :param inplace: Apply the selection in place if True, defaults to False :type inplace: bool, optional :return: None if inplace, otherwise returns a new InteractionMatrix object :rtype: Union[InteractionMatrix, None]

Source code in src/recnexteval/matrix/filters.py 
101
102
103
104
105
106
107
108
109
110
111
112
def timestamps_lt(self: T, timestamp: float, inplace: bool = False) -> None | T:
    """Select interactions up to a given timestamp.

    :param timestamp: The timestamp with which
        the interactions timestamp is compared.
    :type timestamp: float
    :param inplace: Apply the selection in place if True, defaults to False
    :type inplace: bool, optional
    :return: None if `inplace`, otherwise returns a new InteractionMatrix object
    :rtype: Union[InteractionMatrix, None]
    """
    return self._timestamps_cmp(operator.lt, timestamp, inplace)

timestamps_lte(timestamp, inplace=False) 

timestamps_lte(timestamp: float) -> T

timestamps_lte(
    timestamp: float, inplace: Literal[True]
) -> None

Select interactions up to and including a given timestamp.

:param timestamp: The timestamp with which the interactions timestamp is compared. :type timestamp: float :param inplace: Apply the selection in place if True, defaults to False :type inplace: bool, optional :return: None if inplace, otherwise returns a new InteractionMatrix object :rtype: Union[InteractionMatrix, None]

Source code in src/recnexteval/matrix/filters.py 
118
119
120
121
122
123
124
125
126
127
128
129
def timestamps_lte(self: T, timestamp: float, inplace: bool = False) -> None | T:
    """Select interactions up to and including a given timestamp.

    :param timestamp: The timestamp with which
        the interactions timestamp is compared.
    :type timestamp: float
    :param inplace: Apply the selection in place if True, defaults to False
    :type inplace: bool, optional
    :return: None if `inplace`, otherwise returns a new InteractionMatrix object
    :rtype: Union[InteractionMatrix, None]
    """
    return self._timestamps_cmp(operator.le, timestamp, inplace)

get_users_n_last_interaction(n_seq_data=1, t_upper=None, user_in=None, inplace=False)

Source code in src/recnexteval/matrix/filters.py 
143
144
145
146
147
148
def get_users_n_last_interaction(
    self: T, n_seq_data: int = 1, t_upper: None | int = None, user_in: None | set[int] = None, inplace: bool = False
) -> T:
    return self._get_last_n_interactions(
        by=ItemUserBasedEnum.USER, n_seq_data=n_seq_data, t_upper=t_upper, id_in=user_in, inplace=inplace
    )

get_items_n_last_interaction(n_seq_data=1, t_upper=None, item_in=None, inplace=False)

Source code in src/recnexteval/matrix/filters.py 
150
151
152
153
def get_items_n_last_interaction(
    self: T, n_seq_data: int = 1, t_upper: None | int = None, item_in: None | set[int] = None, inplace: bool = False
) -> T:
    return self._get_last_n_interactions(by=ItemUserBasedEnum.ITEM, n_seq_data=n_seq_data, t_upper=t_upper, id_in=item_in, inplace=inplace)

get_users_n_first_interaction(n_seq_data=1, t_lower=None, inplace=False)

Source code in src/recnexteval/matrix/filters.py 
155
156
157
158
def get_users_n_first_interaction(
    self: T, n_seq_data: int = 1, t_lower: None | int = None, inplace: bool = False
) -> T:
    return self._get_first_n_interactions(ItemUserBasedEnum.USER, n_seq_data, t_lower, inplace)

get_items_n_first_interaction(n_seq_data=1, t_lower=None, inplace=False)

Source code in src/recnexteval/matrix/filters.py 
160
161
162
163
def get_items_n_first_interaction(
    self: T, n_seq_data: int = 1, t_lower: None | int = None, inplace: bool = False
) -> T:
    return self._get_first_n_interactions(ItemUserBasedEnum.ITEM, n_seq_data, t_lower, inplace)