MetaDatabaseLoader¶

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

What you get by inheriting from MetaDatabaseLoader¶

MetaDatabaseLoader is the base class for loading the data written by a MetaDatabaseWriter subclass instance or any other method that produces an equivalent format.

Poriscope ships with SQLiteDBLoader, a subclass of MetaDatabaseLoader that reads data written by the SQLiteDBWriter subclass. While additional subclasses can read almost any format you desire, we strongly encourage standardization around this format. Think twice before creating additional subclasses of this base class. It is not sufficient to write just a MetaEventLoader subclass. In addition to this base class, you will also need a paired MetaDatabaseWriter subclass to write data in your target format.

Public Methods¶

Abstract Methods¶

These methods must be implemented by subclasses.

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

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

Purpose: Adds new columns from a pandas DataFrame to an existing SQLite table

Create new columns in the specified table and populate them with the procided data, matching on the ‘id’ column against the primary id in the target table

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

Purpose: Run a given list of queries on the database. 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

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

Purpose: Clean up any open file handles or memory on app exit.

This is called during app exit or plugin deletion to ensure proper cleanup of resources that could otherwise leak. Do this for all channels if no channel is specified, otherwise limit your closure to the specified channel. If no such operation is needed, it suffices to pass.

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

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

Purpose: Retrieve a list of all channel identifiers (the identifier, not the primary key of the channels table) 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]]

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

Purpose: Retrieve the column names available in a specified table, all columns in the database is table is not specified, or None on failure

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

abstractmethod MetaDatabaseLoader.get_column_type(column_name: str) → str | None¶

Purpose: 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]

abstractmethod MetaDatabaseLoader.get_column_units(column_name: str) → str | None¶

Purpose: 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.
Return type:: Optional[str]

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

Purpose: Return the number of events in the database matching the experiment name and channel identifier.

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]

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

Purpose: Retrieve a list of all unique experiment names registered in the database, or a singleton list if an id 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]]

abstractmethod MetaDatabaseLoader.get_llm_prompt() → str | None¶

Purpose: Return a prompt that will tell the LLM the structure of the database to be queried to assist users in accessing the data written in your format

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

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

Purpose: Retrieve the sampling rate for a given experiment and channel id, or None on failure

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]

abstractmethod MetaDatabaseLoader.get_table_by_column(column: str) → str | None¶

Purpose: Retrieve the names 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, or None on failure.
Return type:: Optional[str]

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

Purpose: Retrieve the names of available tables in the database or None on failure.

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

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

Purpose: Reset the state of a specific channel for a new operation or run.

This is called any time an operation on a channel needs to be cleaned up or reset for a new run. If channel is not None, handle only that channel, else reset all of them. In most cases for MetaDatabaseLoaders there is no need to reset and you can simplt pass.

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

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

Purpose: Validate a SQL query without executing it.

Return True, "" if the query is valid, and False, "[[helpful explanation]]" if it is not

Parameters:: query (str) – The SQL query string.
Returns:: True, "" if the query is valid, and False, "[[helpful explanation]]" if it is not
Return type:: Tuple[bool, str]

Concrete Methods¶

MetaDatabaseLoader.construct_event_data_query(conditions: str | None = None, experiments_and_channels: Dict[str, List[int] | None] | None = None) → Tuple[str, str]¶

Construct a query that will get all event data matching a set of conditions

Parameters:

conditions (Optional[str]) – Optional filter condition for query.
experiments_and_channels (Optional[Dict[str, Optional[List[int]]]]) – a dict of experiment names as keys as lists of channels to include as values. Can be None, and individual channel lists can be None to include all channels for that experiment

Returns:

a valid SQL query and an empty string, or an empty string and a debug message

Return type:

Tuple[str, str]

MetaDatabaseLoader.construct_metadata_query(columns: List[str], conditions: str | None = None, experiments_and_channels: Dict[str, List[int] | None] | None = None) → Tuple[str, str, str]¶

Build a SELECT over the metadata tables, joining whatever the request needs.

The shape is derived rather than enumerated. Every query is anchored on events (or on sublevels, when no events column is involved) and aliased e/s/exp; the other tables are joined only when the selected columns or the conditions refer to them:

SELECT [[DISTINCT]] a.id, a.experiment_id, a.channel_id, a.event_id, [[columns]]
FROM events e
[[JOIN sublevels s ON e.id = s.event_db_id]]
[[JOIN experiments exp ON exp.id = e.experiment_id]]
WHERE [[conditions]]

DISTINCT appears when sublevels is joined purely to filter an events plot, since that repeats each event once per sublevel. The returned id column belongs to the returned table name, which callers rely on to write derived columns back.

