opcua.server package¶
Subpackages¶
- opcua.server.standard_address_space package
- Submodules
- opcua.server.standard_address_space.standard_address_space module
- opcua.server.standard_address_space.standard_address_space_part10 module
- opcua.server.standard_address_space.standard_address_space_part11 module
- opcua.server.standard_address_space.standard_address_space_part13 module
- opcua.server.standard_address_space.standard_address_space_part3 module
- opcua.server.standard_address_space.standard_address_space_part4 module
- opcua.server.standard_address_space.standard_address_space_part5 module
- opcua.server.standard_address_space.standard_address_space_part8 module
- opcua.server.standard_address_space.standard_address_space_part9 module
- Module contents
Submodules¶
opcua.server.address_space module¶
-
class
opcua.server.address_space.
AddressSpace
[source]¶ Bases:
object
The address space object stores all the nodes of the OPC-UA server and helper methods. The methods are thread safe
-
dump
(path)[source]¶ Dump address space as binary to file; note that server must be stopped for this method to work DO NOT DUMP AN ADDRESS SPACE WHICH IS USING A SHELF (load_aspace_shelf), ONLY CACHED NODES WILL GET DUMPED!
-
load
(path)[source]¶ Load address space from a binary file, overwriting everything in the current address space
-
load_aspace_shelf
(path)[source]¶ Load the standard address space nodes from a python shelve via LazyLoadingDict as needed. The dump() method can no longer be used if the address space is being loaded from a shelf
Note: Intended for slow devices, such as Raspberry Pi, to greatly improve start up time
-
make_aspace_shelf
(path)[source]¶ Make a shelf for containing the nodes from the standard address space; this is typically only done on first start of the server. Subsequent server starts will load the shelf, nodes are then moved to a cache by the LazyLoadingDict class when they are accessed. Saving data back to the shelf is currently NOT supported, it is only used for the default OPC UA standard address space
Note: Intended for slow devices, such as Raspberry Pi, to greatly improve start up time
-
opcua.server.binary_server module¶
opcua.server.binary_server_asyncio module¶
Socket server forwarding request to internal server
-
class
opcua.server.binary_server_asyncio.
BinaryServer
(internal_server, hostname, port)[source]¶ Bases:
object
-
class
opcua.server.binary_server_asyncio.
OPCUAProtocol
[source]¶ Bases:
asyncio.protocols.Protocol
Interface for OPCUA protocol.
-
clients
= None¶
-
connection_lost
(ex)[source]¶ Called when the connection is lost or closed.
The argument is an exception object or None (the latter meaning a regular EOF is received or the connection was aborted or closed).
-
connection_made
(transport)[source]¶ Called when a connection is made.
The argument is the transport representing the pipe connection. To receive data, wait for data_received() calls. When the connection is closed, connection_lost() is called.
-
iserver
= None¶
-
logger
= None¶
-
loop
= None¶
-
policies
= None¶
-
opcua.server.internal_server module¶
Internal server implementing opcu-ua interface. Can be used on server side or to implement binary/https opc-ua servers
-
class
opcua.server.internal_server.
InternalServer
(shelffile=None, user_manager=None, session_cls=None)[source]¶ Bases:
object
-
disable_history_data_change
(node)[source]¶ Set attribute Historizing of node to False and stop storing data for history
-
disable_history_event
(source)[source]¶ Set attribute History Read of node to False and stop storing data for history
-
enable_history_data_change
(node, period=datetime.timedelta(days=7), count=0)[source]¶ Set attribute Historizing of node to True and start storing data for history
-
enable_history_event
(source, period=datetime.timedelta(days=7), count=0)[source]¶ Set attribute History Read of object events to True and start storing data for history
-
local_discovery_service
¶
-
set_attribute_value
(nodeid, datavalue, attr=<AttributeIds.Value: 13>)[source]¶ directly write datavalue to the Attribute, bypasing some checks and structure creation so it is a little faster
-
thread_loop
¶
-
opcua.server.internal_subscription module¶
server side implementation of a subscription object
-
class
opcua.server.internal_subscription.
InternalSubscription
(subservice, data, addressspace, callback)[source]¶ Bases:
object
-
class
opcua.server.internal_subscription.
MonitoredItemService
(isub, aspace)[source]¶ Bases:
object
implement monitoreditem service for 1 subscription
opcua.server.server module¶
High level interface to pure python OPC-UA server
-
class
opcua.server.server.
Server
(shelffile=None, iserver=None)[source]¶ Bases:
object
High level Server class
This class creates an opcua server with default values
Create your own namespace and then populate your server address space using use the get_root() or get_objects() to get Node objects. and get_event_object() to fire events. Then start server. See example_server.py All methods are threadsafe
If you need more flexibility you call directly the Ua Service methods on the iserver or iserver.isession object members.
During startup the standard address space will be constructed, which may be time-consuming when running a server on a less powerful device (e.g. a Raspberry Pi). In order to improve startup performance, a optional path to a cache file can be passed to the server constructor. If the parameter is defined, the address space will be loaded from the cache file or the file will be created if it does not exist yet. As a result the first startup will be even slower due to the cache file generation but all further start ups will be significantly faster.
Variables: - product_uri (uri) –
- name (string) –
- default_timeout (InternalServer) – timeout in milliseconds for sessions and secure channel
- iserver – internal server object
- bserver (BinaryServer) – binary protocol server
- nodes (Shortcuts) – shortcuts to common nodes
-
create_custom_object_type
(idx, name, basetype=58, properties=None, variables=None, methods=None)[source]¶
-
create_custom_variable_type
(idx, name, basetype=62, properties=None, variables=None, methods=None)[source]¶
-
create_subscription
(period, handler)[source]¶ Create a subscription. returns a Subscription object which allow to subscribe to events or data on server period is in milliseconds handler is a python object with following methods:
def datachange_notification(self, node, val, data): def event_notification(self, event): def status_change_notification(self, status):
-
dehistorize_node_data_change
(node)[source]¶ Stop historizing supplied nodes; see history module Args:
node: node or list of nodes that can be historized (UA variables/properties)Returns:
-
dehistorize_node_event
(node)[source]¶ Stop historizing events from node (typically a UA object); see history module Args:
node: node or list of nodes that can be historized (UA objects)Returns:
-
disable_clock
(val=True)[source]¶ for debugging you may want to disable clock that write every second to address space
-
export_xml_by_ns
(path, namespaces=None)[source]¶ Export nodes of one or more namespaces to an XML file. Namespaces used by nodes are always exported for consistency. Args:
server: opc ua server to use path: name of the xml file to write namespaces: list of string uris or int indexes of the namespace to export, if not provide all ns are used except 0Returns:
-
get_event_generator
(etype=None, emitting_node=2253)[source]¶ Returns an event object using an event type from address space. Use this object to fire events
-
historize_node_data_change
(node, period=datetime.timedelta(days=7), count=0)[source]¶ Start historizing supplied nodes; see history module Args:
node: node or list of nodes that can be historized (variables/properties) period: time delta to store the history; older data will be deleted from the storage count: number of changes to store in the historyReturns:
-
historize_node_event
(node, period=datetime.timedelta(days=7), count=0)[source]¶ Start historizing events from node (typically a UA object); see history module Args:
node: node or list of nodes that can be historized (UA objects) period: time delta to store the history; older data will be deleted from the storage count: number of events to store in the historyReturns:
-
link_method
(node, callback)[source]¶ Link a python function to a UA method in the address space; required when a UA method has been imported to the address space via XML; the python executable must be linked manually Args:
node: UA method node callback: python function that the UA method will callReturns:
-
load_enums
()[source]¶ load UA structures and generate python Enums in ua module for custom enums in server
-
load_type_definitions
(nodes=None)[source]¶ load custom structures from our server. Server side this can be used to create python objects from custom structures imported through xml into server
-
local_discovery_service
¶
-
set_application_uri
(uri)[source]¶ Set application/server URI. This uri is supposed to be unique. If you intent to register your server to a discovery server, it really should be unique in your system! default is : “urn:freeopcua:python:server”
-
set_attribute_value
(nodeid, datavalue, attr=<AttributeIds.Value: 13>)[source]¶ directly write datavalue to the Attribute, bypasing some checks and structure creation so it is a little faster
-
set_build_info
(product_uri, manufacturer_name, product_name, software_version, build_number, build_date)[source]¶ Update the servers build information. This needs to be added to the ServerStatus, BuildInfo and all underlying nodes
-
set_security_IDs
(policyIDs)[source]¶ Method setting up the security endpoints for identification of clients. During server object initialization, all possible endpoints are enabled:
self._policyIDs = [“Anonymous”, “Basic256Sha256”, “Username”]
E.g. to limit the number of IDs and disable anonymous clients:
set_security_IDs([“Basic256Sha256”])(Implementation for ID check is currently not finalized…)
-
set_security_policy
(security_policy)[source]¶ Method setting up the security policies for connections to the server, where security_policy is a list of integers. During server initialization, all endpoints are enabled:
- security_policy = [
ua.SecurityPolicyType.NoSecurity, ua.SecurityPolicyType.Basic256Sha256_SignAndEncrypt, ua.SecurityPolicyType.Basic256Sha256_Sign
]
E.g. to limit the number of endpoints and disable no encryption:
- set_security_policy([
- ua.SecurityPolicyType.Basic256Sha256_SignAndEncrypt])
-
user_manager
¶
opcua.server.subscription_service module¶
server side implementation of subscription service