Coverage for labnirs2snirf / snirf.py: 100%

1""" 

2Functions related to writing NIRS data to SNIRF files using HDF5. 

3""" 

4 

5import logging 

6from pathlib import Path 

7 

8import h5py # type: ignore 

9import numpy as np 

10 

11from . import model 

12from .error import Labnirs2SnirfError 

13 

14 

15class SnirfWriteError(Labnirs2SnirfError): 

16 """Custom error class for SNIRF writing errors.""" 

17 

18 

19log = logging.getLogger(__name__) 

20 

21 

22def write_snirf(nirs: model.Nirs, output_file: Path) -> None: 

23 """ 

24 Write the NIRS data to a SNIRF file. 

25 

26 Parameters 

27 ---------- 

28 nirs : model.Nirs 

29 NIRS data model containing metadata, data, stim and probe information. 

30 output_file : Path 

31 Path to the output SNIRF file. A warning is shown if the file does not 

32 have a ".snirf" extension. 

33 

34 Notes 

35 ----- 

36 - Stimulus onsets are available, but durations, pre, and post rest periods 

37 are not stored in the data exported by LABNIRS. As such, onsets are written 

38 correctly, durations are set to 0, and rest periods are not included. 

39 These timings are stored in a separate .csv file by LABNIRS, reading which 

40 may be implemented in the future. 

41 - While the SNIRF specification allows for both indexed and non-indexed nirs 

42 groups (e.g. /nirs, /nirs1, /nirs2, ...), some tools (e.g. MNE) only accept 

43 a single non-indexed /nirs group. Therefore, we always write it as such. 

44 - These tools (may) have other restrictions as well, such as expecting only 

45 a single data group (/nirs/data1), but as far as I am aware, this is not an 

46 issue for LABNIRS data. 

47 """ 

48 

49 SPECS_FORMAT_VERSION = "1.1" 

50 

51 # Warn about incorrect file extension, but continue 

52 if output_file.suffix != ".snirf": 

53 log.warning("Output file doesn't have the .snirf extension: %s", output_file) 

54 

55 log.debug("Writing SNIRF file: %s", output_file) 

56 with h5py.File(output_file, "w") as f: 

57 log.debug("Created HDF5 file, writing SNIRF structure") 

58 # While the SNIRF specification expects the nirs group to be indexed (/nirs1), 

59 # the use of a non-indexed entry is also allowed if there's only one. 

60 # Some tools (e.g. MNE) only accept a non-indexed /nirs. 

61 nirs_group = f.create_group("/nirs") 

62 f.create_dataset("formatVersion", data=_str_encode(SPECS_FORMAT_VERSION)) 

63 _write_metadata_group(nirs.metadata, nirs_group.create_group("metaDataTags")) 

64 _write_data_group(nirs.data[0], nirs_group.create_group("data1")) 

65 _write_probe_group(nirs.probe, nirs_group.create_group("probe")) 

66 if nirs.stim is not None: 

67 _write_stim_group(nirs.stim, nirs_group) 

68 else: 

69 log.debug("No stimulus data to write") 

70 log.debug("SNIRF file write completed") 

71 

72 

73def _str_encode(s: str) -> bytes: 

74 """ 

75 Encode a string to bytes using UTF-8 encoding. 

76 

77 Parameters 

78 ---------- 

79 s : str 

80 String to encode. 

81 

82 Returns 

83 ------- 

84 bytes 

85 UTF-8 encoded bytes representation of the string. 

86 """ 

87 return s.encode("utf-8") 

88 

89 

90def _write_metadata_group(metadata: model.Metadata, group: h5py.Group) -> None: 

91 """ 

92 Write metadata to a HDF5 group following SNIRF specification. 

93 

94 Parameters 

95 ---------- 

96 metadata : model.Metadata 

97 Metadata object containing subject ID, measurement date/time, units, and additional fields. 

98 group : h5py.Group 

99 HDF5 group where metadata will be written (must be /nirs/metaDataTags). 

100 

101 Raises 

102 ------ 

103 SnirfWriteError 

104 If the group path is not /nirs/metaDataTags. 

105 """ 

106 log.info("Writing metadata entries into %s", group.name) 

107 if group.name != "/nirs/metaDataTags": 

108 log.error("Metadata group must be at /nirs/metaDataTags, got %s", group.name) 

109 raise SnirfWriteError("Metadata group must be at /nirs/metaDataTags") 

110 

111 group.create_dataset("SubjectID", data=_str_encode(metadata.SubjectID)) 

112 group.create_dataset("MeasurementDate", data=_str_encode(metadata.MeasurementDate)) 

113 group.create_dataset("MeasurementTime", data=_str_encode(metadata.MeasurementTime)) 

114 group.create_dataset("LengthUnit", data=_str_encode(metadata.LengthUnit)) 

115 group.create_dataset("TimeUnit", data=_str_encode(metadata.TimeUnit)) 

116 group.create_dataset("FrequencyUnit", data=_str_encode(metadata.FrequencyUnit)) 

117 for field, text in metadata.additional_fields.items(): 

118 group.create_dataset(field, data=_str_encode(text)) 

119 

120 

121def _write_data_group(data: model.Data, group: h5py.Group) -> None: 

122 """ 

123 Write experimental data to a HDF5 group following SNIRF specification. 

124 

125 Parameters 

126 ---------- 

127 data : model.Data 

128 Data object containing time, data time series, and measurement list. 

129 group : h5py.Group 

130 HDF5 group where data will be written (must be /nirs/data1). 

131 

132 Raises 

133 ------ 

134 SnirfWriteError 

135 If the group path is not /nirs/data1. 

136 """ 

