Skip to content

Preparing

Functions to be used for preparing the experimental data for batch processing.

check_column_headers(data_dir, exception_headers=None)

Check that all files in a directory have the same column headers and that column headers don't contain spaces. A ValueError will be raised if the column headers don't match or if a column header contains a space.

Parameters:

Name Type Description Default
data_dir str

Path to the directory containing the files to be checked.

 required
exception_headers List[str]

List of column headers that are allowed to be different between files.

 None
Source code in paramaterial/preparing.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def check_column_headers(data_dir: str, exception_headers: List[str] = None):
    """Check that all files in a directory have the same column headers and that column headers don't contain spaces.
    A ValueError will be raised if the column headers don't match or if a column header contains a space.

    Args:
        data_dir: Path to the directory containing the files to be checked.
        exception_headers: List of column headers that are allowed to be different between files.
    """
    file_list = os.listdir(data_dir)
    first_file = pd.read_csv(f'{data_dir}/{file_list[0]}')
    first_columns = first_file.columns
    if exception_headers is not None:
        for exception_header in exception_headers:
            if exception_header in first_columns:
                first_columns = first_columns.drop(exception_header)
    print("Checking column headers...")
    for column_header in first_file.columns:
        if len(column_header.split(' ')) > 1:
            raise ValueError(f'Column header "{column_header}" contains a space.')
    print(f'First file headers:\n\t{list(first_file.columns)}')
    for file in file_list[1:]:
        df = pd.read_csv(f'{data_dir}/{file}')
        df_columns = df.columns
        if exception_headers is not None:
            for exception_header in exception_headers:
                if exception_header in df_columns:
                    df_columns = df_columns.drop(exception_header)
        if not df_columns.equals(first_columns):
            raise ValueError(f'Column headers in {file} don\'t match column headers of first file.'
                             f'{file} headers:\n\t{list(df.columns)}')
    print(f'Headers in all files are the same as in the first file, except for {exception_headers}.')

check_for_duplicate_files(data_dir)

Check that there are no duplicate files in a directory by hashing the contents of the files. A ValueError will be raised if there are duplicate files.

Parameters:

Name Type Description Default
data_dir str

Path to the directory containing the files to be checked.

 required
Source code in paramaterial/preparing.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
def check_for_duplicate_files(data_dir: str):
    """Check that there are no duplicate files in a directory by hashing the contents of the files.
    A ValueError will be raised if there are duplicate files.

    Args:
        data_dir: Path to the directory containing the files to be checked.
    """
    print('Checking for duplicate files...')
    hashes = [hash(open(f'{data_dir}/{file}', 'rb').read()) for file in os.listdir(data_dir)]
    if len(hashes) != len(set(hashes)):
        duplicates = [file for file, filehash in zip(os.listdir(data_dir), hashes) if hashes.count(filehash) > 1]
        raise ValueError(f'There are duplicate files in {data_dir}.\n'
                         'The duplicates are:' + '\n\t'.join(duplicates))
    else:
        print(f'No duplicate files found in "{data_dir}".')

check_formatting(data_dir, info_path=None, skip_headers=None)

Performs various checks on the formatting of the data files and info table. Will print out formatting checks and warnings.

Parameters:

Name Type Description Default
data_dir Union[str, DataSet]

Path to folder containing data files. A DataSet is also accepted, as in paramaterial 0.1.x, in which case the 0.1.x checks are run on its data_dir (check_column_headers and check_for_duplicate_files).

 required
info_path str

Path to info table spreadsheet.

 None
skip_headers List[str]

Headers that can be different across data files. Other headers must be consistent.

 None
