Backends

Base squirrel data storage backend interface

class squirrel.backends.core.SearchTerm(attr, operator, value)[source]

Methods

`count`(value, /)	Return number of occurrences of value.
`index`(value[, start, stop])	Return first index of value.

attr: str: Alias for field number 0

operator: str: Alias for field number 1

value: None | float | str | bool | int | Container[None | float | str | bool | int] | tuple[None | float | str | bool | int, ...]: Alias for field number 2

class squirrel.backends.mongo.MongoBackend(address: str)[source]

An integration layer between the Client and a MongoDB instance

Attributes:

root: Return the Root Entry in this backend

Methods

`add_pv`(setpoint, readback, description[, ...])	Add PV to the backend.
`add_snapshot`(snapshot)	Add snapshot to the backend.
`add_tag_group`(name, description)	Add new tag group.
`add_tag_to_group`(group_id, name[, description])	Add new tag to a tag group.
`archive_pv`(pv_id)	Archive PV in the backend.
`compare`(op, data, target)	Return whether data and target satisfy the op comparator, typically during application of a search filter.
`delete_entry`(entry)	Delete `entry` from the system (all instances) Throws BackendError if backend contains an entry with the same uuid as `entry` but different data
`delete_snapshot`(snapshot)	Delete the given snapshot from the backend.
`delete_tag_from_group`(group_id, tag_id)	Delete tag from a tag group
`delete_tag_group`(group_id)	Delete entire tag group.
`get_all_pvs`()	Get all PVs from the backend
`get_entry`(meta_id)	Get entry with `meta_id` Throws EntryNotFoundError
`get_meta_pvs`()	Return the PVs used as Snapshot metadata
`get_paged_pvs`(limit[, token, search_string])	Get PVs with setpoint or readback matching search_string
`get_pvs`([search_string])	Get PVs with setpoint or readback matching search_string
`get_snapshot`(uuid)	Fetch specific snapshot with uuid from the backend
`get_snapshots`([title, tags, meta_pvs])	Fetch snapshots from the backend.
`get_snapshots_in_date_range`()	TODO: Returns snapshots with timestamps in a given range.
`get_snapshots_in_index_range`()	TODO: Returns snapshots with indicies in a given range, indexed by reverse-chronological position in the complete list of snapshots.
`get_tags`()	Fetch tag definition from the backend.
`save_entry`(entry)	Save `entry` into the database Throws EntryExistsError
`search`(*search_terms[, meta_pvs])	Search for all entries matching the passed search terms.
`set_meta_pvs`(meta_pvs)	Set the PVs used as Snapshot metadata
`set_tags`(tags)	Set the definition of valid entry tags
`update_entry`(entry)	Update `entry` in the backend.
`update_pv`(pv_id[, setpoint, readback, ...])	Update PV in the backend.
`update_tag_group`(group_id[, name, description])	Update name and/or description of a tag group.
`update_tag_in_group`(group_id, tag_id[, ...])	Update tag in a tag group.

add_multiple_pvs

add_pv(setpoint, readback, description, device='', tags: dict[int, set[int]] | None = None, abs_tolerance=0, rel_tolerance=0, config_address=None) → PV[source]

Add PV to the backend. The new PV will have data according to the passed parameters.

Parameters:

setpointstr: EPICS address to use as a setpoint. May be empty, but must be set explicitly.
readbackstr: EPICS address to use as a readback. May be empty, but must be set explicitly.
descriptionstr: Description of the new PV
devicestr, optional: Alias device name of the new PV, defaults to empty string
tagsTagSet, optional: Set of tags assigned to the PV
abs_tolerancefloat, optional: Absolute tolerance
rel_tolerancefloat, optional: Relative tolerance
config_addressstr, optional: EPICS address for reading last config value. Defaults to empty.

Returns:

PV: PV instance containing all data received from the backend

Raises:

BackendError

add_snapshot(snapshot: Snapshot) → None[source]

Add snapshot to the backend.

Parameters:

snapshotSnapshot

Raises:

BackendError

add_tag_group(name, description) → int[source]

Add new tag group.

Parameters:

namestr: Name of the new tag group
descriptionstr: Description of the new tag group. May be empty, but must be set explicitly.

Returns:

str: ID of the newly created tag group, received from the backend

Raises:

BackendError

add_tag_to_group(group_id: str, name: str, description='') → str[source]

