SQLiteEventWriter¶

class SQLiteEventWriter(settings: Optional[dict] = None)

Save events into a single file with baseline between them discarded

Public Methods¶

SQLiteEventWriter.close_resources(channel: int | None = None) → None¶

Do whatever needs doing to gracefully shut down on app exit

Parameters:: channel (Optional[int]) – channel ID

SQLiteEventWriter.get_empty_settings(globally_available_plugins: Dict[str, List[str]] | None = None, standalone: bool = False) → Dict[str, Dict[str, Any]]¶

Get a dict populated with keys needed to initialize the filter if they are not set yet. This dict must have the following structure, but Min, Max, and Options can be skipped or explicitly set to None if they are not used. Type is required; Value may be omitted or set to None, both meaning there is no default and the user must supply one. All values provided must be consistent with Type. EventFinder objects MUST include a MetaReader object in settings

settings = {'Parameter 1': {'Type': <int, float, str, bool>,
                                 'Value': <value> or None,
                                 'Options': [<option_1>, <option_2>, ... ] or None,
                                 'Min': <min_value> or None,
                                 'Max': <max_value> or None
                                },
                ...
                }

Parameters:

globally_available_plugins (Optional[Dict[str, List[str]]]) – a dict containing all data plugins that exist to date, keyed by metaclass. Must include “MetaReader” as a key, with explicitly set Type MetaReader.
standalone (bool) – False if this is called as part of a GUI, True otherwise. Default False

Returns:

the dict that must be filled in to initialize the filter

Return type:

Dict[str, Dict[str, Any]]

SQLiteEventWriter.get_output_file_name() → Path¶: get the name of the output file

SQLiteEventWriter.reset_channel(channel: int | None = None) → None¶

Permanently delete the given channel’s row (and, via cascading foreign keys, its associated event rows) from the database, so a subsequent write starts from a clean slate. This is destructive, not a resource-cleanup step.

Parameters:

channel (Optional[int]) – channel ID. Note that channel=None does not reset all channels; SQL channel_id = NULL never matches, so no rows are deleted.

Raises:

ValueError – if settings have not been initialized or the output file path is not set in settings
sqlite3.Error – if the delete fails, so that the caller cannot treat an unreset channel as a clean slate

Private Methods¶

SQLiteEventWriter._finalize_initialization() → None¶

Apply the provided paramters and intialize any internal structures needed Should Raise if initialization fails.

This function is called at the end of the class constructor to perform additional initialization specific to the algorithm being implemented. kwargs provided to the base class constructor are available as class attributes.

SQLiteEventWriter._init() → None¶: called at the start of base class initialization

SQLiteEventWriter._initialize_database(channel: int) → None¶

Open a database or file handle for writing events - this function will be called from every channel in the reader

Parameters:

channel (int) – the channel for which to initialize the database

Raises:

ValueError – if the output file path is not set in settings
RuntimeError – if database initialization fails at the SQL level
sqlite3.Error – if a database operation fails
Exception – if an unexpected error occurs during initialization

SQLiteEventWriter._rescale_data_to_adc(data: ~numpy.ndarray[tuple[int, ...], ~numpy.dtype[~numpy.number]], scale: float | None = None, offset: float | None = None, raw_data: bool = False, dtype: ~numpy.dtype[~typing.Any] | None | type[~typing.Any] | ~numpy._typing._dtype_like._SupportsDType[~numpy.dtype[~typing.Any]] | str | tuple[~typing.Any, int] | tuple[~typing.Any, ~typing.SupportsIndex | ~collections.abc.Sequence[~typing.SupportsIndex]] | list[~typing.Any] | ~numpy._typing._dtype_like._DTypeDict | tuple[~typing.Any, ~typing.Any] = <class 'numpy.uint16'>, adc_min: int = -32768, adc_max: int = 32767) → tuple[ndarray[tuple[int, ...], dtype[number]], float | None, float | None]¶

Not used by this writer

Parameters:

data (npt.NDArray[np.number]) – 1D numpy array of data to write to the active file in the specified channel.
scale (Optional[float]) – Scaling between provided data type and encoded form for storage. If None, scale is calculated based on the data to maximally use the available adc range.
offset (Optional[float]) – Offset between provided data type and encoded form for storage. If None, offset is calculated based on the data to maximally use the available adc range.
raw_data (bool) – True means to simply write data as-is to file, False indicates to first rescale it. Default False.
dtype (npt.DTypeLike) – Numpy dtype to use for storage. Defaults to 16-bit unsigned int.
adc_min (int) – Integer encoding the minimum adc code for the adc conversion.
adc_max (int) – Integer encoding the maximum adc code for the adc conversion.

Returns:

Rescaled data as numpy array, scale factor, and offset.

Return type:

tuple[npt.NDArray[np.number], Optional[float], Optional[float]]

SQLiteEventWriter._set_output_dtype() → str¶

set the output dtype - should be a numpy numeric type:

self.output_dtype = ‘<u2’

SQLiteEventWriter._validate_settings(settings: dict) → None¶

Validate that the settings dict contains the correct information for use by the subclass.

Parameters:: settings (dict) – Parameters for event detection.
Raises:: KeyError – If the settings dict does not contain the correct information.

SQLiteEventWriter._write_channel_metadata(channel: int) → None¶

Write any information you need to save about the channel.

Parameters:

channel (int) – int indicating which output to flush

Raises:

ValueError – if settings are not initialized or required settings are missing
RuntimeError – if the channel’s database ID cannot be determined after insertion
sqlite3.Error – if a database operation fails

SQLiteEventWriter._write_data(data: ndarray[tuple[int, ...], dtype[number]], channel: int, index: int, scale: float | None = None, offset: float | None = None, start_sample: int | None = 0, padding_before: int | None = 0, padding_after: int | None = None, baseline_mean: float | None = None, baseline_std: float | None = None, raw_data: bool = False, abort: bool | None = False, last_call: bool | None = False) → bool¶

Append data and metadata to the active file handle.

Parameters:

data (npt.NDArray[np.number]) – 1D numpy array of data to write to the active file in the specified channel.
channel (int) – Int indicating the channel from which it was acquired.
index (int) – event index
scale (Optional[float]) – Float indicating scaling between provided data type and encoded form for storage, default None.
offset (Optional[float]) – Float indicating offset between provided data type and encoded form for storage, default None.
start_sample (Optional[int]) – Integer index of the starting point of the provided array relative to the start of the experimental run, default 0.
padding_before (Optional[int]) – the length of the padding before the actual event start
padding_after (Optional[int]) – the length of the padding after the actual event end
baseline_mean (Optional[float]) – The local baseline, if available
baseline_std (Optional[float]) – the local standard deviation, if available
raw_data (bool) – True means to simply write data as-is to file, False indicates to first rescale it. Default False.
abort (Optional[bool]) – If True, roll back and close the active connection without writing, default False.
last_call (Optional[bool]) – If True, close the shared connection after this write, default False.

Returns:

success of the write operation.

Return type:

bool

Raises:

ValueError – if a database connection cannot be opened, or if start_sample, padding_before, or padding_after is None
sqlite3.Error – if a database operation fails
Exception – if an unexpected error occurs during the write