black formatting

Author: zm711

neo/io/biocamio.py

@@ -7,6 +7,5 @@ class BiocamIO(BiocamRawIO, BaseFromRaw):
     mode = "file"
 
     def __init__(self, filename, fill_gaps_strategy=None):
-        BiocamRawIO.__init__(self, filename=filename,
-                             fill_gaps_strategy=fill_gaps_strategy)
+        BiocamRawIO.__init__(self, filename=filename, fill_gaps_strategy=fill_gaps_strategy)
         BaseFromRaw.__init__(self, filename)

neo/rawio/biocamrawio.py

@@ -145,20 +145,21 @@ def _get_analogsignal_chunk(self, block_index, seg_index, i_start, i_stop, strea
         # read functions are different based on the version of biocam
         if self._read_function is readHDF5t_brw4_sparse:
             if self._fill_gaps_strategy is None:
-                raise ValueError(
-                    "Please set `fill_gaps_strategy` to 'zeros' or 'synthetic_noise'."
-                )
+                raise ValueError("Please set `fill_gaps_strategy` to 'zeros' or 'synthetic_noise'.")
             if self._fill_gaps_strategy == "synthetic_noise":
-                warnings.warn("Event-based compression : gaps will be filled with synthetic noise. "
-                              "Set `fill_gaps_strategy` to 'zeros' to fill gaps with 0s.")
+                warnings.warn(
+                    "Event-based compression : gaps will be filled with synthetic noise. "
+                    "Set `fill_gaps_strategy` to 'zeros' to fill gaps with 0s."
+                )
                 use_synthetic_noise = True
             elif self._fill_gaps_strategy == "zeros":
                 use_synthetic_noise = False
             else:
                 raise ValueError("`fill_gaps_strategy` must be 'zeros' or 'synthetic_noise'")
 
-            data = self._read_function(self._filehandle, i_start, i_stop, self._num_channels,
-                                       use_synthetic_noise=use_synthetic_noise)
+            data = self._read_function(
+                self._filehandle, i_start, i_stop, self._num_channels, use_synthetic_noise=use_synthetic_noise
+            )
         else:
             data = self._read_function(self._filehandle, i_start, i_stop, self._num_channels)
 
@@ -273,7 +274,7 @@ def open_biocam_file_header(filename) -> dict:
         min_digital = experiment_settings["ValueConverter"]["MinDigitalValue"]
         scale_factor = experiment_settings["ValueConverter"]["ScaleFactor"]
         sampling_rate = experiment_settings["TimeConverter"]["FrameRate"]
-        num_frames = rf['TOC'][-1,-1]
+        num_frames = rf["TOC"][-1, -1]
 
         num_channels = None
         well_ID = None
@@ -282,7 +283,9 @@ def open_biocam_file_header(filename) -> dict:
                 num_channels = len(rf[well_ID]["StoredChIdxs"])
                 if "Raw" in rf[well_ID]:
                     if len(rf[well_ID]["Raw"]) % num_channels:
-                        raise NeoReadWriteError(f"Length of raw data array is not multiple of channel number in {well_ID}")
+                        raise NeoReadWriteError(
+                            f"Length of raw data array is not multiple of channel number in {well_ID}"
+                        )
                     num_frames = len(rf[well_ID]["Raw"]) // num_channels
                     break
                 elif "EventsBasedSparseRaw" in rf[well_ID]:
@@ -360,7 +363,7 @@ def readHDF5t_brw4_sparse(rf, t0, t1, nch, use_synthetic_noise=False):
         data.fill(2048)
     else:
         # fill the data collection with Gaussian noise if requested
-        data = generate_synthetic_noise(rf, data, well_ID, start_frame, num_frames) #, std=noise_std)
+        data = generate_synthetic_noise(rf, data, well_ID, start_frame, num_frames)  # , std=noise_std)
     # fill the data collection with the decoded event based sparse raw data
     data = decode_event_based_raw_data(rf, data, well_ID, start_frame, num_frames)
 
@@ -376,38 +379,38 @@ def decode_event_based_raw_data(rf, data, well_ID, start_frame, num_frames):
     # from the given start position and duration in frames, localize the corresponding event positions
     # using the TOC
     toc_start_idx = np.searchsorted(toc[:, 1], start_frame)
-    toc_end_idx = min(
-            np.searchsorted(toc[:, 1], start_frame + num_frames, side="right") + 1,
-            len(toc) - 1)
+    toc_end_idx = min(np.searchsorted(toc[:, 1], start_frame + num_frames, side="right") + 1, len(toc) - 1)
     events_start_pos = events_toc[toc_start_idx]
     events_end_pos = events_toc[toc_end_idx]
     # decode all data for the given well ID and time interval
     binary_data = rf[well_ID]["EventsBasedSparseRaw"][events_start_pos:events_end_pos]
     binary_data_length = len(binary_data)
     pos = 0
     while pos < binary_data_length:
-        ch_idx = int.from_bytes(binary_data[pos:pos + 4], byteorder="little")
+        ch_idx = int.from_bytes(binary_data[pos : pos + 4], byteorder="little")
         pos += 4
-        ch_data_length = int.from_bytes(binary_data[pos:pos + 4], byteorder="little")
+        ch_data_length = int.from_bytes(binary_data[pos : pos + 4], byteorder="little")
         pos += 4
         ch_data_pos = pos
         while pos < ch_data_pos + ch_data_length:
-            from_inclusive = int.from_bytes(binary_data[pos:pos + 8], byteorder="little")
+            from_inclusive = int.from_bytes(binary_data[pos : pos + 8], byteorder="little")
             pos += 8
-            to_exclusive = int.from_bytes(binary_data[pos:pos + 8], byteorder="little")
+            to_exclusive = int.from_bytes(binary_data[pos : pos + 8], byteorder="little")
             pos += 8
             range_data_pos = pos
             for j in range(from_inclusive, to_exclusive):
                 if j >= start_frame + num_frames:
                     break
                 if j >= start_frame:
                     data[ch_idx][j - start_frame] = int.from_bytes(
-                            binary_data[range_data_pos:range_data_pos + 2], byteorder="little")
+                        binary_data[range_data_pos : range_data_pos + 2], byteorder="little"
+                    )
                 range_data_pos += 2
             pos += (to_exclusive - from_inclusive) * 2
 
     return data
 