Add new tag to a tag group.

Parameters:

group_idstr: ID of the tag group
namestr: Name to use for the new tag
descriptionstr, optional: Description to use for the new tag

Returns:

str: ID of the newly created tag, received from the backend

Raises:

BackendError

archive_pv(pv_id) → None[source]

Archive PV in the backend. The PV will still appear in Snapshots that already exist, but will not appear in the list of current PVs nor will data for this PV be saved in future Snapshots.

Parameters:

pv_idstr: ID of the PV to archive

Raises:

BackendError

delete_snapshot(snapshot: Snapshot) → None[source]

Delete the given snapshot from the backend.

Raises:

BackendError

delete_tag_from_group(group_id, tag_id) → None[source]

Delete tag from a tag group

Parameters:

group_idint: ID of the tag group
tag_idint: ID of the tag to remove from the group

Raises:

BackendError

delete_tag_group(group_id) → None[source]

Delete entire tag group.

Parameters:

group_idint: ID of the tag group to delete

Raises:

BackendError

get_all_pvs() → Iterable[PV][source]

Get all PVs from the backend

Returns:

Iterable[PV]

Raises:

BackendError

get_paged_pvs(limit: int, token='', search_string='') → tuple[Iterable[PV], str][source]

Get PVs with setpoint or readback matching search_string

Parameters:

limitint: Maximum number of PVs to receive at a time
tokenstr: Continuation token received from the most recent call
search_stringstr

Returns:

tupe[Iterable[PV], str): First element is the PV payload; second element is the continuation token

Raises:

BackendError

get_pvs(search_string='') → Iterable[PV][source]

Get PVs with setpoint or readback matching search_string

Parameters:

search_stringstr

Returns:

Iterable[PV]

Raises:

BackendError

get_snapshot(uuid) → Snapshot[source]

Fetch specific snapshot with uuid from the backend

Parameters:

uuidstr: ID of the individual Snapshot to fetch

Returns:

Snapshot: Snapshot instance with the specified uuid

Raises:

BackendError

get_snapshots(title='', tags=None, meta_pvs=None) → Iterable[Snapshot][source]

Fetch snapshots from the backend.

Parameters:

titlestr, optional: Substring contained in the title field of desired Snapshots
tagsTagSet, optional: Tags contained in the tags field of desired Snapshots
meta_pvs: Not Implemented

Returns:

Iterable[Snapshot]: Snapshot instances that match the selected filters

Raises:

BackendError

get_snapshots_in_date_range() → None[source]: TODO: Returns snapshots with timestamps in a given range. This method is potentially valuable for limiting how much data is sent by the backend at once.

get_snapshots_in_index_range() → None[source]: TODO: Returns snapshots with indicies in a given range, indexed by reverse-chronological position in the complete list of snapshots. This method is potentially valuable for limiting how much data is sent by the backend at once.

get_tags() → dict[int, list[str | dict[int, str]]][source]

Fetch tag definition from the backend. Caches data for one minute.

Returns:

TagDef: Full tag definition received from the backend

Raises:

BackendError

Search for all entries matching the passed search terms.

Raises:

BackendError

update_pv(pv_id, setpoint='', readback='', description='', device='', tags=None, abs_tolerance=None, rel_tolerance=None) → None[source]

Update PV in the backend. Data will be updated for any passed parameter; data for other parameters will not be affected.

Parameters:

pv_idstr: ID of the PV to update
setpointstr, optional: A new setpoint address
readbackstr, optional: A new readback address
descriptionstr, optional: A new description
devicestr, optional: A new device name
tagsTagSet, optional: A new set of tags
abs_tolerancefloat, optional: A new absolute tolerance
rel_tolerancefloat, optional: A new relative tolerance

Raises:

BackendError

update_tag_group(group_id, name='', description='') → None[source]

Update name and/or description of a tag group.

Parameters:

group_idstr: ID of the tag group
namestr, optional: New name to use for the tag group
descriptionstr, optional: New description to use for the tag group

Raises:

BackendError

update_tag_in_group(group_id, tag_id, name='', description='') → None[source]

Update tag in a tag group. Name and/or description will be updated if passed as parameters.

Parameters:

group_idint: ID of the tag group
tag_idstr: ID of the tag to in the tag group
namestr, optional: New name to use for the tag
descriptionstr, optional: New description to use for the tag

Raises:

BackendError