Skip to content

Loader

A class used to load data from 'csv', 'sas7bdat' and 'parquet' file formats.

Source code in bozio_wasmer_simulations/datasets/base.py 
 11
 12
 13
 14
 15
 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
 60
 61
 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
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
class Loader:
    """
    A class used to load data from 'csv', 'sas7bdat' and 'parquet' file formats.

    """

    # Initialisation
    def __init__(self) -> None:
        """
        Constructs all the necessary attributes for the Loader object.
        """
        pass

    # Fonction auxiliaire de déduction d'un masque à partir de filtres
    @staticmethod
    def _build_mask_from_filters(
        data: pd.DataFrame, filters: Union[List[Tuple], List[List[Tuple]]]
    ) -> pd.DataFrame:
        """
        Builds a mask from filters for a given dataframe.

        Args:
            data (pd.DataFrame):
                The dataframe to build the mask for.
            filters (Union[List[Tuple], List[List[Tuple]]]):
                The filters to apply. Each filter is a tuple of (column, operator, value).
                If filters is a list of tuples, all filters are applied in conjunction.
                If filters is a list of lists of tuples, each list is applied in disjunction.

        Returns:
            (pd.DataFrame): The mask that can be used to filter the dataframe.
        """
        # Si filters est une liste de tuples
        if all(isinstance(i, tuple) for i in filters) and isinstance(filters, list):
            # Initialisation de la série résultat
            mask = pd.Series(True, index=data.index)
            # Parcours des filters
            for conjonction_filter in filters:
                col, op, val = conjonction_filter
                mask &= eval(f"data['{col}'] {op} {val}")
        # Si filters est une liste de liste de tuples
        elif all(
            isinstance(i, list) and all(isinstance(j, tuple) for j in i)
            for i in filters
        ) and isinstance(filters, list):
            # Initialisation de la série résultat
            mask = pd.Series(False, index=data.index)
            # Parcours des filtres
            for disjonction_filter in filters:
                if isinstance(disjonction_filter, tuple):
                    col, op, val = disjonction_filter
                    mask |= eval(f"data['{col}'] {op} {val}")
                elif isinstance(disjonction_filter, list):
                    submask = pd.Series(True, index=data.index)
                    for conjonction_filter in disjonction_filter:
                        col, op, val = conjonction_filter
                        submask &= eval(f"data['{col}'] {op} {val}")
                    mask |= submask
        else:
            raise TypeError(
                f"Invalid type for 'filters' : {filters}. Shoud be in [List[Tuple], List[List[Tuple]]]"
            )
        return mask

    # Fonction de chargement des données SAS
    def read_sas(
        self,
        path: os.PathLike,
        columns: Optional[Union[None, List[str]]] = None,
        filters: Optional[Union[None, List[str], List[List[str]]]] = None,
        **kwargs,
    ) -> pd.DataFrame:
        """
        Loads data from a SAS file.

        Args:
            path (os.PathLike):
                The path to the SAS file.
            columns (Optional[Union[None, List[str]]], optional):
                The columns to load. If None, all columns are loaded.
            filters (Optional[Union[None, List[str], List[List[str]]]], optional):
                The filters to apply. Each filter is a tuple of (column, operator, value).
                If filters is a list of tuples, all filters are applied in conjunction.
                If filters is a list of lists of tuples, each list is applied in disjunction.
            **kwargs
                Additional arguments to pass to the pyreadstat.read_sas7bdat function.

        Returns:
            (pd.DataFrame): The loaded data.
        """
        # Chargement des données
        data, _ = pyreadstat.read_sas7bdat(path, usecols=columns, **kwargs)
        # Filtrage des données
        if filters is not None:
            data = data.loc[self._build_mask_from_filters(data=data, filters=filters)]
        return data

    # Fonction de chargement des données CSV
    def read_csv(
        self,
        path: os.PathLike,
        columns: Optional[Union[None, List[str]]] = None,
        filters: Optional[Union[None, List[str], List[List[str]]]] = None,
        **kwargs,
    ) -> pd.DataFrame:
        """
        Loads data from a CSV file.

        Args:
            path (os.PathLike):
                The path to the CSV file.
            columns (Optional[Union[None, List[str]]], optional):
                The columns to load. If None, all columns are loaded.
            filters (Optional[Union[None, List[str], List[List[str]]]], optional):
                The filters to apply. Each filter is a tuple of (column, operator, value).
                If filters is a list of tuples, all filters are applied in conjunction.
                If filters is a list of lists of tuples, each list is applied in disjunction.
            **kwargs
                Additional arguments to pass to the pd.read_csv function.

        Returns:
            (pd.DataFrame): The loaded data.
        """
        # Chargement des données
        data = pd.read_csv(path, usecols=columns, **kwargs)
        # Filtrage des données
        if filters is not None:
            data = data.loc[self._build_mask_from_filters(data=data, filters=filters)]
        return data

    # Fonction générale de chargement des données
    def load(
        self,
        path: os.PathLike,
        columns: Optional[Union[None, List[str]]] = None,
        filters: Optional[Union[None, List[str], List[List[str]]]] = None,
        **kwargs,
    ) -> pd.DataFrame:
        """
        Loads data from a file based on its extension.

        Args:
            path (os.PathLike):
                The path to the file.
            columns (Optional[Union[None, List[str]]], optional):
                The columns to load. If None, all columns are loaded.
            filters (Optional[Union[None, List[str], List[List[str]]]], optional):
                The filters to apply. Each filter is a tuple of (column, operator, value).
                If filters is a list of tuples, all filters are applied in conjunction.
                If filters is a list of lists of tuples, each list is applied in disjunction.
            **kwargs
                Additional arguments to pass to the appropriate function.

        Returns:
            (pd.DataFrame): The loaded data.

        Raises:
            ValueError: If the file extension is not 'parquet', 'sas7bdat', or 'csv'.
        """
        # Extraction de l'extension du fichier à charger
        extension = path.split(".")[-1]
        # Distinction de la méthode de chargement suivant l'extension
        if (extension == "parquet") | (os.path.isdir(path)):
            data = pd.read_parquet(path, columns=columns, filters=filters, **kwargs)
        elif extension == "sas7bdat":
            data = self.read_sas(path=path, columns=columns, filters=filters, **kwargs)
        elif extension == "csv":
            data = self.read_csv(path=path, columns=columns, filters=filters, **kwargs)
        else:
            raise ValueError(
                f"Unknown extention : {extension}. Should be in ['parquet', 'sas7bdat', 'csv']"
            )

        return data