Conditions are qualified against the tables the query actually joins, so a caller passes a plain WHERE clause body - sublevel_duration < 100 AND voltage > 50 - without knowing the shape. Text inside single-quoted literals is never rewritten, and neither is a subquery, which names its own tables and so needs nothing from the outer aliases - a condition whose only reference to another table sits inside a subquery therefore joins nothing. A bare id is refused, with guidance returned as the debug message, because it means a different row in each table.

Parameters:

columns (List[str]) – List of column names to retrieve.
conditions (Optional[str]) – Optional filter condition for query.
experiments_and_channels (Optional[Dict[str, Optional[List[int]]]]) – a dict of experiment names as keys as lists of channels to include as values. Can be None, and individual channel lists can be None to include all channels for that experiment

Raises:

KeyError – if any of the requested experiment names cannot be found in the database
ValueError – if columns is empty, or a column cannot be mapped to a table

Returns:

a valid SQL query and an empty string, or an empty string and a debug message, and the table name of the affected id column

Return type:

Tuple[str, str, str]

MetaDatabaseLoader.export_subset_to_csv(output_folder: str, subset_name: str = '', conditions: str | None = None, experiments_and_channels: Dict[str, List[int] | None] | None = None) → Generator[float, bool | None, None]¶

Return a generator that shows progress toward outputting a csv version of the subset of the database satisfying the conditions, including both data and metadata

Parameters:

output_folder (str) – The folder to which the subset should be printed. This is assumed to exist already and will raise an error if it does not.
subset_name (str) – Optional string to append to filenames in the subset
conditions (Optional[str]) – Optional filter condition for query.
experiments_and_channels (Optional[Dict[str, Optional[List[int]]]]) – a dict of experiment names as keys as lists of channels to include as values. Can be None, and individual channel lists can be None to include all channels for that experiment

Raises:

KeyError – if any of the requested experiment names cannot be found in the database
ValueError – if the SQL string constructed from the given conditions is invalid, or no matching data is found

Yield:

a float between 0 and 1 representing progress toward completion

Ytype:

float

MetaDatabaseLoader.force_serial_channel_operations() → bool¶

Purpose: Indicate whether operations on different channels must be serialized (not run in parallel).

Returns:: True if only one channel can run at a time, False otherwise
Return type:: bool

MetaDatabaseLoader.get_channel_db_id(experiment_name: str, channel_id: int) → int | None¶

Get the channel primary key (channel_db_id) for a given experiment name and channel identifier.

Parameters:

experiment_name (str) – The name of the experiment.
channel_id (int) – The channel identifier (not the primary key).

Returns:

The primary key of the channel, or None on failure.

Return type:

Optional[int]

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

Purpose: Provide a list of settings details to users to assist in instantiating an instance of your MetaWriter subclass.

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
                               },
               ...
               }

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

This function must implement returning of a dictionary of settings required to initialize the filter, in the specified format. Values in this dictionary can be accessed downstream through the self.settings class variable. This structure is a nested dictionary that supplies both values and a variety of information about those values, used by poriscope to perform sanity and consistency checking at instantiation.

While this function is technically not abstract in MetaEventLoader, which already has an implementation of this function that ensures that settings will have the required Input File key available to users, in most cases you will need to override it to add any other settings required by your subclass or to specify which files types are allowed. If you need additional settings, which you almost certainly do, you MUST call super().get_empty_settings(globally_available_plugins, standalone) before any additional code that you add. For example, your implementation could look like this, to limit it to sqlite files:

settings = super().get_empty_settings(globally_available_plugins, standalone)
settings["Input File"]["Options"] = [
                        "SQLite3 Files (*.sqlite3)",
                        "Database Files (*.db)",
                        "SQLite Files (*.sqlite)",
                        ]
return settings

which will ensure that your have the Input File key and limit visible options to sqlite3 files. By default, it will accept any file type as output, hence the specification of the Options key for the relevant plugin in the example above.

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

MetaDatabaseLoader.get_experiment_id_by_name(experiment_name: str) → int | None¶

Look up the database primary key of the experiment with the given name.

Parameters:: experiment_name (str) – the name of the experiment for which to fetch the id
Returns:: The experiment’s database id, or None if no name was given or no matching experiment was found
Return type:: Optional[int]
Raises:: Exception – if the underlying database query fails

MetaDatabaseLoader.get_experiments_and_channels() → Dict[str, List[int] | None]¶

Retrieve a mapping of experiment names to their associated channel lists.

Calls get_experiment_names() to fetch all experiment identifiers, then maps each experiment to its corresponding list of channels using get_channels_by_experiment().

Returns:: Dictionary mapping experiment names to lists of channel indices.
Return type:: Dict[str, Optional[List[int]]]

Get a list of horizontal and vertical lines and associated labels to overlay on the graph generated by construct_fitted_event()