+
 def generate_synthetic_noise(rf, data, well_ID, start_frame, num_frames):
     # Source: Documentation by 3Brain
     # https://gin.g-node.org/NeuralEnsemble/ephy_testing_data/src/master/biocam/documentation_brw_4.x_bxr_3.x_bcmp_1.x_in_brainwave_5.x_v1.1.3.pdf
@@ -451,10 +454,10 @@ def generate_synthetic_noise(rf, data, well_ID, start_frame, num_frames):
     # fill with Gaussian noise
     for ch_idx in range(len(data)):
         if ch_idx in noise_info:
-            data[ch_idx] = np.array(np.random.normal(noise_info[ch_idx][0], noise_info[ch_idx][1],
-                num_frames), dtype=np.uint16)
+            data[ch_idx] = np.array(
+                np.random.normal(noise_info[ch_idx][0], noise_info[ch_idx][1], num_frames), dtype=np.uint16
+            )
         else:
-            data[ch_idx] = np.array(np.random.normal(median_mean, median_std, num_frames),
-                    dtype=np.uint16)
+            data[ch_idx] = np.array(np.random.normal(median_mean, median_std, num_frames), dtype=np.uint16)
 
     return data

neo/rawio/blackrockrawio.py

@@ -133,7 +133,6 @@ class BlackrockRawIO(BaseRawIO):
     # We need to document the origin of this value
     main_sampling_rate = 30000.0
 
-
     def __init__(
         self, filename=None, nsx_override=None, nev_override=None, nsx_to_load=None, load_nev=True, verbose=False
     ):