Source code in paramaterial/preparing.py 
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
def check_formatting(data_dir: Union[str, DataSet], info_path: str = None, skip_headers: List[str] = None):
    """Performs various checks on the formatting of the data files and info table.
    Will print out formatting checks and warnings.

    Args:
        data_dir: Path to folder containing data files. A DataSet is also accepted, as in paramaterial 0.1.x, in which
            case the 0.1.x checks are run on its data_dir (`check_column_headers` and `check_for_duplicate_files`).
        info_path: Path to info table spreadsheet.
        skip_headers: Headers that can be different across data files. Other headers must be consistent.
    """
    if isinstance(data_dir, DataSet):
        check_column_headers(data_dir.data_dir)
        check_for_duplicate_files(data_dir.data_dir)
        return

    # check column headers
    file_list = os.listdir(data_dir)
    first_file = pd.read_csv(f'{data_dir}/{file_list[0]}')
    first_columns = first_file.columns

    if skip_headers is not None:
        for exception_header in skip_headers:
            if exception_header in first_columns:
                first_columns = first_columns.drop(exception_header)

    mismatch = False
    for file in file_list[1:]:
        df = pd.read_csv(f'{data_dir}/{file}')
        df_columns = df.columns

        if skip_headers is not None:
            for exception_header in skip_headers:
                if exception_header in df_columns:
                    df_columns = df_columns.drop(exception_header)

        if not df_columns.equals(first_columns):
            print(f'Column headers in {file} don\'t match column headers of first file.'
                  f'{file} headers:\n\t{list(df.columns)}')
            mismatch = True

    if not mismatch and skip_headers is not None:
        print(f'Headers, except for {skip_headers}, are the same in all files as in the first file.')
        print('Consistent headers are: ' + ', '.join([col for col in first_columns if col not in skip_headers]))

    elif not mismatch and skip_headers is None:
        print(f'Headers are the same in all files as in the first file.')
        print(f'Consistent headers are: ' + ', '.join([col for col in first_columns]))

    # check for duplicate files
    hashes = [hash(open(f'{data_dir}/{file}', 'rb').read()) for file in os.listdir(data_dir)]
    if len(hashes) != len(set(hashes)):
        duplicates = [file for file, filehash in zip(os.listdir(data_dir), hashes) if hashes.count(filehash) > 1]
        raise ValueError(f'Duplicate files found in {data_dir} by hashing contents.\n'
                         'The duplicates are:' + '\n\t'.join(duplicates))
    else:
        print(f'No duplicate files (by hashing contents) found in "{data_dir}".')

convert_gleeble_output_files_to_csv(directory_path)

Convert all files in a directory from Gleeble output format to csv format.

Source code in paramaterial/preparing.py 
277
278
279
280
281
282
283
284
def convert_gleeble_output_files_to_csv(directory_path: str):
    """Convert all files in a directory from Gleeble output format to csv format."""
    for file in os.listdir(directory_path):
        if not file.endswith('.csv'):
            df = pd.read_csv(f'{directory_path}/{file}', header=[0, 1], delimiter='\t')
            df.columns = \
                [col[0] if str(col[1]).startswith('Unnamed') else ' '.join(col).strip() for col in df.columns]
            df.to_csv(f'{directory_path}/{file[:-4]}.csv', index=False)

copy_data_and_rename_by_test_id(data_in, data_out, info_table, test_id_col='test_id')

Rename files in data directory by test_id in info table and copy to new directory. The info_table must have a column named 'old_filename' containing the original filenames and a column named 'test_id'. The new filenames will be the test_ids with the extension '.csv'.

Parameters:

Name Type Description Default
data_in str

Path to the directory containing the data to be copied.

 required
data_out str

Path to the directory where the data will be copied.

 required
info_table DataFrame

DataFrame containing the metadata for the tests.

 required
test_id_col

Column in the info table containing the test_ids.

 'test_id'

Returns:

Type Description

None
Source code in paramaterial/preparing.py 
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
def copy_data_and_rename_by_test_id(data_in: str, data_out: str, info_table: pd.DataFrame, test_id_col='test_id'):
    """Rename files in data directory by test_id in info table and copy to new directory. The info_table must have a column
    named 'old_filename' containing the original filenames and a column named 'test_id'. The new filenames will be the
    test_ids with the extension '.csv'.

    Args:
        data_in: Path to the directory containing the data to be copied.
        data_out: Path to the directory where the data will be copied.
        info_table: DataFrame containing the metadata for the tests.
        test_id_col: Column in the info table containing the test_ids.

    Returns:
        None
    """
    # make data directory if it doesn't exist
    if not os.path.exists(data_out):
        os.mkdir(data_out)

    # check info table
    if 'old_filename' not in info_table.columns:
        raise ValueError(f'There is no "old_filename" column in the info table.')
    if test_id_col not in info_table.columns:
        raise ValueError(f'There is no "{test_id_col}" column in the info table.')
    if info_table[test_id_col].duplicated().any():
        raise ValueError(f'There are duplicate test_ids.')
    if info_table['old_filename'].duplicated().any():
        raise ValueError(f'There are duplicate old_filenames.')

    for filename, test_id in zip(info_table['old_filename'], info_table[test_id_col]):
        # check that file exists
        if not os.path.exists(f'{data_in}/{filename}'):
            raise FileNotFoundError(f'File {filename} does not exist in {data_in}.')
        # copy and rename file
        shutil.copy(f'{data_in}/{filename}', f'{data_out}/{test_id}.csv')

    print(f'Copied {len(info_table)} files in {data_in} to {data_out}.')

