SQLiteDBLoader¶

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

Subclass of MetaDatabaseLoader for reading and querying SQLite database files.

Provides functionality to load metadata, event data, and other relevant structures from a structured SQLite database generated by Poriscope.

Public Methods¶

SQLiteDBLoader.add_columns_to_table(df: DataFrame, units: List[str | None], table_name: str) → bool¶

Adds new columns from a pandas DataFrame to an existing SQLite table and populates them with data, matching on the ‘id’ column.

Parameters:

df (pd.DataFrame) – A pandas DataFrame. Must contain an ‘id’ column corresponding to the primary key of the target table, and one or more additional columns to be added.
units (List[Optional[str]]) – A list of strings specifying units for the new columns to be added. Must have length equal to the number of new cols, but can contain None values
table_name (str) – The name of the SQLite table to modify. This table must already exist in the databse.

Returns:

True on success, False otherwise

Return type:

bool

Raises:

ValueError – If the DataFrame does not contain an ‘id’ column or if the specified table does not exist.
sqlite3.Error – If a database operation fails
Exception – If an unexpected error occurs while adding the columns

SQLiteDBLoader.alter_database(queries: List[str]) → bool¶

Run a given list of queries on the DB. There is no validation here, use it sparingly.

Parameters:

queries (List[str]) – a list of queries to run on the database

Returns:

True if the operation succeeded, False otherwise

Return type:

bool

Raises:

sqlite3.Error – if a database operation fails
ValueError – if a value error occurs while running the queries
Exception – if an unexpected error occurs while running the queries

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

Close resources gracefully before application exit.

Parameters:: channel (Optional[int]) – Channel ID to close. If None, close all resources.

SQLiteDBLoader.get_channels_by_experiment(experiment: str) → List[int] | None¶

Retrieve a list of all channel IDs associated with a given experiment name or None on failure

Parameters:: experiment (str) – The name of the experiment.
Returns:: List of channel IDs.
Return type:: Optional[List[int]]

SQLiteDBLoader.get_column_names_by_table(table: str | None = None) → List[str] | None¶

Retrieve the column names available in a specified table.

Parameters:: table (Optional[str]) – The name of the table.
Returns:: List of column names.
Return type:: Optional[List[str]]

SQLiteDBLoader.get_column_type(column_name: str) → str | None¶

Retrieve the datatype associated with a specific column name or None on failure

Parameters:: column_name (str) – The name of the column.
Returns:: The datatype of the column.
Return type:: Optional[str]

SQLiteDBLoader.get_column_units(column_name: str) → str | None¶

Retrieve the units associated with a specific column name, or None on failure

Parameters:: column_name (str) – The name of the column.
Returns:: The units of the column, empty string is units is NULL
Return type:: Optional[str]

SQLiteDBLoader.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.

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,
                                 'Units': <unit str> or None
                                },
                ...
                }

Several parameter keywords are reserved: these are

‘Input File’ ‘Output File’ ‘Folder’

These must have Type str and will cause the GUI to generate widgets to allow selection of these elements when used

Parameters:

globally_available_plugins (Optional[Dict[str, List[str]]]) – a dict containing all data plugins that exist to date, keyes by metaclass
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]]

SQLiteDBLoader.get_event_counts_by_experiment_and_channel(experiment: str | None = None, channel: int | None = None) → int | None¶

Return the number of events in the database matching the experiment name and channel name. If no channel name is provided, count across all channels for that experiment. If no experiment is provided, ignore channel and return the number of events in the entire database

Parameters:

experiment (Optional[str]) – The name of the experiment.
channel (Optional[int]) – The index of the channel

Returns:

event count matching the conditions, or None on failure

Return type:

Optional[int]

SQLiteDBLoader.get_experiment_names(experiment_id: int | None = None) → List[str] | None¶

Retrieve a list of all unique experiment names registered in the database or a singleton list if a name is given.

Parameters:: experiment_id (Optional[int]) – the id of the experiment for which to fetch the name
Returns:: List of experiment names, or None on failure
Return type:: Optional[List[str]]

SQLiteDBLoader.get_llm_prompt() → str | None¶

Return a prompt that will tell the LLM the structure of the database to be queried