Parameters:

experiment (int) – get only events from this experiment
channel (int) – analyze only events from this channel
index (int) – the index of the target event

Returns:

a list of x locations to plot vertical lines and a list of y locations to plot horizontal lines, list of tuples to plot little x’s, labels for the vertical lines, labels for the horizontal lines, labels for x’s. Must be lists of equal length, or None

Return type:

Tuple[Optional[List[float]], Optional[List[float]], Optional[List[Tuple[float, float]]], Optional[List[str]], Optional[List[str]], Optional[List[str]]]

MetaDatabaseLoader.load_event_data(conditions: str | None = None, experiments_and_channels: Dict[str, List[int] | None] | None = None) → Generator[Dict[str, Any], 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 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:

conditions (Optional[str]) – Optional filter condition for query.
experiments_and_channels (Optional[Dict[str, Optional[List[int]]]]) – a dict of experiment names as keys as lists of channels to include as values. Can be None, and individual channel lists can be None to include all channels for that experiment

Returns:

a generator that returns primary database id, experiment_id, channel_id, event_id, samplerate, padding_before, padding_after, samplerate, and a numpy array with event data

Return type:

Generator[Dict[str, Any], bool, None]

MetaDatabaseLoader.load_metadata(columns: List[str], conditions: str | None = None, experiments_and_channels: Dict[str, List[int] | None] | None = None) → DataFrame | None¶

Fetch specified columns from the metadata database given a query

Will always include experiment_id, channel_id, and event_id in the dataframe in addition to requested columns.

Parameters:

columns (List[str]) – List of column names to retrieve.
conditions (Optional[str]) – Optional filter condition for query.
experiments_and_channels (Optional[Dict[str, Optional[List[int]]]]) – a dict of experiment names as keys as lists of channels to include as values. Can be None, and individual channel lists can be None to include all channels for that experiment

Returns:

pandas dataframe containing retrieved data, empty if the query matched no rows, or None if the query could not be built or run

Return type:

Optional[pd.DataFrame]

MetaDatabaseLoader.load_metadata_raw(conditions: str | None = None) → DataFrame | None¶

Execute a raw SQL query directly, bypassing all query construction.

Parameters:: conditions (Optional[str]) – A complete SQL query string.
Returns:: pandas dataframe containing retrieved data, empty if the query matched no rows, or None if the query failed or did not validate
Return type:: Optional[pd.DataFrame]

MetaDatabaseLoader.query_database_directly(query: str) → DataFrame | None¶

Run a given query on the DB after basic validation.

Parameters:: query (str) – query to run on the database
Returns:: List of numpy arrays containing retrieved data.
Return type:: Optional[pd.DataFrame]

MetaDatabaseLoader.query_database_directly_and_get_generator(query: str) → Generator[DataFrame, bool, None]¶

Run a given querry on the DB after basic validation and return a generator that feeds out one row at a time

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, bool, None]

MetaDatabaseLoader.report_channel_status(channel: int | None = None, init: bool = False) → str¶

Return a string detailing event counts per experiment and channel.

Parameters:

channel (Optional[int]) – channel ID. Currently unused at the base class level but retained for API compatibility with subclasses that may filter by channel.
init (bool) – True if the function is being called as part of plugin initialization. Default False.

Returns:

a formatted string listing the number of experiments and the event count per channel for each experiment, or "No experiments found." if the database is empty.

Return type:

str

Private Methods¶

Abstract Methods¶

These methods must be implemented by subclasses.

abstractmethod MetaDatabaseLoader._ensure_event_counts() → None¶

Ensure the event_counts summary table exists and is populated. If the table does not exist, create it, populate it from existing events, and add the appropriate triggers to keep it in sync going forward.

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

abstractmethod MetaDatabaseLoader._init() → None¶

Purpose: Perform generic class construction operations.

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.

abstractmethod MetaDatabaseLoader._load_event_data(query: str) → Any¶

Load data and return a generator that gives a one-row dataframe corresponding one row returned by query Make sure you exhaust 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 5 colums: event_id, channel_id, experiment_id, data_format, data, baseline, stdev, padding_before, padding_after, 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 yields one item per row matching the query, with id, event_id, channel_id, experiment_id, samplerate, padding_before, padding_after, and numpy array with event data for raw, filtered, and fitted data. The exact shape of each yielded item (e.g. dict vs. tuple) is defined by the concrete subclass; see its own docstring for the precise structure.
Return type:: Any

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

Purpose: Load and return the data specified by a valid SQL query, or None if the query could not be run

The data should be formatted as a pandas Dataframe object. A query that matched no rows must return an empty dataframe rather than None, so that callers can tell an empty result from a failed query.

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]

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

