The OPCClient class provides methods for interacting with an OPC UA server.
_instances
,
_client
,
_ip
,
_port
,
_timeout
,
_opc_url
,
_keep_connection
,
_connected
__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
,
_instances
: A dictionary holding singleton instances of OPCClients._client
: An instance of opcua.client.Client._ip
: IP address of the OPC server._port
: Port of the OPC server._timeout
: Timeout in seconds for the OPC server._opc_url
: Connection string (URL) for the OPC server._keep_connection
: Whether to keep the connection open._connected
: Whether the connection is open.__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:
__new__
method has been called.server_name
is already present in the _instances
dictionary.server_name
is not present, it creates a new instance of OPCClient using the super().__new__(cls)
method._instances
dictionary with the server_name
as the key.__init__(server_name: str)
Initializes the OPCClient instance.
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.
value
: The value to be written to the OPC tag.var_type
: The type of the value. It should be one of the available types from ua.VariantType
, such as ua.VariantType.Int32
, ua.VariantType.Float
, etc.tag
: The symbolic name of the tag to write to. It can be either a single string or a list of strings representing the hierarchical structure of the tag. For example, if the node name is "DB1"."Struct1"."Struct2"."Tag1"
, then the tag
should be ["DB1", "Struct1", "Struct2", "Tag1"]
. Alternatively, you can specify the tag as a single string, e.g., "Tag1"
.db_name
: The name of the datablock to write to (if any).namespace
: The namespace index.
# 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.
values
: A list containing the values to be written to the OPC tags.var_types
: A list containing the variant types for each value in values
. It should match the length of values
.tag_list
: A list of tag names representing the OPC tags to write to.namespaces
(optional): A list containing the namespaces for each tag in tag_list
. It should match the length of tag_list
. Defaults to None.
# 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.
_keep_connection
is True
and _connected
is True
.
# Example of connecting to the OPC UA server
client.connect()
disconnect(force: bool = True)
Closes the connection to the OPC UA server.
_connected
is True
and either _keep_connection
is False
or force
is True
.
# Example of disconnecting from the OPC UA server
client.disconnect(force=True)