experimental_matrix(info_table, index, columns, as_heatmap=False, title=None, xlabel=None, ylabel=None, tick_params=None, **kwargs)

Make an experimental matrix showing the distribution of test across metadata categories.

Parameters:

Name Type Description Default
info_table DataFrame

DataFrame containing the metadata for the tests.

 required
index Union[str, List[str]]

Column(s) of the info_table to use as the index of the matrix.

 required
columns Union[str, List[str]]

Column(s) of the info_table to use as the columns of the matrix.

 required
as_heatmap bool

If True, return a heatmap of the matrix. If False, return the matrix as a DataFrame.

 False
title str

Title of the heatmap.

 None
xlabel str

Label for the x-axis of the heatmap.

 None
ylabel str

Label for the y-axis of the heatmap.

 None
tick_params Dict

Parameters to pass to the ax.tick_params method of matplotlib.

 None
**kwargs

Additional keyword arguments to pass to the heatmap function.

 {}

Returns:

Type Description
Union[DataFrame, Axes]

If as_heatmap is False, returns a DataFrame of the experimental matrix.
Union[DataFrame, Axes]

If as_heatmap is True, returns a heatmap of the experimental matrix.
Source code in paramaterial/preparing.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
def experimental_matrix(info_table: pd.DataFrame, index: Union[str, List[str]], columns: Union[str, List[str]],
                        as_heatmap: bool = False, title: str = None, xlabel: str = None,
                        ylabel: str = None, tick_params: Dict = None, **kwargs) -> Union[pd.DataFrame, plt.Axes]:
    """Make an experimental matrix showing the distribution of test across metadata categories.

    Args:
        info_table: DataFrame containing the metadata for the tests.
        index: Column(s) of the info_table to use as the index of the matrix.
        columns: Column(s) of the info_table to use as the columns of the matrix.
        as_heatmap: If True, return a heatmap of the matrix. If False, return the matrix as a DataFrame.
        title: Title of the heatmap.
        xlabel: Label for the x-axis of the heatmap.
        ylabel: Label for the y-axis of the heatmap.
        tick_params: Parameters to pass to the ax.tick_params method of matplotlib.
        **kwargs: Additional keyword arguments to pass to the heatmap function.

    Returns:
        If as_heatmap is False, returns a DataFrame of the experimental matrix.
        If as_heatmap is True, returns a heatmap of the experimental matrix.
    """
    # make sure index and columns are lists
    if isinstance(index, str):
        index = [index]
    if isinstance(columns, str):
        columns = [columns]

    # make the experimental matrix
    exp_matrix = info_table.groupby(index + columns).size().unstack(columns).fillna(0).astype(int)

    # return the matrix as a DataFrame if as_heatmap is False
    if not as_heatmap:
        return exp_matrix

    # update the default kwargs
    default_kwargs = dict(linewidths=2, cbar=False, annot=True, square=True, cmap='Blues')
    default_kwargs.update(kwargs)

    ax = sns.heatmap(exp_matrix, **default_kwargs)
    if title:
        ax.set_title(title)
    if xlabel:
        ax.set_xlabel(xlabel)
    if ylabel:
        ax.set_ylabel(ylabel)
    if tick_params:
        ax.tick_params(**tick_params)
    return ax

format_data_file_contents(raw_data_dir, prepared_data_dir, header_rows=None, delimiter=',', header_rename_dict=None)

Format CSV files in a directory by combining column headers into one row and changing delimeter to a comma.

Args: - directory_path: Path to the directory containing the original data files. - formatted_directory_path: Path to the directory where formatted files will be saved. - rename_dict: Dictionary mapping original column names to new column names. - header_rows: List of rows to use as headers (0-indexed). Rows above the highest-indexed row will be dropped. The original headers will be combined into a single row in the formatted files. - delimiter: Delimiter used in the CSV file. This will be changed to a comma in the formatted files.

Example usage:

format_data_file_contents('data/00 raw data', 'data/01 prepared data', header_rename_dict={'Total Strain': 'Strain', 'True Stress': 'Stress'}, header_rows=[0, 1], delimiter=' ')

