asyncua.sync.stop_thread_loop() - Code Metrics - Inspection of "WIP: add 'with' support in sync API, this is much..." - FreeOpcUa/opcua-asyncio - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#51)

by Olivier

created 2019-05-15 10:54 UTC

asyncua.sync.stop_thread_loop() A

↳ Parent: asyncua.sync

Complexity

Conditions

Size

Total Lines	4
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	4
nop	0
dl	0
loc	4
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 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()
        with self._cond:
            self._cond.notify_all()
        self.loop.run_forever()

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

    def post(self, coro):
        futur = asyncio.run_coroutine_threadsafe(coro, loop=self.loop)
        return futur.result()

    def __enter__(self):
        self.start()
        global _tloop
        _tloop = self
        return self

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

#@ipcmethod

_ref_count = 0
_tloop = None


def start_thread_loop():
    global _tloop
    _tloop = ThreadLoop()
    _tloop.start()
    return _tloop


def stop_thread_loop():
    global _tloop
    _tloop.stop()
    _tloop.join()


def get_thread_loop():
    global _tloop
    if _tloop is None:
        start_thread_loop()
    global _ref_count
    _ref_count += 1
    return _tloop


def release_thread_loop():
    global _tloop
    if _tloop is None:
        return
    global _ref_count
    if _ref_count == 0:
        _ref_count -= 1
    stop_thread_loop()


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
        aio_func = getattr(self.aio_obj, func.__name__)
        global _tloop
        result = _tloop.post(aio_func(*args, **kwargs))
        if isinstance(result, node.Node):
            return Node(result)
        if isinstance(result, list) and len(result) > 0 and isinstance(result[0], node.Node):
            return [Node(i) for i in result]
        if isinstance(result, server.event_generator.EventGenerator):
            return EventGenerator(result)
        if isinstance(result, subscription.Subscription):
            return Subscription(result)
        return result

    return wrapper


class Client:
    def __init__(self, url: str, timeout: int = 4):
        global _tloop
        self.aio_obj = client.Client(url, timeout, loop=_tloop.loop)
        self.nodes = Shortcuts(self.aio_obj.uaclient)

    @syncmethod
    def connect(self):
        pass

    @syncmethod
    def disconnect(self):
        pass

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

    @syncmethod
    async def create_subscription(self, period, handler):
        pass

    @syncmethod
    def get_namespace_index(self, url):
        pass

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

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

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


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


class Server:
    def __init__(self, shelf_file=None):
        global _tloop
        self.aio_obj = server.Server(loop=_tloop.loop)
        _tloop.post(self.aio_obj.init(shelf_file))
        self.nodes = Shortcuts(self.aio_obj.iserver.isession)

    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

    @syncmethod
    def stop(self):
        pass

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

    def get_node(self, nodeid):
        return Node(server.Server.get_node(self, nodeid))

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

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


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

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

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


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

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

    @syncmethod
    def get_browse_name(self):
        pass

    @syncmethod
    def get_children(self, refs=ua.ObjectIds.HierarchicalReferences, nodeclassmask=ua.NodeClass.Unspecified):
        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 set_value(self, val):
        pass

    @syncmethod
    def get_value(self, val):
        pass

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

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


class Subscription:
    def __init__(self, sub):
        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 ThreadLoop(Thread):
