quarchpy.connection_specific package

Subpackages

Submodules

quarchpy.connection_specific.connection_QIS module

class quarchpy.connection_specific.connection_QIS.QisInterface(host='127.0.0.1', port=9722, connectionMessage=True)

Bases: object

GetQisModuleSelection(favouriteOnly=True, additionalOptions=['rescan', 'all con types', 'ip scan'], scan=True)

deprecated:: 2.2.13 Use get_qis_module_selection instead.

closeConnection(sock=None, conString: str = None) → str

deprecated:: 2.2.13 Use close_connection instead.

close_connection(sock=None, con_string: str = None) → str

Orders QIS to release a given device (or all devices if no connection string is specified) This is more important for TCP connected devices as the socket is held open until specifically released.

Parameters:

• sock – The socket object to close the connection to. Defaults to the existing socket.

• con_string – Specify the device ID to close, otherwise all devices will be closed

Raises:

ConnectionResetError – Raised if the socket connection has already been reset.

Returns:

‘OK’

Return type:

The response received after sending the close command. On success, this will be

connect(connection_message: bool = True) → str

Opens the connection to the QIS backend

Parameters:

connection_message – Defaults to True. If set to False, suppresses the warning message about an instance already running on the specified port. This can be useful when using isQisRunning() from qisFuncs.

Raises: Exception:

If the connection fails or the welcome string is not received an exception is raised

Returns:

The welcome string received from the backend server upon a successful connection. This will confirm the QIS version but is generally not used other than for debugging

convertStreamAverage(streamAveraging)

deprecated:: 2.2.13 No indication this is used anywhere

device_control_index(device) → int

Returns the index of the device in the control lists. If the device is not registered, then it is added first. This is a key part of allowing us to track the status of multiple streaming devices and manage them from outside their streaming thread

Parameters:

device – Device ID string

Returns:

Index of the device in the various control lists

device_dict_setup(module) → None

Adds a dictionary entry for a new module we are connecting to (including the base QIS connection) This is used for tracking the status of modules throughout the streaming process

Parameters:

module –

Returns:

None

disconnect() → str

Disconnects the current connection to the QIS backend.

This method attempts to gracefully disconnect from the backend server and updates the connection state in the device dictionary. If an error occurs during the disconnection process, the state is updated to indicate the failure, and the exception is re-raised

Returns:

A message indicating that the disconnection process has started.

Return type:

str

Raises:

Exception – Propagates any exception that occurs during the disconnection process.

getDeviceList(sock=None)

deprecated:: 2.2.13 Use start_stream_thread_qps instead.

get_device_list(sock=None) → filter

returns a list of device IDs that are available for connection

Parameters:

sock – Optional connection socket

Returns:

A filtered iterable list of devices

get_list_details(sock=None) → list

Extended version of get_device_list which also returns the additional details fields for each module

Parameters:

sock – Optional connection socket

Returns:

Iterable list of strings containing the details of each device available for connection.

get_qis_module_selection(preferred_connection_only=True, additional_options=['rescan', 'all con types', 'ip scan'], scan=True) → str

Scans for available modules and allows the user to select one through an interactive selection process. Can also handle additional custom options and some built-in ones such as rescanning

Parameters:

• preferred_connection_only – bool by default (True), returns only one preferred connection eg: USB for simplicity

• additional_options – list Additional operational options provided during module selection, such as rescan, all connection types, and IP scan. Defaults to [‘rescan’, ‘all con types’, ‘ip scan’]. These allow the additional options to be given to the user and handled in the top level script

• scan – bool Indicates whether to initiate a rescanning process for devices before listing. Defaults to True and will take longer to return

Returns:

The identifier of the selected module, or the action selected from the additional options.

Return type:

str

Raises:

• KeyError – Raised when unexpected keys are found in the scanned device data.

• ValueError – Raised if no valid selection is made or the provided IP address is invalid.

get_stream_xml_header(device, sock=None) → Element

Gets the XML format header from an attached device (which must have run or be running a stream) Parses the string into XML and returns the root element

Parameters:

• device – Device ID to return from

• sock – Optional QIS socket to use for communication.

Returns:

