PriDatabase

class vallenae.io.PriDatabase(filename, mode='ro')

IO wrapper for pridb database file.

__init__(filename, mode='ro')

Open pridb database file.

Parameters:

• filename (str) – Path to pridb database file

• mode (str) – Define database access: “ro” (read-only), “rw” (read-write), “rwc” (read-write and create empty database if it does not exist)

Methods

__init__(filename[, mode])	Open pridb database file.
channel()	Get list of channels.
close()	Close database connection.
columns()	Columns of data table.
connection()	Get SQLite connection object.
create(filename)	Create empty pridb.
fieldinfo()	Read fieldinfo table.
globalinfo()	Read globalinfo table.
iread_hits(*[, channel, time_start, ...])	Stream hits with returned iterable.
iread_markers(*[, time_start, time_stop, ...])	Stream markers with returned iterable.
iread_parametric(*[, time_start, time_stop, ...])	Stream parametric data with returned iterable.
iread_status(*[, channel, time_start, ...])	Stream status data with returned iterable.
listen([existing, wait, query_filter])	Listen to database changes and return new records.
read(**kwargs)	Read all data set types (hits, markers, parametric data, status data) from pridb to Pandas DataFrame.
read_hits(**kwargs)	Read hits to Pandas DataFrame.
read_markers(**kwargs)	Read marker to Pandas DataFrame.
read_parametric(**kwargs)	Read parametric data to Pandas DataFrame.
read_status(**kwargs)	Read status data to Pandas DataFrame.
rows()	Number of rows in data table.
tables()	Get table names.
to_datetime(time)	Convert a relative time to an absolute datetime object.
write_fieldinfo(field, info)	Write to fieldinfo table.
write_hit(hit)	Write hit to pridb.
write_marker(marker)	Write marker to pridb.
write_parametric(parametric)	Write parametric data to pridb.
write_status(status)	Write status data to pridb.

Attributes

connected	Check if connected to SQLite database.
filename	Filename of database.

static create(filename)

Create empty pridb.

Parameters:

filename (str) – Path to new pridb database file

channel()

Get list of channels.

Return type:

set[int]

to_datetime(time)

Convert a relative time to an absolute datetime object.

Parameters:

timestamp – Relative time in seconds

Return type:

datetime

read_hits(**kwargs)

Read hits to Pandas DataFrame.

Parameters:

**kwargs – Arguments passed to iread_hits

Return type:

DataFrame

Returns:

Pandas DataFrame with hit data

read_markers(**kwargs)

Read marker to Pandas DataFrame.

Parameters:

**kwargs – Arguments passed to iread_markers

Return type:

DataFrame

Returns:

Pandas DataFrame with marker data

read_parametric(**kwargs)

Read parametric data to Pandas DataFrame.

Parameters:

**kwargs – Arguments passed to iread_parametric

Return type:

DataFrame

Returns:

Pandas DataFrame with parametric data

read_status(**kwargs)

Read status data to Pandas DataFrame.

Parameters:

**kwargs – Arguments passed to iread_status

Return type:

DataFrame

Returns:

Pandas DataFrame with status data

read(**kwargs)

Read all data set types (hits, markers, parametric data, status data) from pridb to Pandas DataFrame.

Parameters:

**kwargs – Arguments passed to iread_hits, iread_markers, iread_parametric and iread_status

Return type:

DataFrame

Returns:

Pandas DataFrame with all pridb data set types

iread_hits(*, channel=None, time_start=None, time_stop=None, set_id=None, query_filter=None)

Stream hits with returned iterable.

Parameters:

• channel (Union[int, Sequence[int], None]) – None if all channels should be read. Otherwise specify the channel number or a list of channel numbers

• time_start (float | None) – Start reading at relative time (in seconds). Start at beginning if None

• time_stop (float | None) – Stop reading at relative time (in seconds). Read until end if None

• set_id (Union[int, Sequence[int], None]) – Read by SetID

• query_filter (str | None) – Optional query filter provided as SQL clause, e.g. “Amp > 5000 AND RiseT < 1000”

Return type:

SizedIterable[HitRecord]

Returns:

Sized iterable to sequential read hits

iread_markers(*, time_start=None, time_stop=None, set_id=None, query_filter=None)

