asyncua.sync.Node.__set_nodeid() - Code Metrics - Inspection of "Extended sync wrapper to increase compatibility re..." - FreeOpcUa/opcua-asyncio - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( f08c39...c2cd4c )

by Olivier

created 2020-08-11 11:55 UTC

asyncua.sync.Node.__set_nodeid() A

↳ Parent: asyncua.sync

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	2
dl	0
loc	2
rs	10
c	0
b	0
f	0

"""
sync API of asyncua
"""
import asyncio
from threading import Thread, Condition
import logging

from asyncua import ua
from asyncua import client
from asyncua import server
from asyncua.common import node
from asyncua.common import subscription, shortcuts

logger = logging.getLogger(__name__)


class ThreadLoopNotRunning(Exception):
    pass


class ThreadLoop(Thread):
    def __init__(self):
        Thread.__init__(self)
        self.loop = None
        self._cond = Condition()

    def start(self):
        with self._cond:
            Thread.start(self)
            self._cond.wait()

    def run(self):
        self.loop = asyncio.new_event_loop()
        logger.debug("Threadloop: %s", self.loop)
        self.loop.call_soon_threadsafe(self._notify_start)
        self.loop.run_forever()

    def _notify_start(self):
        with self._cond:
            self._cond.notify_all()

    def stop(self):
        self.loop.call_soon_threadsafe(self.loop.stop)
        self.join()
        self.loop.close()

    def post(self, coro):
        if not self.loop or not self.loop.is_running():
            raise ThreadLoopNotRunning(f"could not post {coro}")
        futur = asyncio.run_coroutine_threadsafe(coro, loop=self.loop)
        return futur.result()

    def __enter__(self):
        self.start()
        return self

    def __exit__(self, exc_t, exc_v, trace):
        self.stop()


def syncmethod(func):
    def wrapper(self, *args, **kwargs):
        args = list(args)  # FIXME: might be very inefficient...
        for idx, arg in enumerate(args):
            if isinstance(arg, Node):
                args[idx] = arg.aio_obj
        for k, v in kwargs.items():
            if isinstance(v, Node):
                kwargs[k] = v.aio_obj
        aio_func = getattr(self.aio_obj, func.__name__)
        result = self.tloop.post(aio_func(*args, **kwargs))
        if isinstance(result, node.Node):
            return Node(self.tloop, result)
        if isinstance(result, list) and len(result) > 0 and isinstance(result[0], node.Node):
            return [Node(self.tloop, i) for i in result]
        if isinstance(result, server.event_generator.EventGenerator):
            return EventGenerator(self.tloop, result)
        if isinstance(result, subscription.Subscription):
            return Subscription(self.tloop, result)
        return result

    return wrapper


class _SubHandler:
    def __init__(self, tloop, sync_handler):
        self.tloop = tloop
        self.sync_handler = sync_handler

    def datachange_notification(self, node, val, data):
        self.sync_handler.datachange_notification(Node(self.tloop, node), val, data)

    def event_notification(self, event):
        self.sync_handler.event_notification(event)


class Client:
    def __init__(self, url: str, timeout: int = 4, tloop=None):
        self.tloop = tloop
        self.close_tloop = False
        if not self.tloop:
            self.tloop = ThreadLoop()
            self.tloop.start()
            self.close_tloop = True
        self.aio_obj = client.Client(url, timeout, loop=self.tloop.loop)
        self.nodes = Shortcuts(self.tloop, self.aio_obj.uaclient)

    def __str__(self):
        return "Sync" + self.aio_obj.__str__()
    __repr__ = __str__

    @syncmethod
    def connect(self):
        pass

    def disconnect(self):
        self.tloop.post(self.aio_obj.disconnect())
        if self.close_tloop:
            self.tloop.stop()

    def set_user(self, username: str):
        self.aio_obj.set_user(username)

    def set_password(self, pwd: str):
        self.aio_obj.set_password(pwd)

    @syncmethod
    def load_type_definitions(self, nodes=None):
        pass

    @syncmethod
    def set_security(self):
        pass

    @syncmethod
    def load_enums(self):
        pass

    def create_subscription(self, period, handler):
        coro = self.aio_obj.create_subscription(period, _SubHandler(self.tloop, handler))
        aio_sub = self.tloop.post(coro)
        return Subscription(self.tloop, aio_sub)

    @syncmethod
    def get_namespace_index(self, url):
        pass

    def get_node(self, nodeid):
        return Node(self.tloop, self.aio_obj.get_node(nodeid))

    @syncmethod
    def connect_and_get_server_endpoints(self):
        pass

    def __enter__(self):
        self.connect()
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.disconnect()


