hvl_ccb.comm package

Submodules

hvl_ccb.comm.base module

Module with base classes for communication protocols.

class hvl_ccb.comm.base.CommunicationProtocol(config)

Bases: hvl_ccb.configuration.ConfigurationMixin, abc.ABC

Communication protocol abstract base class.

Specifies the methods to implement for communication protocol, as well as implements some default settings and checks.

access_lock = None

Access lock to use with context manager when accessing the communication protocol (thread safety)

close()

Close the communication protocol

open()

Open communication protocol

hvl_ccb.comm.labjack_ljm module

Communication protocol for LabJack using the LJM Library. Originally developed and tested for LabJack T7-PRO.

Makes use of the LabJack LJM Library Python wrapper. This wrapper needs an installation of the LJM Library for Windows, Mac OS X or Linux. Go to: https://labjack.com/support/software/installers/ljm and https://labjack.com/support/software/examples/ljm/python

class hvl_ccb.comm.labjack_ljm.LJMCommunication(configuration)

Bases: hvl_ccb.comm.base.CommunicationProtocol

Communication protocol implementing the LabJack LJM Library Python wrapper.

close() → None

Close the communication port.

static config_cls()

Return the default configdataclass class.

Returns:a reference to the default configdataclass class

open() → None

Open the communication port.

read_name(*names) → Union[str, Tuple[str]]

Read one or more inputs by name.

Parameters:names – one or more names to read out from the LabJack Returns:answer of the LabJack, either single answer or multiple answers in a tuple

write_address(address: Union[Sequence[int], int], value: Union[Sequence[object], object]) → None

NOT IMPLEMENTED. Write one or more values to Modbus addresses.

Parameters:

• address – One or more Modbus address on the LabJack.
• value – One or more values to be written to the addresses.

write_name(name: Union[Sequence[str], str], value: Union[Sequence[object], object]) → None

Write one value to a named output.

Parameters:

• name – String or with name of LabJack IO
• value – is the value to write to the named IO port

write_names(names: Sequence[str], values: Sequence[object]) → None

Write more than one value at once to named outputs.

Parameters:

• names – is a sequence of strings with names of LabJack IO
• values – is a sequence of values corresponding to the list of names

class hvl_ccb.comm.labjack_ljm.LJMCommunicationConfig(device_type: (<class 'str'>, <aenum 'DeviceType'>) = 'ANY', connection_type: (<class 'str'>, <aenum 'ConnectionType'>) = 'ANY', identifier: str = 'ANY')

Bases: object

Configuration dataclass for LJMCommunication.

class ConnectionType(*args, **kwds)

Bases: hvl_ccb.utils.enum.AutoNumberNameEnum

LabJack connection type.

ANY = 1

ETHERNET = 4

TCP = 3

USB = 2

WIFI = 5

class DeviceType(*args, **kwds)

Bases: hvl_ccb.utils.enum.AutoNumberNameEnum

LabJack device type.

ANY = 1

DIGIT = 4

T4 = 3

T7 = 2

clean_values() → None

Performs value checks on device_type and connection_type.

connection_type = 'ANY'

Can be either string or of enum ConnectionType.

device_type = 'ANY'

Can bei either string ‘ANY’, ‘T7’, ‘T4’, ‘DIGIT’ or of enum DeviceType.

force_value(fieldname, value)

Forces a value to a dataclass field despite the class being frozen.

Parameters:

• fieldname – name of the field
• value – value to assign

identifier = 'ANY'