Source code in paramaterial/preparing.py 
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
def format_data_file_contents(raw_data_dir: str, prepared_data_dir: str, header_rows=None,
                              delimiter: str = ',', header_rename_dict: Dict[str, str] = None):
    """
    Format CSV files in a directory by combining column headers into one row and changing delimeter to a comma.

    Args:
    - directory_path: Path to the directory containing the original data files.
    - formatted_directory_path: Path to the directory where formatted files will be saved.
    - rename_dict: Dictionary mapping original column names to new column names.
    - header_rows: List of rows to use as headers (0-indexed). Rows above the highest-indexed row will be
    dropped. The original headers will be combined into a single row in the formatted files.
    - delimiter: Delimiter used in the CSV file. This will be changed to a comma in the formatted files.

    Example usage:
    >>> format_data_file_contents('data/00 raw data', 'data/01 prepared data', header_rename_dict={'Total Strain':
    'Strain', 'True Stress': 'Stress'}, header_rows=[0, 1], delimiter='\t')
    """
    # Default header rows to the first row
    if header_rows is None:
        header_rows = [0]

    # Check if the directory exists
    if not os.path.exists(raw_data_dir):
        raise ValueError(f'Directory not found: {raw_data_dir}')

    # Iterate through each file in the directory
    for file in os.listdir(raw_data_dir):

        if file.endswith('.csv'):
            # Read the CSV file with specified headers
            file_path = os.path.join(raw_data_dir, file)
            df = pd.read_csv(file_path, header=header_rows, delimiter=delimiter)

            # Combine multi-level columns and rename as specified
            # blank header cells come back from pandas as 'Unnamed: ...' and are left out of the combined header
            new_headers = [' '.join(str(c) for c in col if not str(c).startswith('Unnamed')) if isinstance(col, tuple)
                           else str(col) for col in df.columns]
            rename = header_rename_dict or {}
            df.columns = [rename.get(col, col) for col in new_headers]

            # Save the formatted DataFrame to the new directory
            new_file_path = os.path.join(prepared_data_dir, file)
            df.to_csv(new_file_path, index=False)

rename_data_files(data_dir, rename_table, old_filename_col='old_filename', test_id_col='test_id')

Rename data files in a directory based on test ID mappings provided in an info table.

Args: - directory_path (str): Path to the directory containing the data files to be renamed. - info_table (pd.DataFrame): DataFrame containing the metadata for the tests. - old_filename_col (str): Column name in the info table for the old filenames. - test_id_col (str): Column name in the info table for the test IDs.

Example usage:

rename_data_files(data_dir='data/01 prepared data', rename_table='info/00 rename list.xlsx', old_filename_col='old filename', test_id_col='test id')

Source code in paramaterial/preparing.py 
 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
def rename_data_files(data_dir: str, rename_table: Union[pd.DataFrame, str],
                      old_filename_col='old_filename', test_id_col='test_id'):
    """
    Rename data files in a directory based on test ID mappings provided in an info table.

    Args:
    - directory_path (str): Path to the directory containing the data files to be renamed.
    - info_table (pd.DataFrame): DataFrame containing the metadata for the tests.
    - old_filename_col (str): Column name in the info table for the old filenames.
    - test_id_col (str): Column name in the info table for the test IDs.

    Example usage:
    >>> rename_data_files(data_dir='data/01 prepared data', rename_table='info/00 rename list.xlsx',
    old_filename_col='old filename', test_id_col='test id')
    """
    # Read the info table if it is a string
    if type(rename_table) == str and rename_table.endswith('.xlsx'):
        rename_table = pd.read_excel(rename_table)
    elif type(rename_table) == str and rename_table.endswith('.csv'):
        rename_table = pd.read_csv(rename_table)
    else:
        assert isinstance(rename_table, pd.DataFrame)

    # Check rename table
    if old_filename_col not in rename_table.columns:
        raise ValueError(f'There is no "{old_filename_col}" column in the rename table.')
    if test_id_col not in rename_table.columns:
        raise ValueError(f'There is no "{test_id_col}" column in the rename table.')
    if rename_table[test_id_col].duplicated().any():
        raise ValueError(f'There are duplicate test_ids.')
    if rename_table[old_filename_col].duplicated().any():
        raise ValueError(f'There are duplicate old_filenames.')

    # Iterate over the rows in the info table
    for _, row in rename_table.iterrows():
        old_filename = row[old_filename_col] + '.csv'
        new_filename = f'{row[test_id_col]}.csv'
        old_filepath = os.path.join(data_dir, old_filename)
        new_filepath = os.path.join(data_dir, new_filename)

        # Rename the file if old file exists, and replace new file if it already exists
        try:
            if os.path.exists(old_filepath):
                os.rename(old_filepath, new_filepath)
                print(f'Renamed {old_filepath} to {new_filepath}.')
            else:
                print(f"File not found: {old_filepath}")
        except FileExistsError:
            os.remove(new_filepath)
            os.rename(old_filepath, new_filepath)
            print(f'Renamed {old_filepath} to {new_filepath} and replaced {new_filepath}.')