class Shortcuts:
    def __init__(self, tloop, aio_server):
        self.tloop = tloop
        self.aio_obj = shortcuts.Shortcuts(aio_server)
        for k, v in self.aio_obj.__dict__.items():
            setattr(self, k, Node(self.tloop, v))


class Server:
    def __init__(self, shelf_file=None, tloop=None):
        self.tloop = tloop
        self.close_tloop = False
        if not self.tloop:
            self.tloop = ThreadLoop()
            self.tloop.start()
            self.close_tloop = True
        self.aio_obj = server.Server(loop=self.tloop.loop)
        self.tloop.post(self.aio_obj.init(shelf_file))
        self.nodes = Shortcuts(self.tloop, self.aio_obj.iserver.isession)

    def __str__(self):
        return "Sync" + self.aio_obj.__str__()
    __repr__ = __str__

    def __enter__(self):
        self.start()
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.stop()

    def set_endpoint(self, url):
        return self.aio_obj.set_endpoint(url)

    def set_server_name(self, name):
        return self.aio_obj.set_server_name(name)

    def set_security_policy(self, security_policy):
        return self.aio_obj.set_security_policy(security_policy)

    def disable_clock(self, boolean):
        return self.aio_obj.disable_clock(boolean)

    @syncmethod
    def register_namespace(self, url):
        return self.aio_obj.register_namespace(url)

    @syncmethod
    def start(self):
        pass

    def stop(self):
        self.tloop.post(self.aio_obj.stop())
        if self.close_tloop:
            self.tloop.stop()

    def link_method(self, node, callback):
        return self.aio_obj.link_method(node, callback)

    @syncmethod
    def get_event_generator(self, etype=None, emitting_node=ua.ObjectIds.Server):
        pass

    def get_node(self, nodeid):
        return Node(self.tloop, self.aio_obj.get_node(nodeid))

    @syncmethod
    def import_xml(self, path=None, xmlstring=None):
        pass

    @syncmethod
    def get_namespace_index(self, url):
        pass

    @syncmethod
    def load_enums(self):
        pass

    @syncmethod
    def load_type_definitions(self):
        pass

    def write_attribute_value(self, nodeid, datavalue, attr=ua.AttributeIds.Value):
        return self.tloop.post(self.aio_obj.write_attribute_value(nodeid, datavalue, attr))


class EventGenerator:
    def __init__(self, tloop, aio_evgen):
        self.aio_obj = aio_evgen
        self.tloop = tloop

    @property
    def event(self):
        return self.aio_obj.event

    def trigger(self, time=None, message=None):
        return self.tloop.post(self.aio_obj.trigger(time, message))


