Coverage for C: \ Users \ peaco \ OneDrive \ Documents \ GitHub \ mth5 \ mth5 \ clients \ base.py: 100%

1# -*- coding: utf-8 -*- 

2""" 

3Created on Fri Oct 11 11:36:26 2024 

4 

5@author: jpeacock 

6""" 

7 

8# ============================================================================= 

9# Imports 

10# ============================================================================= 

11from pathlib import Path 

12from typing import Any, Optional, Sequence, Union 

13 

14from loguru import logger 

15 

16 

17# ============================================================================= 

18 

19 

20class ClientBase: 

21 def __init__( 

22 self, 

23 data_path: Union[str, Path], 

24 sample_rates: Sequence[float] = [1], 

25 save_path: Optional[Union[str, Path]] = None, 

26 mth5_filename: str = "from_client.h5", 

27 **kwargs: Any, 

28 ) -> None: 

29 """ 

30 Initialize the ClientBase. 

31 

32 Any h5 parameters should be in kwargs as `h5_parameter_name`. 

33 

34 Parameters 

35 ---------- 

36 data_path : str or Path 

37 Directory where data files are located. 

38 sample_rates : Sequence[float], optional 

39 List of sample rates to look for. Default is [1]. 

40 save_path : str or Path, optional 

41 Directory to save the mth5 file. If None, uses data_path. 

42 mth5_filename : str, optional 

43 Name of the mth5 file to create. Default is 'from_client.h5'. 

44 **kwargs : Any 

45 Additional keyword arguments for h5 parameters. 

46 

47 Examples 

48 -------- 

49 >>> client = ClientBase(data_path="./data", sample_rates=[1, 8, 256]) 

50 >>> client.save_path 

51 PosixPath('data/from_client.h5') 

52 """ 

53 self.logger = logger 

54 

55 self.data_path = data_path 

56 self.mth5_filename = mth5_filename 

57 self.sample_rates = sample_rates 

58 self.save_path = save_path 

59 

60 self.interact = False 

61 

62 self.mth5_version = "0.2.0" 

63 self.h5_compression = "gzip" 

64 self.h5_compression_opts = 4 

65 self.h5_shuffle = True 

66 self.h5_fletcher32 = True 

67 self.h5_data_level = 1 

68 self.mth5_file_mode = "w" 

69 

70 self.collection: Optional[Any] = None 

71 

72 for key, value in kwargs.items(): 

73 if value is not None: 

74 setattr(self, key, value) 

75 

76 @property 

77 def h5_kwargs(self) -> dict[str, Any]: 

78 """ 

79 Dictionary of HDF5 keyword arguments for file creation. 

80 

81 Returns 

82 ------- 

83 dict 

84 Dictionary of HDF5 file creation parameters. 

85 

86 Examples 

87 -------- 

88 >>> client = ClientBase(data_path="./data") 

89 >>> client.h5_kwargs["compression"] 

90 'gzip' 

91 """ 

92 h5_params = dict( 

93 file_version=self.mth5_version, 

94 compression=self.h5_compression, 

95 compression_opts=self.h5_compression_opts, 

96 shuffle=self.h5_shuffle, 

97 fletcher32=self.h5_fletcher32, 

98 data_level=self.h5_data_level, 

99 ) 

100 

101 for key, value in self.__dict__.items(): 

102 if key.startswith("h5"): 

103 h5_params[key[3:]] = value 

104 

105 return h5_params 

106 

107 @property 

108 def data_path(self) -> Path: 

109 """ 

110 Path to data directory. 

111 

112 Returns 

113 ------- 

114 Path 

115 Path to the data directory. 

116 

117 Examples 

118 -------- 

119 >>> client = ClientBase(data_path="./data") 

120 >>> client.data_path 

121 PosixPath('data') 

122 """ 

123 return self._data_path 

124 

125 @data_path.setter 

126 def data_path(self, value: Union[str, Path]) -> None: 

127 """ 

128 Set the data path. 

129 

130 Parameters 

131 ---------- 

132 value : str or Path 

133 Directory where files are located. 

134 

135 Raises 

136 ------ 

137 IOError 

138 If the path does not exist. 

139 ValueError 

140 If value is None. 

141 

142 Examples 

143 -------- 

144 >>> client = ClientBase(data_path="./data") 

145 >>> client.data_path 

146 PosixPath('data') 

147 """ 