@@ -254,7 +253,6 @@ def __init__(
 
     def _parse_header(self):
 
-
         event_channels = []
         spike_channels = []
         signal_buffers = []

neo/rawio/openephysbinaryrawio.py

@@ -76,7 +76,8 @@ def __init__(self, dirname="", load_sync_channel=False, experiment_names=None):
             warn(
                 "The load_sync_channel=True option is deprecated and will be removed in version 0.15. "
                 "Use load_sync_channel=False instead, which will add sync channels as separate streams.",
-                DeprecationWarning, stacklevel=2
+                DeprecationWarning,
+                stacklevel=2,
             )
         self.folder_structure = None
         self._use_direct_evt_timestamps = None
@@ -152,7 +153,7 @@ def _parse_header(self):
                     # Every stream sync channel is added as its own stream
                     sync_stream_id = f"{stream_name}SYNC"
                     sync_stream_id_to_buffer_id[sync_stream_id] = buffer_id
-                    
+
                     # We save this mapping for the buffer description protocol
                     normal_stream_id_to_sync_stream_id[stream_id] = sync_stream_id
                     # We then set the stream_id to the sync stream id
@@ -188,19 +189,19 @@ def _parse_header(self):
         signal_buffers = []
 
         unique_streams_ids = np.unique(signal_channels["stream_id"])
-        
+
         # This is getting too complicated, we probably should just have a table which would be easier to read
         # And for users to understand
         for stream_id in unique_streams_ids:
-            
+
             # Handle sync channel on a special way
             if "SYNC" in stream_id:
                 # This is a sync channel and should not be added to the signal streams
                 buffer_id = sync_stream_id_to_buffer_id[stream_id]
                 stream_name = stream_id
                 signal_streams.append((stream_name, stream_id, buffer_id))
                 continue
-            
+
             # Neural signal
             stream_index = int(stream_id)
             if stream_index < self._num_of_signal_streams:
@@ -279,9 +280,9 @@ def _parse_header(self):
                         if has_sync_trace and not self.load_sync_channel:
                             # Exclude the sync channel from the main stream
                             self._stream_buffer_slice[stream_id] = slice(None, -1)
-                            
+
                             # Add a buffer slice for the sync channel
-                            sync_stream_id = normal_stream_id_to_sync_stream_id[stream_id]                            
+                            sync_stream_id = normal_stream_id_to_sync_stream_id[stream_id]
                             self._stream_buffer_slice[sync_stream_id] = slice(-1, None)
                         else:
                             self._stream_buffer_slice[stream_id] = None
@@ -294,9 +295,9 @@ def _parse_header(self):
                         if has_sync_trace and not self.load_sync_channel:
                             # Exclude the sync channel from the non-neural stream
                             self._stream_buffer_slice[stream_id_non_neural] = slice(num_neural_channels, -1)
-                            
+
                             # Add a buffer slice for the sync channel
-                            sync_stream_id = normal_stream_id_to_sync_stream_id[stream_id]                            
+                            sync_stream_id = normal_stream_id_to_sync_stream_id[stream_id]
                             self._stream_buffer_slice[sync_stream_id] = slice(-1, None)
                         else:
                             self._stream_buffer_slice[stream_id_non_neural] = slice(num_neural_channels, None)

neo/test/rawiotest/test_openephysbinaryrawio.py

@@ -43,26 +43,26 @@ def test_sync(self):
             block_index=0, seg_index=0, i_start=0, i_stop=100, stream_index=stream_index
         )
         assert chunk.shape[1] == 384
-        
+
     def test_sync_channel_access(self):
         """Test that sync channels can be accessed as separate streams when load_sync_channel=False."""
         rawio = OpenEphysBinaryRawIO(
             self.get_local_path("openephysbinary/v0.6.x_neuropixels_with_sync"), load_sync_channel=False
         )
         rawio.parse_header()
-        
+
         # Find sync channel streams
         sync_stream_names = [s_name for s_name in rawio.header["signal_streams"]["name"] if "SYNC" in s_name]
         assert len(sync_stream_names) > 0, "No sync channel streams found"
-        
+
         # Get the stream index for the first sync channel
         sync_stream_index = list(rawio.header["signal_streams"]["name"]).index(sync_stream_names[0])
-        
+
         # Check that we can access the sync channel data
         chunk = rawio.get_analogsignal_chunk(
             block_index=0, seg_index=0, i_start=0, i_stop=100, stream_index=sync_stream_index
         )
-        
+
         # Sync channel should have only one channel
         assert chunk.shape[1] == 1, f"Expected sync channel to have 1 channel, got {chunk.shape[1]}"