Package pys2opc

The pys2opc package is a Python3 wrapper of the library s2opc_clientserver, based on S2OPC. It uses Cython to bind Python with C. The wrapper provides Python classes for OPC UA concepts.

The PyS2OPC represents the SOPC_CommonHelper_* API and gathers its top level functionalities, which are split in PyS2OPC_Client represents the SOPC_Client*Helper*, and PyS2OPC_Server represents the SOPC_Server*Helper*.

The module can be used simultaneously as a server and as a client, both with several connection configurations.

Client use

First initialize the client with PyS2OPC_Client.initialize(). Then, configure it by passing an XML configuration file to PyS2OPC_Client.load_client_configuration_from_file().
For more details on XML configuration file, see s2opc_clientserver_config.xsd Finally, the client can establish a connection to a server using the (PyS2OPC_Client.connect()) function by selecting the user to be used in the previous XML file.

With connection, you can BaseClientConnectionHandler.read_nodes(), BaseClientConnectionHandler.write_nodes() and BaseClientConnectionHandler.browse_nodes(). You can also BaseClientConnectionHandler.add_nodes_to_subscription(), and receive notifications through BaseClientConnectionHandler.on_datachanged().

>>> from pys2opc import PyS2OPC_Client
>>> PyS2OPC_Client.get_version()
>>> with PyS2OPC_Client.initialize():
>>>     configs = PyS2OPC_Client.load_client_configuration_from_file('client_config.xml')
>>>     with PyS2OPC_Client.connect(configs["read"]) as connection:
>>>         ReadReponse = connection.read_nodes(nodeIds=NODES_TO_READ)

client_config.xml extract: connection example with id = "read"

<Connection serverURL="opc.tcp://localhost:4841" id="read">
  <ServerCertificate path="server_public/server_4k_cert.der"/>
  <SecurityPolicy uri="http://opcfoundation.org/UA/SecurityPolicy#Basic256Sha256"/>
  <SecurityMode mode="Sign"/>
</Connection>

Server use

Server is mainly configured by XML files: the first for the server endpoints configuration, which specifies security parameter, session and users parameters, the next file is the NodeSet describing the structure and content of the address space, the last file to define allowed users and their authorization (read, write, ..).
The configuration function migh also register callback to be notified on server (see BaseAddressSpaceHandler).

For more details on XML server endpoint configuration file, see s2opc_clientserver_config.xsd. For more details on XML address space configuration file, see UANodeSet.xsd. For more details on XML user authorization file, see s2opc_clientserver_users_config.xsd.

There are two main ways to start the server once configured. An "all-in-one" PyS2OPC_Server.serve_forever(), and a more permissive one using a with statement PyS2OPC_Server.serve(). In the with statement, the application code can be started alongside the S2OPC server.

>>> from pys2opc import PyS2OPC_Server
>>> PyS2OPC_Server.get_version()
>>> with PyS2OPC_Server.initialize():
>>>     PyS2OPC.load_server_configuration_from_files('server_config.xml', 'address_space.xml', 'user_config.xml')
>>>     with PyS2OPC_Server.serve():
>>>         # The main loop of the application

server_config.xml extract: endpoint configuration example

  <Endpoint url="opc.tcp://localhost:4841">
    <SecurityPolicies>
      <SecurityPolicy uri="http://opcfoundation.org/UA/SecurityPolicy#Basic256Sha256">
        <SecurityModes>
          <SecurityMode mode="Sign"/>
        </SecurityModes>
        <UserPolicies>
          <UserPolicy policyId="username_Basic256Sha256" tokenType="username" securityUri="http://opcfoundation.org/UA/SecurityPolicy#Basic256Sha256"/>
        </UserPolicies>
      </SecurityPolicy>
  </Endpoint>

address_space.xml extract: variable definition example

  <UAVariable AccessLevel="99" BrowseName="1:LK" DataType="Boolean" NodeId="ns=1;s=BALA_RDLS_W1.RM.LK">
    <DisplayName>LK</DisplayName>
    <Description>Switch Locally Locked</Description>
    <References>
      <Reference IsForward="false" ReferenceType="HasComponent">ns=1;s=BALA_RDLS_W1.RM</Reference>
      <Reference ReferenceType="HasTypeDefinition">i=63</Reference>
    </References>
    <Value>
      <uax:Boolean>false</uax:Boolean>
    </Value>
  </UAVariable>

user_config.xml extract: user authorizatoin configuration example

  <UserPasswordConfiguration hash_iteration_count="10000" hash_length="32" salt_length="16">
    <!-- "me" pwd=1234 has all right accesses. -->
    <UserPassword user="me" hash="847d892ffaccb9822d417866f9d491389b29134b3c73c3a429ac95c627f9d40a" salt="17faf802f81c2503d3043042e79004b4">
      <UserAuthorization write="true" read="true" execute="true" addnode="true"/>
    </UserPassword>
  </UserPasswordConfiguration>

NodeId concept

Throughout the module (e.g. BaseClientConnectionHandler.read_nodes()), when the interface requires a NodeId, the following syntax is used:

>>> node_id = 'ns=1;x=qualifier'

