qslib package

Submodules

qslib.base module

class qslib.base.AccessLevel(value)

Bases: enum.Enum

An enumeration.

Administrator = 'Administrator'

Controller = 'Controller'

Full = 'Full'

Guest = 'Guest'

Observer = 'Observer'

class qslib.base.BaseStatus

Bases: abc.ABC

classmethod from_bytes(out: bytes) → qslib.base.T

classmethod from_machine(connection: Machine) → T

class qslib.base.MachineStatus(drawer: 'str', cover: 'str', lamp_status: 'str')

Bases: qslib.base.BaseStatus

cover: str

drawer: str

lamp_status: str

class qslib.base.RunStatus(name: 'str', stage: 'int', num_stages: 'int', cycle: 'int', num_cycles: 'int', step: 'int', point: 'int', state: 'str')

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

qslib.common module

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

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

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.

change_protocol(new_protocol: Protocol, machine: Machine | None = None) → None

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

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

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

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

filterdata: pd.DataFrame

classmethod from_file(file: str | os.PathLike[str] | IO[bytes]) → Experiment

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

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

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

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

Return the status of the experiment, if currently running.

Requires Observer access on the machine.

Raises

NotRunningError – the experiment is not currently running

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

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.

protocol: Protocol

Temperature and reading protocol for the experiment.

resume(machine: Machine | None = None) → None

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

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

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

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

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

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) → None

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

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.

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)

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

Abort (stop immediately) the current run.

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

connect() → None

Open the connection.

cover_lower() → None

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

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

Cleanly disconnect from the machine.

drawer_close(lower_cover: bool = False) → None

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

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]

get_running_protocol() → qslib.tcprotocol.Protocol

host: str

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

list_files(path: str, *, leaf: str = "'FILE'", verbose: Literal[False], recursive: bool = 'False') → list[str]

list_runs_in_storage() → list[str]

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

machine_status() → qslib.base.MachineStatus

Return information on the status of the machine.

max_access_level: AccessLevel | str = 'Observer'

password: str | None = None

pause_current_run() → None

Pause the current run now.

pause_current_run_at_temperature() → 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.

read_dir_as_zip(path: str, leaf: str = 'FILE') → zipfile.ZipFile

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

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

Resume the current run.

run_command(command: str) → str

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

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

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

Return information on the status of any run.

save_run_from_storage(machine_path: str, download_path: str | IO[bytes], overwrite: bool = False) → None

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

property status: qslib.base.RunStatus

Return the current status of the run.

stop_current_run() → None

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

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

Bases: qslib.base.BaseStatus

cover: str

drawer: str

lamp_status: str

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

Bases: object

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

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

sample_wells: Dict[str, List[str]]

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

to_lineprotocol(timestamp, run_name=None)

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

update_xml(root: xml.etree.ElementTree.Element)

property well_sample

well_samples_as_array() → numpy.ndarray

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')

Bases: qslib.tcprotocol.XMLable

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_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)

copy() → qslib.tcprotocol.Protocol

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

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

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

name: str

runmode: str = 'standard'

stages: Iterable[Stage]

tcplot(ax: Optional['plt.Axes'] = None) → Tuple['plt.Axes', Tuple[List['plt.Line2D'], List['plt.Line2D']]]

A plot of the temperature and data collection points.

to_command() → str

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

volume: float = 50.0

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

Bases: qslib.base.BaseStatus

cycle: int

name: str

num_cycles: int

num_stages: int

point: int

stage: int

state: str

step: int

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

Bases: qslib.tcprotocol.XMLable

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

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

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

index: int | None = None

info_str(index) → str

label: str | None = None

repeat: int = 1

property steps: list[qslib.tcprotocol.BaseStep]

to_command(stageindex=None, **kwargs)

to_xml() → xml.etree.ElementTree.Element

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')

Bases: qslib.tcprotocol.BaseStep, qslib.tcprotocol.XMLable

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

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

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

property identifier: None

info_str(index, repeats: int = 1) → str

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]

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

total_duration(repeats: int = 1)

qslib.data module

class qslib.data.FilterDataReading(pde: ET.Element, timestamp: Optional[float] = None, sds_dir: Optional[str | PathLike[str]] = None, set_temperatures: Union[Literal['auto'], None, List[float]] = 'auto')

