This file is indexed.

/usr/lib/python3/dist-packages/PyTango/client.py is in python3-pytango 8.1.8-1.

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
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
# ------------------------------------------------------------------------------
# This file is part of PyTango (http://www.tinyurl.com/PyTango)
#
# 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.
# ------------------------------------------------------------------------------

"""
High Level API for writting Tango clients

This is an experimental module. Not part of the official API.
"""

import weakref
import functools

import PyTango
from PyTango import DeviceProxy as Device
from PyTango import CmdArgType
from PyTango.codec import loads
from PyTango.codec import dumps as _dumps

_FMT = "pickle"

dumps = functools.partial(_dumps, _FMT)

def _command(device, cmd_info, *args, **kwargs):
    name = cmd_info.cmd_name
    if cmd_info.in_type == CmdArgType.DevEncoded:
        result = device.command_inout(name, dumps((args, kwargs)))
    else:
        result = device.command_inout(name, *args, **kwargs)
    if cmd_info.out_type == CmdArgType.DevEncoded:
        result = loads(*result)
    return result


class _DeviceHelper(object):

    __CMD_FILTER = set(("init", "state", "status"))
    __ATTR_FILTER = set(("state", "status"))

    def __init__(self, dev_name, *args, **kwargs):
        self.dev_name = dev_name
        self.device = Device(dev_name, *args, **kwargs)
        self.slots = weakref.WeakKeyDictionary()

    def connect(self, signal, slot, event_type=PyTango.EventType.CHANGE_EVENT):
        i = self.device.subscribe_event(signal, event_type, slot)
        self.slots[slot] = i
        return i

    def disconnect(self, signal, slot):
        i = self.slots.pop(slot)
        self.device.unsubscribe_event(i)

    def get_attr_cache(self, refresh=False):
        try:
            cache = self.__attr_cache
            if not cache:
                refresh = True
        except AttributeError:
            refresh = True
        if refresh:
            cache = {}
            dev = self.device
            try:
                for attr_info in dev.attribute_list_query_ex():
                    attr_name = attr_info.name
                    if attr_name.lower() in self.__ATTR_FILTER:
                        continue
                    cache[attr_name] = attr_info
            except PyTango.DevFailed:
                pass
            self.__attr_cache = cache
        return cache

    def get_attr_info(self, name):
        cache = self.get_attr_cache()
        result = cache.get(name)
        if result:
            return result
        else:
            cache = self.get_attr_cache(refresh=True)
            return cache.get(name)

    def get_cmd_cache(self, refresh=False):
        try:
            cache = self.__cmd_cache
            if not cache:
                refresh = True
        except AttributeError:
            refresh = True
        if refresh:
            cache = {}
            dev = self.device
            try:
                for cmd_info in dev.command_list_query():
                    cmd_name = cmd_info.cmd_name
                    if cmd_name.lower() in self.__CMD_FILTER:
                        continue
                    cmd_func = functools.partial(_command, dev, cmd_info)
                    cmd_func.__name__ = cmd_name
                    cmd_func.__doc__ = cmd_info.in_type_desc
                    cmd_info.func = cmd_func
                    cache[cmd_name] = cmd_info
            except PyTango.DevFailed:
                pass
            self.__cmd_cache = cache
        return cache

    def get_cmd_info(self, name):
        cache = self.get_cmd_cache()
        result = cache.get(name)
        if result:
            return result
        else:
            cache = self.get_cmd_cache(refresh=True)
            return cache.get(name)

    def is_cmd(self, name):
        return name.lower() in self.get_cmd_cache()

    def members(self):
        result = self.get_attr_cache().keys()
        result.extend(self.get_cmd_cache().keys())
        return result

    def get(self, name):
        dev = self.device
        result = self.get_attr_info(name)
        if result:
            result = dev.read_attribute(name)
            value = result.value
            if result.type == PyTango.DevEncoded:
                result = loads(*value)
            else:
                result = value
            return result
        result = self.get_cmd_info(name)
        if result is None:
            raise KeyError("Unknown %s" % name)
        return result

    def set(self, name, value):
        result = self.get_attr_info(name)
        if result is None:
            raise KeyError("Unknown attribute %s" % name)
        if result.data_type == PyTango.DevEncoded:
            self.device.write_attribute(name, dumps(value))
        else:
            self.device.write_attribute(name, value)

    def get_info(self):
        try:
            return self.__info
        except AttributeError:
            pass
        try:
            info = self.device.info()
            self.__dict__["__info"] = info
            return info
        except PyTango.DevFailed:
            return None

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

    def __repr__(self):
        return str(self)

    def dstr(self):
        info = self.get_info()
        klass = "Device"
        if info:
            klass = info.dev_class
        return "{0}({1})".format(klass, self.dev_name)


class Object(object):
    """Tango object"""

    def __init__(self, dev_name, *args, **kwargs):
        helper = _DeviceHelper(dev_name, *args, **kwargs)
        self.__dict__["_helper"] = helper

    def __getattr__(self, name):
        r = self._helper.get(name)
        if isinstance(r, PyTango.CommandInfo):
            self.__dict__[name] = r.func
            return r.func
        return r

    def __setattr__(self, name, value):
        try:
            return self._helper.set(name, value)
        except KeyError:
            object.__setattr__(self, name, value)

    def __str__(self):
        return str(self._helper)

    def __repr__(self):
        return repr(self._helper)

    def __dir__(self):
        return self._helper.members()


def get_object_proxy(obj):
    """Experimental function. Not part of the official API"""
    return obj._helper.device


def get_object_db(obj):
    """Experimental function. Not part of the official API"""
    return get_object_proxy(obj).get_device_db()


def get_object_name(obj):
    """Experimental function. Not part of the official API"""
    return get_object_proxy(obj).get_name()


def get_object_info(obj):
    """Experimental function. Not part of the official API"""
    return get_object_proxy(obj).info()


def get_attributes_config(obj, refresh=False):
    """Experimental function. Not part of the official API"""
    return obj._helper.get_attr_cache(refresh=refresh)


def get_commands_config(obj, refresh=False):
    """Experimental function. Not part of the official API"""
    return obj._helper.get_cmd_cache(refresh=refresh)


def connect(obj, signal, slot, event_type=PyTango.EventType.CHANGE_EVENT):
    """Experimental function. Not part of the official API"""
    return obj._helper.connect(signal, slot, event_type=event_type)


def disconnect(obj, signal, slot):
    """Experimental function. Not part of the official API"""
    return obj._helper.disconnect(signal, slot)