The namespace part is optional (ns=1;), and its value is 0 by default. The namespace is a 16-bits unsigned integer. The NodeId type x= is either:

  • i= followed by a 32-bits unsigned integer,
  • s= followed by a zero-terminated string,
  • g= followed by an OPC UA GUID (16 bytes hex string, e.g. "72962B91-FA75-4ae6-8D28-B404DC7DAF63"),
  • b= followed by a bytestring.

Types in OPC UA

All values in OPC UA are typed. You cannot write a value of a variable if you don't send the correctly typed value to write. Mostly used types are signed or unsigned integers of various sizes (8, 16, 32, 64 bits), single or double precision floating point values, and string or bytes (bytes may contain \0 values).

Also, arrays are handled at the value level, and values can be array of integers (or other types).

However, values in Python are more flexible. Integers are unbound, floating point are always double, arrays may contain different types of values. The class Variant represent objects that have the properties of python values (e.g. addition/difference/multiplication/division of numbers), but the types of OPC UA values (see pys2opc.SOPC_BuiltinId).

When creating Variant, there is no need to specify its type. It can be changed later, and will be checked only when encoding the value.

DataValue is another OPC UA concept that encapsulates Variant to add timestamps and a quality (see DataValue). It is DataValues that are used to store values (when reading or writing values). There are helpers to convert Python values to OPC UA values (DataValue.from_python()). The other way around is simpler (just use the DataValue variant field as if it was a Python value).

Sub-modules

pys2opc.pys2opc

Classes

class AsyncResponse (asyncReqHdlr: _AsyncRequestHandler)

A special Response subtype to provide a handler for asynchronous OPC UA service call and response access.

Ancestors

Methods

def get_response(self) ‑> Response | None

Returns the response to the request if there is an available response (received). Otherwise returns None. The response type depends on the service used: ReadResponse, WriteResponse, BrowseResponse. Note: if a service failure occurs, raise an Exception sub class : ServiceFailure

class AttributeId

This class or its subclass is capable of returning the name of a member of this class from an id. NamedMembers.get_name_from_id() is particularly useful to translate OPC UA constants to readable strings. The function NamedMembers.get_both_from_id() returns a string which contains both the formatted code and its string decode.

Ancestors

Class variables

var AccessLevel
var AccessLevelEx
var AccessRestrictions
var ArrayDimensions
var BrowseName
var ContainsNoLoops
var DataType
var DataTypeDefinition
var Description
var DisplayName
var EventNotifier
var Executable
var Historizing
var Invalid
var InverseName
var IsAbstract
var MinimumSamplingInterval
var NodeClass
var NodeId
var RolePermissions
var Symmetric
var UserAccessLevel
var UserExecutable
var UserRolePermissions
var UserWriteMask
var Value
var ValueRank
var WriteMask

Inherited members

class BaseAddressSpaceHandler (...)

Base class for the Address Space notification callback. You should derive this class and re-implement its BaseAddressSpaceHandler.on_datachanged() method.

Methods

def on_datachanged(self,
nodeId: str,
attrId: AttributeId,
dataValue: DataValue,
indexRange: str,
status: SOPC_StatusCode)

This is the main callback for the address space notifications (write events). The notification is called each time a write attribute operation succeeded on the server.

There are no notifications for local writes using PyS2OPC_Server.write_nodes().

This callback shall be re-implemented.

Args

nodeId
The written NodeId as a string (see "NodeId concept").
attrId
The written AttributeId
dataValue
The new DataValue
indexRange
An index range (string) selection on the value
status
The StatusCode of this operation that will be returned to the client. This differs from the status code of the value, which is contained in the DataValue.
class BaseClientConnectionHandler (c_cliConHandler: _C_BaseClientConnectionHandler)

Base class giving the prototypes of the callbacks, implements Read/Write/Browse client methods, and implements the subscription-library connection wrappers.

The class supports Python's "with" statements. In this case, the connection is automatically closed upon exit of the context.

An exception ClientConnectionFailure might be raised when calling OPC UA services if the connection with the server has been lost which might be verified using BaseClientConnectionHandler.connected.

Instance variables

prop connected

Returns whether this connection is still active and usable.

Methods

def add_nodes_to_subscription(self, nodeIds: list[str]) ‑> list[bool]

Add a list of NodeIds in the OPC UA format to the subscription (see "NodeId concept"). If there is no subscription created, this function creates one beforehand using default parameters (see BaseClientConnectionHandler.create_subscription()). Stores context for each node, which is reused by the callback function and to close the subscription correctly.

The callback BaseClientConnectionHandler.on_datachanged() will be called once for each new value of a subscribed node. In particular, the callback is at least called once for the initial value.

The returned boolean list indicates if creation of the monitored item for the nodeId at same index succeeded

def browse_nodes(self,
nodeIds: list[str],
maxReferencesPerNode: int = 1000,
bWaitResponse: bool = True) ‑> Response

Requests to execute the OPC UA Browse service on server and returns the Browse response content. See BrowseResponse for result content.

Args

nodeIds
The list of nodeIds to browse (see "NodeId concept").
maxReferencesPerNode
Indicates the maximum number of references to return for each starting node specified in the request (0 means no limitation).
bWaitResponse
True, wait the response and returns a BrowseResponse. Otherwise return an AsyncResponse to retrieve it later (see AsyncResponse.get_response()).
def call_method(self,
objectNodeId: str,
methodNodeId: str,
inputArgList: list[Variant] = [],
bWaitResponse: bool = True) ‑> Response