interruptList()

deprecated:: 2.2.13 No indication this is used anywhere

is_xml_header(header_text) → bool

Checks if the given header string is in XML format (as apposed to legacy text format) or an invalid string

Parameters:

header_text – The header string to evaluate

Returns:

True if the header is in XML form

qis_scan_devices(scan=True, preferred_connection_only=True, ip_address=None) → list

Begins a scan for devices and returns a simple list of devices

Parameters:

• scan – Should a scan be initiated? If False, the function will return immediately with the list

• preferred_connection_only – The default (True), returns only one preferred connection eg: USB for simplicity

• ip_address – IP address of the module you are looking for eg ‘192.168.123.123’

Returns:

List of module strings found during scan

Return type:

list

receive_text(sock) → str

Received bytes from the socket and converts to a test string :param sock: Socket to communicate over

Returns:

rx_bytes(sock) → bytes

Reads an array of bytes from the socket as part of handling a command response

Parameters:

sock – Socket to communicate over

Returns:

Bytes read

scanIP(ipAddress)

deprecated:: 2.2.13 Use scan_ip instead.

scan_ip(ip_address) → str

Triggers QIS to look at a specific IP address for a module Arguments

ipAddressstr

The IP address of the module you are looking for eg ‘192.168.123.123’

sendAndReceiveCmd(sock=None, cmd='$help', device='', readUntilCursor=True, betweenCommandDelay=0.0) → str

deprecated:: 2.2.13 Use send_command instead.

sendAndReceiveText(sock, sendText='$help', device='', readUntilCursor=True) → str

deprecated:: 2.2.13 Use send_and_receive_text instead.

sendCmd(device='', cmd='$help', sock=None, readUntilCursor=True, betweenCommandDelay=0.0, expectedResponse=True) → str

deprecated:: 2.2.13 Use send_command instead.

sendCommand(cmd, device='', timeout=20, sock=None, readUntilCursor=True, betweenCommandDelay=0.0, expectedResponse=True) → str

deprecated:: 2.2.13 Use send_command instead.

send_and_receive_text(sock, send_text='$help', device='', read_until_cursor=True) → str

Internal function for command handling. This handles complex cases such as timeouts and XML response formatting, which conflicts with the default cursor

Parameters:

• sock – The socket to communicate over

• send_text – The command text to send

• device – Optional device ID to send the command to

• read_until_cursor – Flag to indicate if we should read until the cursor is returned

Returns:

Response string from the module

send_command(command: str, device: str = '', qis_socket: socket = None, cursor_expected: bool = True, response_expected: bool = True, command_delay: float = 0.0) → str

Sends a command and returns the response as a string. Multiple lines are escaped with CRLF. The command is sent to the QIS socket, and depending on the command will be replied by either QIS or the hardware module.

Parameters:

• command – Command string

• device – Optional Device ID string to send the command to. Use default/blank for QIS direct commands

• qis_socket – Optional Socket to use for the command, if the default is not wanted

• cursor_expected – Optional Flag true if the command returns a cursor, so we should wait for it

• response_expected – Optional Flag true if the command returns a response, so we should wait for it.

• command_delay – Optional delay to prevent commands running in close succession. Timed in seconds.

Returns:

Command response string or None if no response expected

send_text(sock, message='$help', device='') → bool

Parameters:

• sock – Socket to communicate over

• message – text command to send

• device – Optional device ID to target with the command

Returns:

sort_favourite(found_devices) → list

Reduces the list of located devices by referencing to the preferred type of connection. Only one connection type will be returned for each module for easier user selection. ie: A module connected on both USB and TCP will now only return with USB

Parameters:

found_devices – List of located devices from a scan operation

Returns:

List of devices filtered/sorted devices

startStream(module: str, fileName: str, fileMaxMB: int, releaseOnData: bool, separator: str, streamDuration: int = None, inMemoryData=None, outputFileHandle=None, useGzip: bool = None, gzipCompressLevel: int = 9)

deprecated:: 2.2.13 Use start_stream instead.

start_stream(module: str, file_name: str, max_file_size: int, release_on_data: bool, separator: str, stream_duration: float = None, in_memory_data: StringIO = None, output_file_handle=None, use_gzip: bool = None, gzip_compress_level: int = 9)

