InternalServer.subscribe_server_callback() - Code Metrics - Inspection of "Issue #226 (#232)" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 641228...9a97a0 )

by Olivier

created 2016-07-07 20:06 UTC

InternalServer.subscribe_server_callback() A

↳ Parent: InternalServer

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	5
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
c	0
b	0
f	0
dl	0
loc	5
rs	9.4285
ccs	5
cts	5
cp	1
crap	1

"""
Internal server implementing opcu-ua interface. can be used on server side or to implement binary/https opc-ua servers
"""

from datetime import datetime
from copy import copy, deepcopy
from datetime import timedelta
from os import path
import logging
from threading import Lock
from enum import Enum
try:
    from urllib.parse import urlparse
except ImportError:
    from urlparse import urlparse


from opcua import ua
from opcua.common import utils
from opcua.common.callback import CallbackType, ServerItemCallback, CallbackDispatcher
from opcua.common.node import Node
from opcua.server.history import HistoryManager
from opcua.server.address_space import AddressSpace
from opcua.server.address_space import AttributeService
from opcua.server.address_space import ViewService
from opcua.server.address_space import NodeManagementService
from opcua.server.address_space import MethodService
from opcua.server.subscription_service import SubscriptionService
from opcua.server.standard_address_space import standard_address_space
from opcua.server.users import User
from opcua.common import xmlimporter


class SessionState(Enum):
    Created = 0
    Activated = 1
    Closed = 2


class ServerDesc(object):
    def __init__(self, serv, cap=None):
        self.Server = serv
        self.Capabilities = cap