The identifier specifies information for the connection to be used. This can be an IP address, serial number, or device name. See the LabJack docs ( https://labjack.com/support/software/api/ljm/function-reference/ljmopens/identifier-parameter) for more information.

is_configdataclass = True

classmethod keys() → Sequence[str]

Returns a list of all configdataclass fields key-names.

Returns:a list of strings containing all keys.

classmethod optional_defaults() → Dict[str, object]

Returns a list of all configdataclass fields, that have a default value assigned and may be optionally specified on instantiation.

Returns:a list of strings containing all optional keys.

classmethod required_keys() → Sequence[str]

Returns a list of all configdataclass fields, that have no default value assigned and need to be specified on instantiation.

Returns:a list of strings containing all required keys.

exception hvl_ccb.comm.labjack_ljm.LJMCommunicationError

Bases: Exception

Errors coming from LJMCommunication.

hvl_ccb.comm.modbus_tcp module

Communication protocol for modbus TCP ports. Makes use of the pymodbus library.

class hvl_ccb.comm.modbus_tcp.ModbusTcpCommunication(configuration)

Bases: hvl_ccb.comm.base.CommunicationProtocol

Implements the Communication Protocol for modbus TCP.

close()

Close the Modbus TCP connection.

static config_cls()

Return the default configdataclass class.

Returns:a reference to the default configdataclass class

open() → None

Open the Modbus TCP connection.

Raises:ModbusTcpConnectionFailedException – if the connection fails.

read_holding_registers(address: int, count: int) → List[int]

Read specified number of register starting with given address and return the values from each register.

Parameters:

• address – address of the first register
• count – count of registers to read

Returns:

list of int values

read_input_registers(address: int, count: int) → List[int]

Read specified number of register starting with given address and return the values from each register in a list.

Parameters:

• address – address of the first register
• count – count of registers to read

Returns:

list of int values

write_registers(address: int, values: Union[List[int], int])

Write values from the specified address forward.

Parameters:

• address – address of the first register
• values – list with all values

class hvl_ccb.comm.modbus_tcp.ModbusTcpCommunicationConfig(host: str, unit: int, port: int = 502)

Bases: object

Configuration dataclass for ModbusTcpCommunication.

clean_values()

force_value(fieldname, value)

Forces a value to a dataclass field despite the class being frozen.

Parameters:

• fieldname – name of the field
• value – value to assign

host = None

Host is the IP address of the connected device.

is_configdataclass = True

classmethod keys() → Sequence[str]

Returns a list of all configdataclass fields key-names.

Returns:a list of strings containing all keys.

classmethod optional_defaults() → Dict[str, object]

Returns a list of all configdataclass fields, that have a default value assigned and may be optionally specified on instantiation.

Returns:a list of strings containing all optional keys.

port = 502

TCP port

classmethod required_keys() → Sequence[str]

Returns a list of all configdataclass fields, that have no default value assigned and need to be specified on instantiation.

Returns:a list of strings containing all required keys.

unit = None

Unit number to be used when connecting with Modbus/TCP. Typically this is used when connecting to a relay having Modbus/RTU-connected devices.

exception hvl_ccb.comm.modbus_tcp.ModbusTcpConnectionFailedException(string='')

Bases: pymodbus.exceptions.ConnectionException

Exception raised when the connection failed.

hvl_ccb.comm.opc module

Communication protocol implementing an OPC UA connection. This protocol is used to interface with the “Supercube” PLC from Siemens.

class hvl_ccb.comm.opc.OpcUaCommunication(config)

Bases: hvl_ccb.comm.base.CommunicationProtocol

Communication protocol implementing an OPC UA connection. Makes use of the package python-opcua.

close() → None

Close the connection to the OPC UA server.

static config_cls()

Return the default configdataclass class.

Returns:a reference to the default configdataclass class

init_monitored_nodes(node_id: Union[str, Iterable[str]], ns_index: int) → None

Initialize monitored nodes.

Parameters:

• node_id – one or more strings of node IDs.
• ns_index – the namespace index the nodes belong to.

open() → None

Open the communication to the OPC UA server.

read(node_id, ns_index)

Read a value from a node with id and namespace index.

Parameters:

• node_id – the ID of the node to read the value from
• ns_index – the namespace index of the node

Returns:

the value of the node object.

write(node_id, ns_index, value) → None

Write a value to a node with name name.

Parameters:

• node_id – the id of the node to write the value to.
• ns_index – the namespace index of the node.
• value – the value to write.

class hvl_ccb.comm.opc.OpcUaCommunicationConfig(host: str, endpoint_name: str, port: int = 4840, sub_handler: hvl_ccb.comm.opc.OpcUaSubHandler = <hvl_ccb.comm.opc.OpcUaSubHandler object>, update_period: int = 500)

Bases: object

Configuration dataclass for OPC UA Communciation.

clean_values()

Cleans and enforces configuration values. Does nothing by default, but may be overridden to add custom configuration value checks.

endpoint_name = None

Endpoint of the OPC server, this is a path like ‘OPCUA/SimulationServer’

force_value(fieldname, value)

Forces a value to a dataclass field despite the class being frozen.

Parameters:

• fieldname – name of the field
• value – value to assign

host = None

Hostname or IP-Address of the OPC UA server.

is_configdataclass = True

classmethod keys() → Sequence[str]

Returns a list of all configdataclass fields key-names.

Returns:a list of strings containing all keys.

classmethod optional_defaults() → Dict[str, object]

Returns a list of all configdataclass fields, that have a default value assigned and may be optionally specified on instantiation.

Returns:a list of strings containing all optional keys.

port = 4840

Port of the OPC UA server to connect to.

classmethod required_keys() → Sequence[str]

Returns a list of all configdataclass fields, that have no default value assigned and need to be specified on instantiation.

Returns:a list of strings containing all required keys.

sub_handler = <hvl_ccb.comm.opc.OpcUaSubHandler object>

object to use for handling subscriptions.

update_period = 500

Update period for generating datachange events in OPC UA [milli seconds]

class hvl_ccb.comm.opc.OpcUaSubHandler

Bases: object

Base class for subscription handling of OPC events and data change events. Override methods from this class to add own handling capabilities.

To receive events from server for a subscription data_change and event methods are called directly from receiving thread. Do not do expensive, slow or network operation there. Create another thread if you need to do such a thing.

datachange_notification(node, val, data)

event_notification(event)

hvl_ccb.comm.serial module

Communication protocol for serial ports. Makes use of the pySerial library.

class hvl_ccb.comm.serial.SerialCommunication(configuration)

Bases: hvl_ccb.comm.base.CommunicationProtocol

Implements the Communication Protocol for serial ports.

ENCODING = 'utf-8'

UNICODE_HANDLING = 'replace'

close()

Close the serial connection.

static config_cls()

Return the default configdataclass class.

Returns:a reference to the default configdataclass class

open()

Open the serial connection.

read_text() → str

Read one line of text from the serial port. The input buffer may hold additional data afterwards, since only one line is read.

Returns:String read from the serial port.

write_text(text: str)

Write text to the serial port. The text is encoded and terminated by the configured terminator.

Parameters:text – Text to send to the port.

class hvl_ccb.comm.serial.SerialCommunicationConfig(port: str, baudrate: int, parity: (<class 'str'>, <aenum 'Parity'>), stopbits: (<class 'int'>, <class 'float'>, <aenum 'Stopbits'>), bytesize: (<class 'int'>, <aenum 'Bytesize'>), terminator: bytes = b'rn', timeout: (<class 'int'>, <class 'float'>) = 2)

Bases: object

Configuration dataclass for SerialCommunication.

class Bytesize(*args, **kwds)

Bases: hvl_ccb.utils.enum.ValueEnum

Serial communication bytesize.

EIGHTBITS = 8

FIVEBITS = 5

SEVENBITS = 7

SIXBITS = 6

class Parity(*args, **kwds)

Bases: hvl_ccb.utils.enum.ValueEnum

Serial communication parity.

EVEN = 'E'

MARK = 'M'

NAMES = {'E': 'Even', 'M': 'Mark', 'N': 'None', 'O': 'Odd', 'S': 'Space'}

NONE = 'N'

ODD = 'O'

SPACE = 'S'

class Stopbits(*args, **kwds)

Bases: hvl_ccb.utils.enum.ValueEnum

Serial communication stopbits.

ONE = 1

ONE_POINT_FIVE = 1.5

TWO = 2

baudrate = None

Baudrate of the serial port

bytesize = None

Size of a byte, 5 to 8

clean_values()

force_value(fieldname, value)

Forces a value to a dataclass field despite the class being frozen.

Parameters:

• fieldname – name of the field
• value – value to assign

is_configdataclass = True

classmethod keys() → Sequence[str]

Returns a list of all configdataclass fields key-names.

Returns:a list of strings containing all keys.

classmethod optional_defaults() → Dict[str, object]

Returns a list of all configdataclass fields, that have a default value assigned and may be optionally specified on instantiation.

Returns:a list of strings containing all optional keys.

parity = None

Parity to be used for the connection.

port = None

Port is a string referring to a COM-port (e.g. 'COM3') or a URL. The full list of capabilities is found on the pyserial documentation.

classmethod required_keys() → Sequence[str]

Returns a list of all configdataclass fields, that have no default value assigned and need to be specified on instantiation.

Returns:a list of strings containing all required keys.

stopbits = None

Stopbits setting, can be 1, 1.5 or 2.

terminator = b'\r\n'

The terminator character. Typically this is b'\r\n' or b'\n', but can also be b'\r' or other combinations.

timeout = 2

Timeout in seconds for the serial port

hvl_ccb.comm.visa module

Communication protocol for VISA. Makes use of the pyvisa library. The backend can be NI-Visa or pyvisa-py.

Information on how to install a VISA backend can be found here: https://pyvisa.readthedocs.io/en/master/getting_nivisa.html

So far only TCPIP SOCKET and TCPIP INSTR interfaces are supported.

class hvl_ccb.comm.visa.VisaCommunication(configuration)

Bases: hvl_ccb.comm.base.CommunicationProtocol

Implements the Communication Protocol for VISA / SCPI.

MULTI_COMMANDS_MAX = 5

The maximum of commands that can be sent in one round is 5 according to the VISA standard.

MULTI_COMMANDS_SEPARATOR = ';'

The character to separate two commands is ; according to the VISA standard.

WAIT_AFTER_WRITE = 0.08

Small pause in seconds to wait after write operations, allowing devices to really do what we tell them before continuing with further tasks.

close() → None

Close the VISA connection and invalidates the handle.

static config_cls()

Return the default configdataclass class.

Returns:a reference to the default configdataclass class

open() → None

Open the VISA connection and create the resource.

query(*commands) → Union[str, Tuple[str, ...]]

A combination of write(message) and read.

Parameters:commands – list of commands Returns:list of values

spoll() → int

Execute serial poll on the device. Reads the status byte register STB. This is a fast function that can be executed periodically in a polling fashion.

Returns:integer representation of the status byte

write(*commands) → None

Write commands. No answer is read or expected.

Parameters:commands – one or more commands to send

class hvl_ccb.comm.visa.VisaCommunicationConfig(host: str, interface_type: (<class 'str'>, <aenum 'InterfaceType'>), board: int = 0, port: int = 5025, timeout: int = 5000, chunk_size: int = 204800, open_timeout: int = 1000, write_termination: str = 'n', read_termination: str = 'n', visa_backend: str = '')

Bases: object

VisaCommunication configuration dataclass.

class InterfaceType(*args, **kwds)

Bases: hvl_ccb.utils.enum.AutoNumberNameEnum

Supported VISA Interface types.

TCPIP_INSTR = 2

VXI-11 protocol

TCPIP_SOCKET = 1

VISA-RAW protocol

board = 0

Board number is typically 0 and comes from old bus systems.

chunk_size = 204800

Chunk size is the allocated memory for read operations. The standard is 20kB, and is increased per default here to 200kB. It is specified in bytes.

clean_values()

force_value(fieldname, value)

Forces a value to a dataclass field despite the class being frozen.

Parameters:

• fieldname – name of the field
• value – value to assign

host = None

IP address of the VISA device. DNS names are currently unsupported.

interface_type = None

Interface type of the VISA connection, being one of InterfaceType.

is_configdataclass = True

classmethod keys() → Sequence[str]

Returns a list of all configdataclass fields key-names.

Returns:a list of strings containing all keys.

open_timeout = 1000

Timeout for opening the connection, in milli seconds.

classmethod optional_defaults() → Dict[str, object]

Returns a list of all configdataclass fields, that have a default value assigned and may be optionally specified on instantiation.

Returns:a list of strings containing all optional keys.

port = 5025

TCP port, standard is 5025.

read_termination = '\n'

Read termination character.

classmethod required_keys() → Sequence[str]

Returns a list of all configdataclass fields, that have no default value assigned and need to be specified on instantiation.

Returns:a list of strings containing all required keys.

timeout = 5000

Timeout for commands in milli seconds.

visa_backend = ''

Specifies the path to the library to be used with PyVISA as a backend. Defaults to None, which is NI-VISA (if installed), or pyvisa-py (if NI-VISA is not found). To force the use of pyvisa-py, specify ‘@py’ here.

write_termination = '\n'

Write termination character.

exception hvl_ccb.comm.visa.VisaCommunicationError

Bases: Exception

Base class for VisaCommunication errors.

Module contents

Communication protocols subpackage.