/usr/lib/python2.7/dist-packages/tango/asyncio.py is in python-tango 9.2.0-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 | # -----------------------------------------------------------------------------
# This file is part of PyTango (http://pytango.rtfd.io)
#
# Copyright 2006-2012 CELLS / ALBA Synchrotron, Bellaterra, Spain
# Copyright 2013-2014 European Synchrotron Radiation Facility, Grenoble, France
#
# Distributed under the terms of the GNU Lesser General Public License,
# either version 3 of the License, or (at your option) any later version.
# See LICENSE.txt for more info.
# -----------------------------------------------------------------------------
"""This module exposes a asyncio version of :class:`PyTango.DeviceProxy` and
:class:`PyTango.AttributeProxy"""
from __future__ import absolute_import
__all__ = ["DeviceProxy", "AttributeProxy", "check_requirements",
"get_event_loop"]
from functools import partial
from PyTango import GreenMode
from PyTango.device_proxy import get_device_proxy
from PyTango.attribute_proxy import get_attribute_proxy
from PyTango.tango_asyncio import get_event_loop
def check_requirements():
try:
import asyncio
except ImportError:
try:
import trollius as asyncio
except ImportError:
raise ImportError(
"Not able to import asyncio or its backport trollius")
try:
import concurrent.futures
except ImportError:
raise ImportError(
"No module named concurrent. You need to "
"install the futures backport module to have "
"access to PyTango futures green mode")
return asyncio, concurrent.futures
check_requirements()
DeviceProxy = partial(get_device_proxy,
green_mode=GreenMode.Asyncio,
wait=False)
DeviceProxy.__doc__ = """
DeviceProxy(self, dev_name, wait=False, timeout=None)
-> DeviceProxy
DeviceProxy(self, dev_name, need_check_acc, wait=False, timeout=None)
-> DeviceProxy
Creates a *asyncio* enabled :class:`~PyTango.DeviceProxy`.
The DeviceProxy constructor internally makes some network calls which makes
it *slow*. By using the asyncio *green mode* you may give the control back
to the asyncio event loop using the *yield from* or *await* synthax.
.. note::
The timeout parameter has no relation with the tango device client side
timeout (gettable by :meth:`~PyTango.DeviceProxy.get_timeout_millis`
and settable through :meth:`~PyTango.DeviceProxy.set_timeout_millis`)
:param dev_name: the device name or alias
:type dev_name: str
:param need_check_acc: in first version of the function it defaults to True
Determines if at creation time of DeviceProxy it
should check for channel access (rarely used)
:type need_check_acc: bool
:param wait: whether or not to wait for result of creating a DeviceProxy.
:type wait: bool
:param timeout: The number of seconds to wait for the result.
If None, then there is no limit on the wait time.
Ignored when wait is False.
:type timeout: float
:returns:
if wait is True:
:class:`~PyTango.DeviceProxy`
else:
:class:`concurrent.futures.Future`
:throws:
* a *DevFailed* if wait is True and there is an error creating
the device.
* an *asyncio.TimeoutError* if wait is False, timeout is not
None and the time to create the device has expired.
New in PyTango 8.1.0
"""
AttributeProxy = partial(get_attribute_proxy,
green_mode=GreenMode.Asyncio,
wait=False)
AttributeProxy.__doc__ = """
AttributeProxy(self, full_attr_name, wait=False, timeout=False)
-> AttributeProxy
AttributeProxy(self, device_proxy, attr_name, wait=False, timeout=False)
-> AttributeProxy
Creates a *futures* enabled :class:`~PyTango.AttributeProxy`.
The AttributeProxy constructor internally makes some network calls which
makes it *slow*. By using the asyncio *green mode* you may give the control
back to the asyncio event loop using the *yield from* or *await* synthax.
:param full_attr_name: the full name of the attribute
:type full_attr_name: str
:param device_proxy: the :class:`~PyTango.DeviceProxy`
:type device_proxy: DeviceProxy
:param attr_name: attribute name for the given device proxy
:type attr_name: str
:param wait: whether or not to wait for result of creating an
AttributeProxy.
:type wait: bool
:param timeout: The number of seconds to wait for the result.
If None, then there is no limit on the wait time.
Ignored when wait is False.
:type timeout: float
:returns:
if wait is True:
:class:`~PyTango.AttributeProxy`
else:
:class:`concurrent.futures.Future`
:throws:
* a *DevFailed* if wait is True and there is an error creating the
attribute.
* a *asyncio.TimeoutError* if wait is False, timeout is not
None and the time to create the attribute has expired.
New in PyTango 8.1.0
"""
Device = DeviceProxy
Attribute = AttributeProxy
del GreenMode
del get_device_proxy
del get_attribute_proxy
|