ihkapy.fileio.utils

 1import os
 2
 3# fileio utility methods
 4
 5def get_all_valid_session_basenames(dir_path):
 6    """Returns list of all base-filenames with edf and txt pair."""
 7    valid_basenames = []
 8    for fname in os.listdir(dir_path):
 9        basename,ext = os.path.splitext(fname)
10        if ext == ".edf" and basename+".txt" in os.listdir(dir_path):
11            valid_basenames.append(basename)
12    return valid_basenames
13
14def check_session_basenames_are_valid(basenames,dir_path):
15    """Raises error if edf and txt files not found in dir_path"""
16    if not basenames: print("Warning: no basenames supplied, True by default")
17    for basename in basenames:
18        if basename+".txt" not in os.listdir(dir_path) \
19                or basename+".edf" not in os.listdir(dir_path):
20            return False
21    return True
22
23### FORMAT PATHS
24# All the properly formatted naming conventions
25def fmt_binary_chan_raw_path(
26        wavelet_binaries_path:str,
27        session_basename:str,
28        raw_ch_idx:int):
29    """Format multiplexed wavelet bank binary file."""
30    return os.path.join(wavelet_binaries_path,f"{session_basename}_ch_{str(raw_ch_idx).zfill(3)}.dat")
31
32def fmt_binary_cache_wav_path(
33        cache_path:str, # path to directory of cache
34        session_basename:str,
35        raw_ch_idx:int,
36        amp_phase:str,
37        freq_ch_idx:int=None):
38    """
39    Format path to single channel wavelet binary temporarily cached
40    while computing transform bank.
41    """
42    assert amp_phase in ("RAW","AMP-PHASE"), "The Amplitude / Phase must be denoted AMP or PHASE"
43    if amp_phase == "RAW":
44        fname_raw = f"{session_basename}_ch_{str(raw_ch_idx).zfill(3)}_0RAW.dat"
45        return os.path.join(cache_path,fname_raw)
46        # Above, the zero infront of RAW is intentional, it's for sorting
47        # "0..." < "freqidx..."
48    else:
49        fname_amp = f"{session_basename}_ch_{str(raw_ch_idx).zfill(3)}_freqidx_{str(freq_ch_idx).zfill(2)}_AMP.dat"
50        fname_phase = f"{session_basename}_ch_{str(raw_ch_idx).zfill(3)}_freqidx_{str(freq_ch_idx).zfill(2)}_PHASE.dat"
51        return os.path.join(cache_path,fname_amp),os.path.join(cache_path,fname_phase) 
52
53def fmt_features_df_csv_path(
54        features_path:str,
55        basename:str):
56    """Filepath for csv of features DataFrame of a session."""
57    return os.path.jion(features_path,f"{basename}.csv")
58
59
60
61# TESTS
62if __name__ == "__main__":
63    # Test fmt_binary_chan_raw_path
64    wav_path = "/path/to/wavelet/"
65    sb = "sb"
66    rch_idx = 5
67    assert fmt_binary_chan_raw_path(wav_path,sb,rch_idx) == "/path/to/wavelet/sb_ch_005.dat"
68    print("Test passed: fmt_binary_chan_raw_path()")
69
70    # Test fmt_binary_cache_wav_path()
71    cache_path = "path/to/cache"
72    assert fmt_binary_cache_wav_path(cache_path,sb,4,6,"P") == f"path/to/cache/sb_ch_004_freqidx_06_P.dat"
73    print("Test passed: fmt_binary_cache_wav_path()")
74
75    # Test