Purpose: Load and yield the data specified by a valid SQL query one row at a time. Useful in cases where _load_metadata() returns too much data for memory.

Data should be formatted as a pandas dataframe in line with _load_metadata(). Make sure you exhaust the generator when done with it, or else database 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]

abstractmethod MetaDatabaseLoader._validate_settings(settings: dict) → None¶

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

Parameters:: settings (dict) – Parameters required to configure this database loader.
Raises:: ValueError – If the settings dict does not contain the correct information.

Concrete Methods¶

MetaDatabaseLoader.__init__(settings: dict | None = None) → None¶: Initialize and set up the plugin, if settings are available at this stage

MetaDatabaseLoader._end_of_subquery(fragment: str, start: int) → int¶

Find the offset just past the ) that closes a subquery.

Parameters:

fragment (str) – A fragment of SQL, typically a WHERE clause body.
start (int) – Offset of the ( that opens the subquery.

Returns:

The offset just past the matching ), or the length of the fragment if the parentheses never balance.

Return type:

int

MetaDatabaseLoader._finalize_initialization() → None¶

Purpose: Apply application-specific settings to the plugin, if needed.

If additional initialization operations are required beyond the defaults provided in BaseDataPlugin or MetaDatabaseLoader that must occur after settings have been applied to the reader instance, you can override this function to add those operations.

MetaDatabaseLoader._find_ambiguous_id(conditions: str, aliases: Dict[str, str]) → str | None¶

Explain how to disambiguate a bare id in a condition, if there is one.

id is the only column a joined query cannot resolve on the user’s behalf: it means a different row in every table, so guessing one would silently filter against the wrong thing. Run this after _qualify_conditions(), since anything still bare at that point is genuinely unqualified. An id inside a subquery belongs to that subquery and is not reported, since it resolves against the tables the subquery names.

Parameters:

conditions (str) – A WHERE clause body that has already been qualified.
aliases (Dict[str, str]) – Table name to alias, for every table in the FROM clause.

Returns:

Guidance to show the user, or None if there is nothing ambiguous.

Return type:

Optional[str]

MetaDatabaseLoader._format_debug_msg(debug: str) → str¶

Strip out newlines and unnecessary whitespace from SQL queries for printing

Parameters:: debug (str) – a string containing an error message and an SQL string for correction
Returns:: the input string with whitepsace removed and newlines in it to format for export
Return type:: str

MetaDatabaseLoader._qualify_conditions(conditions: str, aliases: Dict[str, str]) → str¶

Qualify bare column references against the tables a query actually joins.

Lets a user write sublevel_duration < 100 AND experiment_id = 2 and have it work against FROM events e JOIN sublevels s. Only the rewritable parts of the condition are touched; anything inside a string literal or a subquery is left exactly as the user typed it, and a reference the user already qualified is left alone.

Parameters:

conditions (str) – A WHERE clause body, without the leading WHERE.
aliases (Dict[str, str]) – Table name to alias, for every table in the FROM clause, in join order. The first entry is the anchor.

Returns:

The condition with its bare column references qualified.

Return type:

str

MetaDatabaseLoader._references_column(fragment: str, column: str) → bool¶

Report whether a SQL fragment refers to a column in its own right.

Matches a qualified reference (s.duration) as well as a bare one, since the column name is sought on a word boundary rather than a token boundary. A reference inside a string literal or a subquery does not count, for the reasons _split_on_opaque_spans() gives, which is what keeps this in agreement with _qualify_conditions() about what the outer query refers to.

Parameters:

fragment (str) – A fragment of SQL, typically a WHERE clause body.
column (str) – The column name to look for.

Returns:

True if the fragment refers to the column outside any opaque span.

Return type:

bool

MetaDatabaseLoader._split_on_opaque_spans(fragment: str) → List[str]¶

Split a SQL fragment into alternating rewritable and opaque segments.

Even indices hold code that may be rewritten, odd indices hold spans that must be passed through untouched, and "".join of the result reproduces the input exactly. Two kinds of span are opaque:

A single-quoted string literal, so a column name that also appears as a value - sequence = 'sublevel_current' - is not rewritten along with the real column references.
A parenthesised subquery, from the ( of (SELECT through its matching ). A subquery names its own tables, so its column references resolve against those rather than against the outer query’s aliases, and rewriting them silently correlates the subquery to the outer row. A user who wants a correlated subquery qualifies the reference with the outer alias, which is what SQL requires of them anyway.

An unterminated literal or an unbalanced paren makes the rest of the fragment opaque, which leaves malformed input for the query validator to reject rather than rewriting it into something that parses.

Parameters:: fragment (str) – A fragment of SQL, typically a WHERE clause body.
Returns:: The alternating rewritable and opaque segments of the fragment.
Return type:: List[str]