137 log.info("Writing data entries into %s", group.name) 

138 if group.name != "/nirs/data1": 

139 log.error("Data group must be at /nirs/data1, got %s", group.name) 

140 raise SnirfWriteError("Data group must be at /nirs/data1") 

141 

142 log.debug("Writing time with %d points", len(data.time)) 

143 group.create_dataset("time", data=data.time, compression="gzip") 

144 log.debug("Writing dataTimeSeries with shape %s", data.dataTimeSeries.shape) 

145 group.create_dataset("dataTimeSeries", data=data.dataTimeSeries, compression="gzip") 

146 log.debug("Writing measurementList with %d entries", len(data.measurementList)) 

147 for i, row in enumerate(data.measurementList, start=1): 

148 ml = group.create_group(f"measurementList{i}") 

149 ml.create_dataset("sourceIndex", data=row.sourceIndex, dtype="int32") 

150 ml.create_dataset("detectorIndex", data=row.detectorIndex, dtype="int32") 

151 ml.create_dataset("wavelengthIndex", data=row.wavelengthIndex, dtype="int32") 

152 ml.create_dataset("dataType", data=row.dataType, dtype="int32") 

153 ml.create_dataset("dataTypeIndex", data=row.dataTypeIndex, dtype="int32") 

154 if row.dataTypeLabel is not None: 

155 ml.create_dataset("dataTypeLabel", data=_str_encode(row.dataTypeLabel)) 

156 

157 

158def _write_stim_group(stims: list[model.Stim], group: h5py.Group) -> None: 

159 """ 

160 Write stimulus information to HDF5 groups following SNIRF specification. 

161 

162 Parameters 

163 ---------- 

164 stims : list[model.Stim] 

165 List of Stim objects containing stimulus names and onset times. 

166 group : h5py.Group 

167 HDF5 group where stimuli will be written (must be /nirs). 

168 

169 Raises 

170 ------ 

171 SnirfWriteError 

172 If the group path is not /nirs. 

173 

174 Notes 

175 ----- 

176 Stimulus data is written as Nx3 arrays where N is the number of events. 

177 Column 0 contains onset times, column 2 contains amplitude (set to 1), 

178 and column 1 (duration) is set to 0 as duration information is not available. 

179 """ 

180 log.info("Writing stimulus entries into %s", group.name) 

181 if group.name != "/nirs": 

182 log.error("Stimulus group must be at /nirs, got %s", group.name) 

183 raise SnirfWriteError("Stimulus group must be at /nirs") 

184 

185 for i, stim in enumerate(stims, start=1): 

186 st = group.create_group(f"stim{i}") 

187 log.debug( 

188 "Writing stimulus %02d/%02d: %s with %d data points into %s", 

189 i, 

190 len(stims), 

191 stim.name, 

192 len(stim.data), 

193 st.name, 

194 ) 

195 st.create_dataset("name", data=_str_encode(stim.name)) 

196 d = np.zeros((len(stim.data), 3), dtype=stim.data.dtype) 

197 d[:, 0] = stim.data 

198 d[:, 2] = 1 

199 st.create_dataset("data", data=d, compression="gzip") 

200 

201 

202def _write_probe_group(probe: model.Probe, group: h5py.Group) -> None: 

203 """ 

204 Write probe information to a HDF5 group following SNIRF specification. 

205 

206 Parameters 

207 ---------- 

208 probe : model.Probe 

209 Probe object containing wavelengths, source/detector positions, and labels. 

210 group : h5py.Group 

211 HDF5 group where probe information will be written (must be /nirs/probe). 

212 

213 Raises 

214 ------ 

215 SnirfWriteError 

216 If the group path is not /nirs/probe. 

217 """ 

218 log.info("Writing probe information into %s", group.name) 

219 if group.name != "/nirs/probe": 

220 log.error("Probe group must be at /nirs/probe, got %s", group.name) 

221 raise SnirfWriteError("Probe group must be at /nirs/probe") 

222 

223 log.debug("Writing %d wavelengths", len(probe.wavelengths)) 

224 group.create_dataset("wavelengths", data=probe.wavelengths) 

225 

226 log.debug("Writing source positions with shape %s", probe.sourcePos3D.shape) 

227 group.create_dataset("sourcePos3D", data=probe.sourcePos3D, compression="gzip") 

228 

229 log.debug("Writing detector positions with shape %s", probe.detectorPos3D.shape) 

230 group.create_dataset("detectorPos3D", data=probe.detectorPos3D, compression="gzip") 

231 

232 if probe.sourceLabels is not None: 

233 log.debug("Writing %d source labels", len(probe.sourceLabels)) 

234 group.create_dataset( 

235 "sourceLabels", 

236 data=np.array( 

237 probe.sourceLabels, 

238 dtype=h5py.string_dtype(encoding="utf-8"), 

239 ), 

240 ) 

241 else: 

242 log.debug("No source labels to write") 

243 if probe.detectorLabels is not None: 

244 log.debug("Writing %d detector labels", len(probe.detectorLabels)) 

245 group.create_dataset( 

246 "detectorLabels", 

247 data=np.array( 

248 probe.detectorLabels, 

249 dtype=h5py.string_dtype(encoding="utf-8"), 

250 ), 

251 ) 

252 else: 

253 log.debug("No detector labels to write")