Bases: object

cycle: int

exposure: int

property filename_reading_string: str

filter_set: FilterSet

property plate_fluorescence: numpy.ndarray

property plate_set_temperatures: numpy.ndarray

property plate_temperatures: numpy.ndarray

point: int

set_temperatures: np.ndarray | None

set_timestamp_by_quantdata(qstring: str) → float

stage: int

step: int

temperatures: np.ndarray

timestamp: float | None

to_lineprotocol(run_name: Optional[str] = None, sample_array=None) → List[str]

well_fluorescence: np.ndarray

property well_set_temperatures: numpy.ndarray

property well_temperatures: numpy.ndarray

class qslib.data.FilterSet(ex: int, em: int, quant: bool = True)

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: str | FilterSet) → FilterSet

property hacform: str

property lowerform: str

quant: bool = True

to_xml() → xml.etree.ElementTree.Element

property upperform: str

qslib.data.df_from_readings(readings: List[FilterDataReading], start_time: float | None = None) → pd.DataFrame

qslib.experiment module

exception qslib.experiment.AlreadyExistsError

Bases: ValueError

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

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

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.

change_protocol(new_protocol: Protocol, machine: Machine | None = None) → None

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

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

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

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

filterdata: pd.DataFrame

classmethod from_file(file: str | os.PathLike[str] | IO[bytes]) → Experiment

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

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

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

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

Return the status of the experiment, if currently running.

Requires Observer access on the machine.

Raises

NotRunningError – the experiment is not currently running

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

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.

protocol: Protocol

Temperature and reading protocol for the experiment.

resume(machine: Machine | None = None) → None

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

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

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

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

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

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) → None

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

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.

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

Bases: qslib.experiment.MachineError

current: RunStatus

exception qslib.experiment.MachineError(machine: 'InitVar[Machine]')

Bases: Exception

host: str

machine: InitVar[Machine]

port: int | str

exception qslib.experiment.NotRunningError(machine: 'InitVar[Machine]', run: 'str', current: 'RunStatus')

Bases: qslib.experiment.MachineError

current: RunStatus

run: str

exception qslib.experiment.RunAlreadyExistsError(machine: 'InitVar[Machine]', name: 'str')

Bases: qslib.experiment.MachineError

name: str

qslib.machine module

class qslib.machine.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)

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

Abort (stop immediately) the current run.

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

connect() → None

Open the connection.

cover_lower() → None

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

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

Cleanly disconnect from the machine.

drawer_close(lower_cover: bool = False) → None

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

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]

get_running_protocol() → qslib.tcprotocol.Protocol

host: str

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

list_files(path: str, *, leaf: str = "'FILE'", verbose: Literal[False], recursive: bool = 'False') → list[str]

list_runs_in_storage() → list[str]

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

machine_status() → qslib.base.MachineStatus

Return information on the status of the machine.

max_access_level: AccessLevel | str = 'Observer'

password: str | None = None

pause_current_run() → None

Pause the current run now.

pause_current_run_at_temperature() → 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.

read_dir_as_zip(path: str, leaf: str = 'FILE') → zipfile.ZipFile

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

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

Resume the current run.

run_command(command: str) → str

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

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

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

Return information on the status of any run.

save_run_from_storage(machine_path: str, download_path: str | IO[bytes], overwrite: bool = False) → None

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

property status: qslib.base.RunStatus

Return the current status of the run.

stop_current_run() → None

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

qslib.monitor module

class qslib.monitor.Collector(config)

Bases: object

async docollect(args: Dict[str, Union[str, int]], state: qslib.monitor.State, connection: qslib.qsconnection_async.QSConnectionAsync) → None

async handle_led(topic, msg, timestamp)

async handle_msg(state, c, topic, msg, timestamp)

async handle_run_msg(state, c, topic, message, timestamp)

inject(t)

async matrix_announce(msg: str) → None

async monitor()

async reliable_monitor()

class qslib.monitor.MachineState(zone_targets: 'List[float]', zone_controls: 'List[bool]', cover_target: 'float', cover_control: 'bool', drawer: 'str')

Bases: object

cover_control: bool

cover_target: float

drawer: str

async classmethod from_machine(c: qslib.qsconnection_async.QSConnectionAsync)

async refresh(c: qslib.qsconnection_async.QSConnectionAsync)