class Node:
    def __init__(self, tloop, aio_node):
        self.aio_obj = aio_node
        self.tloop = tloop

    def __eq__(self, other):
        return other != None and self.aio_obj == other.aio_obj

    def __ne__(self, other):
        return not self.__eq__(other)

    def __str__(self):
        return self.aio_obj.__str__()

    def __repr__(self):
        return "Sync" + self.aio_obj.__repr__()

    def __hash__(self):
        return self.aio_obj.__hash__()

    def __get_nodeid(self):
        return self.aio_obj.nodeid

    def __set_nodeid(self, value):
        self.aio_obj.nodeid = value

    nodeid = property(__get_nodeid, __set_nodeid)

    @syncmethod
    def read_browse_name(self):
        pass

    @syncmethod
    def read_display_name(self):
        pass

    get_display_name = read_display_name  # legacy

    @syncmethod
    def get_children(
        self, refs=ua.ObjectIds.HierarchicalReferences, nodeclassmask=ua.NodeClass.Unspecified
    ):
        pass

    @syncmethod
    def get_properties(self):
        pass

    @syncmethod
    def get_children_descriptions(
        self,
        refs=ua.ObjectIds.HierarchicalReferences,
        nodeclassmask=ua.NodeClass.Unspecified,
        includesubtypes=True,
    ):
        pass

    @syncmethod
    def get_user_access_level(self):
        pass

    @syncmethod
    def get_child(self, path):
        pass

    @syncmethod
    def set_modelling_rule(self, mandatory: bool):
        pass

    @syncmethod
    def add_variable(self, ns, name, val):
        pass

    @syncmethod
    def add_property(self, ns, name, val):
        pass

    @syncmethod
    def add_object(self, ns, name):
        pass

    @syncmethod
    def add_object_type(self, ns, name):
        pass

    @syncmethod
    def add_folder(self, ns, name):
        pass

    @syncmethod
    def add_method(self, *args):
        pass

    @syncmethod
    def set_writable(self, writable=True):
        pass

    @syncmethod
    def write_value(self, val):
        pass

    set_value = write_value  # legacy

    @syncmethod
    def write_params(self, params):
        pass

    @syncmethod
    def read_params(self, params):
        pass

    @syncmethod
    def read_value(self):
        pass

    get_value = read_value  # legacy

    @syncmethod
    def read_data_type_as_variant_type(self):
        pass

    get_data_type_as_variant_type = read_data_type_as_variant_type #legacy

    @syncmethod
    def call_method(self, methodid, *args):
        pass

    @syncmethod
    def get_references(
        self,
        refs=ua.ObjectIds.References,
        direction=ua.BrowseDirection.Both,
        nodeclassmask=ua.NodeClass.Unspecified,
        includesubtypes=True,
    ):
        pass

    @syncmethod
    def read_description(self):
        pass

    @syncmethod
    def get_variables(self):
        pass

    @syncmethod
    def get_path(self):
        pass

    @syncmethod
    def read_node_class(self):
        pass

    @syncmethod
    def read_attributes(self):
        pass


class Subscription:
    def __init__(self, tloop, sub):
        self.tloop = tloop
        self.aio_obj = sub

    @syncmethod
    def subscribe_data_change(self, nodes, attr=ua.AttributeIds.Value, queuesize=0):
        pass

    @syncmethod
    def subscribe_events(
        self,
        sourcenode=ua.ObjectIds.Server,
        evtypes=ua.ObjectIds.BaseEventType,
        evfilter=None,
        queuesize=0,
    ):
        pass

    def _make_monitored_item_request(self, node: Node, attr, mfilter, queuesize) -> ua.MonitoredItemCreateRequest:
        return self.aio_obj._make_monitored_item_request(node, attr, mfilter, queuesize)

    @syncmethod
    def unsubscribe(self, handle):
        pass

    @syncmethod
    async def create_monitored_items(self, monitored_items):
        pass

    @syncmethod
    def delete(self):
        pass