Begins a stream process which will record data from the module to a CSV file or in memory CSV equivalent

Parameters:

• module – The ID of the module for which the stream is being initiated.

• file_name – The target file path+name for storing the streamed data in CSV form.

• max_file_size – The maximum size in megabytes allowed for the output file.

• release_on_data – If set, blocks further streams until this one has started

• separator – The value separator used to format the streamed CSV data.

• stream_duration – The duration (in seconds) for which the streaming process should run. Unlimited if None.

• in_memory_data – An in memory CSV StringIO as an alternate to file output

• output_file_handle – A file handle to an output file where the stream data is written as an alternate to a file name

• use_gzip – A flag indicating whether the output file should be compressed using gzip to reduce disk use

Returns:

None

start_stream_thread(module: str, file_name: str, max_file_size: float, release_on_data: bool, separator: str, stream_duration: int = None, in_memory_data=None, output_file_handle=None, use_gzip: bool = False, gzip_compress_level: int = 9)

Parameters:

• module – The name of the module from which data is to be streamed.

• file_name – The path to the file where streamed data will be written. Mandatory if neither an in-memory buffer (in_memory_data) nor an external file handle (output_file_handle) is provided.

• max_file_size – The maximum permissible file size in MB. After reaching this limit, streaming to the current file will stop

• release_on_data – True to prevent the stream lock from releasing until data has been received

• separator – Custom separator used to CSV data

• stream_duration – Duration of streaming in seconds, relative to the sampling period. Defaults to streaming indefinitely.

• in_memory_data – An in-memory buffer of type StringIO to hold streamed data. If set, data is written here instead of a file.

• output_file_handle – A pre-opened file handle where data will be written. If set, file_name is ignored.

• use_gzip – If True, writes streamed data to a gzip-compressed file.

• gzip_compress_level – (Default: 9) The compression level (0-9) to use for gzip. 1 is fastest with low compression. 9 is slowest with high compression.

Raises: TypeError

If in_memory_data is passed but is not of type StringIO.

ValueError

If file_name is not provided and neither in_memory_data nor output_file_handle is given. Also raised for invalid or undecodable sampling periods.

Returns:

None

stopStream(module, blocking=True)

deprecated:: 2.2.13 Use stop_stream instead.

stop_stream(module, blocking: bool = True)

Parameters:

• module – The quarchPPM module instance for which the streaming process is to be stopped.

• blocking – If set to True, the function will block and wait until the module has completely stopped streaming. Defaults to True.

Returns:

None

streamBufferStatus(device: str) → str

deprecated:: 2.2.13 Use stream_buffer_status instead.

streamHeaderFormat(device, sock=None) → str

deprecated:: 2.2.13 Use stream_header_format instead.

streamInterrupt() → bool

deprecated:: 2.2.13 No indication this is used anywhere

streamRunningStatus(device: str) → str

deprecated:: 2.2.13 Use stream_running_status instead.

stream_buffer_status(device, sock=None) → str

returns the info on the stripes buffered during the stream

Arguments

devicestr

The device ID to target

sock:

The socket to communicate over, or None to use the default.

Returns:

String with the numbers of stripes buffered

Return type:

str

stream_get_stripes_text(sock, device: str) → Tuple[str, str]

Retrieve and process text data from a QIS stream. We try to ready a block of data and also check for end of data and error cases

Parameters:

• sock – The socket instance used for communication with the device.

• device – The device ID string

Returns:

• The status of the data stream as a comma seperated list of status items

• The retrieved text data from the stream.

Return type:

A tuple containing

stream_header_average(device, sock=None) → str

Gets the averaging used on the current stream, required for processing the stripe data returned from QIS

Arguments

devicestr

The device ID to target

sock:

The socket to communicate over, or None to use the default.

Returns:

String with the rate and unit

Return type:

str

stream_header_format(device, sock=None) → str

Formats the stream header for use at the top of a CSV file. This adds the appropriate time column and each of the channel data columns

Arguments

device:

The device ID to target

sock:

The socket to communicate over, or None to use the default.

