17
18
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
62
63
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
115
116
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 | class fileLoader_abf(fileLoader_base):
"""Load ABF metadata eagerly and raw sweep arrays only when requested."""
loadFileType = ".abf"
def loadFile(self) -> None:
"""Load ABF headers and optionally materialize raw sweep arrays."""
self._loadAbf(loadData=self._loadData)
def _loadAbf(
self,
byteStream: Any | None = None,
loadData: bool = True,
stimulusFileFolder: str | None = None,
) -> None:
"""Load ABF metadata and optionally its raw sweeps.
Args:
byteStream: Optional pyABF-compatible byte stream.
loadData: Whether to materialize raw sweep arrays.
stimulusFileFolder: Optional alternate folder for stimulus files.
"""
try:
if byteStream is not None:
logger.info("Loading byte stream")
self._abf = pyabf.ABF(byteStream, loadData=loadData)
self._isBytesIO = True
else:
self._abf = pyabf.ABF(
self.filepath,
loadData=loadData,
stimulusFileFolder=stimulusFileFolder,
)
except NotImplementedError as error:
logger.error(f" did not load abf file: {self.filepath}")
logger.error(f" NotImplementedError exception was: {error}")
self._loadError = True
self._abf = None
except Exception as error:
logger.error(f" did not load abf file: {self.filepath}")
logger.error(f" unknown Exception was: {error}")
self._loadError = True
self._abf = None
if self._loadError or self._abf is None:
self.myFileType = "abf"
return
self._sweepList = list(self._abf.sweepList)
self._numSweeps = len(self._sweepList)
self._sweepLengthSec = float(self._abf.sweepLengthSec)
self._dataPointsPerMs = float(self._abf.dataPointsPerMs)
self._epochTableList = self._build_epoch_tables(self._abf)
abf_datetime = self._abf.abfDateTime
self.setAcqDate(abf_datetime.strftime("%Y-%m-%d"))
self.setAcqTime(abf_datetime.strftime("%H:%M:%S"))
self.setAcqDateTime(abf_datetime.isoformat(sep=" "))
self._numChannels = len(self._abf.adcUnits)
if self._numChannels > 1:
logger.warning(
f"SanPy defaults to channel 0 for {self._numChannels}-channel "
f"recording {self._path}"
)
adc_units = self._abf.adcUnits[0]
self._sweepLabelX = "sec"
self._sweepLabelY = adc_units
if adc_units in ["A", "pA", "nA", "uA"]:
self._recordingMode = recordingModes.vclamp
epoch_level_unit = "mV"
elif adc_units in ["mV", "uv", "V"]:
self._recordingMode = recordingModes.iclamp
epoch_level_unit = "pA"
else:
logger.warning(f'did not understand adcUnit "{adc_units}"')
epoch_level_unit = None
if self._epochTableList is not None and epoch_level_unit is not None:
for table in self._epochTableList:
table.set_level_unit(epoch_level_unit)
self._userList = self._abf.userList
if loadData:
self._load_sweep_arrays(self._abf)
self.myFileType = "abf"
self._abf = None
def _build_epoch_tables(
self, abf: pyabf.ABF
) -> list[SanPyEpochTable] | None:
"""Build SanPy epoch tables from ABF header structures.
Args:
abf: Header-loaded pyABF recording.
Returns:
One epoch table per sweep, or ``None`` when headers have no usable
epoch description.
"""
try:
pyabf_table = PyAbfEpochTable(abf, 0)
except (AttributeError, IndexError, ValueError) as error:
logger.warning(
f"Could not read ABF epoch headers from {self.filepath}: {error}"
)
return None
tables: list[SanPyEpochTable] = []
for sweep, waveform in enumerate(pyabf_table.epochWaveformsBySweep):
abf.sweepNumber = sweep
abf.sweepEpochs = waveform
tables.append(SanPyEpochTable(abf))
return tables
def _load_sweep_arrays(self, abf: pyabf.ABF) -> None:
"""Materialize ABF time, recording, and command arrays.
Args:
abf: pyABF recording whose signal data may be loaded on demand.
"""
abf.setSweep(0, channel=0)
row_count = abf.sweepX.shape[0]
sweep_count = len(self._sweepList)
self._sweepX = np.zeros((row_count, 1))
self._sweepY = np.zeros((row_count, sweep_count))
self._sweepC = np.zeros((row_count, sweep_count))
for sweep in self._sweepList:
abf.setSweep(sweepNumber=sweep, channel=0)
if sweep == 0:
self._sweepX[:, 0] = abf.sweepX
self._sweepY[:, sweep] = abf.sweepY
try:
self._sweepC[:, sweep] = abf.sweepC
except ValueError as error:
logger.warning(
f"Could not load command sweep {sweep} from "
f"{self.filepath}: {error}"
)
abf.setSweep(0, channel=0)
|