class InternalServer(object):

    def __init__(self, cacheFile = None):
        self.logger = logging.getLogger(__name__)
        
        self.server_callback_dispatcher = CallbackDispatcher()
        
        self.endpoints = []
        self._channel_id_counter = 5
        self.allow_remote_admin = True
        self.disabled_clock = False  # for debugging we may want to disable clock that writes too much in log
        self._known_servers = {}  # used if we are a discovery server

        self.aspace = AddressSpace()
        self.attribute_service = AttributeService(self.aspace)
        self.view_service = ViewService(self.aspace)
        self.method_service = MethodService(self.aspace)
        self.node_mgt_service = NodeManagementService(self.aspace)

        if cacheFile and path.isfile(cacheFile):
            # import address space from shelve
            self.aspace.load(cacheFile)
        else:
            # import address space from code generated from xml
            standard_address_space.fill_address_space(self.node_mgt_service)
            # import address space directly from xml, this has preformance impact so disabled
            #importer = xmlimporter.XmlImporter(self.node_mgt_service)
            #importer.import_xml("/path/to/python-opcua/schemas/Opc.Ua.NodeSet2.xml")

            if cacheFile:
                self.aspace.dump(cacheFile)

        self.loop = utils.ThreadLoop()
        self.asyncio_transports = []
        self.subscription_service = SubscriptionService(self.loop, self.aspace)

        self.history_manager = HistoryManager(self)

        # create a session to use on server side
        self.isession = InternalSession(self, self.aspace, self.subscription_service, "Internal", user=User.Admin)
        self.current_time_node = Node(self.isession, ua.NodeId(ua.ObjectIds.Server_ServerStatus_CurrentTime))
        uries = ["http://opcfoundation.org/UA/"]
        ns_node = Node(self.isession, ua.NodeId(ua.ObjectIds.Server_NamespaceArray))
        ns_node.set_value(uries)
        
        

    def load_address_space(self, path):
        self.aspace.load(path)

    def dump_address_space(self, path):
        self.aspace.dump(path)

    def start(self):
        self.logger.info("starting internal server")
        for edp in self.endpoints:
            self._known_servers[edp.Server.ApplicationUri] = ServerDesc(edp.Server)
        self.loop.start()
        Node(self.isession, ua.NodeId(ua.ObjectIds.Server_ServerStatus_State)).set_value(0)
        Node(self.isession, ua.NodeId(ua.ObjectIds.Server_ServerStatus_StartTime)).set_value(datetime.utcnow())
        if not self.disabled_clock:
            self._set_current_time()

    def stop(self):
        self.logger.info("stopping internal server")
        self.loop.stop()
        self.history_manager.stop()

    def _set_current_time(self):
        self.current_time_node.set_value(datetime.utcnow())
        self.loop.call_later(1, self._set_current_time)

    def get_new_channel_id(self):
        self._channel_id_counter += 1
        return self._channel_id_counter

    def add_endpoint(self, endpoint):
        self.endpoints.append(endpoint)

    def get_endpoints(self, params=None, sockname=None):
        self.logger.info("get endpoint")
        if sockname:
            #return to client the ip address it has access to
            edps = []
            for edp in self.endpoints:
                edp1 = copy(edp)
                url = urlparse(edp1.EndpointUrl)
                url = url._replace(netloc=sockname[0] + ":" + str(sockname[1]))
                edp1.EndpointUrl = url.geturl()
                edps.append(edp1)
            return edps
        return self.endpoints[:]

    def find_servers(self, params):
        if not params.ServerUris:
            return [desc.Server for desc in self._known_servers.values()]
        servers = []
        for serv in self._known_servers.values():
            serv_uri = serv.Server.ApplicationUri.split(":")
            for uri in params.ServerUris:
                uri = uri.split(":")
                if serv_uri[:len(uri)] == uri:
                    servers.append(serv.Server)
                    break
        return servers

    def register_server(self, server, conf=None):
        appdesc = ua.ApplicationDescription()
        appdesc.ApplicationUri = server.ServerUri
        appdesc.ProductUri = server.ProductUri
        appdesc.ApplicationName = server.ServerNames[0]  # FIXME: select name from client locale
        appdesc.ApplicationType = server.ServerType
        appdesc.GatewayServerUri = server.GatewayServerUri
        appdesc.DiscoveryUrls = server.DiscoveryUrls  # FIXME: select discovery uri using reachability from client network
        self._known_servers[server.ServerUri] = ServerDesc(appdesc, conf)

    def register_server2(self, params):
        return self.register_server(params.Server, params.DiscoveryConfiguration)

    def create_session(self, name, user=User.Anonymous, external=False):
        return InternalSession(self, self.aspace, self.subscription_service, name, user=user, external=external)

    def enable_history_data_change(self, node, period=timedelta(days=7), count=0):
        """
        Set attribute Historizing of node to True and start storing data for history
        """
        node.set_attribute(ua.AttributeIds.Historizing, ua.DataValue(True))
        node.set_attr_bit(ua.AttributeIds.AccessLevel, ua.AccessLevel.HistoryRead)
        node.set_attr_bit(ua.AttributeIds.UserAccessLevel, ua.AccessLevel.HistoryRead)
        self.history_manager.historize_data_change(node, period, count)

    def disable_history_data_change(self, node):
        """
        Set attribute Historizing of node to False and stop storing data for history
        """
        node.set_attribute(ua.AttributeIds.Historizing, ua.DataValue(False))
        node.unset_attr_bit(ua.AttributeIds.AccessLevel, ua.AccessLevel.HistoryRead)
        node.unset_attr_bit(ua.AttributeIds.UserAccessLevel, ua.AccessLevel.HistoryRead)
        self.history_manager.dehistorize(node)

    def enable_history_event(self, source, period=timedelta(days=7), count=0):
        """
        Set attribute History Read of object events to True and start storing data for history
        """
        event_notifier = source.get_event_notifier()
        if ua.EventNotifier.SubscribeToEvents not in event_notifier:
            raise ua.UaError("Node does not generate events", event_notifier)
        if ua.EventNotifier.SubscribeToEvents not in event_notifier:
            event_notifier.append(ua.EventNotifier.HistoryRead)
            source.set_event_notifier(event_notifier)
        self.history_manager.historize_event(source, period, count)

    def disable_history_event(self, source):
        """
        Set attribute History Read of node to False and stop storing data for history
        """
        source.unset_attr_bit(ua.AttributeIds.EventNotifier, ua.EventNotifier.HistoryRead)
        self.history_manager.dehistorize(source)
        
    def subscribe_server_callback(self, event, handle):
        """
        Create a subscription from event to handle
        """
        self.server_callback_dispatcher.addListener(event, handle)
        
    def unsubscribe_server_callback(self, event, handle):
        """
        Remove a subscription from event to handle
        """
        self.server_callback_dispatcher.removeListener(event, handle)