Requests to execute the OPC UA CallMethod service on server and returns the Call response content. See CallResponse for result content.

Args

objectNodeId
The nodeId of the object node as string on which method call will be executed. Shall not be NULL
methodNodeId
The nodeId of the method node to execute as string (it might be object's method node instance or method node of its ObjectType). Shall not be NULL
inputArgList
The Variant list containing the input arguments for the method call.
bWaitResponse
True, wait the response and returns a BrowseResponse. Otherwise return an AsyncResponse to retrieve it later (see AsyncResponse.get_response()).

Note : For more information on "NodeId concept".

def close_subscription(self) ‑> None

Close the subscription (no more call to BaseClientConnectionHandler.on_datachanged() will be done)

def create_subscription(self,
requestedPublishingInterval: float = 500.0,
requestedLifetimeCount: int = 10,
requestedMaxKeepAliveCount: int = 3,
maxNotificationsPerPublish: int = 1000,
priority: int = 0,
nbPublishTokens: int = 3) ‑> None

Creates a new subscription with customizable parameters. Only one subscription per connection is available for the moment.

If default values of parameters are the one expected, there is no need to call this function as BaseClientConnectionHandler.add_nodes_to_subscription() will automatically call it.

Args

requestedPublishingInterval
Requested publishing interval in milliseconds.
requestedLifetimeCount
Requested lifetime count for the subscription (number of publishing cycles). When the subscription publishing interval expired this count of times without Publish token available the subscription is deleted by the server. It shall be at least 3 times requestedMaxKeepAliveCount.
requestedMaxKeepAliveCount
Requested max keep alive count for the subscription (number of publishing cycles). When the subscription publishing interval expired this count of times without notification to send, the server sends a keep alive Publish response. requestedLifetimeCount shall be at least 3 times this value.
maxNotificationsPerPublish
Maximum number of notifications sent in a Publish response.
priority
Priority of the subscription (0 means not special priority).
nbPublishTokens
The number of publish tokens to be used by the client.
def disconnect(self) ‑> bool

Disconnect the current connection. Return True if the connection was established and disconnection succeeded, False otherwise.

def on_datachanged(self,
nodeId: str,
dataValue: DataValue)

This callback is called upon reception of a value change for a subscribed node. See BaseClientConnectionHandler.add_nodes_to_subscription().

Args

nodeId
the string containing the NodeId of the changed node.
dataValue
the new value (see DataValue).
def read_nodes(self,
nodeIds: list[str],
attributes: list[AttributeId] = [],
bWaitResponse: bool = True) ‑> Response

Requests to execute the OPC UA Read service on server and returns the Read response content. See ReadResponse for result content.

Args

nodeIds
The list of nodeIds to read (see "NodeId concept").
attributes
The list of attributes to be read in the previous given nodes (1 attribute per node). See AttributeId enum values. Automatically set to AttributeId.Value if empty.
bWaitResponse
True, wait the response and returns a ReadResponse. Otherwise return an AsyncResponse to retrieve it later (see AsyncResponse.get_response()).
def write_nodes(self,
nodeIds: list[str],
datavalues: list[DataValue],
attributes: list[AttributeId] = [],
types=[],
bWaitResponse: bool = True,
autoTypeWithRead: bool = True) ‑> Response

Requests to execute the OPC UA Write service on server and returns the Write response content. See WriteResponse for result content.

Args

nodeIds
The list of nodeIds to write (see "NodeId concept").
datavalues
The list of DataValues to write (1 per node). see DataValue
attributes
The list of attributes to be written for the given nodeIds (1 attribute per node). See AttributeId enum. Automatically set to AttributeId.Value if empty.
bWaitResponse
True, wait the response and returns a WriteResponse. Otherwise return an AsyncResponse to retrieve it later (see AsyncResponse.get_response()).

Note 1: if the data values are not typed, an attempt to read the values first will be done. If some types are still missing afterward an assertion will be raised. Note 2: if data value server timestamp is set, it will be reset for the write operation as it is not writable in servers

class BrowseResponse (results: list[BrowseResult])

A Browse response provides the status codes resulting of OPC UA Browse service requested treatment.

Ancestors

Instance variables

prop results : list[BrowseResult]

Result browse results codes have the index corresponding to the NodeId index provided in BaseClientConnectionHandler.browse_nodes(). Each status code indicate the success StatusCode.isGoodStatus() or failure of the write operation.

Methods

def is_ok(self) ‑> bool

Returns True if all browses were done successfully.

class BrowseResult (statusCode: SOPC_ReturnStatus,
continuationPoint: bytes,
references: list[Reference])

The BrowseResult is a low-level structures that contains the results of browsing a single node.

Attributes

status
the status code of the browse operation.
continuationPoint
whether the browse is incomplete (continuationPoint not empty) or not.
references
list of outgoing References.
class CallResponse (callResult: StatusCode,
inputArgResults: list[StatusCode],
outputResults: list[Variant])

A Call response provides the status codes and list of variant (output arguments of method called) resulting of OPC UA CallMethod service requested treatment.

Ancestors

Instance variables

prop callResult : StatusCode

Returns the status code of the callmethod.

prop inputArgResults : list[StatusCode]

inputArgResults status codes have the index corresponding to the inputArgList index provided in BaseClientConnectionHandler.call_method(). Each status code indicates the correctness of each inputs argument provided to the callmethod.

prop outputResults : list[Variant]

Returns the result (if any) of the method called. This result is given in the form of Variant list and has the index corresponding to the method's output index provided in BaseClientConnectionHandler.call_method().

Methods

def is_ok(self) ‑> bool

Returns True if the callmethod were done successfully.

class ClientConnectionFailure (*args, **kwargs)

Unexpected connection failure exception meaning the connection has been closed. It is raised when an attempt to use the closed connection it is made.

Ancestors

  • builtins.Exception
  • builtins.BaseException
class ConnectionConfiguration (configs: _C_Configurations, cfgIdx: int)

The connection configuration to be used with PyS2OPC_Client.connect() function.

Instance variables

prop connectionId : str
class DataValue (variant: Variant,
timestampSource: float | None = None,
timestampServer: float | None = None,
statusCode: StatusCode = 0)

Representation of the OPC UA DataValue.

Attributes

variant
The Variant storing the value.
variantType
An accessor to the variantType attribute of Variant.
timestampSource
The last time the value changed, specified by the writer, as a Python timestamp.
timestampServer
The last time the value changed, according to the server, as a Python timestamp.
statusCode
The quality associated to the value OR the reason why there is no available value. It is a value from the StatusCode enum (e.g. StatusCode.BadAttributeIdInvalid).

Static methods

def from_python(val, setTs: bool = True) ‑> DataValue

Creates a DataValue from a Python value or a Variant. Creates the Variant, sets the status code to OK, and set source timestamp to current time (when setTs is True).

Args

val
a builtin Python value or a Variant.
setTs
set the source timestamp to the current timestamp

Instance variables

prop variantType : VariantType

variantType: An accessor to the variantType attribute of the Variant.

Methods

def get_python(self)

Returns the python object wrapped by this DataValue. Accessor to Variant.get_python(). Use this when it is known that the Variant object will not be reused (e.g. by a future call to write_nodes).

Does not copy the object before returning it.

class NamedMembers

This class or its subclass is capable of returning the name of a member of this class from an id. NamedMembers.get_name_from_id() is particularly useful to translate OPC UA constants to readable strings. The function NamedMembers.get_both_from_id() returns a string which contains both the formatted code and its string decode.

Subclasses

Static methods

def get_both_from_id(memberId)

Returns a string "0x01234567 (NAME_OF_ID)" which gives both the hexa-formatted Id ans its name.

def get_name_from_id(memberId)

Returns the name of the class member that has the given memberId. Exclude getters and private members.

class NodeClass

The available node classes as an enum.

Ancestors

Class variables

var DataType
var Method
var Object
var ObjectType
var ReferenceType
var Unspecified
var Variable
var VariableType
var View

Inherited members

class PyS2OPC

Base class for components that are common to both Clients and Servers.

Subclasses

Static methods

def get_version() ‑> str

Returns complete version string (PyS2OPC, S2OPC_Common, S2OPC_ClientServer)

Methods

def initialize(logLevel: SOPC_Log_Level = SOPC_Log_Level.SOPC_LOG_LEVEL_DEBUG,
logPath: str = 'logs/',
logFileMaxBytes: int = 1048576,
logMaxFileNumber: int = 50) ‑> None

Common initialization (Client/Server): configure log, start toolkit threads, etc. Automatically called by PyS2OPC_Client.initialize() or PyS2OPC_Server.initialize() if not done.

Performs toolkit common initialization if this has not already been done

class PyS2OPC_Client

The Client side of the PyS2OPC library. Used to create clients only.

Ancestors

Static methods

def clear() ‑> None

Clear the client configuration and clear/stop the toolkit if applicable.

def connect(connCfg: ConnectionConfiguration,
ConnectionHandlerClass=None,
sc_lifetime: int = 3600000) ‑> BaseClientConnectionHandler

This function must be called after PyS2OPC_Client.load_client_configuration_from_file() and use one of the parsed configurations. Use a configuration instance and provided parameters to connect to the server.

Args

connCfg
The configuration to establish the connection to the server
ConnectionHandlerClass
(optional) class that inherits from BaseClientConnectionHandler, and overrides at least overrides the BaseClientConnectionHandler.on_datachanged() callback for subscription.
sc_lifetime
Requested lifetime in millisecond for the secure channel.

It can be optionally used in a with statement, which automatically disconnects the connection. See BaseClientConnectionHandler

def get_client_key_password() ‑> str

Default method that is called during configuration phase if an encrypted private key is used, it shall return the password to decrypt the client private key.

It uses get_pass library and homonymous function to prompt the user for password.

It is possible to overwrite this function by assiging a new implementation to PyS2OPC_Client.get_client_key_password() to obtain a different behavior.

def get_user_X509_key_password(userCertThumb: str) ‑> str

Default method that is called during configuration phase if the UserPolicy requires a user X509 certificate, it shall return the password to be used to decrypt the user X509 private key.

It uses get_pass library and homonymous function to prompt the user for password.

It is possible to overwrite this function by assiging a new implementation to PyS2OPC_Client.get_user_X509_key_password() to obtain a different behavior.

Args

userCertThumb
the user X509 certificate thumbprint for which the private key password is requested for decryption
def get_username_password(connConfigUserId: str) ‑> tuple[str, str]

Default method that is called during configuration phase if the UserPolicy requires a user, it shall return the username and password associated as a 2-tuples of string.

It uses get_pass library and homonymous function to prompt the user for username and password.

It is possible to overwrite this function by assiging a new implementation to PyS2OPC_Client.get_username_password() to obtain a different behavior.

Args

connConfigUserId
the XML user defined id to retrieve the secure connection configuration
def load_client_configuration_from_file(xml_client_config_path: str) ‑> dict[str, ConnectionConfiguration]

Configure client from XML configuration file. This function must be called after PyS2OPC_Client.initialize().

Args

xml_client_config_path
Path to client configuration XML file (s2opc_clientserver_config.xsd schema)

Methods

def initialize(logLevel: SOPC_Log_Level = SOPC_Log_Level.SOPC_LOG_LEVEL_DEBUG,
logPath: str = 'logs/',
logFileMaxBytes: int = 1048576,
logMaxFileNumber: int = 50)

Toolkit initialization for Client. Call common initialization.

This function supports the context management:

>>> with PyS2OPC_Client.initialize():
...     # Do things here, namely configure then wait
...     pass

When reaching out of the with statement, the Toolkit is automatically cleared. See PyS2OPC_Client.clear().

Args

logLevel
log level (SOPC_Log_Level): ERROR: only errors are traced WARNING: only warnings and errors are traced INFO: information level, warnings and errors are traced DEBUG: all available information is traced
logPath
the path for log files creation (might be relative to current working directory). logPath is created if it does not exist.
logFileMaxBytes
The maximum size of the log files, before changing the log index.
logMaxFileNumber
The maximum number of log indexes before cycling logs and reusing the first log.

Inherited members

class PyS2OPC_Server

The Server side of the PyS2OPC library.

Ancestors

Static methods

def browse_nodes(nodeIds: list[str],
maxReferencesPerNode: int = 1000,
bWaitResponse: bool = True) ‑> Response

Requests to execute the OPC UA Browse service locally and returns the Browse response content. See BrowseResponse for result content.

Args

nodeIds
The list of nodeIds to browse (see "NodeId concept").
maxReferencesPerNode
Indicates the maximum number of references to return for each starting node specified in the request (0 means no limitation).
bWaitResponse
True, wait the response and returns a BrowseResponse. Otherwise return an AsyncResponse to retrieve it later (see AsyncResponse.get_response()).
def clear() ‑> None

Clear the server configuration and clear/stop the toolkit if applicable.

def get_server_key_password() ‑> str

Default method that is called during configuration phase if an encrypted private key is used, it shall return the password to decrypt the server private key.

It uses get_pass library and homonymous function to prompt the user for password.

It is possible to overwrite this function by assiging a new implementation to PyS2OPC_Server.get_server_key_password() to obtain a different behavior.

def load_server_configuration_from_files(xml_server_config_path: str,
xml_address_space_config_path: str,
xml_users_config_path: str,
address_space_handler: BaseAddressSpaceHandler = None)

Configure server from XML configuration files for: server endpoints, address space and users credential and rights. This function must be called after PyS2OPC_Server.initialize().

Args

xml_server_config_path
path to server configuration XML file (s2opc_clientserver_config.xsd schema)
xml_address_space_config_path
path to address space configuration XML file (UANodeSet.xsd schema)
xml_users_config_path
path to users credential and rights configuration XML file (s2opc_clientserver_users_config.xsd schema)
address_space_handler
None (no write notification) or an instance of a subclass of BaseAddressSpaceHandler
def read_nodes(nodeIds: list[str],
attributes: list[AttributeId] = [],
bWaitResponse: bool = True) ‑> Response

Requests to execute the OPC UA Read service locally and returns the Read response content. See ReadResponse for result content.

Args

nodeIds
The list of nodeIds to read (see "NodeId concept").
attributes
The list of attributes to be read in the previous given nodes (1 attribute per node). See AttributeId enum values. Automatically set to AttributeId.Value if empty.
bWaitResponse
True, wait the response and returns a ReadResponse. Otherwise return an AsyncResponse to retrieve it later (see AsyncResponse.get_response()).
def serve_forever() ‑> None

Start server forever. Can be interrupted with a KeyboardInterrupt (SIGINT).

def serving() ‑> bool

Returns true if the server is in service, otherwise false.

def stop_serve()

Stop the server (in a blocking way during shutdown phase). Then wait for _callback_stop_server before return, to avoid clearing the server too early.

Note: Server stops after 5 seconds for shutdown phase to indicate shutdown in ServerState node

def write_nodes(nodeIds: list[str],
datavalues: list[DataValue],
attributes: list[AttributeId] = [],
bWaitResponse: bool = True) ‑> Response

Requests to execute the OPC UA Write service locally and returns the Write response content. See WriteResponse for result content.

Args

nodeIds
The list of nodeIds to write (see "NodeId concept").
datavalues
The list of DataValues to write (1 per node). see DataValue
attributes
The list of attributes to be written for the given nodeIds (1 attribute per node). Automatically set to AttributeId.Value if empty. (only Value is supported for the moment)
bWaitResponse
True, wait the response and returns a WriteResponse. Otherwise return an AsyncResponse to retrieve it later (see AsyncResponse.get_response()).

Methods

def initialize(logLevel: SOPC_Log_Level = SOPC_Log_Level.SOPC_LOG_LEVEL_DEBUG,
logPath: str = 'logs/',
logFileMaxBytes: int = 1048576,
logMaxFileNumber: int = 50) ‑> None

Toolkit initialization for Server. Call common initialization.

This function supports the context management:

>>> with PyS2OPC_Server.initialize():
...     # Do things here, namely configure then wait
...     pass

When reaching out of the with statement, the Toolkit is automatically cleared. See PyS2OPC_Server.clear().

Args

logLevel
log level (SOPC_Log_Level): ERROR: only errors are traced WARNING: only warnings and errors are traced INFO: information level, warnings and errors are traced DEBUG: all available information is traced
logPath
the path for log files creation (might be relative to current working directory). logPath is created if it does not exist.
logFileMaxBytes
The maximum size of the log files, before changing the log index.
logMaxFileNumber
The maximum number of log indexes before cycling logs and reusing the first log.
def serve() ‑> None

Starts the server asynchronously. Server information node is updated and endpoints are asynchronously requested to be opened.

Supports the context management facilities to close the server:

>>> with PyS2OPC_Server.serve():
...     # Do things here
...     pass
When reaching out of the <code>with</code> statement, the server is automatically closed.
See <code><a title="pys2opc.PyS2OPC_Server.stop_serve" href="#pys2opc.PyS2OPC_Server.stop_serve">PyS2OPC\_Server.stop\_serve()</a></code>.

If you don't have applicative application, and callbacks are enough, see instead PyS2OPC_Server.serve_forever().

Inherited members

class ReadResponse (results: list[DataValue])

A ReadResponse provides the data values content resulting of OPC UA Read service requested.

Ancestors

Instance variables

prop results : list[DataValue]

Result data values have the index corresponding to the NodeId index provided in BaseClientConnectionHandler.read_nodes()

class Reference (referenceTypeId: str,
isForward: bool,
nodeId: str,
browseName: tuple[uint16_t, str],
displayName: tuple[str, str],
nodeClass: NodeClass,
typeDefinition: str)

A Python version for the OpcUa_ReferenceDescription, a low level representation of the references. Does not contain the source of the Reference.

Attributes

referenceTypeId
The string NodeId that defines the type of the Reference.
isForward
True when the reference is forward (going from the browsed node to nodeId), False when the reference is in the inverse direction (from the nodeId to the browsed node).
nodeId
Target Expanded nodeId of the Reference. When the node is in the address space, Expanded NodeId and NodeId have the same string representation.
browseName
Browse name of the nodeId, as a qualified name, i.e. a couple (namespaceIndex, str).
displayName
Display name of the nodeId, as a localized text, i.e. a couple (str of the chosen locale, str in this locale).
nodeClass
NodeClass of the nodeId (see NodeClass).
typeDefinition
NodeId of the type of the target node, when the target NodeId is a Variable or an Object. It defines which VariableType or ObjectType node was used to instantiate the target node.
class Response (...)

Parent class of all response types ReadResponse, WriteResponse, BrowseResponse, AsyncResponse.

Subclasses

class ReturnStatus

This class or its subclass is capable of returning the name of a member of this class from an id. NamedMembers.get_name_from_id() is particularly useful to translate OPC UA constants to readable strings. The function NamedMembers.get_both_from_id() returns a string which contains both the formatted code and its string decode.

Ancestors

Class variables

var STATUS_CLOSED
var STATUS_ENCODING_ERROR
var STATUS_INVALID_PARAMETERS
var STATUS_INVALID_STATE
var STATUS_NOK
var STATUS_NOT_SUPPORTED
var STATUS_OK
var STATUS_OUT_OF_MEMORY
var STATUS_TIMEOUT
var STATUS_WOULD_BLOCK

Inherited members

class SOPC_Failure (message: str,
status: ReturnStatus | None = None)

Exception sub class to manage failure from S2OPC code

Ancestors

  • builtins.Exception
  • builtins.BaseException
class SOPC_Log_Level (value, names=None, *, module=None, qualname=None, type=None, start=1)

An enumeration.

Ancestors

  • enum.IntFlag
  • builtins.int
  • enum.Flag
  • enum.Enum

Class variables

var SOPC_LOG_LEVEL_DEBUG
var SOPC_LOG_LEVEL_ERROR
var SOPC_LOG_LEVEL_INFO
var SOPC_LOG_LEVEL_WARNING
class SOPC_Log_Module (value, names=None, *, module=None, qualname=None, type=None, start=1)

An enumeration.

Ancestors

  • enum.IntFlag
  • builtins.int
  • enum.Flag
  • enum.Enum

Class variables

var SOPC_LOG_MODULE_CLIENTSERVER
var SOPC_LOG_MODULE_COMMON
var SOPC_LOG_MODULE_PUBSUB
class SOPC_Log_System (value, names=None, *, module=None, qualname=None, type=None, start=1)

An enumeration.

Ancestors

  • enum.IntFlag
  • builtins.int
  • enum.Flag
  • enum.Enum

Class variables

var SOPC_LOG_SYSTEM_FILE
var SOPC_LOG_SYSTEM_NO_LOG
var SOPC_LOG_SYSTEM_USER
class SOPC_SecurityPolicy_ID (value, names=None, *, module=None, qualname=None, type=None, start=1)

An enumeration.

Ancestors

  • enum.IntFlag
  • builtins.int
  • enum.Flag
  • enum.Enum

Class variables

var SOPC_SecurityPolicy_Aes128Sha256RsaOaep_ID
var SOPC_SecurityPolicy_Aes256Sha256RsaPss_ID
var SOPC_SecurityPolicy_Basic256Sha256_ID
var SOPC_SecurityPolicy_Basic256_ID
var SOPC_SecurityPolicy_Invalid_ID
var SOPC_SecurityPolicy_Last_ID
var SOPC_SecurityPolicy_None_ID
var SOPC_SecurityPolicy_PubSub_Aes256_ID
class ServiceFailure (message: str,
status: StatusCode | None = None)

Exception sub class to manage service failure

Ancestors

  • builtins.Exception
  • builtins.BaseException

Subclasses

class ServiceFault (message: str,
status: StatusCode | None = None)

ServiceFailure sub class to manage service fault

Ancestors

class StatusCode

The OpcUa status codes. Directly generated from src/Common/opcua_types/opcua_statuscodes.h. Status codes are used in various places among the OPC UA protocol. They usually represent the quality of a value (see DataValue), or the status of the execution of a service (see WriteResponse, BrowseResponse).

Ancestors

Class variables

var Bad
var BadAggregateConfigurationRejected
var BadAggregateInvalidInputs
var BadAggregateListMismatch
var BadAggregateNotSupported
var BadApplicationSignatureInvalid
var BadArgumentsMissing
var BadAttributeIdInvalid
var BadBoundNotFound
var BadBoundNotSupported
var BadBrowseDirectionInvalid
var BadBrowseNameDuplicated
var BadBrowseNameInvalid
var BadCertificateChainIncomplete
var BadCertificateHostNameInvalid
var BadCertificateInvalid
var BadCertificateIssuerRevocationUnknown
var BadCertificateIssuerRevoked
var BadCertificateIssuerTimeInvalid
var BadCertificateIssuerUseNotAllowed
var BadCertificateRevocationUnknown
var BadCertificateRevoked
var BadCertificateTimeInvalid
var BadCertificateUntrusted
var BadCertificateUriInvalid
var BadCertificateUseNotAllowed
var BadCommunicationError
var BadConditionAlreadyDisabled
var BadConditionAlreadyEnabled
var BadConditionAlreadyShelved
var BadConditionBranchAlreadyAcked
var BadConditionBranchAlreadyConfirmed
var BadConditionDisabled
var BadConditionNotShelved
var BadConfigurationError
var BadConnectionClosed
var BadConnectionRejected
var BadContentFilterInvalid
var BadContinuationPointInvalid
var BadDataEncodingInvalid
var BadDataEncodingUnsupported
var BadDataLost
var BadDataTypeIdUnknown
var BadDeadbandFilterInvalid
var BadDecodingError
var BadDeviceFailure
var BadDialogNotActive
var BadDialogResponseInvalid
var BadDisconnect
var BadDiscoveryUrlMissing
var BadDuplicateReferenceNotAllowed
var BadEncodingError
var BadEncodingLimitsExceeded
var BadEndOfStream
var BadEntryExists
var BadEventFilterInvalid
var BadEventIdUnknown
var BadEventNotAcknowledgeable
var BadExpectedStreamToBlock
var BadFilterElementInvalid
var BadFilterLiteralInvalid
var BadFilterNotAllowed
var BadFilterOperandCountMismatch
var BadFilterOperandInvalid
var BadFilterOperatorInvalid
var BadFilterOperatorUnsupported
var BadHistoryOperationInvalid
var BadHistoryOperationUnsupported
var BadIdentityChangeNotSupported
var BadIdentityTokenInvalid
var BadIdentityTokenRejected
var BadIndexRangeInvalid
var BadIndexRangeNoData
var BadInsufficientClientProfile
var BadInternalError
var BadInvalidArgument
var BadInvalidSelfReference
var BadInvalidState
var BadInvalidTimestamp
var BadInvalidTimestampArgument
var BadLicenseNotAvailable
var BadMaxAgeInvalid
var BadMaxConnectionsReached
var BadMessageNotAvailable
var BadMethodInvalid
var BadMonitoredItemFilterInvalid
var BadMonitoredItemFilterUnsupported
var BadMonitoredItemIdInvalid
var BadMonitoringModeInvalid
var BadNoCommunication
var BadNoContinuationPoints
var BadNoData
var BadNoDataAvailable
var BadNoDeleteRights
var BadNoEntryExists
var BadNoMatch
var BadNoSubscription
var BadNodeAttributesInvalid
var BadNodeClassInvalid
var BadNodeIdExists
var BadNodeIdInvalid
var BadNodeIdUnknown
var BadNodeNotInView
var BadNonceInvalid
var BadNotConnected
var BadNotFound
var BadNotImplemented
var BadNotReadable
var BadNotSupported
var BadNotTypeDefinition
var BadNotWritable
var BadNothingToDo
var BadObjectDeleted
var BadOperationAbandoned
var BadOutOfMemory
var BadOutOfRange
var BadOutOfService
var BadParentNodeIdInvalid
var BadProtocolVersionUnsupported
var BadQueryTooComplex
var BadReferenceLocalOnly
var BadReferenceNotAllowed
var BadReferenceTypeIdInvalid
var BadRefreshInProgress
var BadRequestCancelledByClient
var BadRequestCancelledByRequest
var BadRequestHeaderInvalid
var BadRequestInterrupted
var BadRequestNotAllowed
var BadRequestTimeout
var BadRequestTooLarge
var BadRequestTypeInvalid
var BadResourceUnavailable
var BadResponseTooLarge
var BadSecureChannelClosed
var BadSecureChannelIdInvalid
var BadSecureChannelTokenUnknown
var BadSecurityChecksFailed
var BadSecurityModeInsufficient
var BadSecurityModeRejected
var BadSecurityPolicyRejected
var BadSempahoreFileMissing
var BadSensorFailure
var BadSequenceNumberInvalid
var BadSequenceNumberUnknown
var BadServerHalted
var BadServerIndexInvalid
var BadServerNameMissing
var BadServerNotConnected
var BadServerUriInvalid
var BadServiceUnsupported
var BadSessionClosed
var BadSessionIdInvalid
var BadSessionNotActivated
var BadShelvingTimeOutOfRange
var BadShutdown
var BadSourceNodeIdInvalid
var BadStateNotActive
var BadStructureMissing
var BadSubscriptionIdInvalid
var BadSyntaxError
var BadTargetNodeIdInvalid
var BadTcpEndpointUrlInvalid
var BadTcpInternalError
var BadTcpMessageTooLarge
var BadTcpMessageTypeInvalid
var BadTcpNotEnoughResources
var BadTcpSecureChannelUnknown
var BadTcpServerTooBusy
var BadTimeout
var BadTimestampsToReturnInvalid
var BadTooManyArguments
var BadTooManyMatches
var BadTooManyMonitoredItems
var BadTooManyOperations
var BadTooManyPublishRequests
var BadTooManySessions
var BadTooManySubscriptions
var BadTypeDefinitionInvalid
var BadTypeMismatch
var BadUnexpectedError
var BadUnknownResponse
var BadUserAccessDenied
var BadUserSignatureInvalid
var BadViewIdUnknown
var BadViewParameterMismatch
var BadViewTimestampInvalid
var BadViewVersionInvalid
var BadWaitingForInitialData
var BadWaitingForResponse
var BadWouldBlock
var BadWriteNotSupported
var Good
var GoodCallAgain
var GoodClamped
var GoodCommunicationEvent
var GoodCompletesAsynchronously
var GoodDataIgnored
var GoodDependentValueChanged
var GoodEdited
var GoodEntryInserted
var GoodEntryReplaced
var GoodLocalOverride
var GoodMoreData
var GoodNoData
var GoodNonCriticalTimeout
var GoodOverload
var GoodPostActionFailed
var GoodShutdownEvent
var GoodSubscriptionTransferred
var Uncertain
var UncertainDataSubNormal
var UncertainDominantValueChanged
var UncertainEngineeringUnitsExceeded
var UncertainInitialValue
var UncertainLastUsableValue
var UncertainNoCommunicationLastUsableValue
var UncertainNotAllNodesAvailable
var UncertainReferenceNotDeleted
var UncertainReferenceOutOfServer
var UncertainSensorNotAccurate
var UncertainSubNormal
var UncertainSubstituteValue

Static methods

def isGoodStatus(status: StatusCode) ‑> bool

Inherited members

class Variant (python_value,
variantType: VariantType | None = None)

A Variant is the Pythonic representation of a SOPC_Variant. The SOPC_Variant is a C-structure that can contain multiple built-in types, such as integers, floats, strings.

A Variant instance supports the arithmetic, comparison operations, and increment operators of the Python types. It can be used as any Python value. For instance:

>>> Variant(2) + Variant(6)  # Produces a new Variant
Variant(8)
>>> Variant('foo') + Variant('bar')
Variant('foobar')
>>> Variant(b'to') * 2  # ByteString
Variant(b'toto')
>>> v = Variant(2); v *= 2; v
Variant(4)
>>> Variant(2) == Variant(2) and Variant(2.) == Variant(2) and Variant(2.) == 2
True

Attributes

value
The python value of the Variant.
variantType
Optional: The type of the Variant.

Methods

def get_python(self)

Returns The python value of the Variant

class VariantType

This class or its subclass is capable of returning the name of a member of this class from an id. NamedMembers.get_name_from_id() is particularly useful to translate OPC UA constants to readable strings. The function NamedMembers.get_both_from_id() returns a string which contains both the formatted code and its string decode.

Ancestors

Class variables

var Boolean
var Byte
var ByteString
var DataValue
var DateTime
var DiagnosticInfo
var Double
var ExpandedNodeId
var ExtensionObject
var Float
var Guid
var Int16
var Int32
var Int64
var LocalizedText
var NodeId
var Null
var QualifiedName
var SByte
var StatusCode
var String
var UInt16
var UInt32
var UInt64
var Variant
var XmlElement

Inherited members

class WriteResponse (results: list[StatusCode])

A WriteResponse provides the status codes resulting of OPC UA Write service requested

Ancestors

Instance variables

prop results : list[StatusCode]

Result status codes have the index corresponding to the NodeId index provided in BaseClientConnectionHandler.write_nodes(). Each status code indicate the success StatusCode.isGoodStatus() or failure of the write operation.

Methods

def is_ok(self) ‑> bool

Returns True if all writes were done successfully.