zone_controls: List[bool]

zone_targets: List[float]

class qslib.monitor.RunState(name: 'Optional[str]' = None, stage: 'Optional[int]' = None, cycle: 'Optional[int]' = None, step: 'Optional[int]' = None, plate_setup: 'Optional[PlateSetup]' = None)

Bases: object

cycle: Optional[int] = None

async classmethod from_machine(c: qslib.qsconnection_async.QSConnectionAsync)

name: Optional[str] = None

plate_setup: Optional[qslib.plate_setup.PlateSetup] = None

async refresh(c: qslib.qsconnection_async.QSConnectionAsync)

stage: Optional[int] = None

statemsg(timestamp)

step: Optional[int] = None

class qslib.monitor.State(run: 'RunState', machine: 'MachineState')

Bases: object

async classmethod from_machine(c: qslib.qsconnection_async.QSConnectionAsync)

machine: qslib.monitor.MachineState

run: qslib.monitor.RunState

async qslib.monitor.get_runinfo(c: qslib.qsconnection_async.QSConnectionAsync)

qslib.monitor.index_to_filename_ref(i: Tuple[str, int, int, int, int]) → str

qslib.monitor.parse_fd_fn(x: str) → Tuple[str, int, int, int, int]

qslib.monitor_cli module

qslib.monitor_cli.main(args: List[str])

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: List[str]) → argparse.Namespace

qslib.monitor_cli.run()

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)

Setup basic logging

Parameters

loglevel (int) – minimum loglevel for emitting messages

qslib.parser module

qslib.parser.make_multi_keyword(kwd_str, kwd_value)

qslib.plate_setup module

Code for handling plate setup.

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

Bases: object

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

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

sample_wells: Dict[str, List[str]]

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

to_lineprotocol(timestamp, run_name=None)

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

update_xml(root: xml.etree.ElementTree.Element)

property well_sample

well_samples_as_array() → numpy.ndarray

class qslib.plate_setup.Sample(name: 'str', uuid: 'str' = <factory>, color: 'Tuple[int, int, int, int]' = (255, 0, 0, 255))

Bases: object

color: Tuple[int, int, int, int] = (255, 0, 0, 255)

classmethod from_platesetup_sample(se: xml.etree.ElementTree.Element) → qslib.plate_setup.Sample

name: str

to_xml() → xml.etree.ElementTree.Element

uuid: str

qslib.qs_is_protocol module

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

Bases: qslib.qs_is_protocol.Error

command: Optional[str]

ref_index: Optional[str]

response: str

exception qslib.qs_is_protocol.Error

Bases: Exception

class qslib.qs_is_protocol.QS_IS_Protocol

Bases: asyncio.protocols.Protocol

connection_lost(exc: Optional[Exception]) → None

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

connection_made(transport: Any) → None

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.

data_received(data: bytes) → None

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:

async disconnect() → None

async handle_sub_message(message: bytes) → None

async parse_message(ds: bytes) → None

async run_command(comm: str | bytes, ack_timeout: int = 300, just_ack: bool = True, uid: bool = True) → bytes

exception qslib.qs_is_protocol.ReplyError

Bases: OSError

class qslib.qs_is_protocol.SubHandler(*args, **kwargs)

Bases: Protocol

qslib.qsconnection_async module

class qslib.qsconnection_async.QSConnectionAsync(host: str = 'localhost', port: int = 7000, authenticate_on_connect: bool = True, initial_access_level: qslib.base.AccessLevel = AccessLevel.Observer, password: Optional[str] = None)

Bases: object

Class for connection to a QuantStudio instrument server, using asyncio

async authenticate(password: str)

async connect(authenticate: Optional[bool] = None, initial_access_level: AccessLevel | None = None, password: Optional[str] = None) → str

async disconnect() → None

async get_all_filterdata(run: Optional[str], as_list: Literal[True]) → List[qslib.data.FilterDataReading]

async get_all_filterdata(run: Optional[str], as_list: Literal[False]) → pandas.core.frame.DataFrame

async get_exp_file(path: str, encoding: Literal['plain', 'base64'] = 'base64') → bytes

async get_expfile_list(glob: str) → List[str]

async get_file(path: str, encoding: Literal['plain', 'base64'] = 'base64') → bytes

