asyncua.sync.Node.get_description() - Code Metrics - Inspection of "Fixes" - FreeOpcUa/opcua-asyncio - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#238)

by Olivier

created 2020-07-20 19:51 UTC

asyncua.sync.Node.get_description() A

↳ Parent: asyncua.sync

Complexity

Conditions

Size

Total Lines	3
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	3
nop	1
dl	0
loc	3
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()

    @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 self.aio_obj == other.aio_obj

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

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

    __repr__ = __str__

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

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

    @syncmethod
    def read_browse_name(self):
        pass

    @syncmethod
    def read_display_name(self):
        pass

    @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_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 read_value(self):
        pass

    get_value = read_value  # 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

    @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			@syncmethod
122			def load_type_definitions(self, nodes=None):
123			pass
124
125			@syncmethod
126			def set_security(self):
127			pass
128
129			@syncmethod
130			def load_enums(self):
131			pass
132
133			def create_subscription(self, period, handler):
134			coro = self.aio_obj.create_subscription(period, _SubHandler(self.tloop, handler))
135			aio_sub = self.tloop.post(coro)
136			return Subscription(self.tloop, aio_sub)
137
138			@syncmethod
139			def get_namespace_index(self, url):
140			pass
141
142			def get_node(self, nodeid):
143			return Node(self.tloop, self.aio_obj.get_node(nodeid))
144
145			@syncmethod
146			def connect_and_get_server_endpoints(self):
147			pass
148
149			def __enter__(self):
150			self.connect()
151			return self
152
153			def __exit__(self, exc_type, exc_value, traceback):
154			self.disconnect()
155
156
157			class Shortcuts:
158			def __init__(self, tloop, aio_server):
159			self.tloop = tloop
160			self.aio_obj = shortcuts.Shortcuts(aio_server)
161			for k, v in self.aio_obj.__dict__.items():
162			setattr(self, k, Node(self.tloop, v))
163
164
165			class Server:
166			def __init__(self, shelf_file=None, tloop=None):
167			self.tloop = tloop
168			self.close_tloop = False
169			if not self.tloop:
170			self.tloop = ThreadLoop()
171			self.tloop.start()
172			self.close_tloop = True
173			self.aio_obj = server.Server(loop=self.tloop.loop)
174			self.tloop.post(self.aio_obj.init(shelf_file))
175			self.nodes = Shortcuts(self.tloop, self.aio_obj.iserver.isession)
176
177			def __str__(self):
178			return "Sync" + self.aio_obj.__str__()
179			__repr__ = __str__
180
181			def __enter__(self):
182			self.start()
183			return self
184
185			def __exit__(self, exc_type, exc_value, traceback):
186			self.stop()
187
188			def set_endpoint(self, url):
189			return self.aio_obj.set_endpoint(url)
190
191			def set_server_name(self, name):
192			return self.aio_obj.set_server_name(name)
193
194			def set_security_policy(self, security_policy):
195			return self.aio_obj.set_security_policy(security_policy)
196
197			def disable_clock(self, boolean):
198			return self.aio_obj.disable_clock(boolean)
199
200			@syncmethod
201			def register_namespace(self, url):
202			return self.aio_obj.register_namespace(url)
203
204			@syncmethod
205			def start(self):
206			pass
207
208			def stop(self):
209			self.tloop.post(self.aio_obj.stop())
210			if self.close_tloop:
211			self.tloop.stop()
212
213			def link_method(self, node, callback):
214			return self.aio_obj.link_method(node, callback)
215
216			@syncmethod
217			def get_event_generator(self, etype=None, emitting_node=ua.ObjectIds.Server):
218			pass
219
220			def get_node(self, nodeid):
221			return Node(self.tloop, self.aio_obj.get_node(nodeid))
222
223			@syncmethod
224			def import_xml(self, path=None, xmlstring=None):
225			pass
226
227			@syncmethod
228			def get_namespace_index(self, url):
229			pass
230
231			@syncmethod
232			def load_enums(self):
233			pass
234
235			@syncmethod
236			def load_type_definitions(self):
237			pass
238
239			def write_attribute_value(self, nodeid, datavalue, attr=ua.AttributeIds.Value):
240			return self.tloop.post(self.aio_obj.write_attribute_value(nodeid, datavalue, attr))
241
242
243			class EventGenerator:
244			def __init__(self, tloop, aio_evgen):
245			self.aio_obj = aio_evgen
246			self.tloop = tloop
247
248			@property
249			def event(self):
250			return self.aio_obj.event
251
252			def trigger(self, time=None, message=None):
253			return self.tloop.post(self.aio_obj.trigger(time, message))
254
255
256			class Node:
257			def __init__(self, tloop, aio_node):
258			self.aio_obj = aio_node
259			self.tloop = tloop
260
261			def __eq__(self, other):
262			return self.aio_obj == other.aio_obj
263
264			def __ne__(self, other):
265			return not self.__eq__(other)
266
267			def __str__(self):
268			return "Sync" + self.aio_obj.__str__()
269
270			__repr__ = __str__
271
272			def __hash__(self):
273			return self.aio_obj.__hash__()
274
275			@property
276			def nodeid(self):
277			return self.aio_obj.nodeid
278
279			@syncmethod
280			def read_browse_name(self):
281			pass
282
283			@syncmethod
284			def read_display_name(self):
285			pass
286
287			@syncmethod
288			def get_children(
289			self, refs=ua.ObjectIds.HierarchicalReferences, nodeclassmask=ua.NodeClass.Unspecified
290			):
291			pass
292
293			@syncmethod
294			def get_properties(self):
295			pass
296
297			@syncmethod
298			def get_children_descriptions(
299			self,
300			refs=ua.ObjectIds.HierarchicalReferences,
301			nodeclassmask=ua.NodeClass.Unspecified,
302			includesubtypes=True,
303			):
304			pass
305
306			@syncmethod
307			def get_child(self, path):
308			pass
309
310			@syncmethod
311			def set_modelling_rule(self, mandatory: bool):
312			pass
313
314			@syncmethod
315			def add_variable(self, ns, name, val):
316			pass
317
318			@syncmethod
319			def add_property(self, ns, name, val):
320			pass
321
322			@syncmethod
323			def add_object(self, ns, name):
324			pass
325
326			@syncmethod
327			def add_object_type(self, ns, name):
328			pass
329
330			@syncmethod
331			def add_folder(self, ns, name):
332			pass
333
334			@syncmethod
335			def add_method(self, *args):
336			pass
337
338			@syncmethod
339			def set_writable(self, writable=True):
340			pass
341
342			@syncmethod
343			def write_value(self, val):
344			pass
345
346			set_value = write_value # legacy
347
348			@syncmethod
349			def read_value(self):
350			pass
351
352			get_value = read_value # legacy
353
354			@syncmethod
355			def call_method(self, methodid, *args):
356			pass
357
358			@syncmethod
359			def get_references(
360			self,
361			refs=ua.ObjectIds.References,
362			direction=ua.BrowseDirection.Both,
363			nodeclassmask=ua.NodeClass.Unspecified,
364			includesubtypes=True,
365			):
366			pass
367
368			@syncmethod
369			def read_description(self):
370			pass
371
372			@syncmethod
373			def get_variables(self):
374			pass
375
376			@syncmethod
377			def get_path(self):
378			pass
379
380			@syncmethod
381			def read_node_class(self):
382			pass
383
384			@syncmethod
385			def read_attributes(self):
386			pass
387
388
389			class Subscription:
390			def __init__(self, tloop, sub):
391			self.tloop = tloop
392			self.aio_obj = sub
393
394			@syncmethod
395			def subscribe_data_change(self, nodes, attr=ua.AttributeIds.Value, queuesize=0):
396			pass
397
398			@syncmethod
399			def subscribe_events(
400			self,
401			sourcenode=ua.ObjectIds.Server,
402			evtypes=ua.ObjectIds.BaseEventType,
403			evfilter=None,
404			queuesize=0,
405			):
406			pass
407
408			@syncmethod
409			def unsubscribe(self, handle):
410			pass
411
412			@syncmethod
413			async def create_monitored_items(self, monitored_items):
414			pass
415
416			@syncmethod
417			def delete(self):
418			pass
419

FreeOpcUa / opcua-asyncio

Pull Request — master (#238)

asyncua.sync.Node.get_description() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like