load 

load(path: PathLike, columns: Optional[Union[None, List[str]]] = None, filters: Optional[Union[None, List[str], List[List[str]]]] = None, **kwargs) -> DataFrame

Loads data from a file based on its extension.

Parameters:

Name Type Description Default
path PathLike

The path to the file.

 required
columns Optional[Union[None, List[str]]]

The columns to load. If None, all columns are loaded.

 None
filters Optional[Union[None, List[str], List[List[str]]]]

The filters to apply. Each filter is a tuple of (column, operator, value). If filters is a list of tuples, all filters are applied in conjunction. If filters is a list of lists of tuples, each list is applied in disjunction.

 None

Returns:

Type Description
DataFrame

The loaded data.

Raises:

Type Description
ValueError

If the file extension is not 'parquet', 'sas7bdat', or 'csv'.
Source code in bozio_wasmer_simulations/datasets/base.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def load(
    self,
    path: os.PathLike,
    columns: Optional[Union[None, List[str]]] = None,
    filters: Optional[Union[None, List[str], List[List[str]]]] = None,
    **kwargs,
) -> pd.DataFrame:
    """
    Loads data from a file based on its extension.

    Args:
        path (os.PathLike):
            The path to the file.
        columns (Optional[Union[None, List[str]]], optional):
            The columns to load. If None, all columns are loaded.
        filters (Optional[Union[None, List[str], List[List[str]]]], optional):
            The filters to apply. Each filter is a tuple of (column, operator, value).
            If filters is a list of tuples, all filters are applied in conjunction.
            If filters is a list of lists of tuples, each list is applied in disjunction.
        **kwargs
            Additional arguments to pass to the appropriate function.

    Returns:
        (pd.DataFrame): The loaded data.

    Raises:
        ValueError: If the file extension is not 'parquet', 'sas7bdat', or 'csv'.
    """
    # Extraction de l'extension du fichier à charger
    extension = path.split(".")[-1]
    # Distinction de la méthode de chargement suivant l'extension
    if (extension == "parquet") | (os.path.isdir(path)):
        data = pd.read_parquet(path, columns=columns, filters=filters, **kwargs)
    elif extension == "sas7bdat":
        data = self.read_sas(path=path, columns=columns, filters=filters, **kwargs)
    elif extension == "csv":
        data = self.read_csv(path=path, columns=columns, filters=filters, **kwargs)
    else:
        raise ValueError(
            f"Unknown extention : {extension}. Should be in ['parquet', 'sas7bdat', 'csv']"
        )

    return data

