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.
IOError – If any write-related error occurs

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

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_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=None, standalone=False)¶

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. Value and Type are required. 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 (Dict[str, List[str]]) – a dict containing all data plugins that exist to date, keyes by metaclass
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=None, channel=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

Return type:

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()¶

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
Return type:: 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]

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

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

Purpose: Retrieve the names of the table in which the given column is found, or None on failure

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=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 (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._finalize_initialization()¶

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 (str) – 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)¶

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 10 colums: event_id, channel_id, experiment_id, data_format, baseline, stdev, padding_before, padding_after, samplerate, data where data is a bytes object to be interpreted using data_format

Parameters:: query (str) – a valid SQL query, checked in the calling function for validity
Returns:: a generator that returns primary database id, experiment_id, channel_id, event_id, samplerate, padding_before, padding_after, samplerate, and three numpy arrays with raw event data, filtered event data, and fitted event data
Return type:: Generator[Dict[str,Union[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 or None on failure
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:: A generator that feeds out onne row at a time in the form of a single-line dataframe
Return type:: Generator[pd.DataFrame, None, 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.