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

Passed

Pull Request — master (#238)

by Olivier

created 2020-07-25 09:11 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 self.aio_obj.__str__()

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

    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 self.aio_obj.__str__()
269
270			def __repr__(self):
271			return "Sync" + self.aio_obj.__repr__()
272
273			def __hash__(self):
274			return self.aio_obj.__hash__()
275
276			@property
277			def nodeid(self):
278			return self.aio_obj.nodeid
279
280			@syncmethod
281			def read_browse_name(self):
282			pass
283
284			@syncmethod
285			def read_display_name(self):
286			pass
287
288			@syncmethod
289			def get_children(
290			self, refs=ua.ObjectIds.HierarchicalReferences, nodeclassmask=ua.NodeClass.Unspecified
291			):
292			pass
293
294			@syncmethod
295			def get_properties(self):
296			pass
297
298			@syncmethod
299			def get_children_descriptions(
300			self,
301			refs=ua.ObjectIds.HierarchicalReferences,
302			nodeclassmask=ua.NodeClass.Unspecified,
303			includesubtypes=True,
304			):
305			pass
306
307			@syncmethod
308			def get_child(self, path):
309			pass
310
311			@syncmethod
312			def set_modelling_rule(self, mandatory: bool):
313			pass
314
315			@syncmethod
316			def add_variable(self, ns, name, val):
317			pass
318
319			@syncmethod
320			def add_property(self, ns, name, val):
321			pass
322
323			@syncmethod
324			def add_object(self, ns, name):
325			pass
326
327			@syncmethod
328			def add_object_type(self, ns, name):
329			pass
330
331			@syncmethod
332			def add_folder(self, ns, name):
333			pass
334
335			@syncmethod
336			def add_method(self, *args):
337			pass
338
339			@syncmethod
340			def set_writable(self, writable=True):
341			pass
342
343			@syncmethod
344			def write_value(self, val):
345			pass
346
347			set_value = write_value # legacy
348
349			@syncmethod
350			def read_value(self):
351			pass
352
353			get_value = read_value # legacy
354
355			@syncmethod
356			def call_method(self, methodid, *args):
357			pass
358
359			@syncmethod
360			def get_references(
361			self,
362			refs=ua.ObjectIds.References,
363			direction=ua.BrowseDirection.Both,
364			nodeclassmask=ua.NodeClass.Unspecified,
365			includesubtypes=True,
366			):
367			pass
368
369			@syncmethod
370			def read_description(self):
371			pass
372
373			@syncmethod
374			def get_variables(self):
375			pass
376
377			@syncmethod
378			def get_path(self):
379			pass
380
381			@syncmethod
382			def read_node_class(self):
383			pass
384
385			@syncmethod
386			def read_attributes(self):
387			pass
388
389
390			class Subscription:
391			def __init__(self, tloop, sub):
392			self.tloop = tloop
393			self.aio_obj = sub
394
395			@syncmethod
396			def subscribe_data_change(self, nodes, attr=ua.AttributeIds.Value, queuesize=0):
397			pass
398
399			@syncmethod
400			def subscribe_events(
401			self,
402			sourcenode=ua.ObjectIds.Server,
403			evtypes=ua.ObjectIds.BaseEventType,
404			evfilter=None,
405			queuesize=0,
406			):
407			pass
408
409			@syncmethod
410			def unsubscribe(self, handle):
411			pass
412
413			@syncmethod
414			async def create_monitored_items(self, monitored_items):
415			pass
416
417			@syncmethod
418			def delete(self):
419			pass
420

FreeOpcUa / opcua-asyncio

Pull Request — master (#238)

asyncua.sync.Node.get_description() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like