148 if value is not None: 

149 self._data_path = Path(value) 

150 if not self._data_path.exists(): 

151 raise IOError(f"Could not find {self._data_path}") 

152 else: 

153 raise ValueError("data_path cannot be None") 

154 

155 @property 

156 def sample_rates(self) -> list[float]: 

157 """ 

158 List of sample rates to look for. 

159 

160 Returns 

161 ------- 

162 list of float 

163 Sample rates. 

164 

165 Examples 

166 -------- 

167 >>> client = ClientBase(data_path="./data", sample_rates=[1, 8, 256]) 

168 >>> client.sample_rates 

169 [1.0, 8.0, 256.0] 

170 """ 

171 return self._sample_rates 

172 

173 @sample_rates.setter 

174 def sample_rates(self, value: Union[int, float, str, Sequence[float]]) -> None: 

175 """ 

176 Set the sample rates as a list of floats. 

177 

178 Parameters 

179 ---------- 

180 value : int, float, str, or Sequence[float] 

181 Sample rates to set. If str, should be comma-separated. 

182 

183 Raises 

184 ------ 

185 TypeError 

186 If value cannot be parsed. 

187 

188 Examples 

189 -------- 

190 >>> client = ClientBase(data_path="./data", sample_rates="1,8,256") 

191 >>> client.sample_rates 

192 [1.0, 8.0, 256.0] 

193 """ 

194 if isinstance(value, (int, float)): 

195 self._sample_rates = [float(value)] 

196 elif isinstance(value, str): 

197 self._sample_rates = [float(v) for v in value.split(",")] 

198 elif isinstance(value, (tuple, list)): 

199 self._sample_rates = [float(v) for v in value] 

200 else: 

201 raise TypeError(f"Cannot parse {type(value)}") 

202 

203 @property 

204 def save_path(self) -> Path: 

205 """ 

206 Path to save the mth5 file. 

207 

208 Returns 

209 ------- 

210 Path 

211 Full path to the mth5 file. 

212 

213 Examples 

214 -------- 

215 >>> client = ClientBase(data_path="./data", save_path="./output", mth5_filename="test.h5") 

216 >>> client.save_path 

217 PosixPath('output/test.h5') 

218 """ 

219 return self._save_path.joinpath(self.mth5_filename) 

220 

221 @save_path.setter 

222 def save_path(self, value: Optional[Union[str, Path]]) -> None: 

223 """ 

224 Set the path to save the mth5 file. 

225 

226 Parameters 

227 ---------- 

228 value : str or Path, optional 

229 Directory or file path to save the mth5 file. If None, uses data_path. 

230 

231 Examples 

232 -------- 

233 >>> client = ClientBase(data_path="./data", save_path="./output/test.h5") 

234 >>> client.save_path 

235 PosixPath('output/test.h5') 

236 """ 

237 if value is not None: 

238 value = Path(value) 

239 if value.exists(): 

240 if value.is_dir(): 

241 self._save_path = value 

242 elif value.is_file(): 

243 self._save_path = value.parent 

244 self.mth5_filename = value.name 

245 else: 

246 if "." in value.name: 

247 self._save_path = value.parent 

248 self.mth5_filename = value.name 

249 else: 

250 self._save_path = value 

251 self._save_path.mkdir(exist_ok=True) 

252 else: 

253 self._save_path = self.data_path 

254 

255 def get_run_dict(self) -> Any: 

256 """ 

257 Get run information from the collection. 

258 

259 Returns 

260 ------- 

261 Any 

262 Run information as returned by the collection's get_runs method. 

263 

264 Examples 

265 -------- 

266 >>> client = ClientBase(data_path="./data") 

267 >>> client.collection = ... # assign a collection with get_runs method 

268 >>> client.get_run_dict() 

269 ... 

270 """ 

271 

272 if self.collection is None: 

273 raise ValueError("Collection is not set.") 

274 return self.collection.get_runs(sample_rates=self.sample_rates)