SQLiteDBWriter¶

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

Abstract base class for database writer that will store metadata and data from fitted events for postprocessing later

Public Methods¶

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

Commit and close the shared database connection/cursor, if open.

Parameters:: channel (Optional[int]) – unused; this writer shares a single connection across all channels, so there is no per-channel resource to close independently.

SQLiteDBWriter.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]]

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

Permanently delete the given channel’s row (and, via cascading foreign keys, its events/sublevels/data 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:

RuntimeError – if the configured experiment cannot be found in the database
sqlite3.Error – if the delete fails, so that the caller cannot treat an unreset channel as a clean slate

Private Methods¶

SQLiteDBWriter._column_exists(cursor: Cursor, table_name: str, column_name: str) → bool¶

Check whether a given column exists in the specified database table.

Parameters:

cursor (sqlite3.Cursor) – SQLite database cursor used to execute the query.
table_name (str) – Name of the table to inspect.
column_name (str) – Name of the column to check for existence.

Returns:

True if the column exists, False otherwise.

Return type:

bool

SQLiteDBWriter._init() → None¶

Purpose: Perform generic class construction operations.

All data plugins have this function and must provide an implementation. This is called immediately at the start of class creation and is used to do whatever is required to set up your reader. Note that no app settings are available when this is called, so this function should be used only for generic class construction operations. Most readers simply pass this function.

SQLiteDBWriter._initialize_database(channel: int | None = None) → None¶

Do whatever you need to do to initialize the database file for a given channel before writing the first event

Parameters:

channel (Optional[int]) – int indicating which output to flush

Raises:

ValueError – if event or sublevel metadata declares an unsupported datatype
RuntimeError – if database initialization fails at the SQL level
sqlite3.Error – if a database operation fails
Exception – if an unexpected error occurs during initialization

SQLiteDBWriter._insert_event(cursor: Cursor, event_metadata: Dict[str, int | float | str | bool], experiment_id: int, channel_db_id: int) → bool¶

Insert event metadata into the ‘events’ table. Return True on success, False on failure.

Parameters:

cursor (sqlite3.Cursor) – The SQLite cursor to execute the query.
event_metadata (Dict[str, Union[int, float, str, bool]]) – A dictionary of metadata associated with the event.
experiment_id (int) – The ID of the experiment for which the event is being logged.
channel_db_id (int) – The database ID of the channel to associate with the event.

Returns:

True on success, False on failure

Return type:

bool

SQLiteDBWriter._insert_event_data(cursor: Cursor, event_metadata: Dict[str, int | float | str | bool], event_data: ndarray[tuple[int, ...], dtype[float64]], raw_data: ndarray[tuple[int, ...], dtype[float64]], fit_data: ndarray[tuple[int, ...], dtype[float64]], experiment_id: int, channel_db_id: int, event_db_id: int) → bool¶

Insert the event data into the ‘data’ table after converting it to the appropriate binary format.

Parameters:

cursor (sqlite3.Cursor) – The SQLite cursor to execute the query.
event_metadata (Dict[str, Union[int, float, str, bool]]) – A dictionary of metadata associated with the event.
event_data (npt.NDArray[np.float64]) – A numpy array of filtered event data to be stored as binary in the database.
raw_data (npt.NDArray[np.float64]) – A numpy array of raw event data to be stored as binary in the database.
fit_data (npt.NDArray[np.float64]) – A numpy array of fitted event data to be stored as binary in the database.
experiment_id (int) – The ID of the experiment to which the data belongs.
channel_db_id (int) – The database ID of the channel to associate with the event data.
event_db_id (int) – The database ID of the event this data belongs to.

Returns:

True on success, False on failure

Return type:

bool

Raises:

ValueError – if event_data is not a numpy array of dtype np.float64

SQLiteDBWriter._insert_sublevels(cursor: Cursor, sublevel_metadata: Dict[str, List[int | float | str | bool]], experiment_id: int, channel_db_id: int, event_db_id: int) → bool¶

Insert sublevel metadata into the ‘sublevels’ table.

Parameters:

cursor (sqlite3.Cursor) – The SQLite cursor to execute the query.
sublevel_metadata (Dict[str, List[Union[int, float, str, bool]]]) – A dictionary of sublevel metadata, where each key corresponds to a list of values.
experiment_id (int) – The ID of the experiment for which the sublevels are being logged.
channel_db_id (int) – The database ID of the channel to associate with the sublevels.
event_db_id (int) – The database ID of the event to associate with the sublevels.

Returns:

True on success, False on failure

Return type:

bool

SQLiteDBWriter._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.

SQLiteDBWriter._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:

RuntimeError – if the configured experiment cannot be found in the database
sqlite3.Error – if a database operation fails

SQLiteDBWriter._write_event(channel: int, event_metadata: Dict[str, int | float | str | bool], sublevel_metadata: Dict[str, List[int | float | str | bool]], event_data: ndarray[tuple[int, ...], dtype[float64]], raw_data: ndarray[tuple[int, ...], dtype[float64]], fit_data: ndarray[tuple[int, ...], dtype[float64]], abort: bool | None = False, last_call: bool | None = False) → bool¶

Write a single event worth of data and metadata to the database. Do NOT commit.

Parameters:

channel (int) – identifier for the channel to write events from
event_metadata (Dict[str, Union[int, float, str, bool]]) – a dict of metadata associated to the event
sublevel_metadata (Dict[str, List[Union[int, float, str, bool]]]) – a dict of lists of metadata associated to sublevels within the event. You can assume they all have the same length.
event_data (npt.NDArray[np.float64]) – the filtered data for the event
raw_data (npt.NDArray[np.float64]) – A numpy array of raw event data to be stored as binary in the database.
fit_data (npt.NDArray[np.float64]) – A numpy array of fitted event data to be stored as binary in the database.
abort (Optional[bool]) – True if an abort request was issued in the caller, perform cleanup as needed
last_call (Optional[bool]) – True if this is the last time the function will be called, commit to file and clean up as needed

Returns:

True on successful write, False if the row already existed (rejected by INSERT OR IGNORE). Any other failure (a genuine database error) is raised rather than returned, so the caller can report the real reason instead of assuming a duplicate row.

Return type:

bool

Raises:

ValueError – if a database connection cannot be opened
RuntimeError – if the experiment or channel cannot be found in the database, or if the event insert reports success without producing a row id
sqlite3.Error – if a database operation fails
Exception – if an unexpected error occurs while writing the event

SQLiteDBWriter._write_experiment_metadata(channel: int | None = None) → None¶

Write any information you need to save about the experiment itself

Parameters:: channel (Optional[int]) – int indicating which output to flush
Raises:: sqlite3.Error – if a database operation fails