read_csv 

read_csv(path: PathLike, columns: Optional[Union[None, List[str]]] = None, filters: Optional[Union[None, List[str], List[List[str]]]] = None, **kwargs) -> DataFrame

Loads data from a CSV file.

Parameters:

Name Type Description Default
path PathLike

The path to the CSV file.

 required
columns Optional[Union[None, List[str]]]

The columns to load. If None, all columns are loaded.

 None
filters Optional[Union[None, List[str], List[List[str]]]]

The filters to apply. Each filter is a tuple of (column, operator, value). If filters is a list of tuples, all filters are applied in conjunction. If filters is a list of lists of tuples, each list is applied in disjunction.

 None

Returns:

Type Description
DataFrame

The loaded data.
Source code in bozio_wasmer_simulations/datasets/base.py 
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
def read_csv(
    self,
    path: os.PathLike,
    columns: Optional[Union[None, List[str]]] = None,
    filters: Optional[Union[None, List[str], List[List[str]]]] = None,
    **kwargs,
) -> pd.DataFrame:
    """
    Loads data from a CSV file.

    Args:
        path (os.PathLike):
            The path to the CSV file.
        columns (Optional[Union[None, List[str]]], optional):
            The columns to load. If None, all columns are loaded.
        filters (Optional[Union[None, List[str], List[List[str]]]], optional):
            The filters to apply. Each filter is a tuple of (column, operator, value).
            If filters is a list of tuples, all filters are applied in conjunction.
            If filters is a list of lists of tuples, each list is applied in disjunction.
        **kwargs
            Additional arguments to pass to the pd.read_csv function.

    Returns:
        (pd.DataFrame): The loaded data.
    """
    # Chargement des données
    data = pd.read_csv(path, usecols=columns, **kwargs)
    # Filtrage des données
    if filters is not None:
        data = data.loc[self._build_mask_from_filters(data=data, filters=filters)]
    return data

read_sas 

read_sas(path: PathLike, columns: Optional[Union[None, List[str]]] = None, filters: Optional[Union[None, List[str], List[List[str]]]] = None, **kwargs) -> DataFrame

Loads data from a SAS file.

Parameters:

Name Type Description Default
path PathLike

The path to the SAS file.

 required
columns Optional[Union[None, List[str]]]

The columns to load. If None, all columns are loaded.

 None
filters Optional[Union[None, List[str], List[List[str]]]]

The filters to apply. Each filter is a tuple of (column, operator, value). If filters is a list of tuples, all filters are applied in conjunction. If filters is a list of lists of tuples, each list is applied in disjunction.

 None

Returns:

Type Description
DataFrame

The loaded data.
Source code in bozio_wasmer_simulations/datasets/base.py 
 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
def read_sas(
    self,
    path: os.PathLike,
    columns: Optional[Union[None, List[str]]] = None,
    filters: Optional[Union[None, List[str], List[List[str]]]] = None,
    **kwargs,
) -> pd.DataFrame:
    """
    Loads data from a SAS file.

    Args:
        path (os.PathLike):
            The path to the SAS file.
        columns (Optional[Union[None, List[str]]], optional):
            The columns to load. If None, all columns are loaded.
        filters (Optional[Union[None, List[str], List[List[str]]]], optional):
            The filters to apply. Each filter is a tuple of (column, operator, value).
            If filters is a list of tuples, all filters are applied in conjunction.
            If filters is a list of lists of tuples, each list is applied in disjunction.
        **kwargs
            Additional arguments to pass to the pyreadstat.read_sas7bdat function.

    Returns:
        (pd.DataFrame): The loaded data.
    """
    # Chargement des données
    data, _ = pyreadstat.read_sas7bdat(path, usecols=columns, **kwargs)
    # Filtrage des données
    if filters is not None:
        data = data.loc[self._build_mask_from_filters(data=data, filters=filters)]
    return data