Skip to content

util

logger = logging.getLogger(__name__) module-attribute

ProgressBar

Progress bar as visual.

Source code in src/recnexteval/utils/util.py
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
class ProgressBar:
    """Progress bar as visual."""

    def __init__(self) -> None:
        self.pbar = None

    def __call__(self, block_num, block_size, total_size) -> None:
        if not self.pbar:
            self.pbar = progressbar.ProgressBar(maxval=total_size)
            self.pbar.start()

        downloaded = block_num * block_size
        if downloaded < total_size:
            self.pbar.update(downloaded)
        else:
            self.pbar.finish()

pbar = None instance-attribute

to_tuple(element)

Whether single element or tuple, always returns as tuple.

Source code in src/recnexteval/utils/util.py
12
13
14
15
16
17
def to_tuple(element) -> tuple:
    """Whether single element or tuple, always returns as tuple."""
    if isinstance(element, tuple):
        return element
    else:
        return (element,)

arg_to_str(arg)

Converts a type to its name or returns the string.

:param arg: Argument to convert to string. :type arg: Union[type, str] :return: String representation of the argument. :rtype: str :raises TypeError: If the argument is not a string or a type.

Source code in src/recnexteval/utils/util.py
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def arg_to_str(arg: type | str) -> str:
    """Converts a type to its name or returns the string.

    :param arg: Argument to convert to string.
    :type arg: Union[type, str]
    :return: String representation of the argument.
    :rtype: str
    :raises TypeError: If the argument is not a string or a type.
    """
    if isinstance(arg, type):
        return arg.__name__
    elif not isinstance(arg, str):
        raise TypeError(f"Argument should be string or type, not {type(arg)}!")
    return arg

df_to_sparse(df, item_ix, user_ix, value_ix=None, shape=None)

Source code in src/recnexteval/utils/util.py
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
def df_to_sparse(
    df,
    item_ix,
    user_ix,
    value_ix=None,
    shape=None,
) -> csr_matrix:
    if value_ix is not None and value_ix in df:
        values = df[value_ix]
    else:
        if value_ix is not None:
            # value_ix provided, but not in df
            logger.warning(f"Value column {value_ix} not found in dataframe. Using ones instead.")

        num_entries = df.shape[0]
        # Scipy sums up the entries when an index-pair occurs more than once,
        # resulting in the actual counts being stored. Neat!
        values = np.ones(num_entries)

    indices = list(zip(*df.loc[:, [user_ix, item_ix]].values))

    if indices == []:
        indices = [[], []]  # Empty zip does not evaluate right

    if shape is None:
        shape = df[user_ix].max() + 1, df[item_ix].max() + 1
    sparse_matrix = csr_matrix((values, indices), shape=shape, dtype=values.dtype)

    return sparse_matrix

to_binary(X)

Converts a matrix to binary by setting all non-zero values to 1.

:param X: Matrix to convert to binary. :type X: csr_matrix :return: Binary matrix. :rtype: csr_matrix

Source code in src/recnexteval/utils/util.py
67
68
69
70
71
72
73
74
75
76
77
def to_binary(X: csr_matrix) -> csr_matrix:
    """Converts a matrix to binary by setting all non-zero values to 1.

    :param X: Matrix to convert to binary.
    :type X: csr_matrix
    :return: Binary matrix.
    :rtype: csr_matrix
    """
    X_binary = X.astype(bool).astype(X.dtype)

    return X_binary

invert(x)

Invert an array.

:param x: [description] :type x: [type] :return: [description] :rtype: [type]

Source code in src/recnexteval/utils/util.py
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def invert(x: Union[np.ndarray, csr_matrix]) -> Union[np.ndarray, csr_matrix]:
    """Invert an array.

    :param x: [description]
    :type x: [type]
    :return: [description]
    :rtype: [type]
    """
    if isinstance(x, np.ndarray):
        ret = np.zeros(x.shape)
    elif isinstance(x, csr_matrix):
        ret = csr_matrix(x.shape)
    else:
        raise TypeError("Unsupported type for argument x.")
    ret[x.nonzero()] = 1 / x[x.nonzero()]
    return ret

add_rows_to_csr_matrix(matrix, n=1)

Add a row of zeros to a csr_matrix.

ref: https://stackoverflow.com/questions/52299420/scipy-csr-matrix-understand-indptr

:param matrix: Matrix to add a row of zeros to. :type matrix: csr_matrix :return: Matrix with a row of zeros added. :rtype: csr_matrix

Source code in src/recnexteval/utils/util.py
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def add_rows_to_csr_matrix(matrix: csr_matrix, n: int = 1) -> csr_matrix:
    """Add a row of zeros to a csr_matrix.

    ref: https://stackoverflow.com/questions/52299420/scipy-csr-matrix-understand-indptr

    :param matrix: Matrix to add a row of zeros to.
    :type matrix: csr_matrix
    :return: Matrix with a row of zeros added.
    :rtype: csr_matrix
    """
    new_shape = (matrix.shape[0] + n, matrix.shape[1])
    new_indptr = np.append(matrix.indptr, [matrix.indptr[-1]] * n)
    matrix = csr_matrix((matrix.data, matrix.indices, new_indptr), shape=new_shape, copy=False)
    return matrix

add_columns_to_csr_matrix(matrix, n=1)

Add a column of zeros to a csr_matrix.

https://stackoverflow.com/questions/30691160/effectively-change-dimension-of-scipy-spare-csr-matrix

:param matrix: Matrix to add a column of zeros to. :type matrix: csr_matrix :return: Matrix with a column of zeros added. :rtype: csr_matrix

Source code in src/recnexteval/utils/util.py
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def add_columns_to_csr_matrix(matrix: csr_matrix, n: int = 1) -> csr_matrix:
    """Add a column of zeros to a csr_matrix.

    https://stackoverflow.com/questions/30691160/effectively-change-dimension-of-scipy-spare-csr-matrix

    :param matrix: Matrix to add a column of zeros to.
    :type matrix: csr_matrix
    :return: Matrix with a column of zeros added.
    :rtype: csr_matrix
    """
    new_shape = (matrix.shape[0], matrix.shape[1] + n)
    matrix = csr_matrix(
        (matrix.data, matrix.indices, matrix.indptr),
        shape=new_shape,
        copy=False,
    )
    return matrix

set_row_csr_addition(A, row_idx, new_row)

Set row of a csr_matrix to a new row.

ref: https://stackoverflow.com/questions/28427236/set-row-of-csr-matrix

:param A: Matrix to set a row of. :type A: csr_matrix :param row_idx: Index of the row to set. :type row_idx: int :param new_row: New row to set. :type new_row: np.ndarray

Source code in src/recnexteval/utils/util.py
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def set_row_csr_addition(A: csr_matrix, row_idx: int, new_row: np.ndarray) -> None:
    """Set row of a csr_matrix to a new row.

    ref: https://stackoverflow.com/questions/28427236/set-row-of-csr-matrix

    :param A: Matrix to set a row of.
    :type A: csr_matrix
    :param row_idx: Index of the row to set.
    :type row_idx: int
    :param new_row: New row to set.
    :type new_row: np.ndarray
    """
    indptr = np.zeros(A.shape[1] + 1)
    indptr[row_idx + 1 :] = A.shape[1]
    indices = np.arange(A.shape[1])
    A += csr_matrix((new_row, indices, indptr), shape=A.shape)