OPCClient Class

The OPCClient class provides methods for interacting with an OPC UA server.

Attributes

_instances, _client, _ip, _port, _timeout, _opc_url, _keep_connection, _connected

Methods

__new__ __init__, name, server_name, set_config, set_connection, keep_connected, set_timeout, init_client, construct_full_tag_name, read, read_multiple, write, write_multiple, connect, disconnect,

Class Attributes

Class construction

__new__(cls, server_name: str, *args, **kwargs) -> 'OPCClient':

The __new__ method in the OPCClient class is a special method in Python that is used for object creation. It is responsible for creating and returning a new instance of the class when the class is called as a function. In the provided context, the __new__ method in the OPCClient class is used to ensure that only one instance of the OPCClient is created for each OPC Server. This is achieved through the use of a dictionary _instances which keeps track of the existing instances of OPCClient. Here's a breakdown of how the __new__ method works in this code:

__init__(server_name: str)

Initializes the OPCClient instance.

Methods

name()

Returns the OPC server name for this instance.

server_name = client.name()

set_config()

Configures the OPC Client with actual parameters from the configuration file.

client.set_config()

set_connection(*args)

The set_connection(*args) method is responsible for setting the connection parameters of the OPC Client. It accepts a variable number of arguments, typically an IP address and a port number. The method first checks if the provided arguments are valid. If the arguments are valid, it sets the connection parameters of the OPC Client to the provided values. After setting the connection parameters, it logs a success message indicating that the connection has been successfully configured. If the provided arguments are invalid, it logs an error message indicating the failure to configure the connection.

client.set_connection("192.168.0.1", 4840)

keep_connected(keep: bool = True)

Sets if the client should keep the connection alive or reconnect every time.

client.keep_connected(True)

set_timeout(timeout: int)

Sets the timeout for the OPC UA client in seconds.

client.set_timeout(10)

init_client()

Initializes the OPC UA client.

client.init_client()

construct_full_tag_name(namespace, tags, db_name=None)

Constructs a full node name from namespace, tag components, and optional datablock name.

tag_name = client.construct_full_tag_name(3, ["Sensor", "Temperature"], "DB1")

read(tag: List[str] or str, db_name: str = '', namespace: str = '3')

This method reads data from an OPC server. It takes a tag parameter, which can be a string or a list of strings representing the tag name. The db_name parameter is the symbolic name of the datablock to read from, and namespace is the namespace index. The method connects to the OPC server, constructs the full node name using the construct_full_tag_name method, reads the value from the node, logs the information, handles any exceptions, disconnects from the OPC server, and returns the read value.


    # Example of reading data from the OPC server
    value = client.read(["DB1", "Sensor", "Temperature"])

    # Alternatively, you can use:
    value = client.read(["Sensor", "Temperature"], "DB1")
    

read_multiple(tag_list: List[str], default=None, namespace: str = '3')

This method reads multiple tags from an OPC server. It takes a tag_list parameter, which is a list of strings representing the tags to read from. The default parameter specifies the value to return if the read fails, and namespace is the namespace index. The method connects to the OPC server, iterates over each tag in the list, constructs the full node name using the construct_full_tag_name method, reads the value from the node, logs the information, handles any exceptions, disconnects from the OPC server, and returns a list of the read values.


    # Example of reading multiple tags from the OPC server
    values = client.read_multiple(["Sensor1", "Sensor2", "Sensor3"])

    # Alternatively, you can specify a default value:
    values = client.read_multiple(["Sensor1", "Sensor2", "Sensor3"], default=numpy.NaN)
    

Note: The default value is used if the read fails, in example it could be e.g., [1.5, nan, 3.5] if the Sensor2 tag is not available.

write(value, var_type: ua.VariantType, tag: List[str] or str, db_name: str = '', namespace: str = '3')

Writes a value to an OPC tag.


    # Example of writing a value to an OPC tag
    success = client.write(100, ua.VariantType.Float, ["DB1", "Sensor", "Temperature"])
    if success:
        print("Write operation successful")
    else:
        print("Write operation failed")
    

write_multiple(values: List[any], var_types: List[ua.VariantType], tag_list: List[str], namespaces: List[str] = None) -> List[bool]

Writes multiple values into OPC tags.


    # Example of writing multiple values into OPC tags
    success_list = client.write_multiple([100, 20.5, True], [ua.VariantType.Int32, ua.VariantType.Float, ua.VariantType.Boolean], ["Tag1", "Tag2", "Tag3"])
    for success in success_list:
        if success:
            print("Write operation successful")
        else:
            print("Write operation failed")

    # Alternatively, you can specify namespaces:
    success_list = client.write_multiple([100, 20.5, True], [ua.VariantType.Int32, ua.VariantType.Float, ua.VariantType.Boolean], ["Tag1", "Tag2", "Tag3"])
    

Note: The length of values and var_types must be the same as the length of tag_list.

Note: Also default values could be used.

connect()

Establishes a connection to the OPC UA server if it is not already connected.


    # Example of connecting to the OPC UA server
    client.connect()
    

disconnect(force: bool = True)

Closes the connection to the OPC UA server.


    # Example of disconnecting from the OPC UA server
    client.disconnect(force=True)