-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmodule.py
84 lines (69 loc) · 2.43 KB
/
module.py
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
import ctypes
import dbus
import subprocess
class Module(object):
def __init__(self, qmf_object=None, dbus_object=None, lib_object=None, impl_object=None):
self.qmf_object = qmf_object
self.dbus_object = dbus_object
self.lib_object = lib_object
self.impl_object = impl_object
def check(self):
raise NotImplementedError("Method test must be implemented in subclass")
def test(self, name, *values):
# Ignore all None values
vals = []
for v in values:
if v is not None:
vals.append(v)
# Test the rest
value = vals[0]
for v in vals[1:]:
if v != value:
self.failed(name, values)
def failed(self, name, values):
print "Test %s failed | " % name,
for value in values:
print "%s | " % value,
print
class Library(object):
def __init__(self, lib):
self.lib = ctypes.CDLL(lib)
class DBus(object):
def __init__(self, obj, iface):
self.iface = dbus.Interface(obj, dbus_interface=iface)
self.props = dbus.Interface(obj, dbus_interface='org.freedesktop.DBus.Properties').GetAll(iface)
def __getattr__(self, name):
if name in self.props.keys():
return self.props[name]
return self.iface.get_dbus_method(name)
class QMF(object):
props = dict()
def __init__(self, agent, class_):
agent.loadSchemaInfo()
self.agent = agent
data = self.agent.query("{class: '%s', package:'%s'}" % (class_, agent.getPackages()[0]))
if len(data) == 0:
print "Query returned no data for class: %s" % class_
self = None
return
self.data = data[0]
# This should be done automatically, but isn't
self.data._schema = self.agent.getSchema(self.data.getSchemaId())
self.props = self.data.getProperties()
def __getattr__(self, name):
# Parameters
if name in self.props.keys():
return self.props[name]
# Methods
def func(*args):
return self.data._invoke(name, args, {'timeout_':10})
return func
class ImplObject(object):
def _readFile(self, fileName):
f = open(fileName)
s = f.readlines()
f.close()
return s
def _readOut(self, process):
p = subprocess.Popen(process, stdout=subprocess.PIPE)
return p.communicate()[0].strip()