class InternalSession(object):
    _counter = 10
    _auth_counter = 1000

    def __init__(self, internal_server, aspace, submgr, name, user=User.Anonymous, external=False):
        self.logger = logging.getLogger(__name__)
        self.iserver = internal_server
        self.external = external  # define if session is external, we need to copy some objects if it is internal
        self.aspace = aspace
        self.subscription_service = submgr
        self.name = name
        self.user = user
        self.nonce = None
        self.state = SessionState.Created
        self.session_id = ua.NodeId(self._counter)
        InternalSession._counter += 1
        self.authentication_token = ua.NodeId(self._auth_counter)
        InternalSession._auth_counter += 1
        self.subscriptions = []
        self.logger.info("Created internal session %s", self.name)
        self._lock = Lock()

    def __str__(self):
        return "InternalSession(name:{}, user:{}, id:{}, auth_token:{})".format(self.name, self.user, self.session_id, self.authentication_token)

    def get_endpoints(self, params=None, sockname=None):
        return self.iserver.get_endpoints(params, sockname)

    def create_session(self, params, sockname=None):
        self.logger.info("Create session request")

        result = ua.CreateSessionResult()
        result.SessionId = self.session_id
        result.AuthenticationToken = self.authentication_token
        result.RevisedSessionTimeout = params.RequestedSessionTimeout
        result.MaxRequestMessageSize = 65536
        self.nonce = utils.create_nonce(32)
        result.ServerNonce = self.nonce
        result.ServerEndpoints = self.get_endpoints(sockname=sockname)

        return result

    def close_session(self, delete_subs):
        self.logger.info("close session %s with subscriptions %s", self, self.subscriptions)
        self.state = SessionState.Closed
        self.delete_subscriptions(self.subscriptions[:])

    def activate_session(self, params):
        self.logger.info("activate session")
        result = ua.ActivateSessionResult()
        if self.state != SessionState.Created:
            raise utils.ServiceError(ua.StatusCodes.BadSessionIdInvalid)
        self.nonce = utils.create_nonce(32)
        result.ServerNonce = self.nonce
        for _ in params.ClientSoftwareCertificates:
            result.Results.append(ua.StatusCode())
        self.state = SessionState.Activated
        id_token = params.UserIdentityToken
        if isinstance(id_token, ua.UserNameIdentityToken):
            if self.iserver.allow_remote_admin and id_token.UserName in ("admin", "Admin"):
                self.user = User.Admin
        self.logger.info("Activated internal session %s for user %s", self.name, self.user)
        return result

    def read(self, params):
        results = self.iserver.attribute_service.read(params)
        if self.external:
            return results
        return [deepcopy(dv) for dv in results]

    def history_read(self, params):
        return self.iserver.history_manager.read_history(params)

    def write(self, params):
        if not self.external:
            # If session is internal we need to store a copy og object, not a reference,
            #otherwise users may change it and we will not generate expected events
            params.NodesToWrite = [deepcopy(ntw) for ntw in params.NodesToWrite]
        return self.iserver.attribute_service.write(params, self.user)

    def browse(self, params):
        return self.iserver.view_service.browse(params)

    def translate_browsepaths_to_nodeids(self, params):
        return self.iserver.view_service.translate_browsepaths_to_nodeids(params)

    def add_nodes(self, params):
        return self.iserver.node_mgt_service.add_nodes(params, self.user)

    def delete_nodes(self, params):
        return self.iserver.node_mgt_service.delete_nodes(params, self.user)

    def add_references(self, params):
        return self.iserver.node_mgt_service.add_references(params, self.user)

    def delete_references(self, params):
        return self.iserver.node_mgt_service.delete_references(params, self.user)

    def add_method_callback(self, methodid, callback):
        return self.aspace.add_method_callback(methodid, callback)

    def call(self, params):
        return self.iserver.method_service.call(params)

    def create_subscription(self, params, callback):
        result = self.subscription_service.create_subscription(params, callback)
        with self._lock:
            self.subscriptions.append(result.SubscriptionId)
        return result

    def create_monitored_items(self, params):        
        subscription_result = self.subscription_service.create_monitored_items(params)
        self.iserver.server_callback_dispatcher.dispatch(CallbackType.ItemSubscriptionCreated, ServerItemCallback(params, subscription_result))
        return subscription_result

    def modify_monitored_items(self, params):
        subscription_result = self.subscription_service.modify_monitored_items(params)
        self.iserver.server_callback_dispatcher.dispatch(CallbackType.ItemSubscriptionModified, ServerItemCallback(params, subscription_result))
        return subscription_result
    
    def republish(self, params):
        return self.subscription_service.republish(params)

    def delete_subscriptions(self, ids):
        for i in ids:
            with self._lock:
                if i in self.subscriptions:
                    self.subscriptions.remove(i)
        return self.subscription_service.delete_subscriptions(ids)

    def delete_monitored_items(self, params):
        subscription_result = self.subscription_service.delete_monitored_items(params)
        self.iserver.server_callback_dispatcher.dispatch(CallbackType.ItemSubscriptionDeleted, ServerItemCallback(params, subscription_result))
        return subscription_result

    def publish(self, acks=None):
        if acks is None:
            acks = []
        return self.subscription_service.publish(acks)


