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
252
|
# -*- coding: utf-8 -*-
#
# File: portato/backend/catapult/system.py
# This file is part of the Portato-Project, a graphical portage-frontend.
#
# Copyright (C) 2007-2008 René 'Necoro' Neumann
# This is free software. You may redistribute copies of it under the terms of
# the GNU General Public License version 2.
# There is NO WARRANTY, to the extent permitted by law.
#
# Written by René 'Necoro' Neumann <necoro@necoro.net>
from __future__ import absolute_import
import re, os
from threading import Event
import dbus
import catapult
from .package import CatapultPackage
from ..system_interface import SystemInterface
from ...helper import debug, info, warning, unique_array
class CatapultSystem (SystemInterface):
def __init__ (self):
SystemInterface.__init__(self)
self.bus = dbus.SessionBus()
# get the system
so = self.bus.get_object(catapult.get_dbus_address(catapult.DEFAULT), catapult.CATAPULT_SYSTEM_BUS, follow_name_owner_changes = True)
self.proxy = dbus.Interface(so, catapult.CATAPULT_SYSTEM_IFACE)
def get_version (self):
admint = dbus.Interface(self.bus.get_object(catapult.get_dbus_address(catapult.DEFAULT), catapult.CATAPULT_BUS), catapult.CATAPULT_ADMIN_IFACE)
return "Catapult: %s v. %s" % (self.proxy.bus_name.split(".")[-1], str(admint.version()))
def geneticize_list (self, list_of_packages, only_cpv = False):
"""Convertes a list of cpv's into L{backend.Package}s.
@param list_of_packages: the list of packages
@type list_of_packages: string[]
@param only_cpv: do nothing - return the passed list
@type only_cpv: boolean
@returns: converted list
@rtype: PortagePackage[]
"""
if not only_cpv:
return [CatapultPackage(x) for x in list_of_packages]
else:
return [str(x) for x in list_of_packages]
def split_cpv (self, cpv):
split = self.proxy.split_cpv(cpv)
if all(split):
return map(str, split)
else:
return None
def cpv_matches (self, cpv, criterion):
return CatapultPackage(cpv).matches(criterion)
def find_best(self, list, only_cpv = False):
if only_cpv:
return str(self.proxy.find_best(list))
else:
return CatapultPackage(self.proxy.find_best(list))
def find_best_match (self, search_key, masked = False, only_installed = False, only_cpv = False):
p = self.proxy.find_best_match(search_key, masked, only_installed)
if p :
if not only_cpv:
return CatapultPackage(p)
else:
return str(p)
return None
def _wrap_find(self, key, masked, set, withVersion, only_cpv):
l = []
try:
l = self.proxy.find_packages(key, set, masked, withVersion)
except dbus.DBusException, e:
name, data = str(e).split("\n")[-2].split(": ")[1:]
if name == catapult.CATAPULT_ERR_AMBIGUOUS_PACKAGE:
debug("Ambigous packages: %s.", data)
l = []
for cp in data.split(","):
l += self.proxy.find_packages(cp, set, masked, withVersion)
else:
raise
return self.geneticize_list(l, not(withVersion) or only_cpv)
def find_packages (self, search_key, masked = False, only_cpv = False):
return self._wrap_find(search_key, masked, "all", True, only_cpv)
def find_installed_packages (self, search_key, masked = False, only_cpv = False):
return self._wrap_find(search_key, masked, "installed", True, only_cpv)
def find_system_packages (self, only_cpv = False):
# result = self.proxy.find_system_packages()
# if only_cpv:
# return result
# else:
# return tuple(map(self.geneticize_list, result))
return (self._wrap_find(search_key, False, "system", True, only_cpv), [])
def find_world_packages (self, only_cpv = False):
# result = self.proxy.find_world_packages()
# if only_cpv:
# return result
# else:
# return tuple(map(self.geneticize_list, result))
return (self._wrap_find(search_key, False, "world", True, only_cpv), [])
def _wrap_find_all (self, key, masked, set, withVersion, only_cpv):
if not key:
key = ""
else:
key = "*%s*" % key
l = self.proxy.find_packages("", set, masked, withVersion)
if key:
l = catapult.filter_list(key, l)
return self.geneticize_list(l, not(withVersion) or only_cpv)
def find_all_installed_packages (self, name = None, withVersion = True, only_cpv = False):
return self._wrap_find_all(name, True, "installed", withVersion, only_cpv)
def find_all_uninstalled_packages (self, name = None, only_cpv = False):
return self._wrap_find_all(name, True, "uninstalled", True, only_cpv)
def find_all_packages (self, name = None, withVersion = True, only_cpv = False):
return self._wrap_find_all(name, True, "all", withVersion, only_cpv)
def find_all_world_packages (self, name = None, only_cpv = False):
return self._wrap_find_all(name, True, "world", withVersion, only_cpv)
def find_all_system_packages (self, name = None, only_cpv = False):
return self._wrap_find_all(name, True, "system", withVersion, only_cpv)
def list_categories (self, name = None):
cats = self.proxy.list_categories()
if name:
cats = catapult.filter_list("*%s*" % name, cats)
return map(str, cats)
def sort_package_list(self, pkglist):
return self.geneticize_list(self.proxy.sort_package_list([x.get_cpv() for x in pkglist]))
def reload_settings (self):
return self.proxy.reload_settings()
def update_world (self, newuse = False, deep = False):
ret = []
e = Event()
def wait (list):
ret.extend([(CatapultPackage(x), CatapultPackage(y)) for x,y in list])
e.set()
def error (ex):
e.set()
raise ex
self.proxy.update_world(newuse, deep, {}, reply_handler = wait, error_handler = error, timeout = 300)
e.wait()
return ret
# return [(CatapultPackage(x), CatapultPackage(y)) for x,y in self.proxy.update_world(newuse, deep, {}, timeout = 300)]
def get_updated_packages (self):
ret = []
e = Event()
def wait (list):
ret.extend([CatapultPackage(x) for x in list])
e.set()
def error (ex):
e.set()
raise ex
self.proxy.get_updated_packages(reply_handler = wait, error_handler = error, timeout = 300)
e.wait()
return ret
def get_use_desc (self, flag, package = None):
if not package:
package = ""
return str(self.proxy.get_use_desc(flag, package))
def get_global_settings(self, key):
return str(self.proxy.get_global_settings(key))
def new_package (self, cpv):
return CatapultPackage(cpv)
def get_config_path (self):
return str(self.proxy.get_config_path())
def get_sync_command (self):
return [str(x) for x in self.proxy.get_sync_command()]
def get_merge_command (self):
return [str(x) for x in self.proxy.get_merge_command()]
def get_oneshot_option (self):
return [str(x) for x in self.proxy.get_oneshot_option()]
def get_newuse_option (self):
return [str(x) for x in self.proxy.get_newuse_option()]
def get_deep_option (self):
return [str(x) for x in self.proxy.get_deep_option()]
def get_update_option (self):
return [str(x) for x in self.proxy.get_update_option()]
def get_pretend_option (self):
return [str(x) for x in self.proxy.get_pretend_option()]
def get_unmerge_option (self):
return [str(x) for x in self.proxy.get_unmerge_option()]
def get_environment (self):
default_opts = self.get_global_settings("EMERGE_DEFAULT_OPTS")
opts = dict(os.environ)
if default_opts:
opt_list = default_opts.split()
changed = False
for option in ["--ask", "-a", "--pretend", "-p"]:
if option in opt_list:
opt_list.remove(option)
changed = True
if changed:
opts.update(EMERGE_DEFAULT_OPTS = " ".join(opt_list))
opts.update(TERM = "xterm")
return opts
|