Returns:: a prompt that gives an LLM context for the database and how to query it, or None on failure
Return type:: Optional[str]

SQLiteDBLoader.get_samplerate_by_experiment_and_channel(experiment: str, channel: int) → float | None¶

Retrieve the sampling rate for a given experiment and channel id

Parameters:

experiment (str) – The name of the experiment in the database.
channel (int) – The channel id to get sampling rate for.

Returns:

sampling rate for the specific expreiment-channel combination, or None on failure

Return type:

Optional[float]

Raises:

ValueError – if no samplerate could be found for the given experiment and channel

SQLiteDBLoader.get_table_by_column(column: str) → str | None¶

Retrieve the name of the table in which the given column is found, or None on failure

Parameters:: column (str) – The name of the column.
Returns:: The name of the table containing the column, or None on failure.
Return type:: Optional[str]

SQLiteDBLoader.get_table_names() → List[str] | None¶

Retrieve the names of available tables in the database.

Returns:: List of table names.
Return type:: Optional[List[str]]

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

Perform any actions necessary to reset a channel to its starting state. If channel is not None, handle only that channel, else reset all of them.

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

SQLiteDBLoader.validate_filter_query(query: str) → Tuple[bool, str]¶

Validate a SQL filter query without executing it.

Parameters:: query (str) – The SQL query string.
Returns:: True if valid query, False otherwise, plus error message if invalid or emnpty string on success
Return type:: Tuple[bool, str]

Private Methods¶

SQLiteDBLoader._ensure_event_counts() → None¶

Ensure event_counts table and its triggers exist, creating and populating them from scratch if they don’t (backwards compatibility for old DBs).

Returns:: None
Return type:: None
Raises:: sqlite3.Error – If a database error occurs during table creation or population.

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

SQLiteDBLoader._get_sqlite_type(dtype: dtype) → str¶

Maps pandas dtype to SQLite data type.

Parameters:: dtype (np.dtype) – The pandas dtype
Returns:: The corresponding SQLite data type as a string.
Return type:: str

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

SQLiteDBLoader._load_event_data(query: str) → Generator[Tuple[int, int, int, int, float, int, int, ndarray[tuple[int, ...], dtype[float64]], ndarray[tuple[int, ...], dtype[float64]], ndarray[tuple[int, ...], dtype[float64]]], bool, None]¶

Load data and return a generator that gives a one-row dataframe corresponding one row returned by query Make sure you exhaust or explicitly abort the generator, or else connections will remain open You can assume that the query was generated by self.construct_event_data_query() and will have 11 columns, in this order: db_id, event_id, channel_id, experiment_id, data_format, samplerate, padding_before, padding_after, raw_data, filtered_data, fit_data where raw_data, filtered_data, and fit_data are bytes objects to be interpreted using data_format

Parameters:: query (str) – a valid SQL query, checked in the calling function for validity
Returns:: a generator that yields tuples of (primary database id, experiment_id, channel_id, event_id, samplerate, padding_before, padding_after, raw event data, filtered event data, and fitted event data), and accepts an abort boolean sent back via generator.send()
Return type:: Generator[Tuple[int, int, int, int, float, int, int, npt.NDArray[np.float64], npt.NDArray[np.float64], npt.NDArray[np.float64]], bool, None]

SQLiteDBLoader._load_metadata(query: str) → DataFrame | None¶

Load and return the data specified by a valid SQL query formatted as a pandas dataframe

Parameters:: query (str) – a valid SQL query, checked in the calling function for validity
Returns:: A dataframe containing the requested event data as columns, empty if the query matched no rows, or None if the query could not be run
Return type:: Optional[pd.DataFrame]

SQLiteDBLoader._load_metadata_generator(query: str) → Generator[DataFrame, None, None]¶

Load and return the data specified by a valid SQL query formatted as a pandas dataframe Make sure you exhaust the generator, or else connections will remain open

Parameters:: query (str) – query to run on the database
Returns:: None. This is a generator that yields one row at a time in the form of a single-line pd.DataFrame; it returns None once exhausted or on failure.
Return type:: None

SQLiteDBLoader._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:: ValueError – If the settings dict does not contain the correct information.