1		"""
2		Internal server implementing opcu-ua interface. can be used on server side or to implement binary/https opc-ua servers
3		"""
4
5	1	from datetime import datetime
6	1	from copy import copy, deepcopy
7	1	from datetime import timedelta
8	1	from os import path
9	1	import logging
10	1	from threading import Lock
11	1	from enum import Enum
12	1	try:
13		from urllib.parse import urlparse
14		except ImportError:
15		from urlparse import urlparse
16
17	1
18	1	from opcua import ua
19	1	from opcua.common import utils
20	1	from opcua.common.callback import CallbackType, ServerItemCallback, CallbackDispatcher
21	1	from opcua.common.node import Node
22	1	from opcua.server.history import HistoryManager
23	1	from opcua.server.address_space import AddressSpace
24	1	from opcua.server.address_space import AttributeService
25	1	from opcua.server.address_space import ViewService
26	1	from opcua.server.address_space import NodeManagementService
27	1	from opcua.server.address_space import MethodService
28	1	from opcua.server.subscription_service import SubscriptionService
29	1	from opcua.server.standard_address_space import standard_address_space
30		from opcua.server.users import User
31		from opcua.common import xmlimporter
32	1
33	1
34	1	class SessionState(Enum):
35	1	Created = 0
36		Activated = 1
37		Closed = 2
38	1
39	1
40	1	class ServerDesc(object):
41	1	def __init__(self, serv, cap=None):
42		self.Server = serv
43		self.Capabilities = cap
44	1
45
46	1	class InternalServer(object):
47	1
48	1	def __init__(self, cacheFile = None):
49	1	self.logger = logging.getLogger(__name__)
50	1
51	1	self.server_callback_dispatcher = CallbackDispatcher()
52	1
53		self.endpoints = []
54	1	self._channel_id_counter = 5
55	1	self.allow_remote_admin = True
56	1	self.disabled_clock = False # for debugging we may want to disable clock that writes too much in log
57	1	self._known_servers = {} # used if we are a discovery server
58	1
59		self.aspace = AddressSpace()
60	1	self.attribute_service = AttributeService(self.aspace)
61		self.view_service = ViewService(self.aspace)
62		self.method_service = MethodService(self.aspace)
63		self.node_mgt_service = NodeManagementService(self.aspace)
64
65		if cacheFile and path.isfile(cacheFile):
66		# import address space from shelve
67		self.aspace.load(cacheFile)
68	1	else:
69	1	# import address space from code generated from xml
70	1	standard_address_space.fill_address_space(self.node_mgt_service)
71		# import address space directly from xml, this has preformance impact so disabled
72	1	#importer = xmlimporter.XmlImporter(self.node_mgt_service)
73		#importer.import_xml("/path/to/python-opcua/schemas/Opc.Ua.NodeSet2.xml")
74
75	1	if cacheFile:
76	1	self.aspace.dump(cacheFile)
77	1
78	1	self.loop = utils.ThreadLoop()
79	1	self.asyncio_transports = []
80		self.subscription_service = SubscriptionService(self.loop, self.aspace)
81	1
82		self.history_manager = HistoryManager(self)
83
84	1	# create a session to use on server side
85		self.isession = InternalSession(self, self.aspace, self.subscription_service, "Internal", user=User.Admin)
86		self.current_time_node = Node(self.isession, ua.NodeId(ua.ObjectIds.Server_ServerStatus_CurrentTime))
87	1	uries = ["http://opcfoundation.org/UA/"]
88	1	ns_node = Node(self.isession, ua.NodeId(ua.ObjectIds.Server_NamespaceArray))
89	1	ns_node.set_value(uries)
90	1
91	1
92	1
93	1	def load_address_space(self, path):
94	1	self.aspace.load(path)
95	1
96		def dump_address_space(self, path):
97	1	self.aspace.dump(path)
98	1
99	1	def start(self):
100	1	self.logger.info("starting internal server")
101		for edp in self.endpoints:
102	1	self._known_servers[edp.Server.ApplicationUri] = ServerDesc(edp.Server)
103	1	self.loop.start()
104	1	Node(self.isession, ua.NodeId(ua.ObjectIds.Server_ServerStatus_State)).set_value(0)
105		Node(self.isession, ua.NodeId(ua.ObjectIds.Server_ServerStatus_StartTime)).set_value(datetime.utcnow())
106	1	if not self.disabled_clock:
107	1	self._set_current_time()
108	1
109		def stop(self):
110	1	self.logger.info("stopping internal server")
111	1	self.loop.stop()
112		self.history_manager.stop()
113	1
114	1	def _set_current_time(self):
115	1	self.current_time_node.set_value(datetime.utcnow())
116		self.loop.call_later(1, self._set_current_time)
117	1
118	1	def get_new_channel_id(self):
119	1	self._channel_id_counter += 1
120	1	return self._channel_id_counter
121	1
122	1	def add_endpoint(self, endpoint):
123	1	self.endpoints.append(endpoint)
124	1
125	1	def get_endpoints(self, params=None, sockname=None):
126		self.logger.info("get endpoint")
127	1	if sockname:
128	1	#return to client the ip address it has access to
129	1	edps = []
130	1	for edp in self.endpoints:
131	1	edp1 = copy(edp)
132	1	url = urlparse(edp1.EndpointUrl)
133	1	url = url._replace(netloc=sockname[0] + ":" + str(sockname[1]))
134	1	edp1.EndpointUrl = url.geturl()
135	1	edps.append(edp1)
136	1	return edps
137	1	return self.endpoints[:]
138	1
139		def find_servers(self, params):
140	1	if not params.ServerUris:
141	1	return [desc.Server for desc in self._known_servers.values()]
142	1	servers = []
143	1	for serv in self._known_servers.values():
144	1	serv_uri = serv.Server.ApplicationUri.split(":")
145	1	for uri in params.ServerUris:
146	1	uri = uri.split(":")
147	1	if serv_uri[:len(uri)] == uri:
148	1	servers.append(serv.Server)
149		break
150	1	return servers
151
152		def register_server(self, server, conf=None):
153	1	appdesc = ua.ApplicationDescription()
154	1	appdesc.ApplicationUri = server.ServerUri
155		appdesc.ProductUri = server.ProductUri
156	1	appdesc.ApplicationName = server.ServerNames[0] # FIXME: select name from client locale
157		appdesc.ApplicationType = server.ServerType
158		appdesc.GatewayServerUri = server.GatewayServerUri
159		appdesc.DiscoveryUrls = server.DiscoveryUrls # FIXME: select discovery uri using reachability from client network
160	1	self._known_servers[server.ServerUri] = ServerDesc(appdesc, conf)
161	1
162	1	def register_server2(self, params):
163	1	return self.register_server(params.Server, params.DiscoveryConfiguration)
164
165	1	def create_session(self, name, user=User.Anonymous, external=False):
166		return InternalSession(self, self.aspace, self.subscription_service, name, user=user, external=external)
167
168		def enable_history_data_change(self, node, period=timedelta(days=7), count=0):
169	1	"""
170	1	Set attribute Historizing of node to True and start storing data for history
171	1	"""
172	1	node.set_attribute(ua.AttributeIds.Historizing, ua.DataValue(True))
173		node.set_attr_bit(ua.AttributeIds.AccessLevel, ua.AccessLevel.HistoryRead)
174	1	node.set_attr_bit(ua.AttributeIds.UserAccessLevel, ua.AccessLevel.HistoryRead)
175		self.history_manager.historize_data_change(node, period, count)
176
177		def disable_history_data_change(self, node):
178		"""
179	1	Set attribute Historizing of node to False and stop storing data for history
180	1	"""
181		node.set_attribute(ua.AttributeIds.Historizing, ua.DataValue(False))
182	1	node.unset_attr_bit(ua.AttributeIds.AccessLevel, ua.AccessLevel.HistoryRead)
183		node.unset_attr_bit(ua.AttributeIds.UserAccessLevel, ua.AccessLevel.HistoryRead)
184	1	self.history_manager.dehistorize(node)
185
186	1	def enable_history_event(self, source, period=timedelta(days=7), count=0):
187		"""
188		Set attribute History Read of object events to True and start storing data for history
189		"""
190	1	event_notifier = source.get_event_notifier()
191	1	if ua.EventNotifier.SubscribeToEvents not in event_notifier:
192		raise ua.UaError("Node does not generate events", event_notifier)
193		if ua.EventNotifier.SubscribeToEvents not in event_notifier:
194	1	event_notifier.append(ua.EventNotifier.HistoryRead)
195	1	source.set_event_notifier(event_notifier)
196	1	self.history_manager.historize_event(source, period, count)
197
198	1	def disable_history_event(self, source):
199	1	"""
200	1	Set attribute History Read of node to False and stop storing data for history
201	1	"""
202	1	source.unset_attr_bit(ua.AttributeIds.EventNotifier, ua.EventNotifier.HistoryRead)
203	1	self.history_manager.dehistorize(source)
204	1
205	1	def subscribe_server_callback(self, event, handle):
206	1	"""
207	1	Create a subscription from event to handle
208	1	"""
209	1	self.server_callback_dispatcher.addListener(event, handle)
210	1
211	1	def unsubscribe_server_callback(self, event, handle):
212	1	"""
213	1	Remove a subscription from event to handle
214	1	"""
215		self.server_callback_dispatcher.removeListener(event, handle)
216	1
217
218		class InternalSession(object):
219	1	_counter = 10
220	1	_auth_counter = 1000
221
222	1	def __init__(self, internal_server, aspace, submgr, name, user=User.Anonymous, external=False):
223	1	self.logger = logging.getLogger(__name__)
224		self.iserver = internal_server
225	1	self.external = external # define if session is external, we need to copy some objects if it is internal
226	1	self.aspace = aspace
227	1	self.subscription_service = submgr
228	1	self.name = name
229	1	self.user = user
230	1	self.nonce = None
231	1	self.state = SessionState.Created
232	1	self.session_id = ua.NodeId(self._counter)
233		InternalSession._counter += 1
234	1	self.authentication_token = ua.NodeId(self._auth_counter)
235		InternalSession._auth_counter += 1
236	1	self.subscriptions = []
237	1	self.logger.info("Created internal session %s", self.name)
238	1	self._lock = Lock()
239	1
240		def __str__(self):
241	1	return "InternalSession(name:{}, user:{}, id:{}, auth_token:{})".format(self.name, self.user, self.session_id, self.authentication_token)
242	1
243	1	def get_endpoints(self, params=None, sockname=None):
244	1	return self.iserver.get_endpoints(params, sockname)
245
246	1	def create_session(self, params, sockname=None):
247	1	self.logger.info("Create session request")
248	1
249		result = ua.CreateSessionResult()
250	1	result.SessionId = self.session_id
251	1	result.AuthenticationToken = self.authentication_token
252	1	result.RevisedSessionTimeout = params.RequestedSessionTimeout
253	1	result.MaxRequestMessageSize = 65536
254	1	self.nonce = utils.create_nonce(32)
255	1	result.ServerNonce = self.nonce
256	1	result.ServerEndpoints = self.get_endpoints(sockname=sockname)
257
258	1	return result
259	1
260	1	def close_session(self, delete_subs):
261	1	self.logger.info("close session %s with subscriptions %s", self, self.subscriptions)
262	1	self.state = SessionState.Closed
263		self.delete_subscriptions(self.subscriptions[:])
264	1
265	1	def activate_session(self, params):
266		self.logger.info("activate session")
267	1	result = ua.ActivateSessionResult()
268	1	if self.state != SessionState.Created:
269		raise utils.ServiceError(ua.StatusCodes.BadSessionIdInvalid)
270		self.nonce = utils.create_nonce(32)
271	1	result.ServerNonce = self.nonce
272	1	for _ in params.ClientSoftwareCertificates:
273		result.Results.append(ua.StatusCode())
274	1	self.state = SessionState.Activated
275	1	id_token = params.UserIdentityToken
276		if isinstance(id_token, ua.UserNameIdentityToken):
277	1	if self.iserver.allow_remote_admin and id_token.UserName in ("admin", "Admin"):
278	1	self.user = User.Admin
279		self.logger.info("Activated internal session %s for user %s", self.name, self.user)
280	1	return result
281	1
282		def read(self, params):
283	1	results = self.iserver.attribute_service.read(params)
284	1	if self.external:
285		return results
286	1	return [deepcopy(dv) for dv in results]
287	1
288		def history_read(self, params):
289	1	return self.iserver.history_manager.read_history(params)
290
291		def write(self, params):
292	1	if not self.external:
293	1	# If session is internal we need to store a copy og object, not a reference,
294		#otherwise users may change it and we will not generate expected events
295	1	params.NodesToWrite = [deepcopy(ntw) for ntw in params.NodesToWrite]
296	1	return self.iserver.attribute_service.write(params, self.user)
297
298	1	def browse(self, params):
299	1	return self.iserver.view_service.browse(params)
300	1
301	1	def translate_browsepaths_to_nodeids(self, params):
302	1	return self.iserver.view_service.translate_browsepaths_to_nodeids(params)
303
304	1	def add_nodes(self, params):
305	1	return self.iserver.node_mgt_service.add_nodes(params, self.user)
306
307	1	def delete_nodes(self, params):
308		return self.iserver.node_mgt_service.delete_nodes(params, self.user)
309
310	1	def add_references(self, params):
311		return self.iserver.node_mgt_service.add_references(params, self.user)
312
313	1	def delete_references(self, params):
314	1	return self.iserver.node_mgt_service.delete_references(params, self.user)
315	1
316	1	def add_method_callback(self, methodid, callback):
317	1	return self.aspace.add_method_callback(methodid, callback)
318	1
319		def call(self, params):
320	1	return self.iserver.method_service.call(params)
321	1
322		def create_subscription(self, params, callback):
323	1	result = self.subscription_service.create_subscription(params, callback)
324	1	with self._lock:
325	1	self.subscriptions.append(result.SubscriptionId)
326	1	return result
327
328		def create_monitored_items(self, params):
329		subscription_result = self.subscription_service.create_monitored_items(params)
330		self.iserver.server_callback_dispatcher.dispatch(CallbackType.ItemSubscriptionCreated, ServerItemCallback(params, subscription_result))
331		return subscription_result
332
333		def modify_monitored_items(self, params):
334		subscription_result = self.subscription_service.modify_monitored_items(params)
335		self.iserver.server_callback_dispatcher.dispatch(CallbackType.ItemSubscriptionModified, ServerItemCallback(params, subscription_result))
336		return subscription_result
337
338		def republish(self, params):
339		return self.subscription_service.republish(params)
340
341		def delete_subscriptions(self, ids):
342		for i in ids:
343		with self._lock:
344		if i in self.subscriptions:
345		self.subscriptions.remove(i)
346		return self.subscription_service.delete_subscriptions(ids)
347
348		def delete_monitored_items(self, params):
349		subscription_result = self.subscription_service.delete_monitored_items(params)
350		self.iserver.server_callback_dispatcher.dispatch(CallbackType.ItemSubscriptionDeleted, ServerItemCallback(params, subscription_result))
351		return subscription_result
352
353		def publish(self, acks=None):
354		if acks is None:
355		acks = []
356		return self.subscription_service.publish(acks)
357

FreeOpcUa / python-opcua

Push — master ( 641228...9a97a0 )

InternalServer.subscribe_server_callback() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like