1			"""
2			sync API of asyncua
3			"""
4			import asyncio
5			from threading import Thread, Condition
6			import logging
7
8			from asyncua import ua
9			from asyncua import client
10			from asyncua import server
11			from asyncua.common import node
12			from asyncua.common import subscription, shortcuts
13
14			logger = logging.getLogger(__name__)
15
16
17			class ThreadLoopNotRunning(Exception):
18			pass
19
20
21			class ThreadLoop(Thread):
22			def __init__(self):
23			Thread.__init__(self)
24			self.loop = None
25			self._cond = Condition()
26
27			def start(self):
28			with self._cond:
29			Thread.start(self)
30			self._cond.wait()
31
32			def run(self):
33			self.loop = asyncio.new_event_loop()
34			logger.debug("Threadloop: %s", self.loop)
35			self.loop.call_soon_threadsafe(self._notify_start)
36			self.loop.run_forever()
37
38			def _notify_start(self):
39			with self._cond:
40			self._cond.notify_all()
41
42			def stop(self):
43			self.loop.call_soon_threadsafe(self.loop.stop)
44			self.join()
45			self.loop.close()
46
47			def post(self, coro):
48			if not self.loop or not self.loop.is_running():
49			raise ThreadLoopNotRunning(f"could not post {coro}")
50			futur = asyncio.run_coroutine_threadsafe(coro, loop=self.loop)
51			return futur.result()
52
53			def __enter__(self):
54			self.start()
55			return self
56
57			def __exit__(self, exc_t, exc_v, trace):
58			self.stop()
59
60
61			def syncmethod(func):
62			def wrapper(self, args, *kwargs):
63			args = list(args) # FIXME: might be very inefficient...
64			for idx, arg in enumerate(args):
65			if isinstance(arg, Node):
66			args[idx] = arg.aio_obj
67			for k, v in kwargs.items():
68			if isinstance(v, Node):
69			kwargs[k] = v.aio_obj
70			aio_func = getattr(self.aio_obj, func.__name__)
71			result = self.tloop.post(aio_func(args, *kwargs))
72			if isinstance(result, node.Node):
73			return Node(self.tloop, result)
74			if isinstance(result, list) and len(result) > 0 and isinstance(result[0], node.Node):
75			return [Node(self.tloop, i) for i in result]
76			if isinstance(result, server.event_generator.EventGenerator):
77			return EventGenerator(self.tloop, result)
78			if isinstance(result, subscription.Subscription):
79			return Subscription(self.tloop, result)
80			return result
81
82			return wrapper
83
84
85			class _SubHandler:
86			def __init__(self, tloop, sync_handler):
87			self.tloop = tloop
88			self.sync_handler = sync_handler
89
90			def datachange_notification(self, node, val, data):
91			self.sync_handler.datachange_notification(Node(self.tloop, node), val, data)
92
93			def event_notification(self, event):
94			self.sync_handler.event_notification(event)
95
96
97			class Client:
98			def __init__(self, url: str, timeout: int = 4, tloop=None):
99			self.tloop = tloop
100			self.close_tloop = False
101			if not self.tloop:
102			self.tloop = ThreadLoop()
103			self.tloop.start()
104			self.close_tloop = True
105			self.aio_obj = client.Client(url, timeout, loop=self.tloop.loop)
106			self.nodes = Shortcuts(self.tloop, self.aio_obj.uaclient)
107
108			def __str__(self):
109			return "Sync" + self.aio_obj.__str__()
110			__repr__ = __str__
111
112			@syncmethod
113			def connect(self):
114			pass
115
116			def disconnect(self):
117			self.tloop.post(self.aio_obj.disconnect())
118			if self.close_tloop:
119			self.tloop.stop()
120
121			def set_user(self, username: str):
122			self.aio_obj.set_user(username)
123
124			def set_password(self, pwd: str):
125			self.aio_obj.set_password(pwd)
126
127			@syncmethod
128			def load_type_definitions(self, nodes=None):
129			pass
130
131			@syncmethod
132			def set_security(self):
133			pass
134
135			@syncmethod
136			def load_enums(self):
137			pass
138
139			def create_subscription(self, period, handler):
140			coro = self.aio_obj.create_subscription(period, _SubHandler(self.tloop, handler))
141			aio_sub = self.tloop.post(coro)
142			return Subscription(self.tloop, aio_sub)
143
144			@syncmethod
145			def get_namespace_index(self, url):
146			pass
147
148			def get_node(self, nodeid):
149			return Node(self.tloop, self.aio_obj.get_node(nodeid))
150
151			@syncmethod
152			def connect_and_get_server_endpoints(self):
153			pass
154
155			def __enter__(self):
156			self.connect()
157			return self
158
159			def __exit__(self, exc_type, exc_value, traceback):
160			self.disconnect()
161
162
163			class Shortcuts:
164			def __init__(self, tloop, aio_server):
165			self.tloop = tloop
166			self.aio_obj = shortcuts.Shortcuts(aio_server)
167			for k, v in self.aio_obj.__dict__.items():
168			setattr(self, k, Node(self.tloop, v))
169
170
171			class Server:
172			def __init__(self, shelf_file=None, tloop=None):
173			self.tloop = tloop
174			self.close_tloop = False
175			if not self.tloop:
176			self.tloop = ThreadLoop()
177			self.tloop.start()
178			self.close_tloop = True
179			self.aio_obj = server.Server(loop=self.tloop.loop)
180			self.tloop.post(self.aio_obj.init(shelf_file))
181			self.nodes = Shortcuts(self.tloop, self.aio_obj.iserver.isession)
182
183			def __str__(self):
184			return "Sync" + self.aio_obj.__str__()
185			__repr__ = __str__
186
187			def __enter__(self):
188			self.start()
189			return self
190
191			def __exit__(self, exc_type, exc_value, traceback):
192			self.stop()
193
194			def set_endpoint(self, url):
195			return self.aio_obj.set_endpoint(url)
196
197			def set_server_name(self, name):
198			return self.aio_obj.set_server_name(name)
199
200			def set_security_policy(self, security_policy):
201			return self.aio_obj.set_security_policy(security_policy)
202
203			def disable_clock(self, boolean):
204			return self.aio_obj.disable_clock(boolean)
205
206			@syncmethod
207			def register_namespace(self, url):
208			return self.aio_obj.register_namespace(url)
209
210			@syncmethod
211			def start(self):
212			pass
213
214			def stop(self):
215			self.tloop.post(self.aio_obj.stop())
216			if self.close_tloop:
217			self.tloop.stop()
218
219			def link_method(self, node, callback):
220			return self.aio_obj.link_method(node, callback)
221
222			@syncmethod
223			def get_event_generator(self, etype=None, emitting_node=ua.ObjectIds.Server):
224			pass
225
226			def get_node(self, nodeid):
227			return Node(self.tloop, self.aio_obj.get_node(nodeid))
228
229			@syncmethod
230			def import_xml(self, path=None, xmlstring=None):
231			pass
232
233			@syncmethod
234			def get_namespace_index(self, url):
235			pass
236
237			@syncmethod
238			def load_enums(self):
239			pass
240
241			@syncmethod
242			def load_type_definitions(self):
243			pass
244
245			def write_attribute_value(self, nodeid, datavalue, attr=ua.AttributeIds.Value):
246			return self.tloop.post(self.aio_obj.write_attribute_value(nodeid, datavalue, attr))
247
248
249			class EventGenerator:
250			def __init__(self, tloop, aio_evgen):
251			self.aio_obj = aio_evgen
252			self.tloop = tloop
253
254			@property
255			def event(self):
256			return self.aio_obj.event
257
258			def trigger(self, time=None, message=None):
259			return self.tloop.post(self.aio_obj.trigger(time, message))
260
261
262			class Node:
263			def __init__(self, tloop, aio_node):
264			self.aio_obj = aio_node
265			self.tloop = tloop
266
267			def __eq__(self, other):
268			return other != None and self.aio_obj == other.aio_obj
269
270			def __ne__(self, other):
271			return not self.__eq__(other)
272
273			def __str__(self):
274			return self.aio_obj.__str__()
275
276			def __repr__(self):
277			return "Sync" + self.aio_obj.__repr__()
278
279			def __hash__(self):
280			return self.aio_obj.__hash__()
281
282			def __get_nodeid(self):
283			return self.aio_obj.nodeid
284
285			def __set_nodeid(self, value):
286			self.aio_obj.nodeid = value
287
288			nodeid = property(__get_nodeid, __set_nodeid)
289
290			@syncmethod
291			def read_browse_name(self):
292			pass
293
294			@syncmethod
295			def read_display_name(self):
296			pass
297
298			get_display_name = read_display_name # legacy
299
300			@syncmethod
301			def get_children(
302			self, refs=ua.ObjectIds.HierarchicalReferences, nodeclassmask=ua.NodeClass.Unspecified
303			):
304			pass
305
306			@syncmethod
307			def get_properties(self):
308			pass
309
310			@syncmethod
311			def get_children_descriptions(
312			self,
313			refs=ua.ObjectIds.HierarchicalReferences,
314			nodeclassmask=ua.NodeClass.Unspecified,
315			includesubtypes=True,
316			):
317			pass
318
319			@syncmethod
320			def get_user_access_level(self):
321			pass
322
323			@syncmethod
324			def get_child(self, path):
325			pass
326
327			@syncmethod
328			def set_modelling_rule(self, mandatory: bool):
329			pass
330
331			@syncmethod
332			def add_variable(self, ns, name, val):
333			pass
334
335			@syncmethod
336			def add_property(self, ns, name, val):
337			pass
338
339			@syncmethod
340			def add_object(self, ns, name):
341			pass
342
343			@syncmethod
344			def add_object_type(self, ns, name):
345			pass
346
347			@syncmethod
348			def add_folder(self, ns, name):
349			pass
350
351			@syncmethod
352			def add_method(self, *args):
353			pass
354
355			@syncmethod
356			def set_writable(self, writable=True):
357			pass
358
359			@syncmethod
360			def write_value(self, val):
361			pass
362
363			set_value = write_value # legacy
364
365			@syncmethod
366			def write_params(self, params):
367			pass
368
369			@syncmethod
370			def read_params(self, params):
371			pass
372
373			@syncmethod
374			def read_value(self):
375			pass
376
377			get_value = read_value # legacy
378
379			@syncmethod
380			def read_data_type_as_variant_type(self):
381			pass
382
383			get_data_type_as_variant_type = read_data_type_as_variant_type #legacy
384
385			@syncmethod
386			def call_method(self, methodid, *args):
387			pass
388
389			@syncmethod
390			def get_references(
391			self,
392			refs=ua.ObjectIds.References,
393			direction=ua.BrowseDirection.Both,
394			nodeclassmask=ua.NodeClass.Unspecified,
395			includesubtypes=True,
396			):
397			pass
398
399			@syncmethod
400			def read_description(self):
401			pass
402
403			@syncmethod
404			def get_variables(self):
405			pass
406
407			@syncmethod
408			def get_path(self):
409			pass
410
411			@syncmethod
412			def read_node_class(self):
413			pass
414
415			@syncmethod
416			def read_attributes(self):
417			pass
418
419
420			class Subscription:
421			def __init__(self, tloop, sub):
422			self.tloop = tloop
423			self.aio_obj = sub
424
425			@syncmethod
426			def subscribe_data_change(self, nodes, attr=ua.AttributeIds.Value, queuesize=0):
427			pass
428
429			@syncmethod
430			def subscribe_events(
431			self,
432			sourcenode=ua.ObjectIds.Server,
433			evtypes=ua.ObjectIds.BaseEventType,
434			evfilter=None,
435			queuesize=0,
436			):
437			pass
438
439			def _make_monitored_item_request(self, node: Node, attr, mfilter, queuesize) -> ua.MonitoredItemCreateRequest:
440			return self.aio_obj._make_monitored_item_request(node, attr, mfilter, queuesize)
441
442			@syncmethod
443			def unsubscribe(self, handle):
444			pass
445
446			@syncmethod
447			async def create_monitored_items(self, monitored_items):
448			pass
449
450			@syncmethod
451			def delete(self):
452			pass
453

FreeOpcUa / opcua-asyncio

Push — master ( f08c39...c2cd4c )

asyncua.sync.Node.__set_nodeid() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like