qslib package

Submodules

qslib.base module

class qslib.base.BaseStatus(**kwargs)[source]

Bases: abc.ABC

classmethod from_bytes(out)[source]
Return type

~T

classmethod from_machine(connection)[source]
Return type

T

class qslib.base.MachineStatus(drawer, cover, lamp_status)[source]

Bases: qslib.base.BaseStatus

cover: str
drawer: str
lamp_status: str
class qslib.base.RunStatus(name, stage, num_stages, cycle, num_cycles, step, point, state)[source]

Bases: qslib.base.BaseStatus

cycle: int
name: str
num_cycles: int
num_stages: int
point: int
stage: int
state: str
step: int

qslib.cli module

exception qslib.cli.NoAccess[source]

Bases: BaseException

exception qslib.cli.NoNewAccess[source]

Bases: BaseException

class qslib.cli.OutP(level)[source]

Bases: object

error(s)[source]
good(s)[source]
level: bool
out(s)[source]
verbose(s)[source]
warn(s)[source]
qslib.cli.add_administrator_password(m, p, newpass)[source]
qslib.cli.add_controller_password(m, p, newpass)[source]
qslib.cli.check_access(host, p, controller_pw, admin_pw, default_controller)[source]
qslib.cli.error_ssh(m, p)[source]
qslib.cli.restart_is(m, p)[source]
qslib.cli.set_default_access(m, p)[source]
qslib.cli.start_ssh_backup(m, p, sshpass)[source]
qslib.cli.stop_ssh_backup(m, p)[source]

qslib.common module

qslib.data module

class qslib.data.FilterDataReading(pde, timestamp=None, sds_dir=None, set_temperatures='auto')[source]

Bases: object

cycle: int
exposure: int
property filename_reading_string: str
Return type

str

filter_set: FilterSet
classmethod from_file(path, sds_dir)[source]
Return type

FilterDataReading

property plate_fluorescence: numpy.ndarray
Return type

ndarray

property plate_set_temperatures: numpy.ndarray
Return type

ndarray

property plate_temperatures: numpy.ndarray
Return type

ndarray

point: int
set_temperatures: npt.NDArray[np.float64] | None
set_timestamp_by_quantdata(qstring)[source]
Return type

float

stage: int
step: int
temperatures: npt.NDArray[np.float64]
timestamp: float | None
to_lineprotocol(run_name=None, sample_array=None)[source]
Return type

List[str]

well_fluorescence: npt.NDArray[np.float64]
property well_set_temperatures: numpy.ndarray
Return type

ndarray

property well_temperatures: numpy.ndarray
Return type

ndarray

class qslib.data.FilterSet(ex, em, quant=True)[source]

Bases: object

Representation of a filter set, potentially including the “quant” parameter used by HACFILT in SCPI protocols.

em: int
ex: int
classmethod fromstring(string)[source]
Return type

FilterSet

property hacform: str
Return type

str

property lowerform: str
Return type

str

quant: bool = True
to_xml()[source]
Return type

Element

property upperform: str
Return type

str

qslib.data.df_from_readings(readings, start_time=None)[source]
Return type

pd.DataFrame

qslib.experiment module

Experiment class and related.

exception qslib.experiment.AlreadyExistsError(machine: InitVar[Machine], name: str)[source]

Bases: qslib.experiment.MachineError

A run already exists in uncollected (experiment:) with the same name.

name: str
class qslib.experiment.Experiment(name=None, protocol=None, plate_setup=None, _create_xml=True)[source]

Bases: object

A QuantStudio experiment / EDS file

This class can create, modify, load and save experiments in several ways, run them, and control and modify them while running.

Experiments can be loaded from a file:

>>> exp: Experiment = Experiment.from_file("experiment.eds")

They can also be loaded from a running experiment:

>>> machine = Machine("localhost", 7000, password="password")
>>> exp = Experiment.from_running(machine)

Or from the machine’s storage:

>>> exp = Experiment.from_machine(machine, "experiment")

They can also be created from scratch:

>>> exp = Experiment("an-experiment-name")
>>> exp.protocol = Protocol([Stage([Step(time=60, temperature=60)])])
>>> exp.plate_setup = PlateSetup({"sample_name": "A5"})

And they can be run on a machine:

>>> exp.run(machine)

Data can be accessed in a few ways:

The (hopefully) easiest way is with welldata, which has multi-indexes for both rows and columns.

>>> exp.welldata.loc[('x1-m4', 4), [('time', 'hours'), ('A05', 'fl')]].plot()

Or for a temperature curve:

>>> exp.welldata.loc[('x1-m4', 4), [('A05', 'st'), ('A05', 'fl')]].plot()

filterdata should still work normally.

Notes

There are a few differences in how QSLib and AB’s software handles experiments.

  • AB’s software considers the run as starting when the machine indicates “Run Starting”. This is stored as runstarttime in QSLib, but as it may include the lamp warmup (3 minutes) and other pre-actual-protocol time, QSLib instead prefers activestarttime, which it sets from the beginning of the first real (not PRERUN) Stage, at which point the machine starts its own active clock and starts ramping to the first temperature. QSLib uses this as the start time reference in its data, and also includes the timestamp from the machine.

  • The machine has a specific language for run protocols. QSLib uses this language. AB’s Design and Analysis software does not, instead using an XML format. Not everything in the machine’s language is possible to express in the XML format (eg, disabling pcr analysis, saving images); the XML format has some concepts not present in the machine format, and is generally more complicated and harder to understand. QSLib uses and trusts the machine’s protocol if at all possible, even for files written by AB D&A (it is stored in the log if the run has started).

  • QSLib will try to write a reasonable XML protocol for AB D&A to see, but it may by an approximation or simply wrong, if the actual protocol can’t be expressed there. It will also store the actual protocol in tcprotocol.xml, and its own representation.

  • By default, creating a step with a per-cycle increment in QSLib starts the change on cycle 2, not cycle 1, as is the default in the software.

  • Immediate pause/resume, mid-run stage addition, and other functions are not supported by AB D&A and experiments using them may confuse the software later.

  • QSLib writes notes to XML files, and tries to create reasonable XML files for AB D&A, but may still cause problems. At the moment, it makes clear that its files are its own (setting software versions in experiment.xml and Manifest.mf).

abort(machine=None)[source]

If this experiment is running, abort it, stopping it immediately.

Requires and takes exclusive Controller access on the machine.

Raises

NotRunningError – the experiment is not currently running

Return type

None

activeendtime: datetime | None

The actual end of the main part of the run, indicated by “Stage POSTRun” or an abort.

activestarttime: datetime | None

The actual beginning of the first stage of the run, defined as the first “Run Stage” message in the log after “Stage PRERUN”. This is not what AB’s software considers the start of a run.

property all_filters: Collection[qslib.data.FilterSet]

All filters used at some point in the experiment.

If the experiment has data, this is based on the existing data. Otherwise, it is based on the experiment protocol.

Return type

Collection[FilterSet]

change_protocol(new_protocol, machine=None, force=False)[source]

For a running experiment and an updated protocol, check compatibility with the current run, and if possible, update the protocol in the experiment file and on the machine, changing the current run.

Changes that should be possible:

  • Changing the number of cycles of the current run to a higher or lower value (but higher or equal to the current cycle number), allowing stages to be lengthened, shortened, or stopped.

  • Adding new stages after the current stage.

  • Arbitrarily changing any stage that hasn’t started.

For safest results, ensure power saving is turned off in the Android software.

Parameters
  • new_protocol (Protocol) – [description]

  • machine (Machine, optional) – [description], by default None

Return type

None

change_protocol_from_now(new_stages, machine=None)[source]

For a running experiment, change the remaining stages to be the provided :param:`new_stages` list. This is a convenience function that:

  1. Gets the currently-running stage and cycle.

  2. Sets the repeat number of the current stage to its current cycle, thus ending it after the end of the current cycle.

  3. Changes the remainder of the stages to be those in the :param:`new_stages` list.

Because this does not impact any current or past stages, there is less of a need to ensure that the stages provided are compatible with the old protocol. The only check done is to ensure that, if the provided stages have any collection commands using default filters, the old protocol has specified default filters. This function does not allow the default filters to be changed: if you want to use filters other than the defaults, or defaults were not provided in the old protocol, then either specify filters explicitly (recommended) for new stages you’d like to be different, or use Experiment.change_protocol directly.

Return type

None

createdtime: datetime

The run creation time.

data_for_sample(sample)[source]

Convenience function to return data for a specific sample.

Finds wells using self.plate_setup.sample_wells[sample], then returns self.welldata.loc[:, wells]

Parameters

sample (str) – sample name

Returns

Slice of welldata. Will have multiple wells if sample is in multiple wells.

Return type

pd.Dataframe

property filter_strings: list[str]

All filters, as x?-m? strings, used at some point in the experiment.

Return type

list[str]

property filterdata: pandas.core.frame.DataFrame
Return type

DataFrame

classmethod from_file(file)[source]

Load an experiment from an EDS file.

Returns

file – The filename or file handle to read.

Return type

str or os.PathLike[str] or IO[bytes]

Raises

ValueError – if the file does not appear to be an EDS file (lacks an experiment.xml).

classmethod from_machine(machine, name)[source]

Create an experiment from data on a machine, checking the running experiment if any, the machine’s public_run_complete storage, and the machine’s uncollected storage.

Parameters
  • machine (Machine | str) – the machine to connect to, either as a Machine or a host name

  • name (str) – the name of the run

Returns

a copy of the experiment

Return type

Experiment

classmethod from_machine_storage(machine, name)[source]

Create an experiment from the one currently running on a machine.

Parameters

machine (Machine) – the machine to connect to

Returns

a copy of the experiment

Return type

Experiment

classmethod from_running(machine)[source]

Create an experiment from the one currently running on a machine.

Parameters

machine (Machine) – the machine to connect to

Returns

a copy of the running experiment

Return type

Experiment

classmethod from_uncollected(machine, name, move=False)[source]

Create an experiment from the uncollected (not yet compressed) storage.

Parameters
  • machine (Machine) – the machine to connect to

  • name (str) – the name of the run to collect.

Returns

a copy of the experiment

Return type

Experiment

get_status(machine=None)[source]

Return the status of the experiment, if currently running.

Requires Observer access on the machine.

Raises

NotRunningError – the experiment is not currently running

Return type

RunStatus

info(format='markdown', plate='list')[source]

Generate a summary of the experiment, with some formatting configuation. str() uses this with default parameters.

Parameters
  • format ("markdown" or "org", optional) – Format of output, currently “markdown” or “org”, and currently matters only when plate is “table”. By default “markdown”. If an unknown value, passed as tablefmt to tabulate.

  • plate ("list" or "table", optional) – Format of plate information. “list” gives a list of samples, “table” outputs a plate layout table (possibly quite wide). By default “list”.

Returns

Summary

Return type

str

info_html()[source]

Create a self-contained HTML summary (returned as a string, but very large) of the experiment.

Return type

str

machine: Machine | None
modifiedtime: datetime

The last modification time. QSLib sets this on write. AB D&A may not.

name: str

Experiment name, also used as file name on the machine.

num_zones: int

The number of temperature zones (excluding cover), or -1 if not known.

pause_now(machine=None)[source]

If this experiment is running, pause it (immediately).

Requires and takes exclusive Controller access on the machine.

Raises

NotRunningError – the experiment is not currently running

Return type

None

plate_setup: PlateSetup

Plate setup for the experiment.

plot_anneal_melt(samples=None, filters=None, anneal_stages=None, melt_stages=None, between_stages=None, process=None, normalization=None, ax=None, marker=None, legend=True, figure_kw=None, line_kw=None)[source]

Plots anneal/melt curves.

This uses solid lines for the anneal, dashed lines for the melt, and dotted lines for anything “between” the anneal and melt (for example, a temperature hold).

Line labels are intended to provide full information when read in combination with the axes title. They will only include information that does not apply to all lines. For example, if every line is from the same filter set, but different samples, then only the sample will be shown. If every line is from the same sample, but different filter sets, then only the filter set will be shown. Wells are shown if a sample has multiple wells.

