Coverage for src/thunderfish/files.py: 14%

1""" Files 

2 

3Helper functions for parsing and reading files, generated by thunderfish. 

4 

5## Functions 

6 

7- `parse_filename()`: parse components of an EOD analysis file name. 

8- `open_from_zip()`: open file from corresponding zip file. 

9- `close_zip()`: close stream and corresponding zip file. 

10 

11""" 

12 

13import io 

14 

15from pathlib import Path 

16from zipfile import ZipFile 

17 

18 

19def parse_filename(file_path): 

20 """Parse components of an EOD analysis file name. 

21 

22 Analysis files generated by the `eodanalysis` module are named 

23 according to 

24 ```plain 

25 PATH/RECORDING-CHANNEL-TIME-FTYPE-N.EXT 

26 ``` 

27 

28 Parameters 

29 ---------- 

30 file_path: str or Path 

31 Path of the file to be parsed. 

32 

33 Returns 

34 ------- 

35 recording: str 

36 Path and basename of the recording, i.e. 'PATH/RECORDING'. 

37 A leading './' is removed. 

38 base_path: Path 

39 Path and basename of the analysis results, 

40 i.e. 'PATH/RECORDING-CHANNEL-TIME'. 

41 channel: int 

42 Channel of the recording 

43 ('CHANNEL' component of the file name if present). 

44 -1 if not present in `file_path`. 

45 time: float 

46 Start time of analysis window in seconds 

47 ('TIME' component of the file name if present). 

48 `None` if not present in `file_path`. 

49 ftype: str 

50 Type of analysis file (e.g. 'wavespectrum', 'pulsephases', etc.), 

51 ('FTYPE' component of the file name if present). 

52 See `file_types` for a list of all supported file types. 

53 Empty string if not present in `file_path`. 

54 index: int 

55 Index of the EOD. 

56 ('N' component of the file name if present). 

57 -1 if not present in `file_path`. 

58 ext: str 

59 File extension *without* leading period 

60 ('EXT' component of the file name). 

61 

62 """ 

63 file_path = Path(file_path) 

64 ext = file_path.suffix 

65 ext = ext[1:] 

66 parts = file_path.stem.split('-') 

67 index = -1 

68 if len(parts) > 0 and parts[-1].isdigit(): 

69 index = int(parts[-1]) 

70 parts = parts[:-1] 

71 ftype = '' 

72 if len(parts) > 0: 

73 ftype = parts[-1] 

74 parts = parts[:-1] 

75 base_path = file_path.parent / '-'.join(parts) 

76 time = None 

77 if len(parts) > 0 and len(parts[-1]) > 0 and \ 

78 parts[-1][0] == 't' and parts[-1][-1] == 's' and \ 

79 parts[-1][1:-1].isdigit(): 

80 time = float(parts[-1][1:-1]) 

81 parts = parts[:-1] 

82 channel = -1 

83 if len(parts) > 0 and len(parts[-1]) > 0 and \ 

84 parts[-1][0] == 'c' and parts[-1][1:].isdigit(): 

85 channel = int(parts[-1][1:]) 

86 parts = parts[:-1] 

87 recording = '-'.join(parts) 

88 return recording, base_path, channel, time, ftype, index, ext 

89 

90 

91def open_from_zip(file_handle): 

92 """Open file from corresponding zip file. 

93  

94 Helper function used by the various load_*() functions 

95 in the waveanalysis and pulseanalysis modules. 

96 

97 Parameters 

98 ---------- 

99 file_handle: str, Path, or stream 

100 The file to be opened for reading. 

101 If str or Path and file does not exist, try to open 

102 possible corresponding zip file for reading. 

103 

104 Returns 

105 ------- 

106 file_handle: str, Path, or stream 

107 The file to be read. If not a stream (e.g. does not have a 

108 readline() function), it still needs to be opened. 

109 If it is a stream from a zip file (`zf` is not None) it needs 

110 to be closed after the file has been read. 

111 zf: ZipFile or None 

112 If zip file was found and opened, this is the zip file handler 

113 that needs to be closed after the file has been read. 

114 """ 

115 zf = None 

116 if not hasattr(file_handle, 'readline') and not Path(file_handle).is_file(): 

117 _, base_path, _, _, _, _, _ = parse_filename(file_handle) 

118 zip_file = base_path.with_suffix('.zip') 

119 if zip_file.is_file(): 

120 zf = ZipFile(zip_file, 'r') 

121 file_pathes = zf.namelist() 

122 if file_handle in file_pathes: 

123 file_handle = io.TextIOWrapper(zf.open(file_handle, 'r')) 

124 else: 

125 raise FileNotFoundError(f'"{file_handle}" is not contained in zip-file "{zip_file}"') 

126 else: 

127 raise FileNotFoundError(f'neither "{file_handle}" nor "{zip_file}" do existis') 

128 return file_handle, zf 

129 

130 

131def close_zip(file_handle, zf): 

132 """Close stream and corresponding zip file. 

133 

134 Helper function used by the various load_*() functions 

135 in the waveanalysis and pulseanalysis modules. 

136 

137 Parameters 

138 ---------- 

139 file_handle: str, Path, or stream 

140 If it is a stream from a zip file (`zf` is not None) 

141 it is closed. 

142 zf: ZipFile or None 

143 If non None, this zip file is closed. 

144 """ 

145 if zf is not None: 

146 file_handle.close() 

147 zf.close() 

148