utils.py 6.64 KB
Newer Older
1
2
3
4
5
6
7
# -*- coding: utf-8 -*-
#
# This file is part of the bliss project
#
# Copyright (c) 2016 Beamline Control Unit, ESRF
# Distributed under the GNU LGPLv3. See LICENSE for more info.

8
9
import inspect
import types
10
import itertools
11
import functools
12

13
class WrappedMethod(object):
14
15
16
17
18
19
20
21
22
23
  def __init__(self, control, method_name):
    self.method_name = method_name
    self.control = control

  def __call__(self, this, *args, **kwargs):
    return getattr(self.control, self.method_name)(*args, **kwargs)

def wrap_methods(from_object, target_object):
   for name in dir(from_object):
       if inspect.ismethod(getattr(from_object, name)):
24
25
26
         if hasattr(target_object, name) and inspect.ismethod(getattr(target_object, name)):
           continue
         setattr(target_object, name, types.MethodType(WrappedMethod(from_object, name), target_object, target_object.__class__))
27

28
29
30
31
32
33
34
def add_property(inst, name, method):
  cls = type(inst)
  if not hasattr(cls, '__perinstance'):
    cls = type(cls.__name__, (cls,), {})
    cls.__perinstance = True
    inst.__class__ = cls
  setattr(cls, name, property(method))
35
36
37
38
39


def grouped(iterable, n):
    "s -> (s0,s1,s2,...sn-1), (sn,sn+1,sn+2,...s2n-1), (s2n,s2n+1,s2n+2,...s3n-1), ..."
    return itertools.izip(*[iter(iterable)]*n)
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


"""
functions to add custom attributes and commands to an object.
"""
def add_object_method(obj, method, pre_call, name=None, args=[], types_info=(None, None)):

    if name is None:
        name = method.im_func.func_name

    def call(self, *args, **kwargs):
        if callable(pre_call):
            pre_call(self, *args, **kwargs)

        return method.im_func(method.im_self, *args, **kwargs)

    obj._add_custom_method(
        types.MethodType(functools.partial(call, *([obj] + args)),
                         obj), name, types_info)


def object_method(method=None, name=None, args=[], types_info=(None, None)):
    """
    Decorator to add a custom method to an object.

    The same as add_object_method but its purpose is to be used as a
    decorator to the controller method which is to be exported as object method.
    """
    if method is None:
        # Passes here if decorator parameters are not present ???.
        # ...
        return functools.partial(object_method, name=name, args=args,
                                 types_info=types_info)

    # Returns a method where _object_method_ attribute is filled with a
    # dict of elements to characterize it.
    method._object_method_ = dict(name=name, args=args, types_info=types_info)

    return method


def add_object_attribute(obj, name=None, fget=None, fset=None, args=[], type_info=None):

    cust_attr_dict = getattr(obj, "_%s__custom_attributes_dict" % obj.__class__.__name__)

    if cust_attr_dict.get(name):
        access_mode = cust_attr_dict[name][1]
        if fget and not 'r' in access_mode:
            access_mode = "rw"

        if fset and not 'w' in access_mode:
            access_mode = "rw"

        cust_attr_dict[name] = (type_info, access_mode)
    else:
        if fget:
            cust_attr_dict[name] = (type_info, "r")
        elif fset:
            cust_attr_dict[name] = (type_info, "w")
        else:
            raise RuntimeError("impossible case: must have fget or fset...")


"""
decorators for set/get methods to access to custom attributes
"""
def object_attribute_get(get_method=None, name=None, args=[], type_info=None):
    if get_method is None:
        return functools.partial(object_attribute_get, name=name, args=args,
                                 type_info=type_info)

111
112
113
114
115
    if name is None:
        name = get_method.func_name
    attr_name = name
    if attr_name.startswith("get_"):
        attr_name = attr_name[4:] # removes leading "get_"
116
117
118
119
120
121
122
123
124
125
126
127
128
129

    get_method._object_method_ = dict(name=name, args=args, types_info=("None", type_info))

    if not hasattr(get_method, "_object_attribute_"):
        get_method._object_attribute_ = dict()
    get_method._object_attribute_.update(name=attr_name, fget=get_method, args=args, type_info=type_info)

    return get_method

def object_attribute_set(set_method=None, name=None, args=[], type_info=None):
    if set_method is None:
        return functools.partial(object_attribute_set, name=name, args=args,
                                 type_info=type_info)

130
131
132
133
134
    if name is None:
        name = set_method.func_name
    attr_name = name
    if attr_name.startswith("set_"):
        attr_name = attr_name[4:] # removes leading "set_"
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150

    set_method._object_method_ = dict(name=name, args=args, types_info=(type_info, "None"))

    if not hasattr(set_method, "_object_attribute_"):
        set_method._object_attribute_ = dict()
    set_method._object_attribute_.update(name=attr_name, fset=set_method, args=args, type_info=type_info)

    return set_method


def set_custom_members(src_obj, target_obj, pre_call=None):
    # Creates custom methods and attributes for <target_obj> object
    # using <src_object> object definitions.
    # Populates __custom_methods_list and __custom_attributes_dict
    # for tango device server.
    for name, member in inspect.getmembers(src_obj):
151
152
153
154
        # Just fills the list.
        if hasattr(member, "_object_attribute_"):
            add_object_attribute(target_obj,  **member._object_attribute_)

155
156
157
158
159
160
161
162
        # For each method of <src_obj>: try to add it as a
        # custom method or as methods to set/get custom
        # attributes.
        try:
            add_object_method(target_obj, member, pre_call, **member._object_method_)
        except AttributeError:
            pass

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

class config_property(object):
    """
    Config descriptor to allow easy access to static configuration options

    Only works on classes which have a `config` member that has `get()` and
    `set()` methods (for example:
    :class:`~bliss.config.motors.beacon_backend.StaticConfig`).

    Example::

        from bliss.comm.tcp import Tcp
        from bliss.common.utils import config_property
        from bliss.controllers.motor import Controller


        class MyMotorController(Controller):

            host_name = config_property('host')
            port = config_property('port', converter=int, default=5000)

            def initialize(self):
                url = '{0}:{1}'.format(self.host_name, self.port)
                self.tcp = Tcp(url=url)
    """

    def __init__(self, name, converter=str, default=None):
        self.__name = name
        self.__converter = converter
        self.__default = default

    def __get__(self, obj, objtype):
        if obj is None:
            return self
        return obj.config.get(self.__name, converter=self.__converter,
                              default=self.__default)

    def __set__(self, obj, value):
        obj.config.set(self.__name, value)