High-level API: Experiment, Protocol, PlateSetup, Protocol¶

Class for accessing and manipulating experiments¶

class qslib.common.Experiment(name: str | None = None, protocol: Protocol | None = None, plate_setup: PlateSetup | None = None, _create_xml: bool = True)[source]

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_storage(machine, "experiment.eds")

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: Machine | None = None) → 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

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]

change_protocol(new_protocol: Protocol, machine: Machine | None = None) → None[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

collect_finished(machine: Machine | None = None) → None[source]

NOT YET IMPLEMENTED

Collect the completed (aborted/etc) experiment from a machine, reliably. This will search for and recover working data if the EDS file generation on the machine failed.

Parameters: machine (Machine, optional) – [description], by default None
Raises: NotImplementedError – [description]

create_new_copy() → qslib.experiment.Experiment[source]

Create a copy of the experiment, with data and run information removed, suitable for rerunning.

Returns
Return type: Experiment

createdtime: datetime: The run creation time.

data_for_sample(sample: str) → pandas.core.frame.DataFrame[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 filterdata: pandas.core.frame.DataFrame

classmethod from_file(file: str | os.PathLike[str] | IO[bytes]) → Experiment[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_storage(machine: qslib.machine.Machine, path: str) → qslib.experiment.Experiment[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 runnig experiment
Return type: Experiment

classmethod from_running(machine: qslib.machine.Machine) → qslib.experiment.Experiment[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 runnig experiment
Return type: Experiment

classmethod from_uncollected(machine: qslib.machine.Machine, name: str, move: bool = False) → qslib.experiment.Experiment[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 runnig experiment

Return type

Experiment

get_status(machine: Machine | None = None) → RunStatus[source]

Return the status of the experiment, if currently running.

Requires Observer access on the machine.

Raises: NotRunningError – the experiment is not currently running

info_html() → str | ET.ElementTree | None[source]

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.

pause_now(machine: Machine | None = None) → 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

plate_setup: PlateSetup: Plate setup for the experiment.

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.

samples: Either a reference to a single sample (a string), or a list of sample names. 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.
filters: 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, melt_stages, 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: Optional. A Normalizer instance to apply to the data. By default, this is NormRaw, which passes through raw fluorescence values. NormToMeanPerWell also works well.

marker

ax: Optional. An axes to put the plot on. If not provided, the function will create a new figure.
legend: Optional. Determines whether a legend is included.
figure_kw: Optional. A dictionary of options passed through as keyword options to the figure creation. Only applies if ax is None.
line_kw: Optional. A dictionary of keywords passed to all three plotting commands.

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.

samples

Either a reference to a single sample (a string), or a list of sample names. 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.

filters

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

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

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

Optional (default False). 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

Optional. An axes to put the plot on. If not provided, the function will create a new figure. If temperatures=”axes”, however, you must provide a list or tuple of two axes, the first for fluorescence, the second for temperature.

legend

Optional. Determines whether a legend is included.

figure_kw

Optional. A dictionary of options passed through as keyword options to the figure creation. Only applies if ax is None.

line_kw

Optional. A dictionary of keywords passed to fluorescence plot commands.

plot_temperatures()[source]

protocol: Protocol: Temperature and reading protocol for the experiment.

property rawdata: pandas.core.frame.DataFrame

resume(machine: Machine | None = None) → 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

run(machine: Machine | str | None = None, password=None, require_exclusive=True) → None[source]

Load the run onto a machine, and start it.

Parameters

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

Raises

MachineBusyError – The machine isn’t idle.
AlreadyExistsError – The machine already has a folder for this run in its working runs folder.

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 “-“.

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

save_file(file: str | os.PathLike[str] | IO[bytes], overwrite: bool = False) → None[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.

save_file_without_changes(file: str | IO[bytes], overwrite: bool = False) → None[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.

stop(machine: Machine | None = None) → 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

summary(format: str = 'markdown', plate: str = 'list') → str[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

sync_from_machine(machine: Machine | None = None, connect: bool = False, log_method: Literal['copy', 'eval'] = 'eval') → None[source]: Try to synchronize the data in the experiment to the current state of the run on a machine, more efficiently than reloading everything.

temperatures: pd.DataFrame | 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.

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

Classes for defining experiments¶

class qslib.common.Protocol(stages: Iterable[Stage], name: str = <factory>, volume: float = 50.0, runmode: str = 'standard', filters: list[str | FilterSet] = <factory>, covertemperature: float = 105.0, _class: str = 'Protocol')[source]

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 :any:`Step, or from SCPI commands (TODO: implement). Steps may repeat their contents as well, but this is not yet implemeted.

Parameters

stages (Iterable[Stage]) – The stages of the protocol, likely Stage.
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.

property all_filters: Collection[qslib.data.FilterSet]

property all_points: pandas.core.frame.DataFrame

property all_temperatures: numpy.ndarray: An array of temperature settings at all_times.

property all_times: numpy.ndarray: An array of all start and end times of each step, interleaved.

check_compatible(new: qslib.tcprotocol.Protocol, status: qslib.base.RunStatus)[source]

copy() → qslib.tcprotocol.Protocol[source]

covertemperature: float = 105.0

property dataframe: pandas.core.frame.DataFrame: A DataFrame of the temperature protocol.

filters: list[str | FilterSet]

classmethod from_command(s: str) → qslib.tcprotocol.Protocol[source]

classmethod from_xml(e: xml.etree.ElementTree.Element) → qslib.tcprotocol.Protocol[source]

classmethod fromdict(d: dict[str, typing.Any]) → qslib.tcprotocol.Protocol[source]

name: str

runmode: str = 'standard'

stages: Iterable[Stage]

tcplot(ax: Optional['plt.Axes'] = None) → Tuple['plt.Axes', Tuple[List['plt.Line2D'], List['plt.Line2D']]][source]: A plot of the temperature and data collection points.

to_command() → str[source]

to_xml(e: ET.Element | None = None, covertemperature=105.0) → tuple[ET.ElementTree, ET.ElementTree][source]

volume: float = 50.0

class qslib.common.Stage(steps: 'list[BaseStep] | Step' = <property object at 0x7ff4a8ce4360>, repeat: 'int' = 1, index: 'int | None' = None, label: 'str | None' = None, _class: 'str' = 'Stage')[source]

dataframe(start_time: float = 0, previous_temperatures: list[float] | None = None) → pd.DataFrame[source]

Create a dataframe of the steps in this stage.

Parameters

start_time – The initial start time, in seconds, of the stage (before the ramp to the first step). Default is 0.
previous_temperatures – 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.

classmethod from_xml(e: xml.etree.ElementTree.Element) → qslib.tcprotocol.Stage[source]

classmethod fromdict(d: dict[str, typing.Any]) → qslib.tcprotocol.Stage[source]

index: int | None = None

info_str(index) → str[source]

label: str | None = None

repeat: int = 1

classmethod stepped_ramp(temp_from: float, temp_to: float, total_time: int, nsteps: int, collect: bool = False, filters: Sequence = ())[source]

property steps: list[qslib.tcprotocol.BaseStep]

to_command(stageindex=None, **kwargs)[source]

to_xml() → xml.etree.ElementTree.Element[source]

class qslib.common.Step(time: int, temperature: float | Sequence[float], collect: bool = False, temp_increment: float = 0.0, temp_incrementcycle: int = 2, time_increment: int = 0, time_incrementcycle: int = 2, filters: Sequence[FilterSet | str] = (), pcr: bool = False, quant: bool = True, tiff: bool = False, _class: str = 'Step')[source]

A normal protocol step.

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) – Collect fluorescence data?
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.tcprotocol.ProtoCommand]

collect: bool = False

duration_at_cycle(cycle: int) → float[source]: Duration of the step (excluding ramp) at cycle (from 1)

filters: Sequence[FilterSet | str] = ()

classmethod from_xml(e: xml.etree.ElementTree.Element, *, etc=1, ehtc=1, he=False) → qslib.tcprotocol.Step[source]

classmethod fromdict(d: dict[str, typing.Any]) → qslib.tcprotocol.Step[source]

property identifier: None

info_str(index, repeats: int = 1) → str[source]: String describing the step.

pcr: bool = False

quant: bool = True

property repeat: int

temp_increment: float = 0.0

temp_incrementcycle: int = 2

temperature: float | Sequence[float]

property temperature_list: list[float]

temperatures_at_cycle(cycle: int) → list[float][source]: Temperatures of the step at cycle (from 1)

tiff: bool = False

time: int

time_increment: int = 0

time_incrementcycle: int = 2

to_xml() → xml.etree.ElementTree.Element[source]

total_duration(repeats: int = 1)[source]

class qslib.common.PlateSetup(sample_wells: 'Mapping[str, str | List[str]] | None' = None, samples: 'Iterable[Sample] | Mapping[str, Sample]' = ()) → 'None'[source]

classmethod from_array(array: Union[numpy.ndarray, pandas.core.frame.DataFrame], *, make_unique: bool = False) → qslib.plate_setup.PlateSetup[source]

async classmethod from_machine(c: qslib.qsconnection_async.QSConnectionAsync, runtitle: Optional[str] = None) → qslib.plate_setup.PlateSetup[source]

classmethod from_platesetup_xml(platexml: xml.etree.ElementTree.Element) → qslib.plate_setup.PlateSetup[source]

classmethod from_tsv(tsvstr: str) → qslib.plate_setup.PlateSetup[source]

get_wells(samples_or_wells: str | Sequence[str]) → list[str][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.

sample_wells: Dict[str, List[str]]

samples_by_name: Dict[str, qslib.plate_setup.Sample]

to_lineprotocol(timestamp, run_name=None)[source]

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

update_xml(root: xml.etree.ElementTree.Element)[source]

property well_sample

well_samples_as_array() → numpy.ndarray[source]

Classes for interacting with machines¶

class qslib.common.Machine(host: str, password: str | None = None, max_access_level: AccessLevel | str = AccessLevel.Observer, port: int = 7000, connect_now: bool = False, tunnel_host: str | tuple[str, int] | None = None, tunnel_user: str | None = None, tunnel_key: str | 'paramiko.pkey.PKey' | None = None, _initial_access_level: AccessLevel | str = AccessLevel.Observer)[source]

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 initial connection is as Observer.

For clean access, the class provides a context manager for the connection, and the at_level method provides a context manager for access level:

>>> with Machine('machine', 'password', max_access_level="Controller") as m:
>>>     # Now connected
>>>     print(m.run_status())  # runs at Observer level
>>>     with m.at_level("Controller", exclusive=True):
>>>         # Runs as Controller, and fails if another Controller is connected (exclusive)
>>>         m.abort_run()
>>>         m.drawer_open()
>>>     # Now back to Observer
>>>     print(m.status())
>>> # Now disconnected.

The connection context manager can also be used with with m: form for a Machine instance m that already exists, in which case it will connect and disconnect.

If you don’t want to use these, you can also use connect and disconnect.

Note that there is no supported method on the machine’s server for removing hanging connections other than a reboot, and AB’s software will not start runs when other connections hold Controller level.

Parameters

host (str) – The host name or IP to connect to.
password (str) – 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 Access and Security Considerations for more information.
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 (default = 7000)) – The port to connect to. (Use the normal SCPI port, not the line-editor connection usually on 2323).
tunnel_host (str or tuple[str, int], optional) – If set, to a hostname/IP string or (hostname/IP, port) tuple, create an SSH tunnel to the tunnel_host in order to connect to the machine, rather than connecting directly. This uses paramiko, and will not read your ssh configuration file for any host aliases, but will use your keys if you have an ssh-agent running.
tunnel_user (str, optional) – If set, specify the user for the tunnel connection. If unset, the local user name is used.
tunnel_key (str or paramiko.pkey.PKey, optional) – If set, specify the filename of a private key, or the paramiko-loaded key, to use for the tunnel connection.

Examples

Set up a connection

abort_current_run() → None[source]: Abort (stop immediately) the current run.

at_access(access_level: AccessLevel | str, exclusive: bool = False, stealth: bool = False) → Generator[Machine, None, None][source]

connect() → None[source]: Open the connection.

property connected

cover_lower() → None[source]: Lower/engage the plate cover, closing the drawer if needed.

property cover_position: Literal['Up', 'Down', 'Unknown']: Return the cover position from the ENG? command. Note that this does not always seem to work.

property current_run_name: str | None: Name of current run, or None if no run is active.

define_protocol(protocol: qslib.tcprotocol.Protocol) → None[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

disconnect() → None[source]: Cleanly disconnect from the machine.

drawer_close(lower_cover: bool = False) → None[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 not lower the cover automaticaly after closing, use lower_cover=True to do so.

drawer_open() → None[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.

property drawer_position: Literal['Open', 'Closed', 'Unknown']: Return the drawer position from the DRAW? command.

get_access_level() → tuple[qslib.base.AccessLevel, bool, bool][source]

get_running_protocol() → qslib.tcprotocol.Protocol[source]

host: str

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

list_runs_in_storage() → list[str][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: str) → Experiment[source]

machine_status() → qslib.base.MachineStatus[source]: Return information on the status of the machine.

max_access_level: AccessLevel | str = 'Observer'

password: str | None = None

pause_current_run() → None[source]: Pause the current run now.

pause_current_run_at_temperature() → None[source]

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.

read_dir_as_zip(path: str, leaf: str = 'FILE') → zipfile.ZipFile[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: str, context: str | None = None, leaf: str = 'FILE') → bytes[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

resume_current_run() → None[source]: Resume the current run.

run_command(command: str) → str[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: str | bytes) → bytes[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) – 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: str) → str[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: command (str) – 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() → qslib.base.RunStatus[source]: Return information on the status of any run.

save_run_from_storage(machine_path: str, download_path: str | IO[bytes], overwrite: bool = False) → None[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

set_access_level(access_level: AccessLevel | str, exclusive: bool = False, stealth: bool = False, _log: bool = True) → None[source]

property status: qslib.base.RunStatus: Return the current status of the run.

stop_current_run() → None[source]: Stop (stop after cycle end) the current run.

tunnel_host: str | tuple[str, int] | None = None

tunnel_key: str | 'paramiko.pkey.PKey' | None = None

tunnel_user: str | None = None

write_file(path: str, data: str | bytes) → None[source]

class qslib.common.RunStatus(name: 'str', stage: 'int', num_stages: 'int', cycle: 'int', num_cycles: 'int', step: 'int', point: 'int', state: 'str')[source]

cycle: int

name: str

num_cycles: int

num_stages: int

point: int

stage: int

state: str

step: int

class qslib.common.MachineStatus(drawer: 'str', cover: 'str', lamp_status: 'str')[source]

cover: str

drawer: str

lamp_status: str

High-level API: Experiment, Protocol, PlateSetup, Protocol¶

Class for accessing and manipulating experiments¶

Classes for defining experiments¶

Classes for interacting with machines¶

qslib

Navigation

Related Topics