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
|
# Copyright (C) 2015-2016 Red Hat, Inc. All rights reserved.
#
# This copyrighted material is made available to anyone wishing to use,
# modify, copy, or redistribute it subject to the terms and conditions
# of the GNU General Public License v.2.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import dbus
import dbus.service
from . import cfg
from .utils import get_properties, add_properties, get_object_property_diff, \
log_debug
from .state import State
# noinspection PyPep8Naming,PyUnresolvedReferences
class AutomatedProperties(dbus.service.Object):
"""
This class implements the needed interfaces for:
org.freedesktop.DBus.Properties
Other classes inherit from it to get the same behavior
"""
def __init__(self, object_path, search_method=None):
dbus.service.Object.__init__(self, cfg.bus, object_path)
self._ap_interface = []
self._ap_o_path = object_path
self._ap_search_method = search_method
self.state = None
def dbus_object_path(self):
return self._ap_o_path
def emit_data(self):
props = {}
for i in self.interface():
props[i] = AutomatedProperties._get_all_prop(self, i)
return self._ap_o_path, props
def set_interface(self, interface):
"""
With inheritance, we can't easily tell what interfaces a class provides,
so we will have each class that implements an interface tell the
base AutomatedProperties what it is they do provide. This is kind of
clunky, and perhaps we can figure out a better way to do this later.
:param interface: An interface the object supports
:return:
"""
if interface not in self._ap_interface:
self._ap_interface.append(interface)
# noinspection PyUnusedLocal
def interface(self, all_interfaces=False):
if all_interfaces:
cpy = list(self._ap_interface)
cpy.extend(
["org.freedesktop.DBus.Introspectable",
"org.freedesktop.DBus.Properties"])
return cpy
return self._ap_interface
@staticmethod
def _get_prop(obj, interface_name, property_name):
value = getattr(obj, property_name)
# Note: If we get an exception in this handler we won't know about it,
# only the side effect of no returned value!
log_debug('Get (%s), type (%s), value(%s)' %
(property_name, str(type(value)), str(value)))
return value
# Properties
# noinspection PyUnusedLocal
@dbus.service.method(dbus_interface=dbus.PROPERTIES_IFACE,
in_signature='ss', out_signature='v',
async_callbacks=('cb', 'cbe'))
def Get(self, interface_name, property_name, cb, cbe):
# Note: If we get an exception in this handler we won't know about it,
# only the side effect of no returned value!
r = cfg.create_request_entry(
-1, AutomatedProperties._get_prop,
(self, interface_name, property_name),
cb, cbe, False)
cfg.worker_q.put(r)
@staticmethod
def _get_all_prop(obj, interface_name):
if interface_name in obj.interface(True):
# Using introspection, lets build this dynamically
properties = get_properties(obj)
if interface_name in properties:
return properties[interface_name][1]
return {}
raise dbus.exceptions.DBusException(
obj._ap_interface,
'The object %s does not implement the %s interface'
% (obj.__class__, interface_name))
@dbus.service.method(dbus_interface=dbus.PROPERTIES_IFACE,
in_signature='s', out_signature='a{sv}',
async_callbacks=('cb', 'cbe'))
def GetAll(self, interface_name, cb, cbe):
r = cfg.create_request_entry(
-1, AutomatedProperties._get_all_prop,
(self, interface_name),
cb, cbe, False)
cfg.worker_q.put(r)
@dbus.service.method(dbus_interface=dbus.PROPERTIES_IFACE,
in_signature='ssv')
def Set(self, interface_name, property_name, new_value):
setattr(self, property_name, new_value)
self.PropertiesChanged(interface_name,
{property_name: new_value}, [])
# As dbus-python does not support introspection for properties we will
# get the autogenerated xml and then add our wanted properties to it.
@dbus.service.method(dbus_interface=dbus.INTROSPECTABLE_IFACE,
out_signature='s')
def Introspect(self):
r = dbus.service.Object.Introspect(self, self._ap_o_path, cfg.bus)
# Look at the properties in the class
props = get_properties(self)
for int_f, v in props.items():
r = add_properties(r, int_f, v[0])
return r
@dbus.service.signal(dbus_interface=dbus.PROPERTIES_IFACE,
signature='sa{sv}as')
def PropertiesChanged(self, interface_name, changed_properties,
invalidated_properties):
log_debug(('SIGNAL: PropertiesChanged(%s, %s, %s, %s)' %
(str(self._ap_o_path), str(interface_name),
str(changed_properties), str(invalidated_properties))))
def refresh(self, search_key=None, object_state=None):
"""
Take the values (properties) of an object and update them with what
lvm currently has. You can either fetch the new ones or supply the
new state to be updated with
:param search_key: The value to use to search for
:param object_state: Use this as the new object state
"""
num_changed = 0
# If we can't do a lookup, bail now, this happens if we blindly walk
# through all dbus objects as some don't have a search method, like
# 'Manager' object.
if not self._ap_search_method:
return 0
# Either we have the new object state or we need to go fetch it
if object_state:
new_state = object_state
else:
if search_key:
search = search_key
else:
search = self.lvm_id
new_state = self._ap_search_method([search])[0]
assert isinstance(new_state, State)
assert new_state
# When we refresh an object the object identifiers might have changed
# because LVM allows the user to change them (name & uuid), thus if
# they have changed we need to update the object manager so that
# look-ups will happen correctly
old_id = self.state.identifiers()
new_id = new_state.identifiers()
if old_id[0] != new_id[0] or old_id[1] != new_id[1]:
cfg.om.lookup_update(self, new_id[0], new_id[1])
# Grab the properties values, then replace the state of the object
# and retrieve the new values.
o_prop = get_properties(self)
self.state = new_state
n_prop = get_properties(self)
changed = get_object_property_diff(o_prop, n_prop)
if changed:
for int_f, v in changed.items():
self.PropertiesChanged(int_f, v, [])
num_changed += 1
return num_changed
|