Parameters
  • samples (str | Sequence[str] | None) – A reference to a single sample (a string), a list of sample names, or a Python regular expression as a string, matching sample names (full start-to-end matches only). Well names may also be included, in which case each well will be treated without regard to the sample name that may refer to it. Note this means you cannot give your samples names that correspond with well references. If not provided, all (named) samples will be included.

  • filters (str | FilterSet | Collection[str | FilterSet] | None) – Optional. A filterset (string or FilterSet) or list of filtersets to include in the plot. Multiple filtersets will be plotted on the same axes. Optional; if None, then all filtersets with data in the experiment will be included.

  • anneal_stages (int | Sequence[int] | None) – Optional. A stage or list of stages (integers, starting from 1), corresponding to the anneal, melt, and stages between the anneal and melt (if any). Any of these may be None, in which case the function will try to determine the correct values automatically.

  • melt_stages (int | Sequence[int] | None) – Optional. A stage or list of stages (integers, starting from 1), corresponding to the anneal, melt, and stages between the anneal and melt (if any). Any of these may be None, in which case the function will try to determine the correct values automatically.

  • between_stages (int | Sequence[int] | None) – Optional. A stage or list of stages (integers, starting from 1), corresponding to the anneal, melt, and stages between the anneal and melt (if any). Any of these may be None, in which case the function will try to determine the correct values automatically.

  • normalization (Processor | None) – Optional. A Normalizer instance to apply to the data. By default, this is NormRaw, which passes through raw fluorescence values. NormToMeanPerWell also works well.

  • ax (plt.Axes' | None) – Optional. An axes to put the plot on. If not provided, the function will create a new figure, by default with constrained_layout=True, though this can be modified with figure_kw.

  • marker (str | None) – The marker format for data points, or None for no markers (default).

  • legend (bool | Literal['inset', 'right']) – Whether to add a legend. True (default) decides whether to have the legend as an inset or to the right of the axes based on the number of lines. “inset” and “right” specify the positioning. Note that for “right”, you must use some method to adjust the axes positioning: constrained_layout, tight_layout, or manually reducing the axes width are all options.

  • figure_kw (Mapping[str, Any] | None) – Optional. A dictionary of options passed through as keyword options to the figure creation. Only applies if ax is None.

  • line_kw (Mapping[str, Any] | None) – Optional. A dictionary of keywords passed to all three plotting commands.

Returns

The axes object of the plot.

Return type

plt.Axes

plot_over_time(samples=None, filters=None, stages=slice(None, None, None), process=None, normalization=None, ax=None, legend=True, temperatures='axes', marker=None, stage_lines=True, annotate_stage_lines=False, figure_kw=None, line_kw=None)[source]

Plots fluorescence over time, optionally with temperatures over time.

Line labels are intended to provide full information when read in combination with the axes title. They will only include information that does not apply to all lines. For example, if every line is from the same filter set, but different samples, then only the sample will be shown. If every line is from the same sample, but different filter sets, then only the filter set will be shown. Wells are shown if a sample has multiple wells.

Parameters
  • samples (str | Sequence[str] | None) – A reference to a single sample (a string), a list of sample names, or a Python regular expression as a string, matching sample names (full start-to-end matches only). Well names may also be included, in which case each well will be treated without regard to the sample name that may refer to it. Note this means you cannot give your samples names that correspond with well references. If not provided, all (named) samples will be included.

  • filters (str | FilterSet | Collection[str | FilterSet] | None) – Optional. A filterset (string or FilterSet) or list of filtersets to include in the plot. Multiple filtersets will be plotted on the same axes. Optional; if None, then all filtersets with data in the experiment will be included.

  • stages (slice | int | Sequence[int]) – Optional. A stage, list of stages, or slice (all using integers starting from 1), to include in the plot. By default, all stages are plotted. For example, to plot stage 2, use stages=2; to plot stages 2 and 4, use stages=[2, 4], to plot stages 3 through 15, use stages=slice(3, 16) (Python ranges are exclusive on the end). Note that is a slice, you can use None instead of a number to denote the beginning/end.

  • normalization (Processor | None) – Optional. A Normalizer instance to apply to the data. By default, this is NormRaw, which passes through raw fluorescence values. NormToMeanPerWell also works well.

  • temperatures (Literal[False, 'axes', 'inset', 'twin']) –

    Optional (default “axes”). Several alternatives for displaying temperatures. “axes” uses a separate axes (created if ax is not provided, otherwise ax must be a list of two axes).

    Temperatures are from Experiment.temperature, and are thus the temperatures as recorded during the run, not the set temperatures. Note that this has a very large number of data points, something that should be dealt with at some point.

  • ax (plt.Axes' | 'Sequence[plt.Axes]' | None) – Optional. An axes to put the plot on. If not provided, the function will create a new figure, by default with constrained_layout=True, though this can be modified with figure_kw. If temperatures=”axes”, you must provide a list or tuple of two axes, the first for fluorescence, the second for temperature.

  • marker (str | None) – The marker format for data points, or None for no markers (default).

  • legend (bool | Literal['inset', 'right']) – Whether to add a legend. True (default) decides whether to have the legend as an inset or to the right of the axes based on the number of lines. “inset” and “right” specify the positioning. Note that for “right”, you must use some method to adjust the axes positioning: constrained_layout, tight_layout, or manually reducing the axes width are all options.

  • stage_lines (bool | Literal['fluorescence', 'temperature']) – Whether to include dotted vertical lines on transitions between stages. If “fluorescence” or “temperature”, include only on one of the two axes.

  • annotate_stage_lines (bool | float | Literal['fluorescence', 'temperature'] | Tuple[Literal['fluorescence', 'temperature'], float]) – Whether to include text annotations for stage lines. Float parameter allows setting the minimum duration of stage, as a fraction of total plotted time, to annotate, in order to avoid overlapping annotations (default threshold is 0.05).

  • figure_kw (Mapping[str, Any] | None) – Optional. A dictionary of options passed through as keyword options to the figure creation. Only applies if ax is None.

  • line_kw (Mapping[str, Any] | None) – Optional. A dictionary of keywords passed to fluorescence plot commands.

Return type

Sequence[plt.Axes]

plot_protocol(ax=None)[source]

A plot of the temperature and data collection points in the experiment’s protocol.

Return type

Tuple[plt.Axes, Tuple[List[plt.Line2D], List[plt.Line2D]]]

plot_temperatures(*, sel=slice(None, None, None), hours=None, ax=None, stage_lines=True, annotate_stage_lines=True, legend=False, figure_kw=None, line_kw=None)[source]

Plot sample temperature readings.

Parameters
  • sel (slice | Callable[[pd.DataFrame], bool]) – A selector for the temperature DataFrame. This is not necessarily easy to use; hours is an easier alternative.

  • hours (tuple[float, float] | None) – Constructs a selector to show temperatures for a time range. :param:`sel` should not be set.

  • ax (Optional[plt.Axes]) – Optional. An axes to put the plot on. If not provided, the function will create a new figure, by default with constrained_layout=True, though this can be modified with figure_kw.

  • stage_lines (bool) – Whether to include dotted vertical lines on transitions between stages.

  • annotate_stage_lines (bool | float) – Whether to include text annotations for stage lines. Float parameter allows setting the minimum duration of stage, as a fraction of total plotted time, to annotate, in order to avoid overlapping annotations (default threshold is 0.05).

  • legend (bool) – Whether to add a legend.

  • figure_kw (Mapping[str, Any] | None) – Optional. A dictionary of options passed through as keyword options to the figure creation. Only applies if ax is None.

  • line_kw (Mapping[str, Any] | None) – Optional. A dictionary of keywords passed to plot commands.

Return type

plt.Axes

protocol: Protocol

Temperature and reading protocol for the experiment.

property rawdata: pandas.core.frame.DataFrame
Return type

DataFrame

resume(machine=None)[source]

If this experiment is running, resume it.

Requires and takes exclusive Controller access on the machine.

Raises

NotRunningError – the experiment is not currently running

Return type

None

run(machine=None, require_exclusive=False, require_drawer_check=True)[source]

Load the run onto a machine, and start it.

Parameters

machine (MachineReference | None) – The machine to run on, by default None, in which case the machine associated with the run (if any) is used.

Raises
Return type

None

runendtime: datetime | None

The run end time as a datetime, taken from the log. This is the end of the run,

runstarttime: datetime | None

The run start time as a datetime. This is taken directly from the log, ignoring the software-set value and replacing it on save if possibe. It is defined as the moment the machine records “Run Starting” in its log, using its timestamp. This may be 3 minutes before the start of the protocol if the lamp needs to warm up. It should be the same value as defined by AB’s software.

Use activestarttime for a more accurate value.

None if the file has not been updated since the start of the run

runstate: Literal['INIT', 'RUNNING', 'COMPLETE', 'ABORTED', 'STOPPED', 'UNKNOWN']

Run state, possible values INIT, RUNNING, COMPLETE, ABORTED, STOPPED(?).

property runtitle_safe: str

Run name with ” ” replaced by “_”.

Return type

str

property sample_wells: dict[str, list[str]]

A dictionary of sample names to sample wells (convenience read/write access to the PlateSetup .

Return type

dict[str, list[str]]

save_file(file, overwrite=False)[source]

Save an EDS file of the experiment. This should be readable by AB’s software, but makes no attempt to hide that it was written by QSLib, and contains some other information. By default, this will refuse to overwrite an existing file.

Parameters
  • file (str or IO[bytes]) – A filename or open binary IO.

  • overwrite (bool, optional) – If True, overwrite any existing file without warning. Defaults to False.

Return type

None

save_file_without_changes(file, overwrite=False)[source]

Save an EDS file of the experiment. Unlike save_file, this will not update any parts of the file, so if it has not been modified elsewhere, it will be the same as when it was loaded. By default, this will refuse to overwrite an existing file.

Parameters
  • file (str or IO[bytes]) – A filename or open binary IO.

  • overwrite (bool, optional) – If True, overwrite any existing file without warning. Defaults to False.

Return type

None

stop(machine=None)[source]

If this experiment is running, stop it after the end of the current cycle.

Requires and takes exclusive Controller access on the machine.

Raises

NotRunningError – the experiment is not currently running

Return type

None

summary(format='markdown', plate='list')[source]
Return type

str

sync_from_machine(machine=None, log_method='eval')[source]

Try to synchronize the data in the experiment to the current state of the run on a machine, more efficiently than reloading everything.

Return type

None

temperatures: pd.DataFrame | None = None

A DataFrame of temperature readings, at one second resolution, during the experiment (and potentially slightly before and after, if included in the message log).

Columns (as multi-index):

(“time”, …)float

Time of temperature reading, for choices of “timestamp” (Unix timestamp in seconds), “seconds” (seconds since the active start of the run), or “hours”. The latter two may be negative, and may not be set if the run never became active.

(“sample”, …)float

Sample temperature for blocks 1, 2, …, 6, and average in “avg”.

(“block”, …)float

Block temperature for blocks 1, 2, …, 6, and average in “avg”.

(“other”, “cover”)float

Cover temperature

(“other”, “heatsink”)float

Heatsink temperature

property welldata: pandas.core.frame.DataFrame

A DataFrame with fluorescence reading information.

Indices (multi-index) are (filter_set, stage, cycle, step, point), where filter_set is a string in familiar form (eg, “x1-m4”) and the rest are int.

Columns (as multi-index):

(“time”, …)float

Time of the data collection, taken from the .quant file. May differ for different filter sets. Options are “timestamp” (unix timestamp in seconds), “seconds”, and “hours” (the latter two from the active start of the run).

(well, option)float

Data for a well, with well formatted like “A05”. Options are “rt” (read temperature from .quant file), “st” (more stable temperature), and “fl” (fluorescence).

(“exposure”, “exposure”)float

Exposure time from filterdata.xml. Misleading, because it only refers to the longest exposure of multiple exposures.

Return type

DataFrame

writesoftware: str

A string describing the software and version used to write the file.

exception qslib.experiment.MachineBusyError(machine: InitVar[Machine], current: RunStatus)[source]

Bases: qslib.experiment.MachineError

The machine is busy.

current: RunStatus
exception qslib.experiment.MachineError(machine: InitVar[Machine])[source]

Bases: Exception

Base class for an error from a machine.

host: str
machine: InitVar[Machine]
port: int | str
exception qslib.experiment.NotRunningError(machine: InitVar[Machine], run: str, current: RunStatus)[source]

Bases: qslib.experiment.MachineError

The named experiment is not currently running.

current: RunStatus
run: str

qslib.machine module

class qslib.machine.Machine(host, password=None, automatic=True, max_access_level=AccessLevel.Controller, port=7000, _initial_access_level=AccessLevel.Observer)[source]

Bases: object

A connection to a QuantStudio machine. The connection can be opened and closed, and reused. A maximum access level can be set and changed, which will prevent the access level from going above that level.

By default, the class tries to handle connections and access automatically.

Parameters
  • host (str) – The host name or IP to connect to.

  • password (str | None) – The password to use. Note that this class does not obscure or protect the password at all, because it should not be relied on for security. See Security Considerations for more information.

  • automatic (bool) – Whether or not to automatically handle connection, disconnection, and where possible, access level. Default True.

  • max_access_level ("Observer", "Controller", "Administrator", or "Full") – The maximum access level to allow. This is not the initial access level, which will be Observer. The parameter can be changed later by changing the max_access_level attribute.

  • port (int) – The port to connect to. (Use the normal SCPI port, not the line-editor connection usually on 2323). Default is 7000.

abort_current_run()[source]

Abort (stop immediately) the current run.

Return type

None

property access_level: qslib.scpi_commands.AccessLevel
Return type

AccessLevel

asdict(password=False)[source]
Return type

dict[str, str | int]

at_access(access_level, exclusive=False, stealth=False)[source]
Return type

Generator[Machine, None, None]

automatic: bool = True
connect()[source]

Open the connection manually.

Return type

None

property connected: bool

Whether or not there is a current connection to the machine.

Note that when using automatic connections, this will usually be False, because connections will only be active when running a command.

Return type

bool

property connection: qslib.qsconnection_async.QSConnectionAsync

The QSConnectionAsync for the connection, or a ConnectionError.

Return type

QSConnectionAsync

cover_lower(check=True, ensure_drawer=True)[source]

Lower/engage the plate cover, closing the drawer if needed.

Return type

None

property cover_position: Literal['Up', 'Down', 'Unknown', '']

Return the cover position from the ENG? command. Note that this does not always seem to work.

Return type

Literal[‘Up’, ‘Down’, ‘Unknown’, ‘’]

property current_run_name: str | None

Name of current run, or None if no run is active.

Return type

str | None

define_protocol(protocol)[source]

Send a protocol to the machine. This is not related to a particular experiment. The name on the machine is set by the protocol.

Parameters

protocol (Protocol) – protocol to send

Return type

None

disconnect()[source]

Cleanly disconnect from the machine.

Return type

None

drawer_close(lower_cover=True, check=True)[source]

Close the machine drawer using the OPEN command. This will ensure proper cover/drawer operation. It will not check run status, and will open and close the drawer during runs and potentially during imaging.

By default, it will lower the cover automaticaly after closing, use lower_cover=False to not do so.

Return type

None

drawer_open()[source]

Open the machine drawer using the OPEN command. This will ensure proper cover/drawer operation. It will not check run status, and will open and close the drawer during runs and potentially during imaging.

Return type

None

property drawer_position: Literal['Open', 'Closed', 'Unknown']

Return the drawer position from the DRAW? command.

Return type

Literal[‘Open’, ‘Closed’, ‘Unknown’]

ensured_connection(access_level=AccessLevel.Observer)[source]
Return type

Generator[Machine, None, None]

get_access_level()[source]
Return type

tuple[AccessLevel, bool, bool]

get_running_protocol()[source]
Return type

Protocol

host: str
list_files(path: str, *, leaf: str = "'FILE'", verbose: Literal[True], recursive: bool = 'False') list[dict[str, Any]][source]
list_files(path: str, *, leaf: str = "'FILE'", verbose: Literal[False], recursive: bool = 'False') list[str]
Return type

list[str] | list[dict[str, Any]]

list_runs_in_storage()[source]

List runs in machine storage.

Returns

run filenames. Retrieve with load_run_from_storage (to open as :any`Experiment`) or save_run_from_storage (to download and save it without opening.)

Return type

list[str]

load_run_from_storage(path)[source]
Return type

Experiment

machine_status()[source]

Return information on the status of the machine.

Return type

MachineStatus

property max_access_level: qslib.scpi_commands.AccessLevel
Return type

AccessLevel

password: str | None = None
pause_current_run()[source]

Pause the current run now.

Return type

None

pause_current_run_at_temperature()[source]
Return type

None

port: int = 7000
property power: bool

Get and set the machine’s operational power (lamp, etc) as a bool.

Setting this to False will not turn off the machine, just power down the lamp, temperature control, etc. It will do so even if there is currently a run.

Return type

bool

read_dir_as_zip(path, leaf='FILE')[source]

Read a directory on the

Parameters
  • path (str) – path on the machine

  • leaf (str, optional) – leaf to use, by default “FILE”

Returns

the returned zip file

Return type

zipfile.ZipFile

read_file(path, context=None, leaf='FILE')[source]

Read a file.

Parameters
  • path (str) – File path on the machine.

  • context (str | None (default None)) – Context.

  • leaf (str (default FILE)) –

Returns

returned file

Return type

bytes

restart_system()[source]

Restart the system (both the InstrumentServer and android interface) by killing the zygote process.

Return type

None

resume_current_run()[source]

Resume the current run.

Return type

None

run_command(command)[source]

Run a SCPI command, and return the response as a string. Waits for OK, not just NEXT.

Parameters

command (str) – command to run

Returns

Response message (after “OK”, not including it)

Return type

str

Raises

CommandError – Received an Error response.

run_command_bytes(command)[source]

Run an SCPI command, and return the response as bytes (undecoded). Returns after the command is processed (OK or NEXT), but potentially before it has completed (NEXT).

Parameters

command (str | bytes | SCPICommand) – command to run

Returns

Response message (after “OK” or “NEXT”, likely “” in latter case)

Return type

bytes

Raises

CommandError – Received

run_command_to_ack(command)[source]

Run an SCPI command, and return the response as a string. Returns after the command is processed (OK or NEXT), but potentially before it has completed (NEXT).

Parameters

commands – command to run

Returns

Response message (after “OK” or “NEXT”, likely “” in latter case)

Return type

str

Raises

CommandError – Received an Error response.

run_status()[source]

Return information on the status of any run.

Return type

RunStatus

save_run_from_storage(machine_path, download_path, overwrite=False)[source]

Download a file from run storage on the machine.

Parameters
  • machine_path (str) – filename on the machine

  • download_path (str | IO[bytes]) – filename to download to, or an open file

  • overwrite (bool, optional) – if False and provided a filename rather than an open file, will not overwrite existing filies; by default False

Return type

None

set_access_level(access_level, exclusive=False, stealth=False)[source]
Return type

None

property status: qslib.base.RunStatus

Return the current status of the run.

Return type

RunStatus

stop_current_run()[source]

Stop (stop after cycle end) the current run.

Return type

None

write_file(path, data)[source]
Return type

None

qslib.monitor module

class qslib.monitor.Collector(config)[source]

Bases: object

async compile_eds(connection, name)[source]
Return type

None

async docollect(args, state, connection)[source]
Return type

None

async handle_led(topic, message, timestamp)[source]
Return type

None

async handle_msg(state, c, topic, message, timestamp)[source]
Return type

None

async handle_run_msg(state, c, topic, message, timestamp)[source]
Return type

None

inject(t, flush=False)[source]
Return type

None

property ipdir: Path | None
Return type

Path | None

async matrix_announce(msg)[source]
Return type

None

async monitor(connected_fut=None)[source]
Return type

None

async reliable_monitor(connected_fut=None)[source]
Return type

None

run_ip_path(name)[source]
Return type

Path

async setup_new_rundir(connection, name, *, firstmsg=None, overwrite=False)[source]
Return type

None

async sync_completed(connection, name)[source]
Return type

None

class qslib.monitor.Config(matrix=None, influxdb=None, machine=MachineConfig(password=None, name='localhost', host='localhost', port='7000', retries=3, compile=False), sync=SyncConfig(completed_directory=None, in_progress_directory=None))[source]

Bases: object

influxdb: Optional[qslib.monitor.InfluxConfig] = None
machine: qslib.monitor.MachineConfig = MachineConfig(password=None, name='localhost', host='localhost', port='7000', retries=3, compile=False)
matrix: Optional[qslib.monitor.MatrixConfig] = None
sync: qslib.monitor.SyncConfig = SyncConfig(completed_directory=None, in_progress_directory=None)
class qslib.monitor.InfluxConfig(token, org, bucket, url)[source]

Bases: object

bucket: str
org: str
token: str
url: str
class qslib.monitor.LEDStatus(temperature, current, voltage, junctemp)[source]

Bases: object

current: float
junctemp: float
temperature: float
voltage: float
class qslib.monitor.MachineConfig(password=None, name='localhost', host='localhost', port='7000', retries=3, compile=False)[source]

Bases: object

compile: bool = False
host: str = 'localhost'
name: str = 'localhost'
password: Optional[str] = None
port: str = '7000'
retries: int = 3
class qslib.monitor.MachineState(zone_targets, zone_controls, cover_target, cover_control, drawer)[source]

Bases: object

cover_control: bool
cover_target: float
drawer: str
async classmethod from_machine(c)[source]
Return type

MachineState

async refresh(c)[source]
Return type

None

zone_controls: List[bool]
zone_targets: List[float]
class qslib.monitor.MatrixConfig(password, user, room, host, encryption=False)[source]

Bases: object

encryption: bool = False
host: str
password: str
room: str
user: str
class qslib.monitor.RunState(name=None, stage=None, cycle=None, step=None, plate_setup=None)[source]

Bases: object

cycle: Optional[int] = None
async classmethod from_machine(c)[source]
Return type

RunState

name: Optional[str] = None
plate_setup: Optional[PlateSetup] = None
async refresh(c)[source]
Return type

None

stage: Optional[int | str] = None
statemsg(timestamp)[source]
Return type

str

step: Optional[int] = None
class qslib.monitor.State(run, machine)[source]

Bases: object

async classmethod from_machine(c)[source]
Return type

State

machine: qslib.monitor.MachineState
run: qslib.monitor.RunState
class qslib.monitor.SyncConfig(completed_directory=None, in_progress_directory=None)[source]

Bases: object

completed_directory: Optional[str] = None
in_progress_directory: Optional[str] = None
async qslib.monitor.get_runinfo(c)[source]
Return type

State

qslib.monitor.index_to_filename_ref(i)[source]
Return type

str

qslib.monitor_cli module

qslib.monitor_cli.main(args)[source]

Wrapper allowing fib() to be called with string arguments in a CLI fashion

Instead of returning the value from fib(), it prints the result to the stdout in a nicely formatted message.

Parameters

args (List[str]) – command line parameters as list of strings (for example ["--verbose", "42"]).

qslib.monitor_cli.parse_args(args)[source]
Return type

Namespace

qslib.monitor_cli.run()[source]

Calls main() passing the CLI arguments extracted from sys.argv

This function can be used as entry point to create console scripts with setuptools.

qslib.monitor_cli.setup_logging(loglevel)[source]

Setup basic logging

Parameters

loglevel (int) – minimum loglevel for emitting messages

qslib.plate_setup module

Code for handling plate setup.

class qslib.plate_setup.PlateSetup(sample_wells=None, samples=())[source]

Bases: object

classmethod from_array(array, *, make_unique=False)[source]
Return type

PlateSetup

async classmethod from_machine(c, runtitle=None)[source]
Return type

PlateSetup

classmethod from_platesetup_xml(platexml)[source]
Return type

PlateSetup

classmethod from_tsv(tsvstr)[source]
Return type

PlateSetup

get_descriptive_string(name)[source]
Return type

str

get_wells(samples_or_wells)[source]

Given a sample, well, or list of the two, returns the corresponding wells. Note that this relies on samples not having well-like names.

Return type

list[str]

property sample_wells
samples_by_name: Dict[str, qslib.plate_setup.Sample]
to_lineprotocol(timestamp, run_name=None)[source]
Return type

list[str]

to_table(headers=[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12], tablefmt='orgtbl', showindex=('A', 'B', 'C', 'D', 'E', 'F', 'G', 'H'), **kwargs)[source]
Return type

str

update_xml(root)[source]
Return type

None

property well_sample
well_samples_as_array()[source]
Return type

ndarray[Any, Any]

class qslib.plate_setup.Sample(name, uuid=None, color=(255, 0, 0, 255), properties=None, description=None, wells=None)[source]

Bases: object

color: Tuple[int, int, int, int]
description: str | None
classmethod from_platesetup_sample(se)[source]
Return type

Sample

name: str
properties: dict[str, str]
to_xml()[source]
Return type

Element

property uuid: str
Return type

str

wells: list[str]

qslib.processors module

class qslib.processors.NormRaw[source]

Bases: qslib.processors.Processor

A Processor that takes no arguments, and simply passes through raw fluorescence values.

process(data)[source]
Return type

DataFrame

process_scoped(data, scope)[source]

Filter the data, and return it (possibly not a copy), if scope is the minimum necessary scope for this normalization type. Otherwise, just return the same data.

This is useful for writing scope-agnostic code, provided that you call this for every scope before using the data.

The values for scope are:

  • “all”: the entire welldata array.

  • “limited”: all time points, but limited to the filter sets and samples being plotted.

Return type

DataFrame

ylabel(previous_label=None)[source]
Return type

str

class qslib.processors.NormToMaxPerWell(stage=None, step=None, cycle=None, *, selection=None)[source]

Bases: qslib.processors.Processor

A Processor that divides the fluorescence reading for each (filterset, well) pair by the max value of that pair within a particular selection of data.

The easiest way to use this is to give a particular stage (all data in that stage will be used), or a stage and set of cycles (those cycles in that stage will be used). For example:

  • To normalize to the mean stage 8 values, use NormToMeanPerWell(stage=8).

  • To normalize to the first 5 cycles of stage 2, use

    NormToMeanPerWell(stage=2, cycle=slice(1, 6)).

selection allows arbitrary Pandas indexing (without the filter_set level of the MultiIndex) for unusual cases.

process(data)[source]
Return type

DataFrame

process_scoped(data, scope)[source]

Filter the data, and return it (possibly not a copy), if scope is the minimum necessary scope for this normalization type. Otherwise, just return the same data.

This is useful for writing scope-agnostic code, provided that you call this for every scope before using the data.

The values for scope are:

  • “all”: the entire welldata array.

  • “limited”: all time points, but limited to the filter sets and samples being plotted.

Return type

DataFrame

scope: ClassVar[Literal['all', 'limited']] = 'limited'
selection: Any
ylabel(previous_label=None)[source]
Return type

str

class qslib.processors.NormToMeanPerWell(stage=None, step=None, cycle=None, *, selection=None)[source]

Bases: qslib.processors.Processor

A Processor that divides the fluorescence reading for each (filterset, well) pair by the mean value of that pair within a particular selection of data.

The easiest way to use this is to give a particular stage (all data in that stage will be used), or a stage and set of cycles (those cycles in that stage will be used). For example:

  • To normalize to the mean stage 8 values, use NormToMeanPerWell(stage=8).

  • To normalize to the first 5 cycles of stage 2, use

    NormToMeanPerWell(stage=2, cycle=slice(1, 6)).

selection allows arbitrary Pandas indexing (without the filter_set level of the MultiIndex) for unusual cases.

process(data)[source]
Return type

DataFrame

process_scoped(data, scope)[source]

Filter the data, and return it (possibly not a copy), if scope is the minimum necessary scope for this normalization type. Otherwise, just return the same data.

This is useful for writing scope-agnostic code, provided that you call this for every scope before using the data.

The values for scope are:

  • “all”: the entire welldata array.

  • “limited”: all time points, but limited to the filter sets and samples being plotted.

Return type

DataFrame

scope: ClassVar[Literal['all', 'limited']] = 'limited'
selection: Any
ylabel(previous_label=None)[source]
Return type

str

class qslib.processors.Processor[source]

Bases: object

abstract process(data)[source]
Return type

DataFrame

abstract process_scoped(data, scope)[source]

Filter the data, and return it (possibly not a copy), if scope is the minimum necessary scope for this normalization type. Otherwise, just return the same data.

This is useful for writing scope-agnostic code, provided that you call this for every scope before using the data.

The values for scope are:

  • “all”: the entire welldata array.

  • “limited”: all time points, but limited to the filter sets and samples being plotted.

Return type

DataFrame

abstract ylabel(previous_label=None)[source]
Return type

str

class qslib.processors.SmoothEMWMean(com=None, span=None, halflife=None, alpha=None, min_periods=0, adjust=True, ignore_na=False)[source]

Bases: qslib.processors.Processor

A Processor that smooths fluorescence readings using Pandas’ Exponential Moving Window (ewm / exponentially weighted moving-average).

adjust: bool
alpha: float | None
com: float | None
halflife: float | str | timedelta | None
ignore_na: bool
min_periods: int
process(data)[source]
Return type

DataFrame

process_scoped(data, scope)[source]

Filter the data, and return it (possibly not a copy), if scope is the minimum necessary scope for this normalization type. Otherwise, just return the same data.

This is useful for writing scope-agnostic code, provided that you call this for every scope before using the data.

The values for scope are:

  • “all”: the entire welldata array.

  • “limited”: all time points, but limited to the filter sets and samples being plotted.

Return type

DataFrame

scope: ClassVar[ScopeType] = 'limited'
span: float | None
ylabel(previous_label=None)[source]
Return type

str

class qslib.processors.SmoothWindowMean(window, min_periods=None, center=False, win_type=None, closed=None)[source]

Bases: qslib.processors.Processor

A Processor that smooths fluorescence readings using Pandas’ Rolling, and mean.

center: bool
closed: str | None
min_periods: int | None
process(data)[source]
Return type

DataFrame

process_scoped(data, scope)[source]

Filter the data, and return it (possibly not a copy), if scope is the minimum necessary scope for this normalization type. Otherwise, just return the same data.

This is useful for writing scope-agnostic code, provided that you call this for every scope before using the data.

The values for scope are:

  • “all”: the entire welldata array.

  • “limited”: all time points, but limited to the filter sets and samples being plotted.

Return type

DataFrame

scope: ClassVar[ScopeType] = 'limited'
win_type: str | None
window: int
ylabel(previous_label=None)[source]
Return type

str

class qslib.processors.SubtractByMeanPerWell(stage=None, step=None, cycle=None, *, selection=None)[source]

Bases: qslib.processors.Processor

A Processor that subtracts the fluorescence reading for each (filterset, well) pair by the mean value of that pair within a particular selection of data.

The easiest way to use this is to give a particular stage (all data in that stage will be used), or a stage and set of cycles (those cycles in that stage will be used). For example:

  • To subtract the mean stage 8 values, use NormToMeanPerWell(stage=8).

  • To subtract the mean of the first 5 cycles of stage 2, use

    NormToMeanPerWell(stage=2, cycle=slice(1, 6)).

selection allows arbitrary Pandas indexing (without the filter_set level of the MultiIndex) for unusual cases.

process(data)[source]
Return type

DataFrame

process_scoped(data, scope)[source]

Filter the data, and return it (possibly not a copy), if scope is the minimum necessary scope for this normalization type. Otherwise, just return the same data.

This is useful for writing scope-agnostic code, provided that you call this for every scope before using the data.

The values for scope are:

  • “all”: the entire welldata array.

  • “limited”: all time points, but limited to the filter sets and samples being plotted.

Return type

DataFrame

scope: ClassVar[Literal['all', 'limited']] = 'limited'
selection: Any
ylabel(previous_label=None)[source]
Return type

str

qslib.protocol module

class qslib.protocol.CustomStep(body, identifier=None, repeat=1)[source]

Bases: qslib.protocol.ProtoCommand

A protocol step composed of SCPI/protocol commands.

property body: list[qslib.protocol.ProtoCommand]
Return type

list[ProtoCommand]

property collects: bool
Return type

bool

duration_at_cycle(cycle)[source]
Return type

Quantity[int]

classmethod from_scpicommand(sc)[source]
Return type

CustomStep

property identifier: int | str | None
Return type

int | str | None

info_str(index=None, repeats=1)[source]
Return type

str

repeat: int = 1
temperatures_at_cycle(cycle)[source]
Return type

Quantity[ndarray]

to_scpicommand(*, stepindex=- 1, **kwargs)[source]
Return type

SCPICommand

total_duration(repeat=1)[source]
Return type

Quantity[int]

class qslib.protocol.Exposure(settings, state='HoldAndCollect')[source]

Bases: qslib.protocol.ProtoCommand

Modifies exposure settings.

classmethod from_scpicommand(sc)[source]
Return type

Exposure

settings: Sequence[Tuple[qslib.data.FilterSet, Sequence[int]]]
state: str = 'HoldAndCollect'
to_scpicommand(**kwargs)[source]
Return type

SCPICommand

class qslib.protocol.HACFILT(filters, default_filters=NOTHING)[source]

Bases: qslib.protocol.ProtoCommand

Sets filters for HoldAndCollect .

filters: Sequence[qslib.data.FilterSet]
classmethod from_scpicommand(sc)[source]
Return type

HACFILT

to_scpicommand(default_filters=None, **kwargs)[source]
Return type

SCPICommand

class qslib.protocol.Hold(time, increment=<Quantity(0, 'second')>, incrementcycle=1, incrementstep=1)[source]

Bases: qslib.protocol.ProtoCommand

A protocol hold (for a time) command.

classmethod from_scpicommand(sc)[source]
Return type

Hold

increment: pint.Quantity[int] = <Quantity(0, 'second')>
incrementcycle: int = 1
incrementstep: int = 1
time: pint.Quantity[int] | None
to_scpicommand(**kwargs)[source]
Return type

SCPICommand

class qslib.protocol.HoldAndCollect(time, increment=<Quantity(0, 'second')>, incrementcycle=1, incrementstep=1, tiff=False, quant=True, pcr=False)[source]

Bases: qslib.protocol.ProtoCommand

A protocol hold (for a time) and collect (set by HACFILT) command.

classmethod from_scpicommand(sc)[source]
Return type

HoldAndCollect

increment: pint.quantity.Quantity[int] = <Quantity(0, 'second')>
incrementcycle: int = 1
incrementstep: int = 1
pcr: bool = False
quant: bool = True
tiff: bool = False
time: pint.quantity.Quantity[int]
to_scpicommand(**kwargs)[source]
Return type

SCPICommand

class qslib.protocol.ProtoCommand(*args, **kwargs)[source]

Bases: abc.ABC

classmethod from_scpicommand(sc)[source]
Return type

~T

abstract to_scpicommand(**kwargs)[source]
Return type

SCPICommand

class qslib.protocol.Protocol(stages=NOTHING, name=NOTHING, volume=50.0, runmode='standard', filters=NOTHING, covertemperature=105.0, prerun=NOTHING, postrun=NOTHING, classname='Protocol')[source]

Bases: qslib.protocol.ProtoCommand

A run protocol for the QuantStudio. Protocols encapsulate the temperature and camera controls for an entire run. They are composed of Stage`s, which may repeat for a number of cycles, and the stages are in turn composed of Steps, which may be created for usual cases with :class:`Step, or from SCPI commands. Steps may repeat their contents as well, but this is not yet implemeted.

Parameters
  • stages (Iterable[Stage]) – The stages of the protocol, likely Stage.

  • stage (_NumOrRefIndexer[Stage]) – A more convenient way of accessing the stages of the protocol, with numbering that matches the machine.

  • name (str | None) – A protocol name. If not set, a timestamp will be used, unlike AB’s uuid.

  • volume (float) – The sample volume, in µL.

  • runmode (str | None) – The run mode.

  • covertemperature (float (default 105.0)) – The cover temperature

  • filters (Sequence[str]) – A list of default filters that can be used by any collection commands that don’t specify their own.

  • prerun (Sequence[SCPICommand]) – Sets PRERUN. DO NOT USE THIS UNLESS YOU KNOW WHAT YOU ARE DOING.

  • postrun (Sequence[SCPICommand]) – Sets POSTRUN. DO NOT USE THIS UNLESS YOU KNOW WHAT YOU ARE DOING.

property all_filters: Collection[qslib.data.FilterSet]

A list of all filters used at some point in the protocol.

Return type

Collection[FilterSet]

property all_points: pandas.core.frame.DataFrame
Return type

DataFrame

property all_temperatures: numpy.ndarray

An array of temperature settings at all_times.

Return type

ndarray

property all_times: numpy.ndarray

An array of all start and end times of each step, interleaved.

Return type

ndarray

check_compatible(new, status)[source]

Checks compatibility for changing a running protocol to a new one.

Raises ValueError if incompatible, returns True if compatible.

Parameters
Raises

ValueError – Protocols are incompatible.

Return type

bool

copy()[source]

Returns a new copy (recursively) of the protocol.

Return type

Protocol

covertemperature: float
property dataframe: pandas.core.frame.DataFrame

A DataFrame of the temperature protocol.

Return type

DataFrame

filters: Sequence[qslib.data.FilterSet]
classmethod from_scpicommand(sc)[source]
Return type

Protocol

classmethod from_xml(e)[source]
Return type

Protocol

classmethod fromdict(d)[source]
Return type

Protocol

info()[source]

Generate a (markdown) text protocol description.

Return type

str

name: str
plot_protocol(ax=None)[source]

A plot of the temperature and data collection points.

Return type

Tuple[plt.Axes, Tuple[List[plt.Line2D], List[plt.Line2D]]]

postrun: Sequence[qslib.scpi_commands.SCPICommandLike]
prerun: Sequence[qslib.scpi_commands.SCPICommandLike]
runmode: str
property stage: qslib.protocol._NumOrRefIndexer[qslib.protocol.Stage]

A more convenient view of Protocol.stages. This allows one-indexed access, such that protocol.stage[5] == protocol.stages[6] is stage 5 of the protocol, in the interpretation of tha machine. Indexing can use slices, and is inclusive, so protocol.stage[5:6] returns stages 5 and 6. Getting, setting, and appending stages are all supported through this interface.

Return type

_NumOrRefIndexer[Stage]

stages: list[qslib.protocol.Stage]
to_scpicommand(**kwargs)[source]
Return type

SCPICommand

to_xml(covertemperature=105.0)[source]
Return type

tuple[ElementTree, ElementTree]

volume: float
class qslib.protocol.Ramp(temperature, increment=<Quantity(0.0, 'delta_degree_Celsius')>, incrementcycle=1, incrementstep=1, rate=100.0, cover=None)[source]

Bases: qslib.protocol.ProtoCommand

Ramps temperature to a new setting.

cover: pint.Quantity[float] | None
classmethod from_scpicommand(sc)[source]
Return type

Ramp

increment: pint.Quantity[float]
incrementcycle: int
incrementstep: int
rate: float
temperature: pint.Quantity[np.ndarray]
to_scpicommand(**kwargs)[source]
Return type

SCPICommand

class qslib.protocol.Stage(steps, repeat=1, index=None, label=None, default_filters=())[source]

Bases: qslib.protocol.XMLable, qslib.protocol.ProtoCommand

A Stage in a protocol, composed of Step s with a possible repeat.

dataframe(start_time=0, previous_temperatures=None)[source]

Create a dataframe of the steps in this stage.

Parameters
  • start_time (float) – The initial start time, in seconds, of the stage (before the ramp to the first step). Default is 0.

  • previous_temperatures (list[float] | None) – A list of temperatures at the end of the previous stage, to allow calculation of ramp time. If None, the ramp is assumed to take no time.

Return type

pd.DataFrame

classmethod from_scpicommand(sc, **kwargs)[source]
Return type

Stage

classmethod from_xml(e)[source]
Return type

Stage

classmethod fromdict(d)[source]
Return type

Stage

classmethod hold_at(temperature, total_time, step_time=None, collect=None, filters=())[source]

Hold at a temperature for a set amount of time, with steps of a configurable fixed time.

Parameters
  • temperatures – The temperature or temperatures to hold. If not strings or quantities, the value/values are interpreted as °C.

  • total_time (int | str | pint.Quantity[int]) – Desired total time for the stage. If this is not a multiple of step_time, it may not be the actual total time for the stage. The function will emit a warning if the difference is more than 10%. If an integer, value is interpreted as seconds.

  • step_time (int | str | pint.Quantity[int] | None) – If None (default), the stage will have one step. Otherwise, it will have steps of this time. If an integer, value is interpreted as seconds.

  • collect (bool | None) – Whether or not each step should collect fluorescence data. If None (default), collects data if filters is explicitly set.

  • filters (Sequence[str | FilterSet]) – A list of filters to collect. If empty, and collect is True, then each step will collect the default filters for the Protocol.

Returns

The resulting Stage

Return type

Stage

Raises

ValueError – If step time is larger than total time.

classmethod hold_for(temperature, total_time, step_time=None, collect=None, filters=())

Hold at a temperature for a set amount of time, with steps of a configurable fixed time.

Parameters
  • temperatures – The temperature or temperatures to hold. If not strings or quantities, the value/values are interpreted as °C.

  • total_time (int | str | pint.Quantity[int]) – Desired total time for the stage. If this is not a multiple of step_time, it may not be the actual total time for the stage. The function will emit a warning if the difference is more than 10%. If an integer, value is interpreted as seconds.

  • step_time (int | str | pint.Quantity[int] | None) – If None (default), the stage will have one step. Otherwise, it will have steps of this time. If an integer, value is interpreted as seconds.

  • collect (bool | None) – Whether or not each step should collect fluorescence data. If None (default), collects data if filters is explicitly set.

  • filters (Sequence[str | FilterSet]) – A list of filters to collect. If empty, and collect is True, then each step will collect the default filters for the Protocol.

Returns

The resulting Stage

Return type

Stage

Raises

ValueError – If step time is larger than total time.

index: int | None
info_str(index=None)[source]
Return type

str

label: str | None
repeat: int
property step: qslib.protocol._NumOrRefIndexer[qslib.protocol.CustomStep]
Return type

_NumOrRefIndexer[CustomStep]

classmethod stepped_ramp(from_temperature, to_temperature, total_time, *, n_steps=None, temperature_step=None, collect=None, filters=())[source]

Hold at a series of temperatures, from one to another.

Parameters
  • from_temperature (float | str | pint.Quantity[float] | Sequence[float]) – Initial temperature/s (inclusive).

  • to_temperature (float | str | pint.Quantity[float] | Sequence[float]) – Final temperature/s (inclusive).

  • total_time (int | str | pint.Quantity[int]) – Total time for the stage

  • n_steps (int | None) – Number of steps. If None, uses 1.0 Δ°C steps, or, if doing a multi-temperature change, uses maximum step of 1.0 Δ°C.

  • temperature_step (float | str | pint.Quantity[float] | None) – Step temperature change (optional). Must be None, or correctly match calculation, if n_steps is not None. If both this and n_steps are None, default is 1.0 Δ°C steps. If temperature step does not exactly fit range, it will be adjusted, with a warning if the change is more than 5%. Sign is ignored. If doing a multi-temperature change, then this is the maximum temperature step.

  • collect (bool | None) – Collect data? If None, collects data if filters is set explicitly.

  • filters (Sequence[str | FilterSet]) – Filters to collect.

Returns

The resulting stage.

Return type

Stage

steps: Sequence[CustomStep]
to_scpicommand(stageindex=None, **kwargs)[source]
Return type

SCPICommand

to_xml(**kwargs)[source]
Return type

Element

class qslib.protocol.Step(time, temperature, collect=None, temp_increment=<Quantity(0.0, 'delta_degree_Celsius')>, temp_incrementcycle=2, temp_incrementpoint=2, time_increment=<Quantity(0, 'second')>, time_incrementcycle=2, time_incrementpoint=2, filters=(), pcr=False, quant=True, tiff=False, repeat=1, default_filters=())[source]

Bases: qslib.protocol.CustomStep, qslib.protocol.XMLable

A normal protocol step, of a hold and possible collection.

Parameters
  • time (int) – The step time setting, in seconds.

  • temperature (float | Sequence[float]) – The temperature hold setting, either as a float (all zones the same) or a sequence (of correct length) of floats setting the temperature for each zone.

  • collect (bool | None) – Collect fluorescence data? If None (default), collect only if the Step has an explicit filters setting.

  • temp_increment (float) – Amount to increment all zone temperatures per cycle on and after temp_incrementcycle.

  • temp_incrementcycle (int (default 2)) – First cycle to start the increment changes. Note that the default in QSLib is 2, not 1 (as in AB’s software), so that leaving this alone makes sense (the first cycle will be at temperature, the next at temperature + temp_incrementcycle.

  • time_increment (float) –

  • time_incrementcycle (int) – The same settings for time per cycle.

  • filters (Sequence[FilterSet | str] (default empty)) – A list of filter pairs to collect, either using FilterSet or a string like “x1-m4”. If collect is True and this is empty, then the filters will be set by the Protocol.

Notes

This currently does not support step-level repeats, which do exist on the machine.

property body: list[qslib.protocol.ProtoCommand]
Return type

list[ProtoCommand]

collect: bool | None
property collects
duration_at_cycle(cycle)[source]

Duration of the step (excluding ramp) at cycle (from 1)

Return type

Quantity

filters: Sequence[FilterSet]
classmethod from_scpicommand(sc)[source]
Return type

Step

classmethod from_xml(e, *, etc=1, ehtc=1, he=False)[source]
Return type

Step

classmethod fromdict(d)[source]
Return type

Step

property identifier: None
Return type

None

info_str(index=None, repeats=1)[source]

String describing the step.

Return type

str

pcr: bool
quant: bool
repeat: int
temp_increment: pint.Quantity[float]
temp_incrementcycle: int
temp_incrementpoint: int
temperature: pint.Quantity[Any]
property temperature_list: pint.quantity.Quantity[numpy.ndarray]
Return type

Quantity[ndarray]

temperatures_at_cycle(cycle)[source]

Temperatures of the step at cycle (from 1)

Return type

Quantity[ndarray]

tiff: bool
time: pint.Quantity[int]
time_increment: pint.Quantity[int]
time_incrementcycle: int
time_incrementpoint: int
to_xml(**kwargs)[source]
Return type

Element

total_duration(repeats=1)[source]
Return type

Quantity

class qslib.protocol.XMLable[source]

Bases: abc.ABC

abstract to_xml(**kwargs)[source]
Return type

Element

qslib.protocol.c

alias of qslib.protocol.Protocol

qslib.qs_is_protocol module

exception qslib.qs_is_protocol.AccessLevelExceeded(command: 'str', accessLimit: 'AccessLevel', message: 'str')[source]

Bases: qslib.qs_is_protocol.CommandError

accessLimit: qslib.scpi_commands.AccessLevel
command: str
message: str
classmethod parse(command, ref_index, message)[source]
Return type

AccessLevelExceeded

exception qslib.qs_is_protocol.AuthError(command: 'str', message: 'str')[source]

Bases: qslib.qs_is_protocol.CommandError

command: str
message: str
classmethod parse(command, ref_index, message)[source]
Return type

AuthError

exception qslib.qs_is_protocol.CommandError[source]

Bases: qslib.qs_is_protocol.Error

static parse(command, ref_index, response)[source]
Return type

CommandError

exception qslib.qs_is_protocol.Error[source]

Bases: Exception

exception qslib.qs_is_protocol.InsufficientAccess(command: 'str', requiredAccess: 'AccessLevel', currentAccess: 'AccessLevel', message: 'str')[source]

Bases: qslib.qs_is_protocol.CommandError

command: str
currentAccess: qslib.scpi_commands.AccessLevel
message: str
classmethod parse(command, ref_index, message)[source]
Return type

InsufficientAccess

requiredAccess: qslib.scpi_commands.AccessLevel
exception qslib.qs_is_protocol.InvocationError(command: 'str', message: 'str')[source]

Bases: qslib.qs_is_protocol.CommandError

command: str
message: str
classmethod parse(command, ref_index, message)[source]
Return type

InvocationError

exception qslib.qs_is_protocol.NoMatch(command: 'str', message: 'str')[source]

Bases: qslib.qs_is_protocol.CommandError

command: str
message: str
classmethod parse(command, ref_index, message)[source]
Return type

NoMatch

class qslib.qs_is_protocol.QS_IS_Protocol[source]

Bases: asyncio.protocols.Protocol

connection_lost(exc)[source]

Called when the connection is lost or closed.

The argument is an exception object or None (the latter meaning a regular EOF is received or the connection was aborted or closed).

Return type

None

connection_made(transport)[source]

Called when a connection is made.

The argument is the transport representing the pipe connection. To receive data, wait for data_received() calls. When the connection is closed, connection_lost() is called.

Return type

None

data_received(data)[source]

Process received data packet from instrument, keeping track of quotes. If a newline occurs when the quote stack is empty, create a task to process the message, but continue processing. (TODO: consider threads/processes here.)

Parameters

data (bytes) – bytes:

Return type

None

async disconnect()[source]
Return type

None

async handle_sub_message(message)[source]
Return type

None

last_received: float
lostconnection: Future[Any]
async parse_message(ds)[source]
Return type

None

async run_command(comm, ack_timeout=300, just_ack=True, uid=True)[source]
Return type

bytes

waiting_commands: list[tuple[bytes, None | Future[tuple[bytes, bytes, None | Future[tuple[bytes, bytes, None]]]]]]
exception qslib.qs_is_protocol.ReplyError[source]

Bases: OSError

class qslib.qs_is_protocol.SubHandler(*args, **kwargs)[source]

Bases: Protocol

exception qslib.qs_is_protocol.UnparsedCommandError(command: 'Optional[str]', ref_index: 'Optional[str]', response: 'str')[source]

Bases: qslib.qs_is_protocol.CommandError

command: Optional[str]
ref_index: Optional[str]
response: str

qslib.qsconnection_async module

exception qslib.qsconnection_async.AlreadyCollectedError[source]

Bases: Exception

class qslib.qsconnection_async.FilterDataFilename(filterset, stage, cycle, step, point)[source]

Bases: object

cycle: int
filterset: qslib.data.FilterSet
classmethod fromstring(x)[source]
Return type

FilterDataFilename

is_same_point(other)[source]
Return type

bool

point: int
stage: int
step: int
tostring()[source]
Return type

str

class qslib.qsconnection_async.QSConnectionAsync(host='localhost', port=7000, authenticate_on_connect=True, initial_access_level=AccessLevel.Observer, password=None)[source]

Bases: object

Class for connection to a QuantStudio instrument server, using asyncio

async authenticate(password)[source]
Return type

None

async compile_eds(run_name)[source]

Take a finished run directory in experiments:, compile it into an EDS, and move it to public_run_complete:

Return type

None

async connect(authenticate=None, initial_access_level=None, password=None)[source]
Return type

str

property connected: bool
Return type

bool

async disconnect()[source]
Return type

None

async get_all_filterdata(run: Optional[str], as_list: Literal[True]) List[qslib.data.FilterDataReading][source]
async get_all_filterdata(run: Optional[str], as_list: Literal[False]) pandas.core.frame.DataFrame
Return type

Union[pd.DataFrame, List[data.FilterDataReading]]

async get_exp_file(path, encoding='base64')[source]
Return type

bytes

async get_expfile_list(glob, allow_nomatch=False)[source]
Return type

List[str]

async get_filterdata_one(ref: qslib.qsconnection_async.FilterDataFilename, *, run: Optional[str] = 'None', return_files: Literal[True]) tuple[qslib.data.FilterDataReading, list[tuple[str, bytes]]][source]
async get_filterdata_one(ref: qslib.qsconnection_async.FilterDataFilename, *, run: Optional[str] = 'None', return_files: Literal[False] = 'False') qslib.data.FilterDataReading
Return type

data.FilterDataReading | tuple[data.FilterDataReading, list[tuple[str, bytes]]]

async get_run_start_time()[source]
Return type

float

async get_run_title()[source]
Return type

str

async get_sds_file(path, runtitle=None, encoding='base64')[source]
Return type

bytes

async list_files(path: str, *, leaf: str = "'FILE'", verbose: Literal[True], recursive: bool = 'False') list[dict[str, Any]][source]
async list_files(path: str, *, leaf: str = "'FILE'", verbose: Literal[False], recursive: bool = 'False') list[str]
async list_files(path: str, *, leaf: str = "'FILE'", verbose: bool = 'False', recursive: bool = 'False') list[str] | list[dict[str, Any]]
Return type

list[str] | list[dict[str, Any]]

async read_dir_as_zip(path, leaf='FILE')[source]

Read a directory on the

Parameters
  • path (str) – path on the machine

  • leaf (str, optional) – leaf to use, by default “FILE”

Returns

the returned zip file

Return type

zipfile.ZipFile

async read_file(path, context=None, leaf='FILE', encoding='base64')[source]
Return type

bytes

async run_command(command, just_ack=False)[source]
Return type

str

async run_command_to_bytes(command, just_ack=True)[source]
Return type

bytes

async set_access_level(level)[source]
Return type

None

exception qslib.qsconnection_async.RunNotFinishedError[source]

Bases: Exception

qslib.rawquant_compat module

qslib.scpi_commands module

SCPI Command class and parsing

class qslib.scpi_commands.AccessLevel(value)[source]

Bases: enum.Enum

QS machine access level, with comparisons.

Administrator = 'Administrator'
Controller = 'Controller'
Full = 'Full'
Guest = 'Guest'
Observer = 'Observer'
value: str
class qslib.scpi_commands.ArgList(opts, args)[source]

Bases: object

A representation of an SCPI list of options (-key=value) and arguments.

args: list[bool | int | float | str]
classmethod from_string(argument_string)[source]

Parse an SCPI argument string.

Return type

ArgList

opts: dict[str, bool | int | float | str]
class qslib.scpi_commands.SCPICommand(command, *args, comment=None, **kwargs)[source]

Bases: qslib.scpi_commands.SCPICommandLike

A representation of an SCPI Command.

args: Sequence[str | int | float | np.number[Any] | Sequence[str | int | float | np.number[Any]] | Sequence['SCPICommand']]
command: str
comment: str | None
classmethod from_scpicommand(com)[source]

Try to create the object from an SCPICommand.

Return type

SCPICommand

classmethod from_string(command_string)[source]

Parse (as SCPICommands) an SCPI command string.

Return type

SCPICommand

opts: dict[str, str | int | float | np.number[Any] | Sequence[str | int | float | np.number[Any]]]
specialize()[source]

If possible, convert SCPICommand to QSLib classes for the command.

Return type

SCPICommandLike

to_scpicommand(**kwargs)[source]

Convert the object to an SCPICommand

Return type

SCPICommand

to_string()[source]

Create a usable command string, including terminal newline.

Return type

str

class qslib.scpi_commands.SCPICommandLike[source]

Bases: abc.ABC

Abstract class for an object that can be converted from/to an SCPICommand.

abstract classmethod from_scpicommand(com)[source]

Try to create the object from an SCPICommand.

Return type

~T

abstract to_scpicommand(**kwargs)[source]

Convert the object to an SCPICommand

Return type

SCPICommand

qslib.version module

Module contents

class qslib.AccessLevel(value)[source]

Bases: enum.Enum

QS machine access level, with comparisons.

Administrator = 'Administrator'
Controller = 'Controller'
Full = 'Full'
Guest = 'Guest'
Observer = 'Observer'
value: str
class qslib.CustomStep(body, identifier=None, repeat=1)[source]

Bases: qslib.protocol.ProtoCommand

A protocol step composed of SCPI/protocol commands.

property body: list[qslib.protocol.ProtoCommand]
Return type

list[ProtoCommand]

property collects: bool
Return type

bool

duration_at_cycle(cycle)[source]
Return type

Quantity[int]

classmethod from_scpicommand(sc)[source]
Return type

CustomStep

property identifier: int | str | None
Return type

int | str | None

info_str(index=None, repeats=1)[source]
Return type

str

repeat: int = 1
temperatures_at_cycle(cycle)[source]
Return type

Quantity[ndarray]

to_scpicommand(*, stepindex=- 1, **kwargs)[source]
Return type

SCPICommand

total_duration(repeat=1)[source]
Return type

Quantity[int]

class qslib.Experiment(name=None, protocol=None, plate_setup=None, _create_xml=True)[source]

Bases: object

A QuantStudio experiment / EDS file

This class can create, modify, load and save experiments in several ways, run them, and control and modify them while running.

Experiments can be loaded from a file:

>>> exp: Experiment = Experiment.from_file("experiment.eds")

They can also be loaded from a running experiment:

>>> machine = Machine("localhost", 7000, password="password")
>>> exp = Experiment.from_running(machine)

Or from the machine’s storage:

>>> exp = Experiment.from_machine(machine, "experiment")

They can also be created from scratch:

>>> exp = Experiment("an-experiment-name")
>>> exp.protocol = Protocol([Stage([Step(time=60, temperature=60)])])
>>> exp.plate_setup = PlateSetup({"sample_name": "A5"})

And they can be run on a machine:

>>> exp.run(machine)

Data can be accessed in a few ways:

The (hopefully) easiest way is with welldata, which has multi-indexes for both rows and columns.

>>> exp.welldata.loc[('x1-m4', 4), [('time', 'hours'), ('A05', 'fl')]].plot()

Or for a temperature curve:

>>> exp.welldata.loc[('x1-m4', 4), [('A05', 'st'), ('A05', 'fl')]].plot()

filterdata should still work normally.

Notes

There are a few differences in how QSLib and AB’s software handles experiments.

  • AB’s software considers the run as starting when the machine indicates “Run Starting”. This is stored as runstarttime in QSLib, but as it may include the lamp warmup (3 minutes) and other pre-actual-protocol time, QSLib instead prefers activestarttime, which it sets from the beginning of the first real (not PRERUN) Stage, at which point the machine starts its own active clock and starts ramping to the first temperature. QSLib uses this as the start time reference in its data, and also includes the timestamp from the machine.

  • The machine has a specific language for run protocols. QSLib uses this language. AB’s Design and Analysis software does not, instead using an XML format. Not everything in the machine’s language is possible to express in the XML format (eg, disabling pcr analysis, saving images); the XML format has some concepts not present in the machine format, and is generally more complicated and harder to understand. QSLib uses and trusts the machine’s protocol if at all possible, even for files written by AB D&A (it is stored in the log if the run has started).

  • QSLib will try to write a reasonable XML protocol for AB D&A to see, but it may by an approximation or simply wrong, if the actual protocol can’t be expressed there. It will also store the actual protocol in tcprotocol.xml, and its own representation.

  • By default, creating a step with a per-cycle increment in QSLib starts the change on cycle 2, not cycle 1, as is the default in the software.

  • Immediate pause/resume, mid-run stage addition, and other functions are not supported by AB D&A and experiments using them may confuse the software later.

  • QSLib writes notes to XML files, and tries to create reasonable XML files for AB D&A, but may still cause problems. At the moment, it makes clear that its files are its own (setting software versions in experiment.xml and Manifest.mf).

abort(machine=None)[source]

If this experiment is running, abort it, stopping it immediately.

Requires and takes exclusive Controller access on the machine.

Raises

NotRunningError – the experiment is not currently running

Return type

None

activeendtime: datetime | None

The actual end of the main part of the run, indicated by “Stage POSTRun” or an abort.

activestarttime: datetime | None

The actual beginning of the first stage of the run, defined as the first “Run Stage” message in the log after “Stage PRERUN”. This is not what AB’s software considers the start of a run.

property all_filters: Collection[qslib.data.FilterSet]

All filters used at some point in the experiment.

If the experiment has data, this is based on the existing data. Otherwise, it is based on the experiment protocol.

Return type

Collection[FilterSet]

change_protocol(new_protocol, machine=None, force=False)[source]

For a running experiment and an updated protocol, check compatibility with the current run, and if possible, update the protocol in the experiment file and on the machine, changing the current run.

Changes that should be possible:

  • Changing the number of cycles of the current run to a higher or lower value (but higher or equal to the current cycle number), allowing stages to be lengthened, shortened, or stopped.

  • Adding new stages after the current stage.

  • Arbitrarily changing any stage that hasn’t started.

For safest results, ensure power saving is turned off in the Android software.

Parameters
  • new_protocol (Protocol) – [description]

  • machine (Machine, optional) – [description], by default None

Return type

None

change_protocol_from_now(new_stages, machine=None)[source]

For a running experiment, change the remaining stages to be the provided :param:`new_stages` list. This is a convenience function that:

  1. Gets the currently-running stage and cycle.

  2. Sets the repeat number of the current stage to its current cycle, thus ending it after the end of the current cycle.

  3. Changes the remainder of the stages to be those in the :param:`new_stages` list.

Because this does not impact any current or past stages, there is less of a need to ensure that the stages provided are compatible with the old protocol. The only check done is to ensure that, if the provided stages have any collection commands using default filters, the old protocol has specified default filters. This function does not allow the default filters to be changed: if you want to use filters other than the defaults, or defaults were not provided in the old protocol, then either specify filters explicitly (recommended) for new stages you’d like to be different, or use Experiment.change_protocol directly.

Return type

None

createdtime: datetime

The run creation time.

data_for_sample(sample)[source]

Convenience function to return data for a specific sample.

Finds wells using self.plate_setup.sample_wells[sample], then returns self.welldata.loc[:, wells]

Parameters

sample (str) – sample name

Returns

Slice of welldata. Will have multiple wells if sample is in multiple wells.

Return type

pd.Dataframe

property filter_strings: list[str]

All filters, as x?-m? strings, used at some point in the experiment.

Return type

list[str]

property filterdata: pandas.core.frame.DataFrame
Return type

DataFrame

classmethod from_file(file)[source]

Load an experiment from an EDS file.

Returns

file – The filename or file handle to read.

Return type

str or os.PathLike[str] or IO[bytes]

Raises

ValueError – if the file does not appear to be an EDS file (lacks an experiment.xml).

classmethod from_machine(machine, name)[source]

Create an experiment from data on a machine, checking the running experiment if any, the machine’s public_run_complete storage, and the machine’s uncollected storage.

Parameters
  • machine (Machine | str) – the machine to connect to, either as a Machine or a host name

  • name (str) – the name of the run

Returns

a copy of the experiment

Return type

Experiment

classmethod from_machine_storage(machine, name)[source]

Create an experiment from the one currently running on a machine.

Parameters

machine (Machine) – the machine to connect to

Returns

a copy of the experiment

Return type

Experiment

classmethod from_running(machine)[source]

Create an experiment from the one currently running on a machine.

Parameters

machine (Machine) – the machine to connect to

Returns

a copy of the running experiment

Return type

Experiment

classmethod from_uncollected(machine, name, move=False)[source]

Create an experiment from the uncollected (not yet compressed) storage.

Parameters
  • machine (Machine) – the machine to connect to

  • name (str) – the name of the run to collect.

Returns

a copy of the experiment

Return type

Experiment

get_status(machine=None)[source]

Return the status of the experiment, if currently running.

Requires Observer access on the machine.

Raises

NotRunningError – the experiment is not currently running

Return type

RunStatus

info(format='markdown', plate='list')[source]

Generate a summary of the experiment, with some formatting configuation. str() uses this with default parameters.

Parameters
  • format ("markdown" or "org", optional) – Format of output, currently “markdown” or “org”, and currently matters only when plate is “table”. By default “markdown”. If an unknown value, passed as tablefmt to tabulate.

  • plate ("list" or "table", optional) – Format of plate information. “list” gives a list of samples, “table” outputs a plate layout table (possibly quite wide). By default “list”.

Returns

Summary

Return type

str

info_html()[source]

Create a self-contained HTML summary (returned as a string, but very large) of the experiment.

Return type

str

machine: Machine | None
modifiedtime: datetime

The last modification time. QSLib sets this on write. AB D&A may not.

name: str

Experiment name, also used as file name on the machine.

num_zones: int

The number of temperature zones (excluding cover), or -1 if not known.

pause_now(machine=None)[source]

If this experiment is running, pause it (immediately).

Requires and takes exclusive Controller access on the machine.

Raises

NotRunningError – the experiment is not currently running

Return type

None

plate_setup: PlateSetup

Plate setup for the experiment.

plot_anneal_melt(samples=None, filters=None, anneal_stages=None, melt_stages=None, between_stages=None, process=None, normalization=None, ax=None, marker=None, legend=True, figure_kw=None, line_kw=None)[source]

Plots anneal/melt curves.

This uses solid lines for the anneal, dashed lines for the melt, and dotted lines for anything “between” the anneal and melt (for example, a temperature hold).

Line labels are intended to provide full information when read in combination with the axes title. They will only include information that does not apply to all lines. For example, if every line is from the same filter set, but different samples, then only the sample will be shown. If every line is from the same sample, but different filter sets, then only the filter set will be shown. Wells are shown if a sample has multiple wells.

Parameters
  • samples (str | Sequence[str] | None) – A reference to a single sample (a string), a list of sample names, or a Python regular expression as a string, matching sample names (full start-to-end matches only). Well names may also be included, in which case each well will be treated without regard to the sample name that may refer to it. Note this means you cannot give your samples names that correspond with well references. If not provided, all (named) samples will be included.

  • filters (str | FilterSet | Collection[str | FilterSet] | None) – Optional. A filterset (string or FilterSet) or list of filtersets to include in the plot. Multiple filtersets will be plotted on the same axes. Optional; if None, then all filtersets with data in the experiment will be included.

  • anneal_stages (int | Sequence[int] | None) – Optional. A stage or list of stages (integers, starting from 1), corresponding to the anneal, melt, and stages between the anneal and melt (if any). Any of these may be None, in which case the function will try to determine the correct values automatically.

  • melt_stages (int | Sequence[int] | None) – Optional. A stage or list of stages (integers, starting from 1), corresponding to the anneal, melt, and stages between the anneal and melt (if any). Any of these may be None, in which case the function will try to determine the correct values automatically.

  • between_stages (int | Sequence[int] | None) – Optional. A stage or list of stages (integers, starting from 1), corresponding to the anneal, melt, and stages between the anneal and melt (if any). Any of these may be None, in which case the function will try to determine the correct values automatically.

  • normalization (Processor | None) – Optional. A Normalizer instance to apply to the data. By default, this is NormRaw, which passes through raw fluorescence values. NormToMeanPerWell also works well.

  • ax (plt.Axes' | None) – Optional. An axes to put the plot on. If not provided, the function will create a new figure, by default with constrained_layout=True, though this can be modified with figure_kw.

  • marker (str | None) – The marker format for data points, or None for no markers (default).

  • legend (bool | Literal['inset', 'right']) – Whether to add a legend. True (default) decides whether to have the legend as an inset or to the right of the axes based on the number of lines. “inset” and “right” specify the positioning. Note that for “right”, you must use some method to adjust the axes positioning: constrained_layout, tight_layout, or manually reducing the axes width are all options.

  • figure_kw (Mapping[str, Any] | None) – Optional. A dictionary of options passed through as keyword options to the figure creation. Only applies if ax is None.

  • line_kw (Mapping[str, Any] | None) – Optional. A dictionary of keywords passed to all three plotting commands.

Returns

The axes object of the plot.

Return type

plt.Axes

plot_over_time(samples=None, filters=None, stages=slice(None, None, None), process=None, normalization=None, ax=None, legend=True, temperatures='axes', marker=None, stage_lines=True, annotate_stage_lines=False, figure_kw=None, line_kw=None)[source]

Plots fluorescence over time, optionally with temperatures over time.

Line labels are intended to provide full information when read in combination with the axes title. They will only include information that does not apply to all lines. For example, if every line is from the same filter set, but different samples, then only the sample will be shown. If every line is from the same sample, but different filter sets, then only the filter set will be shown. Wells are shown if a sample has multiple wells.

Parameters
  • samples (str | Sequence[str] | None) – A reference to a single sample (a string), a list of sample names, or a Python regular expression as a string, matching sample names (full start-to-end matches only). Well names may also be included, in which case each well will be treated without regard to the sample name that may refer to it. Note this means you cannot give your samples names that correspond with well references. If not provided, all (named) samples will be included.

  • filters (str | FilterSet | Collection[str | FilterSet] | None) – Optional. A filterset (string or FilterSet) or list of filtersets to include in the plot. Multiple filtersets will be plotted on the same axes. Optional; if None, then all filtersets with data in the experiment will be included.

  • stages (slice | int | Sequence[int]) – Optional. A stage, list of stages, or slice (all using integers starting from 1), to include in the plot. By default, all stages are plotted. For example, to plot stage 2, use stages=2; to plot stages 2 and 4, use stages=[2, 4], to plot stages 3 through 15, use stages=slice(3, 16) (Python ranges are exclusive on the end). Note that is a slice, you can use None instead of a number to denote the beginning/end.

  • normalization (Processor | None) – Optional. A Normalizer instance to apply to the data. By default, this is NormRaw, which passes through raw fluorescence values. NormToMeanPerWell also works well.

  • temperatures (Literal[False, 'axes', 'inset', 'twin']) –

    Optional (default “axes”). Several alternatives for displaying temperatures. “axes” uses a separate axes (created if ax is not provided, otherwise ax must be a list of two axes).

    Temperatures are from Experiment.temperature, and are thus the temperatures as recorded during the run, not the set temperatures. Note that this has a very large number of data points, something that should be dealt with at some point.

  • ax (plt.Axes' | 'Sequence[plt.Axes]' | None) – Optional. An axes to put the plot on. If not provided, the function will create a new figure, by default with constrained_layout=True, though this can be modified with figure_kw. If temperatures=”axes”, you must provide a list or tuple of two axes, the first for fluorescence, the second for temperature.

  • marker (str | None) – The marker format for data points, or None for no markers (default).

  • legend (bool | Literal['inset', 'right']) – Whether to add a legend. True (default) decides whether to have the legend as an inset or to the right of the axes based on the number of lines. “inset” and “right” specify the positioning. Note that for “right”, you must use some method to adjust the axes positioning: constrained_layout, tight_layout, or manually reducing the axes width are all options.

  • stage_lines (bool | Literal['fluorescence', 'temperature']) – Whether to include dotted vertical lines on transitions between stages. If “fluorescence” or “temperature”, include only on one of the two axes.

  • annotate_stage_lines (bool | float | Literal['fluorescence', 'temperature'] | Tuple[Literal['fluorescence', 'temperature'], float]) – Whether to include text annotations for stage lines. Float parameter allows setting the minimum duration of stage, as a fraction of total plotted time, to annotate, in order to avoid overlapping annotations (default threshold is 0.05).

  • figure_kw (Mapping[str, Any] | None) – Optional. A dictionary of options passed through as keyword options to the figure creation. Only applies if ax is None.

  • line_kw (Mapping[str, Any] | None) – Optional. A dictionary of keywords passed to fluorescence plot commands.

Return type

Sequence[plt.Axes]

plot_protocol(ax=None)[source]

A plot of the temperature and data collection points in the experiment’s protocol.

Return type

Tuple[plt.Axes, Tuple[List[plt.Line2D], List[plt.Line2D]]]

plot_temperatures(*, sel=slice(None, None, None), hours=None, ax=None, stage_lines=True, annotate_stage_lines=True, legend=False, figure_kw=None, line_kw=None)[source]

Plot sample temperature readings.

Parameters
  • sel (slice | Callable[[pd.DataFrame], bool]) – A selector for the temperature DataFrame. This is not necessarily easy to use; hours is an easier alternative.

  • hours (tuple[float, float] | None) – Constructs a selector to show temperatures for a time range. :param:`sel` should not be set.

  • ax (Optional[plt.Axes]) – Optional. An axes to put the plot on. If not provided, the function will create a new figure, by default with constrained_layout=True, though this can be modified with figure_kw.

  • stage_lines (bool) – Whether to include dotted vertical lines on transitions between stages.

  • annotate_stage_lines (bool | float) – Whether to include text annotations for stage lines. Float parameter allows setting the minimum duration of stage, as a fraction of total plotted time, to annotate, in order to avoid overlapping annotations (default threshold is 0.05).

  • legend (bool) – Whether to add a legend.

  • figure_kw (Mapping[str, Any] | None) – Optional. A dictionary of options passed through as keyword options to the figure creation. Only applies if ax is None.

  • line_kw (Mapping[str, Any] | None) – Optional. A dictionary of keywords passed to plot commands.

Return type

plt.Axes

protocol: Protocol

Temperature and reading protocol for the experiment.

property rawdata: pandas.core.frame.DataFrame
Return type

DataFrame

resume(machine=None)[source]

If this experiment is running, resume it.

Requires and takes exclusive Controller access on the machine.

Raises

NotRunningError – the experiment is not currently running

Return type

None

run(machine=None, require_exclusive=False, require_drawer_check=True)[source]

Load the run onto a machine, and start it.

Parameters

machine (MachineReference | None) – The machine to run on, by default None, in which case the machine associated with the run (if any) is used.

Raises
Return type

None

runendtime: datetime | None

The run end time as a datetime, taken from the log. This is the end of the run,

runstarttime: datetime | None

The run start time as a datetime. This is taken directly from the log, ignoring the software-set value and replacing it on save if possibe. It is defined as the moment the machine records “Run Starting” in its log, using its timestamp. This may be 3 minutes before the start of the protocol if the lamp needs to warm up. It should be the same value as defined by AB’s software.

Use activestarttime for a more accurate value.

None if the file has not been updated since the start of the run

runstate: Literal['INIT', 'RUNNING', 'COMPLETE', 'ABORTED', 'STOPPED', 'UNKNOWN']

Run state, possible values INIT, RUNNING, COMPLETE, ABORTED, STOPPED(?).

property runtitle_safe: str

Run name with ” ” replaced by “_”.

Return type

str

property sample_wells: dict[str, list[str]]

A dictionary of sample names to sample wells (convenience read/write access to the PlateSetup .

Return type

dict[str, list[str]]

save_file(file, overwrite=False)[source]

Save an EDS file of the experiment. This should be readable by AB’s software, but makes no attempt to hide that it was written by QSLib, and contains some other information. By default, this will refuse to overwrite an existing file.

Parameters
  • file (str or IO[bytes]) – A filename or open binary IO.

  • overwrite (bool, optional) – If True, overwrite any existing file without warning. Defaults to False.

Return type

None

save_file_without_changes(file, overwrite=False)[source]

Save an EDS file of the experiment. Unlike save_file, this will not update any parts of the file, so if it has not been modified elsewhere, it will be the same as when it was loaded. By default, this will refuse to overwrite an existing file.

Parameters
  • file (str or IO[bytes]) – A filename or open binary IO.

  • overwrite (bool, optional) – If True, overwrite any existing file without warning. Defaults to False.

Return type

None

stop(machine=None)[source]

If this experiment is running, stop it after the end of the current cycle.

Requires and takes exclusive Controller access on the machine.

Raises

NotRunningError – the experiment is not currently running

Return type

None

summary(format='markdown', plate='list')[source]
Return type

str

sync_from_machine(machine=None, log_method='eval')[source]

Try to synchronize the data in the experiment to the current state of the run on a machine, more efficiently than reloading everything.

Return type

None

temperatures: pd.DataFrame | None = None

A DataFrame of temperature readings, at one second resolution, during the experiment (and potentially slightly before and after, if included in the message log).

Columns (as multi-index):

(“time”, …)float

Time of temperature reading, for choices of “timestamp” (Unix timestamp in seconds), “seconds” (seconds since the active start of the run), or “hours”. The latter two may be negative, and may not be set if the run never became active.

(“sample”, …)float

Sample temperature for blocks 1, 2, …, 6, and average in “avg”.

(“block”, …)float

Block temperature for blocks 1, 2, …, 6, and average in “avg”.

(“other”, “cover”)float

Cover temperature

(“other”, “heatsink”)float

Heatsink temperature

user: str | None
property welldata: pandas.core.frame.DataFrame

A DataFrame with fluorescence reading information.

Indices (multi-index) are (filter_set, stage, cycle, step, point), where filter_set is a string in familiar form (eg, “x1-m4”) and the rest are int.

Columns (as multi-index):

(“time”, …)float

Time of the data collection, taken from the .quant file. May differ for different filter sets. Options are “timestamp” (unix timestamp in seconds), “seconds”, and “hours” (the latter two from the active start of the run).

(well, option)float

Data for a well, with well formatted like “A05”. Options are “rt” (read temperature from .quant file), “st” (more stable temperature), and “fl” (fluorescence).

(“exposure”, “exposure”)float

Exposure time from filterdata.xml. Misleading, because it only refers to the longest exposure of multiple exposures.

Return type

DataFrame

writesoftware: str

A string describing the software and version used to write the file.

class qslib.Machine(host, password=None, automatic=True, max_access_level=AccessLevel.Controller, port=7000, _initial_access_level=AccessLevel.Observer)[source]

Bases: object

A connection to a QuantStudio machine. The connection can be opened and closed, and reused. A maximum access level can be set and changed, which will prevent the access level from going above that level.

By default, the class tries to handle connections and access automatically.

Parameters
  • host (str) – The host name or IP to connect to.

  • password (str | None) – The password to use. Note that this class does not obscure or protect the password at all, because it should not be relied on for security. See Security Considerations for more information.

  • automatic (bool) – Whether or not to automatically handle connection, disconnection, and where possible, access level. Default True.

  • max_access_level ("Observer", "Controller", "Administrator", or "Full") – The maximum access level to allow. This is not the initial access level, which will be Observer. The parameter can be changed later by changing the max_access_level attribute.

  • port (int) – The port to connect to. (Use the normal SCPI port, not the line-editor connection usually on 2323). Default is 7000.

abort_current_run()[source]

Abort (stop immediately) the current run.

Return type

None

property access_level: qslib.scpi_commands.AccessLevel
Return type

AccessLevel

asdict(password=False)[source]
Return type

dict[str, str | int]

at_access(access_level, exclusive=False, stealth=False)[source]
Return type

Generator[Machine, None, None]

automatic: bool = True
connect()[source]

Open the connection manually.

Return type

None

property connected: bool

Whether or not there is a current connection to the machine.

Note that when using automatic connections, this will usually be False, because connections will only be active when running a command.

Return type

bool

property connection: qslib.qsconnection_async.QSConnectionAsync

The QSConnectionAsync for the connection, or a ConnectionError.

Return type

QSConnectionAsync

cover_lower(check=True, ensure_drawer=True)[source]

Lower/engage the plate cover, closing the drawer if needed.

Return type

None

property cover_position: Literal['Up', 'Down', 'Unknown', '']

Return the cover position from the ENG? command. Note that this does not always seem to work.

Return type

Literal[‘Up’, ‘Down’, ‘Unknown’, ‘’]

property current_run_name: str | None

Name of current run, or None if no run is active.

Return type

str | None

define_protocol(protocol)[source]

Send a protocol to the machine. This is not related to a particular experiment. The name on the machine is set by the protocol.

Parameters

protocol (Protocol) – protocol to send

Return type

None

disconnect()[source]

Cleanly disconnect from the machine.

Return type

None

drawer_close(lower_cover=True, check=True)[source]

Close the machine drawer using the OPEN command. This will ensure proper cover/drawer operation. It will not check run status, and will open and close the drawer during runs and potentially during imaging.

By default, it will lower the cover automaticaly after closing, use lower_cover=False to not do so.

Return type

None

drawer_open()[source]

Open the machine drawer using the OPEN command. This will ensure proper cover/drawer operation. It will not check run status, and will open and close the drawer during runs and potentially during imaging.

Return type

None

property drawer_position: Literal['Open', 'Closed', 'Unknown']

Return the drawer position from the DRAW? command.

Return type

Literal[‘Open’, ‘Closed’, ‘Unknown’]

ensured_connection(access_level=AccessLevel.Observer)[source]
Return type

Generator[Machine, None, None]

get_access_level()[source]
Return type

tuple[AccessLevel, bool, bool]

get_running_protocol()[source]
Return type

Protocol

host: str
list_files(path: str, *, leaf: str = "'FILE'", verbose: Literal[True], recursive: bool = 'False') list[dict[str, Any]][source]
list_files(path: str, *, leaf: str = "'FILE'", verbose: Literal[False], recursive: bool = 'False') list[str]
Return type

list[str] | list[dict[str, Any]]

list_runs_in_storage()[source]

List runs in machine storage.

Returns

run filenames. Retrieve with load_run_from_storage (to open as :any`Experiment`) or save_run_from_storage (to download and save it without opening.)

Return type

list[str]

load_run_from_storage(path)[source]
Return type

Experiment

machine_status()[source]

Return information on the status of the machine.

Return type

MachineStatus

property max_access_level: qslib.scpi_commands.AccessLevel
Return type

AccessLevel

password: str | None = None
pause_current_run()[source]

Pause the current run now.

Return type

None

pause_current_run_at_temperature()[source]
Return type

None

port: int = 7000
property power: bool

Get and set the machine’s operational power (lamp, etc) as a bool.

Setting this to False will not turn off the machine, just power down the lamp, temperature control, etc. It will do so even if there is currently a run.

Return type

bool

read_dir_as_zip(path, leaf='FILE')[source]

Read a directory on the

Parameters
  • path (str) – path on the machine

  • leaf (str, optional) – leaf to use, by default “FILE”

Returns

the returned zip file

Return type

zipfile.ZipFile

read_file(path, context=None, leaf='FILE')[source]

Read a file.

Parameters
  • path (str) – File path on the machine.

  • context (str | None (default None)) – Context.

  • leaf (str (default FILE)) –

Returns

returned file

Return type

bytes

restart_system()[source]

Restart the system (both the InstrumentServer and android interface) by killing the zygote process.

Return type

None

resume_current_run()[source]

Resume the current run.

Return type

None

run_command(command)[source]

Run a SCPI command, and return the response as a string. Waits for OK, not just NEXT.

Parameters

command (str) – command to run

Returns

Response message (after “OK”, not including it)

Return type

str

Raises

CommandError – Received an Error response.

run_command_bytes(command)[source]

Run an SCPI command, and return the response as bytes (undecoded). Returns after the command is processed (OK or NEXT), but potentially before it has completed (NEXT).

Parameters

command (str | bytes | SCPICommand) – command to run

Returns

Response message (after “OK” or “NEXT”, likely “” in latter case)

Return type

bytes

Raises

CommandError – Received

run_command_to_ack(command)[source]

Run an SCPI command, and return the response as a string. Returns after the command is processed (OK or NEXT), but potentially before it has completed (NEXT).

Parameters

commands – command to run

Returns

Response message (after “OK” or “NEXT”, likely “” in latter case)

Return type

str

Raises

CommandError – Received an Error response.

run_status()[source]

Return information on the status of any run.

Return type

RunStatus

save_run_from_storage(machine_path, download_path, overwrite=False)[source]

Download a file from run storage on the machine.

Parameters
  • machine_path (str) – filename on the machine

  • download_path (str | IO[bytes]) – filename to download to, or an open file

  • overwrite (bool, optional) – if False and provided a filename rather than an open file, will not overwrite existing filies; by default False

Return type

None

set_access_level(access_level, exclusive=False, stealth=False)[source]
Return type

None

property status: qslib.base.RunStatus

Return the current status of the run.

Return type

RunStatus

stop_current_run()[source]

Stop (stop after cycle end) the current run.

Return type

None

write_file(path, data)[source]
Return type

None

class qslib.MachineStatus(drawer, cover, lamp_status)[source]

Bases: qslib.base.BaseStatus

cover: str
drawer: str
lamp_status: str
class qslib.NormRaw[source]

Bases: qslib.processors.Processor

A Processor that takes no arguments, and simply passes through raw fluorescence values.

process(data)[source]
Return type

DataFrame

process_scoped(data, scope)[source]

Filter the data, and return it (possibly not a copy), if scope is the minimum necessary scope for this normalization type. Otherwise, just return the same data.

This is useful for writing scope-agnostic code, provided that you call this for every scope before using the data.

The values for scope are:

  • “all”: the entire welldata array.

  • “limited”: all time points, but limited to the filter sets and samples being plotted.

Return type

DataFrame

ylabel(previous_label=None)[source]
Return type

str

class qslib.NormToMaxPerWell(stage=None, step=None, cycle=None, *, selection=None)[source]

Bases: qslib.processors.Processor

A Processor that divides the fluorescence reading for each (filterset, well) pair by the max value of that pair within a particular selection of data.

The easiest way to use this is to give a particular stage (all data in that stage will be used), or a stage and set of cycles (those cycles in that stage will be used). For example:

  • To normalize to the mean stage 8 values, use NormToMeanPerWell(stage=8).

  • To normalize to the first 5 cycles of stage 2, use

    NormToMeanPerWell(stage=2, cycle=slice(1, 6)).

selection allows arbitrary Pandas indexing (without the filter_set level of the MultiIndex) for unusual cases.

process(data)[source]
Return type

DataFrame

process_scoped(data, scope)[source]

Filter the data, and return it (possibly not a copy), if scope is the minimum necessary scope for this normalization type. Otherwise, just return the same data.

This is useful for writing scope-agnostic code, provided that you call this for every scope before using the data.

The values for scope are:

  • “all”: the entire welldata array.

  • “limited”: all time points, but limited to the filter sets and samples being plotted.

Return type

DataFrame

scope: ClassVar[Literal['all', 'limited']] = 'limited'
selection: Any
ylabel(previous_label=None)[source]
Return type

str

class qslib.NormToMeanPerWell(stage=None, step=None, cycle=None, *, selection=None)[source]

Bases: qslib.processors.Processor

A Processor that divides the fluorescence reading for each (filterset, well) pair by the mean value of that pair within a particular selection of data.

The easiest way to use this is to give a particular stage (all data in that stage will be used), or a stage and set of cycles (those cycles in that stage will be used). For example:

  • To normalize to the mean stage 8 values, use NormToMeanPerWell(stage=8).

  • To normalize to the first 5 cycles of stage 2, use

    NormToMeanPerWell(stage=2, cycle=slice(1, 6)).

selection allows arbitrary Pandas indexing (without the filter_set level of the MultiIndex) for unusual cases.

process(data)[source]
Return type

DataFrame

process_scoped(data, scope)[source]

Filter the data, and return it (possibly not a copy), if scope is the minimum necessary scope for this normalization type. Otherwise, just return the same data.

This is useful for writing scope-agnostic code, provided that you call this for every scope before using the data.

The values for scope are:

  • “all”: the entire welldata array.

  • “limited”: all time points, but limited to the filter sets and samples being plotted.

Return type

DataFrame

scope: ClassVar[Literal['all', 'limited']] = 'limited'
selection: Any
ylabel(previous_label=None)[source]
Return type

str

class qslib.PlateSetup(sample_wells=None, samples=())[source]

Bases: object

classmethod from_array(array, *, make_unique=False)[source]
Return type

PlateSetup

async classmethod from_machine(c, runtitle=None)[source]
Return type

PlateSetup

classmethod from_platesetup_xml(platexml)[source]
Return type

PlateSetup

classmethod from_tsv(tsvstr)[source]
Return type

PlateSetup

get_descriptive_string(name)[source]
Return type

str

get_wells(samples_or_wells)[source]

Given a sample, well, or list of the two, returns the corresponding wells. Note that this relies on samples not having well-like names.

Return type

list[str]

property sample_wells
samples_by_name: Dict[str, qslib.plate_setup.Sample]
to_lineprotocol(timestamp, run_name=None)[source]
Return type

list[str]

to_table(headers=[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12], tablefmt='orgtbl', showindex=('A', 'B', 'C', 'D', 'E', 'F', 'G', 'H'), **kwargs)[source]
Return type

str

update_xml(root)[source]
Return type

None

property well_sample
well_samples_as_array()[source]
Return type

ndarray[Any, Any]

class qslib.Protocol(stages=NOTHING, name=NOTHING, volume=50.0, runmode='standard', filters=NOTHING, covertemperature=105.0, prerun=NOTHING, postrun=NOTHING, classname='Protocol')[source]

Bases: qslib.protocol.ProtoCommand

A run protocol for the QuantStudio. Protocols encapsulate the temperature and camera controls for an entire run. They are composed of Stage`s, which may repeat for a number of cycles, and the stages are in turn composed of Steps, which may be created for usual cases with :class:`Step, or from SCPI commands. Steps may repeat their contents as well, but this is not yet implemeted.

Parameters
  • stages (Iterable[Stage]) – The stages of the protocol, likely Stage.

  • stage (_NumOrRefIndexer[Stage]) – A more convenient way of accessing the stages of the protocol, with numbering that matches the machine.

  • name (str | None) – A protocol name. If not set, a timestamp will be used, unlike AB’s uuid.

  • volume (float) – The sample volume, in µL.

  • runmode (str | None) – The run mode.

  • covertemperature (float (default 105.0)) – The cover temperature

  • filters (Sequence[str]) – A list of default filters that can be used by any collection commands that don’t specify their own.

  • prerun (Sequence[SCPICommand]) – Sets PRERUN. DO NOT USE THIS UNLESS YOU KNOW WHAT YOU ARE DOING.

  • postrun (Sequence[SCPICommand]) – Sets POSTRUN. DO NOT USE THIS UNLESS YOU KNOW WHAT YOU ARE DOING.

property all_filters: Collection[qslib.data.FilterSet]

A list of all filters used at some point in the protocol.

Return type

Collection[FilterSet]

property all_points: pandas.core.frame.DataFrame
Return type

DataFrame

property all_temperatures: numpy.ndarray

An array of temperature settings at all_times.

Return type

ndarray

property all_times: numpy.ndarray

An array of all start and end times of each step, interleaved.

Return type

ndarray

check_compatible(new, status)[source]

Checks compatibility for changing a running protocol to a new one.

Raises ValueError if incompatible, returns True if compatible.

Parameters
Raises

ValueError – Protocols are incompatible.

Return type

bool

copy()[source]

Returns a new copy (recursively) of the protocol.

Return type

Protocol

covertemperature: float
property dataframe: pandas.core.frame.DataFrame

A DataFrame of the temperature protocol.

Return type

DataFrame

filters: Sequence[qslib.data.FilterSet]
classmethod from_scpicommand(sc)[source]
Return type

Protocol

classmethod from_xml(e)[source]
Return type

Protocol

classmethod fromdict(d)[source]
Return type

Protocol

info()[source]

Generate a (markdown) text protocol description.

Return type

str

name: str
plot_protocol(ax=None)[source]

A plot of the temperature and data collection points.

Return type

Tuple[plt.Axes, Tuple[List[plt.Line2D], List[plt.Line2D]]]

postrun: Sequence[qslib.scpi_commands.SCPICommandLike]
prerun: Sequence[qslib.scpi_commands.SCPICommandLike]
runmode: str
property stage: qslib.protocol._NumOrRefIndexer[qslib.protocol.Stage]

A more convenient view of Protocol.stages. This allows one-indexed access, such that protocol.stage[5] == protocol.stages[6] is stage 5 of the protocol, in the interpretation of tha machine. Indexing can use slices, and is inclusive, so protocol.stage[5:6] returns stages 5 and 6. Getting, setting, and appending stages are all supported through this interface.

Return type

_NumOrRefIndexer[Stage]

stages: list[qslib.protocol.Stage]
to_scpicommand(**kwargs)[source]
Return type

SCPICommand

to_xml(covertemperature=105.0)[source]
Return type

tuple[ElementTree, ElementTree]

volume: float
class qslib.RunStatus(name, stage, num_stages, cycle, num_cycles, step, point, state)[source]

Bases: qslib.base.BaseStatus

cycle: int
name: str
num_cycles: int
num_stages: int
point: int
stage: int
state: str
step: int
class qslib.SmoothEMWMean(com=None, span=None, halflife=None, alpha=None, min_periods=0, adjust=True, ignore_na=False)[source]

Bases: qslib.processors.Processor

A Processor that smooths fluorescence readings using Pandas’ Exponential Moving Window (ewm / exponentially weighted moving-average).

adjust: bool
alpha: float | None
com: float | None
halflife: float | str | timedelta | None
ignore_na: bool
min_periods: int
process(data)[source]
Return type

DataFrame

process_scoped(data, scope)[source]

Filter the data, and return it (possibly not a copy), if scope is the minimum necessary scope for this normalization type. Otherwise, just return the same data.

This is useful for writing scope-agnostic code, provided that you call this for every scope before using the data.

The values for scope are:

  • “all”: the entire welldata array.

  • “limited”: all time points, but limited to the filter sets and samples being plotted.

Return type

DataFrame

scope: ClassVar[ScopeType] = 'limited'
span: float | None
ylabel(previous_label=None)[source]
Return type

str

class qslib.Stage(steps, repeat=1, index=None, label=None, default_filters=())[source]

Bases: qslib.protocol.XMLable, qslib.protocol.ProtoCommand

A Stage in a protocol, composed of Step s with a possible repeat.

dataframe(start_time=0, previous_temperatures=None)[source]

Create a dataframe of the steps in this stage.

Parameters
  • start_time (float) – The initial start time, in seconds, of the stage (before the ramp to the first step). Default is 0.

  • previous_temperatures (list[float] | None) – A list of temperatures at the end of the previous stage, to allow calculation of ramp time. If None, the ramp is assumed to take no time.

Return type

pd.DataFrame

classmethod from_scpicommand(sc, **kwargs)[source]
Return type

Stage

classmethod from_xml(e)[source]
Return type

Stage

classmethod fromdict(d)[source]
Return type

Stage

classmethod hold_at(temperature, total_time, step_time=None, collect=None, filters=())[source]

Hold at a temperature for a set amount of time, with steps of a configurable fixed time.

Parameters
  • temperatures – The temperature or temperatures to hold. If not strings or quantities, the value/values are interpreted as °C.

  • total_time (int | str | pint.Quantity[int]) – Desired total time for the stage. If this is not a multiple of step_time, it may not be the actual total time for the stage. The function will emit a warning if the difference is more than 10%. If an integer, value is interpreted as seconds.

  • step_time (int | str | pint.Quantity[int] | None) – If None (default), the stage will have one step. Otherwise, it will have steps of this time. If an integer, value is interpreted as seconds.

  • collect (bool | None) – Whether or not each step should collect fluorescence data. If None (default), collects data if filters is explicitly set.

  • filters (Sequence[str | FilterSet]) – A list of filters to collect. If empty, and collect is True, then each step will collect the default filters for the Protocol.

Returns

The resulting Stage

Return type

Stage

Raises

ValueError – If step time is larger than total time.

classmethod hold_for(temperature, total_time, step_time=None, collect=None, filters=())

Hold at a temperature for a set amount of time, with steps of a configurable fixed time.

Parameters
  • temperatures – The temperature or temperatures to hold. If not strings or quantities, the value/values are interpreted as °C.

  • total_time (int | str | pint.Quantity[int]) – Desired total time for the stage. If this is not a multiple of step_time, it may not be the actual total time for the stage. The function will emit a warning if the difference is more than 10%. If an integer, value is interpreted as seconds.

  • step_time (int | str | pint.Quantity[int] | None) – If None (default), the stage will have one step. Otherwise, it will have steps of this time. If an integer, value is interpreted as seconds.

  • collect (bool | None) – Whether or not each step should collect fluorescence data. If None (default), collects data if filters is explicitly set.

  • filters (Sequence[str | FilterSet]) – A list of filters to collect. If empty, and collect is True, then each step will collect the default filters for the Protocol.

Returns

The resulting Stage

Return type

Stage

Raises

ValueError – If step time is larger than total time.

index: int | None
info_str(index=None)[source]
Return type

str

label: str | None
repeat: int
property step: qslib.protocol._NumOrRefIndexer[qslib.protocol.CustomStep]
Return type

_NumOrRefIndexer[CustomStep]

classmethod stepped_ramp(from_temperature, to_temperature, total_time, *, n_steps=None, temperature_step=None, collect=None, filters=())[source]

Hold at a series of temperatures, from one to another.

Parameters
  • from_temperature (float | str | pint.Quantity[float] | Sequence[float]) – Initial temperature/s (inclusive).

  • to_temperature (float | str | pint.Quantity[float] | Sequence[float]) – Final temperature/s (inclusive).

  • total_time (int | str | pint.Quantity[int]) – Total time for the stage

  • n_steps (int | None) – Number of steps. If None, uses 1.0 Δ°C steps, or, if doing a multi-temperature change, uses maximum step of 1.0 Δ°C.

  • temperature_step (float | str | pint.Quantity[float] | None) – Step temperature change (optional). Must be None, or correctly match calculation, if n_steps is not None. If both this and n_steps are None, default is 1.0 Δ°C steps. If temperature step does not exactly fit range, it will be adjusted, with a warning if the change is more than 5%. Sign is ignored. If doing a multi-temperature change, then this is the maximum temperature step.

  • collect (bool | None) – Collect data? If None, collects data if filters is set explicitly.

  • filters (Sequence[str | FilterSet]) – Filters to collect.

Returns

The resulting stage.

Return type

Stage

steps: Sequence[CustomStep]
to_scpicommand(stageindex=None, **kwargs)[source]
Return type

SCPICommand

to_xml(**kwargs)[source]
Return type

Element

class qslib.Step(time, temperature, collect=None, temp_increment=<Quantity(0.0, 'delta_degree_Celsius')>, temp_incrementcycle=2, temp_incrementpoint=2, time_increment=<Quantity(0, 'second')>, time_incrementcycle=2, time_incrementpoint=2, filters=(), pcr=False, quant=True, tiff=False, repeat=1, default_filters=())[source]

Bases: qslib.protocol.CustomStep, qslib.protocol.XMLable

A normal protocol step, of a hold and possible collection.

Parameters
  • time (int) – The step time setting, in seconds.

  • temperature (float | Sequence[float]) – The temperature hold setting, either as a float (all zones the same) or a sequence (of correct length) of floats setting the temperature for each zone.

  • collect (bool | None) – Collect fluorescence data? If None (default), collect only if the Step has an explicit filters setting.

  • temp_increment (float) – Amount to increment all zone temperatures per cycle on and after temp_incrementcycle.

  • temp_incrementcycle (int (default 2)) – First cycle to start the increment changes. Note that the default in QSLib is 2, not 1 (as in AB’s software), so that leaving this alone makes sense (the first cycle will be at temperature, the next at temperature + temp_incrementcycle.

  • time_increment (float) –

  • time_incrementcycle (int) – The same settings for time per cycle.

  • filters (Sequence[FilterSet | str] (default empty)) – A list of filter pairs to collect, either using FilterSet or a string like “x1-m4”. If collect is True and this is empty, then the filters will be set by the Protocol.

Notes

This currently does not support step-level repeats, which do exist on the machine.

property body: list[qslib.protocol.ProtoCommand]
Return type

list[ProtoCommand]

collect: bool | None
property collects
duration_at_cycle(cycle)[source]

Duration of the step (excluding ramp) at cycle (from 1)

Return type

Quantity

filters: Sequence[FilterSet]
classmethod from_scpicommand(sc)[source]
Return type

Step

classmethod from_xml(e, *, etc=1, ehtc=1, he=False)[source]
Return type

Step

classmethod fromdict(d)[source]
Return type

Step

property identifier: None
Return type

None

info_str(index=None, repeats=1)[source]

String describing the step.

Return type

str

pcr: bool
quant: bool
repeat: int
temp_increment: pint.Quantity[float]
temp_incrementcycle: int
temp_incrementpoint: int
temperature: pint.Quantity[Any]
property temperature_list: pint.quantity.Quantity[numpy.ndarray]
Return type

Quantity[ndarray]

temperatures_at_cycle(cycle)[source]

Temperatures of the step at cycle (from 1)

Return type

Quantity[ndarray]

tiff: bool
time: pint.Quantity[int]
time_increment: pint.Quantity[int]
time_incrementcycle: int
time_incrementpoint: int
to_xml(**kwargs)[source]
Return type

Element

total_duration(repeats=1)[source]
Return type

Quantity