src.fairreckitlib.data.set.processor.dataset_processor_lfm

  1"""This module contains the base processor for LastFM datasets.
  2
  3Classes:
  4
  5    DatasetProcessorLFM: the base class for LastFM dataset processors.
  6
  7This program has been developed by students from the bachelor Computer Science at
  8Utrecht University within the Software Project course.
  9© Copyright Utrecht University (Department of Information and Computing Sciences)
 10"""
 11
 12from abc import ABCMeta, abstractmethod
 13from typing import Callable, List, Optional, Tuple
 14
 15import pandas as pd
 16
 17from ..dataset_config import DATASET_RATINGS_IMPLICIT, RatingMatrixConfig
 18from ..dataset_config import DatasetIndexConfig, DatasetMatrixConfig, DatasetTableConfig
 19from ..dataset_constants import TABLE_FILE_PREFIX
 20from .dataset_processor_base import DatasetProcessorBase
 21
 22
 23class DatasetProcessorLFM(DatasetProcessorBase, metaclass=ABCMeta):
 24    """DataProcessor base class for LastFM datasets.
 25
 26    Provides an abstraction for processing the listening event table,
 27    and also for generalizing the user table data. An iterative matrix
 28    processor function is exposed for derived subclasses as the LastFM
 29    dataset matrices tend to be very big.
 30
 31    Abstract methods:
 32
 33    create_listening_events_config
 34    create_user_table_config
 35    """
 36
 37    @abstractmethod
 38    def create_listening_events_config(self) -> Optional[DatasetTableConfig]:
 39        """Create the listening event table configuration.
 40
 41        Returns:
 42            the configuration of the listening event table or None when not available.
 43        """
 44        raise NotImplementedError()
 45
 46    @abstractmethod
 47    def create_user_table_config(self) -> DatasetTableConfig:
 48        """Create the user table configuration.
 49
 50        Returns:
 51            the configuration of the user table.
 52        """
 53        raise NotImplementedError()
 54
 55    def get_event_configs(self) -> List[Tuple[str, Callable[[], Optional[DatasetTableConfig]]]]:
 56        """Get event table configuration processors.
 57
 58        Returns:
 59            a list containing the listening event table processor.
 60        """
 61        return [('listening event', self.process_listening_events)]
 62
 63    def get_table_configs(self) -> List[Tuple[str, Callable[[], Optional[DatasetTableConfig]]]]:
 64        """Get table configuration processors.
 65
 66        Derived implementations are expected to call the super implementation in
 67        order to include the user table in the configuration.
 68
 69        Returns:
 70            a list containing the user table processor.
 71        """
 72        return [('user', self.process_user_table)]
 73
 74    def process_listening_events(self) -> Optional[DatasetTableConfig]:
 75        """Process the listening event table.
 76
 77        Returns:
 78            the listening event table configuration or None on failure.
 79        """
 80        les_table_config = self.create_listening_events_config()
 81        # skip without table configuration
 82        if les_table_config is None:
 83            return None
 84
 85        try:
 86            # count records in chunks as these files are huge
 87            table_iterator = les_table_config.read_table(self.dataset_dir, chunk_size=50000000)
 88            for _, table in enumerate(table_iterator):
 89                les_table_config.num_records += len(table)
 90
 91            return les_table_config
 92        except FileNotFoundError:
 93            return None
 94
 95    def process_matrix(
 96            self,
 97            matrix_table_config: DatasetTableConfig,
 98            user_idx_file: str=None,
 99            item_idx_file: str=None) -> Optional[DatasetMatrixConfig]:
100        """Process the matrix with the specified configuration.
101
102        Args:
103            matrix_table_config: the configuration of the matrix to process.
104            user_idx_file: the file name of the user indices or None when not present.
105            item_idx_file: the file name of the item indices or None when not present.
106
107        Returns:
108            the matrix configuration or None on failure.
109        """
110        user_id = matrix_table_config.primary_key[0]
111        item_id = matrix_table_config.primary_key[1]
112        count_column = matrix_table_config.columns[0]
113
114        unique_users = []
115        unique_items = []
116        rating_min = 1000000000.0
117        rating_max = 0.0
118
119        try:
120            matrix_it = matrix_table_config.read_table(self.dataset_dir, chunk_size=50000000)
121            # process matrix in chunks
122            for _, matrix in enumerate(matrix_it):
123                unique_users = pd.Series(unique_users, dtype='int').append(matrix[user_id]).unique()
124                unique_items = pd.Series(unique_items, dtype='int').append(matrix[item_id]).unique()
125                matrix_table_config.num_records += len(matrix)
126                rating_min = min(rating_min, matrix[count_column].min())
127                rating_max = max(rating_max, matrix[count_column].max())
128        except FileNotFoundError:
129            return None
130
131        return DatasetMatrixConfig(
132            matrix_table_config,
133            RatingMatrixConfig(
134                float(rating_min),
135                float(rating_max),
136                DATASET_RATINGS_IMPLICIT
137            ),
138            DatasetIndexConfig(
139                user_idx_file,
140                user_id,
141                len(unique_users)
142            ),
143            DatasetIndexConfig(
144                item_idx_file,
145                item_id,
146                len(unique_items)
147            )
148        )
149
150    def process_user_table(self) -> Optional[DatasetTableConfig]:
151        """Process the user table.
152
153        Changes the contents of the gender column to be more user-friendly,
154        and the contents of the age column to -1 when above 100.
155
156        Returns:
157            the user table configuration or None on failure.
158        """
159        user_table_config = self.create_user_table_config()
160
161        try:
162            user_table = user_table_config.read_table(self.dataset_dir)
163            user_table_config.num_records = len(user_table)
164        except FileNotFoundError:
165            return None
166
167        # convert gender to more user-friendly names
168        user_table['user_gender'].replace({
169            'm': 'Male',
170            'f': 'Female',
171            'n': 'Neutral'
172        }, inplace=True)
173        # convert age above 100 to -1
174        user_table['user_age'].mask(user_table['user_age'].gt(100), inplace=True)
175        user_table['user_age'].fillna(-1.0, inplace=True)
176        user_table['user_age'] = user_table['user_age'].astype(int)
177
178        # update table configuration
179        user_table_config.file.name = TABLE_FILE_PREFIX + self.dataset_name + '_users.tsv.bz2'
180        user_table_config.file.options.compression = 'bz2'
181        user_table_config.file.options.header = False
182        user_table_config.file.options.sep = None
183
184        # store the resulting user table
185        user_table_config.save_table(user_table, self.dataset_dir)
186
187        return user_table_config