Returns:

Get the CSV formatted header string for the current stream

Return type:

str

stream_running_status(device, sock=None) → str

returns a single word status string for a given device. Generally this will be running, overrun, or stopped

Arguments

devicestr

The device ID to target

sock:

The socket to communicate over, or None to use the default.

Returns:

Single word status string to show the operation of streaming

Return type:

str

waitStop()

deprecated:: 2.2.13 No indication this is used anywhere

quarchpy.connection_specific.connection_QPS module

class quarchpy.connection_specific.connection_QPS.QpsInterface(host='127.0.0.1', port=9822)

Bases: object

closeConnection(conString=None)

connect(targetDevice)

disconnect(targetDevice)

getDeviceList(scan=True, ipAddress=None)

get_list_details(sock=None)

get_qps_module_selection(preferred_connection_only: bool = True, additional_options: List[str] | None = None, scan: bool = True) → Any | None

Scans for QPS devices and prompts the user to select one.

open_recording(file_path, cmdTimeout=5, pollInterval=3, startOpenTimout=5)

recv()

scanIP(ipAddress, sleep=10)

Triggers QPS to look at a specific IP address for a quarch module

Parameters:

• QpsConnection (QpsInterface) – The interface to the instance of QPS you would like to use for the scan.

• ipAddress (str) – The IP address of the module you are looking for eg ‘192.168.123.123’

• sleep (int, optional) – This optional variable sleeps to allow the network to scan for the module before allowing new commands to be sent to QPS.

send(data)

sendCmdVerbose(cmd, timeout=20)

sendCommand(cmd, timeout=20, expectedResponse=True)

quarchpy.connection_specific.connection_ReST module

class quarchpy.connection_specific.connection_ReST.ReSTConn(ConnTarget)

Bases: object

close()

sendCommand(Command, expectedResponse=True, max_retries=2)

quarchpy.connection_specific.connection_Serial module

class quarchpy.connection_specific.connection_Serial.SerialConn(ConnTarget)

Bases: object

close()

sendCommand(Command, expectedResponse=True)

quarchpy.connection_specific.connection_Serial.serial_read_until(Port, Char, Timeout)

quarchpy.connection_specific.connection_TCP module

class quarchpy.connection_specific.connection_TCP.TCPConn(ConnTarget)

Bases: object

close()

sendCommand(Command, readUntilCursor=True, expectedResponse=True)

quarchpy.connection_specific.connection_Telnet module

class quarchpy.connection_specific.connection_Telnet.TelnetConn(ConnTarget)

Bases: object

close()

sendCommand(Command, expectedResponse=True)

quarchpy.connection_specific.connection_USB module

class quarchpy.connection_specific.connection_USB.TQuarchUSB_IF(context)

Bases: object

BulkRead()

BulkReadEP(ep, n)

BulkReadEPTout(ep, n, timeOut)

BulkReadN(n)

CheckComms()

ClosePort()

DebugDump()

FetchCmdReply()

FetchCmdReplyTOut(timeout)

GetExtendedInfo()

GetIdn()

GetLastError()

GetSerialNumber()

IsPortOpen()

OpenPort()

RunCommand(command, expectedResponse=True)

SendCommand(command)

SetTimeout(timeout)

VerboseSendCmd(cmd)

WriteZeroPacketCmd()

clean_and_flush_stuck_usb_comms()

USB comms can become stuck if a command is sent out too quickly after the end of the previous one If this is the case, the module response buffer contains data which needs flushed. This command triggers a buffer return and performs a full read of all data. If possible, it returns the response of the previous command which caused the stuck buffer.

lockUSBStr = '\x02\x00\x00\x01\x04\x01'

unlockUSBStr = '\x02\x00\x00\x01\x03\x00'

quarchpy.connection_specific.connection_USB.USB(SerID='')

class quarchpy.connection_specific.connection_USB.USBConn(ConnTarget)

Bases: object

close()

sendCommand(Command, expectedResponse=True)

quarchpy.connection_specific.connection_USB.getUSBDeviceSerialNo(context, device)

quarchpy.connection_specific.connection_USB.importUSB(version='None')

Module contents