async get_filterdata_one(filterset: Union[qslib.data.FilterSet, str], stage: int, cycle: int, step: int, point: int = 1, run: Optional[str] = None) → qslib.data.FilterDataReading

async get_run_start_time() → float

async get_run_title() → str

async get_sds_file(path: str, runtitle: Optional[str] = None, encoding: Literal['base64', 'plain'] = 'base64') → bytes

async run_command(command: str, just_ack: bool = False) → str

async run_command_to_bytes(command: str, just_ack: bool = True) → bytes

async set_access_level(level: qslib.base.AccessLevel)

qslib.tcprotocol module

class qslib.tcprotocol.BaseStep

Bases: abc.ABC

abstract property body: Iterable[qslib.tcprotocol.ProtoCommand]

abstract property identifier: int | str | None

info_str(index=None, repeats=1) → str

abstract property repeat: int

to_command(*, stepindex, **kwargs)

class qslib.tcprotocol.Exposure(settings: 'Mapping[FilterSet, Sequence[int]]', state: 'str' = 'HoldAndCollect')

Bases: qslib.tcprotocol.ProtoCommand

settings: Mapping[qslib.data.FilterSet, Sequence[int]]

state: str = 'HoldAndCollect'

to_command(**kwargs) → str

class qslib.tcprotocol.HACFILT(filters: 'Iterable[FilterSet | str]') → 'None'

Bases: qslib.tcprotocol.ProtoCommand

filters: Sequence[FilterSet] | None

to_command(filters: Sequence[FilterSet | str] | None = None, **kwargs) → str

class qslib.tcprotocol.Hold(time: 'int | None', increment: 'float | None' = None, incrementcycle: 'int | None' = None, incrementstep: 'int | None' = None)

Bases: qslib.tcprotocol.ProtoCommand

increment: float | None = None

incrementcycle: int | None = None

incrementstep: int | None = None

time: int | None

to_command(**kwargs) → str

class qslib.tcprotocol.HoldAndCollect(time: 'int', increment: 'float | None' = None, incrementcycle: 'int | None' = None, incrementstep: 'int | None' = None, tiff: 'bool' = False, quant: 'bool' = True, pcr: 'bool' = False)

Bases: qslib.tcprotocol.ProtoCommand

increment: float | None = None

incrementcycle: int | None = None

incrementstep: int | None = None

pcr: bool = False

quant: bool = True

tiff: bool = False

time: int

to_command(**kwargs) → str

class qslib.tcprotocol.ProtoCommand

Bases: abc.ABC

abstract to_command(**kwargs) → str

class qslib.tcprotocol.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')

Bases: qslib.tcprotocol.XMLable

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_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)

copy() → qslib.tcprotocol.Protocol

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

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

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

name: str

runmode: str = 'standard'

stages: Iterable[Stage]

tcplot(ax: Optional['plt.Axes'] = None) → Tuple['plt.Axes', Tuple[List['plt.Line2D'], List['plt.Line2D']]]

A plot of the temperature and data collection points.

to_command() → str

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

volume: float = 50.0

class qslib.tcprotocol.Ramp(temperature: 'Sequence[float]', increment: 'float | None' = None, incrementcycle: 'int | None' = None, incrementstep: 'int | None' = None, rate: 'float | None' = None, cover: 'float | None' = None)

Bases: qslib.tcprotocol.ProtoCommand

cover: float | None = None

increment: float | None = None

incrementcycle: int | None = None

incrementstep: int | None = None

rate: float | None = None

temperature: Sequence[float]

to_command(**kwargs) → str

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

Bases: qslib.tcprotocol.XMLable

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

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

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

index: int | None = None

info_str(index) → str

label: str | None = None

repeat: int = 1

property steps: list[qslib.tcprotocol.BaseStep]

to_command(stageindex=None, **kwargs)

to_xml() → xml.etree.ElementTree.Element

class qslib.tcprotocol.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')

Bases: qslib.tcprotocol.BaseStep, qslib.tcprotocol.XMLable

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

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

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

property identifier: None

info_str(index, repeats: int = 1) → str

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]

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

total_duration(repeats: int = 1)

class qslib.tcprotocol.XMLable

Bases: abc.ABC

abstract to_xml() → xml.etree.ElementTree.Element

qslib.util module

Module contents