Stream markers with returned iterable.

Parameters:

• time_start (float | None) – Start reading at relative time (in seconds). Start at beginning if None

• time_stop (float | None) – Stop reading at relative time (in seconds). Read until end if None

• set_id (Union[int, Sequence[int], None]) – Read by SetID

• query_filter (str | None) – Optional query filter provided as SQL clause, e.g. “Number > 11 AND Data LIKE ‘%TimeZone%’”

Return type:

SizedIterable[MarkerRecord]

Returns:

Sized iterable to sequential read markers

iread_parametric(*, time_start=None, time_stop=None, set_id=None, query_filter=None)

Stream parametric data with returned iterable.

Parameters:

• time_start (float | None) – Start reading at relative time (in seconds). Start at beginning if None

• time_stop (float | None) – Stop reading at relative time (in seconds). Read until end if None

• set_id (Union[int, Sequence[int], None]) – Read by SetID

• query_filter (str | None) – Optional query filter provided as SQL clause, e.g. “PA0 >= -5000 AND PA0 < 5000”

Return type:

SizedIterable[ParametricRecord]

Returns:

Sized iterable to sequential read parametric data

iread_status(*, channel=None, time_start=None, time_stop=None, set_id=None, query_filter=None)

Stream status data with returned iterable.

Parameters:

• channel (Union[int, Sequence[int], None]) – None if all channels should be read. Otherwise specify the channel number or a list of channel numbers

• time_start (float | None) – Start reading at relative time (in seconds). Start at beginning if None

• time_stop (float | None) – Stop reading at relative time (in seconds). Read until end if None

• set_id (Union[int, Sequence[int], None]) – Read by SetID

• query_filter (str | None) – Optional query filter provided as SQL clause, e.g. “RMS < 300 OR RMS > 500”

Return type:

SizedIterable[StatusRecord]

Returns:

Sized iterable to sequential read status data

listen(existing=False, wait=False, query_filter=None)

Listen to database changes and return new records.

Parameters:

• existing (bool) – Return already existing records

• wait (bool) – Wait for new records even if no acquisition (writer) is active. Otherwise the function returns after all records are read.

• query_filter (str | None) – Optional query filter provided as SQL clause, e.g. “Time >= 100 AND Chan == 2”

Yields:

New hit/marker/parametric/status data records

Return type:

Iterable[HitRecord | MarkerRecord | ParametricRecord | StatusRecord]

write_hit(hit)

Write hit to pridb.

Caution: HitRecord.set_id is ignored and automatically incremented.

Parameters:

hit (HitRecord) – Hit data set

Returns:

Index (SetID) of inserted row

write_marker(marker)

Write marker to pridb.

Caution: MarkerRecord.set_id is ignored and automatically incremented.

Parameters:

marker (MarkerRecord) – Marker data set

Returns:

Index (SetID) of inserted row

write_status(status)

Write status data to pridb.

Caution: StatusRecord.set_id is ignored and automatically incremented.

Parameters:

status (StatusRecord) – Status data set

Returns:

Index (SetID) of inserted row

write_parametric(parametric)

Write parametric data to pridb.

Caution: ParametricRecord.set_id is ignored and automatically incremented.

Parameters:

parametric (ParametricRecord) – Parametric data set

Returns:

Index (SetID) of inserted row

close()

Close database connection.

columns()

Columns of data table.

Return type:

tuple[str, ...]

property connected: bool

Check if connected to SQLite database.

connection()

Get SQLite connection object.

Raises:

RuntimeError – If connection is closed

Return type:

Connection

fieldinfo()

Read fieldinfo table.

The fieldinfo table stores informations about columns of the data table (like units).

Return type:

dict[str, dict[str, Any]]

Returns:

Dict of column names and informations (again a dict)

property filename: str

Filename of database.

globalinfo()

Read globalinfo table.

Return type:

dict[str, Any]

rows()

Number of rows in data table.

Return type:

int

tables()

Get table names.

Return type:

set[str]

write_fieldinfo(field, info)

Write to fieldinfo table.

Parameters:

• field (str) – Column name of data table

• info (dict[str, Any]) – Dict of properties and values, e.g. {“Unit”: “[Hz]”}

Raises:

ValueError – If field is not a column of data table