18			def __init__(self):
19			Thread.__init__(self)
20			self.loop = None
21			self._cond = Condition()
22
23			def start(self):
24			with self._cond:
25			Thread.start(self)
26			self._cond.wait()
27
28			def run(self):
29			self.loop = asyncio.new_event_loop()
30			with self._cond:
31			self._cond.notify_all()
32			self.loop.run_forever()
33
34			def stop(self):
35			self.loop.call_soon_threadsafe(self.loop.stop)
36
37			def post(self, coro):
38			futur = asyncio.run_coroutine_threadsafe(coro, loop=self.loop)
39			return futur.result()
40
41			def __enter__(self):
42			self.start()
43			global _tloop
44			_tloop = self
45			return self
46
47			def __exit__(self, exc_t, exc_v, trace):
48			self.stop()
49			self.join()
50
51			#@ipcmethod
52
53			_ref_count = 0
54			_tloop = None
55
56
57			def start_thread_loop():
58			global _tloop
59			_tloop = ThreadLoop()
60			_tloop.start()
61			return _tloop
62
63
64			def stop_thread_loop():
65			global _tloop
66			_tloop.stop()
67			_tloop.join()
68
69
70			def get_thread_loop():
71			global _tloop
72			if _tloop is None:
73			start_thread_loop()
74			global _ref_count
75			_ref_count += 1
76			return _tloop
77
78
79			def release_thread_loop():
80			global _tloop
81			if _tloop is None:
82			return
83			global _ref_count
84			if _ref_count == 0:
85			_ref_count -= 1
86			stop_thread_loop()
87
88
89			def syncmethod(func):
90			def wrapper(self, args, *kwargs):
91			args = list(args) #FIXME: might be very inefficient...
92			for idx, arg in enumerate(args):
93			if isinstance(arg, Node):
94			args[idx] = arg.aio_obj
95			aio_func = getattr(self.aio_obj, func.__name__)
96			global _tloop
97			result = _tloop.post(aio_func(args, *kwargs))
98			if isinstance(result, node.Node):
99			return Node(result)
100			if isinstance(result, list) and len(result) > 0 and isinstance(result[0], node.Node):
101			return [Node(i) for i in result]
102			if isinstance(result, server.event_generator.EventGenerator):
103			return EventGenerator(result)
104			if isinstance(result, subscription.Subscription):
105			return Subscription(result)
106			return result
107
108			return wrapper
109
110
111			class Client:
112			def __init__(self, url: str, timeout: int = 4):
113			global _tloop
114			self.aio_obj = client.Client(url, timeout, loop=_tloop.loop)
115			self.nodes = Shortcuts(self.aio_obj.uaclient)
116
117			@syncmethod
118			def connect(self):
119			pass
120
121			@syncmethod
122			def disconnect(self):
123			pass
124
125			@syncmethod
126			def load_type_definitions(self, nodes=None):
127			pass
128
129			@syncmethod
130			async def create_subscription(self, period, handler):
131			pass
132
133			@syncmethod
134			def get_namespace_index(self, url):
135			pass
136
137			def get_node(self, nodeid):
138			return Node(self.aio_obj.get_node(nodeid))
139
140			def __enter__(self):
141			self.connect()
142			return self
143
144			def __exit__(self, exc_type, exc_value, traceback):
145			self.disconnect()
146
147
148			class Shortcuts:
149			def __init__(self, aio_server):
150			self.aio_obj = shortcuts.Shortcuts(aio_server)
151			for k, v in self.aio_obj.__dict__.items():
152			setattr(self, k, Node(v))
153
154
155			class Server:
156			def __init__(self, shelf_file=None):
157			global _tloop
158			self.aio_obj = server.Server(loop=_tloop.loop)
159			_tloop.post(self.aio_obj.init(shelf_file))
160			self.nodes = Shortcuts(self.aio_obj.iserver.isession)
161
162			def __enter__(self):
163			self.start()
164			return self
165
166			def __exit__(self, exc_type, exc_value, traceback):
167			self.stop()
168
169			def set_endpoint(self, url):
170			return self.aio_obj.set_endpoint(url)
171
172			def set_server_name(self, name):
173			return self.aio_obj.set_server_name(name)
174
175			def set_security_policy(self, security_policy):
176			return self.aio_obj.set_security_policy(security_policy)
177
178			def disable_clock(self, boolean):
179			return self.aio_obj.disable_clock(boolean)
180
181			@syncmethod
182			def register_namespace(self, url):
183			return self.aio_obj.register_namespace(url)
184
185			@syncmethod
186			def start(self):
187			pass
188
189			@syncmethod
190			def stop(self):
191			pass
192
193			@syncmethod
194			async def get_event_generator(self, etype=None, emitting_node=ua.ObjectIds.Server):
195			pass
196
197			def get_node(self, nodeid):
198			return Node(server.Server.get_node(self, nodeid))
199
200			@syncmethod
201			def import_xml(self, path=None, xmlstring=None):
202			pass
203
204			def set_attribute_value(self, nodeid, datavalue, attr=ua.AttributeIds.Value):
205			return self.aio_obj.set_attribute_value(nodeid, datavalue, attr)
206
207
208			class EventGenerator:
209			def __init__(self, aio_evgen):
210			self.aio_obj = aio_evgen
211
212			@property
213			def event(self):
214			return self.aio_obj.event
215
216			def trigger(self, time=None, message=None):
217			return self.aio_obj.trigger(time, message)
218
219
220			class Node:
221			def __init__(self, aio_node):
222			self.aio_obj = aio_node
223			global _tloop
224
225			@property
226			def nodeid(self):
227			return self.aio_obj.nodeid
228
229			@syncmethod
230			def get_browse_name(self):
231			pass
232
233			@syncmethod
234			def get_children(self, refs=ua.ObjectIds.HierarchicalReferences, nodeclassmask=ua.NodeClass.Unspecified):
235			pass
236
237			@syncmethod
238			def get_children_descriptions(self,
239			refs=ua.ObjectIds.HierarchicalReferences,
240			nodeclassmask=ua.NodeClass.Unspecified,
241			includesubtypes=True):
242			pass
243
244			@syncmethod
245			def get_child(self, path):
246			pass
247
248			@syncmethod
249			def set_modelling_rule(self, mandatory: bool):
250			pass
251
252			@syncmethod
253			def add_variable(self, ns, name, val):
254			pass
255
256			@syncmethod
257			def add_property(self, ns, name, val):
258			pass
259
260			@syncmethod
261			def add_object(self, ns, name):
262			pass
263
264			@syncmethod
265			def add_object_type(self, ns, name):
266			pass
267
268			@syncmethod
269			def add_folder(self, ns, name):
270			pass
271
272			@syncmethod
273			def add_method(self, *args):
274			pass
275
276			@syncmethod
277			def set_writable(self, writable=True):
278			pass
279
280			@syncmethod
281			def set_value(self, val):
282			pass
283
284			@syncmethod
285			def get_value(self, val):
286			pass
287
288			@syncmethod
289			def call_method(self, methodid, *args):
290			pass
291
292			def __eq__(self, other):
293			return self.aio_obj == other.aio_obj
294
295
296			class Subscription:
297			def __init__(self, sub):
298			self.aio_obj = sub
299
300			@syncmethod
301			def subscribe_data_change(self, nodes, attr=ua.AttributeIds.Value, queuesize=0):
302			pass
303
304			@syncmethod
305			def subscribe_events(self,
306			sourcenode=ua.ObjectIds.Server,
307			evtypes=ua.ObjectIds.BaseEventType,
308			evfilter=None,
309			queuesize=0):
310			pass
311
312			@syncmethod
313			def unsubscribe(self, handle):
314			pass
315
316			@syncmethod
317			async def create_monitored_items(self, monitored_items):
318			pass
319
320			@syncmethod
321			def delete(self):
322			pass
323

FreeOpcUa / opcua-asyncio

Pull Request — master (#51)

asyncua.sync.stop_thread_loop() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like