6 this file is part of the project scolasync. It is a rewrite of
7 usbDisk.py to take in account udisks2.
9 Copyright (C) 2014 Georges Khaznadar <georgesk@ofset.org>
11 This program is free software: you can redistribute it and/or modify
12 it under the terms of the GNU General Public License as published by
13 the Free Software Foundation, either version3 of the License, or
14 (at your option) any later version.
16 This program is distributed in the hope that it will be useful,
17 but WITHOUT ANY WARRANTY; without even the implied warranty of
18 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
19 GNU General Public License for more details.
21 You should have received a copy of the GNU General Public License
22 along with this program. If not, see <http://www.gnu.org/licenses/>.
25 licence[
'en']=licence_en
26 dependances=
"python3-dbus python3-dbus.mainloop.pyqt5"
29 import dbus, subprocess, os, os.path, re, time, threading, logging, inspect
30 from dbus.mainloop.glib
import DBusGMainLoop, threads_init
31 from gi.repository
import Gio, GLib, UDisks
40 logging.basicConfig(level=logging.DEBUG)
43 callerframerecord = inspect.stack()[caller]
44 frame = callerframerecord[0]
45 info = inspect.getframeinfo(frame)
46 return " -- file={0}, function={1}, line={2}".format(
47 info.filename, info.function, info.lineno
60 if type(obj)==type(
""):
63 path= obj.get_object_path()
64 posUnderscore=path.rfind(
"_")
65 posSlash=path.rfind(
"/")
66 if posUnderscore > posSlash:
67 path=path[:posUnderscore]
76 stat = os.statvfs(device)
79 free = stat.f_bsize * stat.f_bavail
80 total = stat.f_bsize * stat.f_blocks
86 no_options = GLib.Variant(
'a{sv}', {})
93 '/org/freedesktop/UDisks2/block_devices/loop',
95 '/org/freedesktop/UDisks2/block_devices/dm_',
97 '/org/freedesktop/UDisks2/block_devices/ram',
98 '/org/freedesktop/UDisks2/block_devices/zram',
100 '/org/freedesktop/UDisks2/drives/',
115 def __init__(self, logger=logging, diskClass=object):
125 DBusGMainLoop(set_as_default=
True)
127 self.
bus = dbus.SystemBus()
128 self.
udisks = UDisks.Client.new_sync(
None)
132 'profile': [
'man',
'obj'],
136 'profile': [
'man',
'obj'],
140 'profile': [
'man',
'obj'],
143 'interface-removed': {
144 'profile': [
'man',
'obj'],
147 'interface-proxy-properties-changed': {
148 'profile': [
'man',
'obj',
'interface'],
157 self.
addHook(
'interface-added',
159 self.
addHook(
'interface-removed',
161 self.
addHook(
'interface-proxy-properties-changed',
173 if inspect.getargspec(func).args == self.
cbHooks[signal][
'profile']:
174 cb=self.
manager.connect(signal,func)
175 self.
cbHooks[signal][
'hooks'].append(cb)
191 return fs.call_mount_sync(no_options,
None)
192 except GLib.GError
as e:
193 if 'UDisks2.Error.AlreadyMounted' in e.message:
194 m=re.match(
r".*already mounted[^/]*([^\']+).*",e.message)
196 elif 'UDisks2.Error.DeviceBusy' in e.message:
200 time.sleep(retryDelay)
201 timeout -= retryDelay
208 for obj
in self.
manager.get_objects():
219 def _interesting_obj(self, obj):
226 for boring
in not_interesting:
227 if path.startswith(boring):
228 return interesting, drive, partition
231 block = obj.get_block()
233 return interesting, drive, partition
236 drive_name = block.get_cached_property(
'Drive').get_string()
237 if drive_name ==
'/':
238 return interesting, drive, partition
240 drive = self.
udisks.get_object(drive_name).get_drive()
243 if drive
and drive.get_cached_property(
'Optical').get_boolean():
244 return interesting, drive, partition
248 partition = obj.get_partition()
249 return interesting, drive, partition
257 def _udisks_obj_added(self, obj):
271 for s
in obj.get_block().get_cached_property(
'Symlinks'):
272 if b
'/dev/disk/by-id/usb' in bytes(s):
285 def _udisks_partition_added(self, obj, drive, partition):
287 block = obj.get_block()
289 fstype = block.get_cached_property(
'IdType').get_string()
290 parent = partition.get_cached_property(
'Table').get_string()
291 total = drive.get_cached_property(
'Size').get_uint64()
294 fs = obj.get_filesystem()
296 mount_points = fs.get_cached_property(
'MountPoints').get_bytestring_array()
297 if len(mount_points)>0:
298 mount= mount_points[0]
299 if not mount
and fstype ==
'vfat':
303 logging.exception(QApplication.translate(
"uDisk",
"Échec au montage du disque : %s",
None) % path)
308 self.
logger.
debug(QApplication.translate(
"uDisk",
"On n'ajoute pas le disque : partition non-USB",
None)+
inspectData())
310 self.
logger.
debug(QApplication.translate(
"uDisk",
"On n'ajoute pas le disque : partition vide",
None)+
inspectData())
313 path=path, mp=mount, isUsb=isUsb,
314 vendor=drive.get_cached_property(
'Vendor').get_string(),
315 model=drive.get_cached_property(
'Model').get_string(),
318 serial=block.get_cached_property(
'Drive').get_string().split(
'_')[-1],
319 uuid=block.get_cached_property(
'IdUUID').get_string(),
322 device=block.get_cached_property(
'Device').get_bytestring().decode(
'utf-8'),
328 def _udisks_drive_added(self, obj, drive, part):
330 block = obj.get_block()
332 self.
logger.
debug(QApplication.translate(
"uDisk",
"Disque déjà ajouté auparavant : %s",
None) % path+
inspectData())
335 size = drive.get_cached_property(
'Size').get_uint64()
340 self.logger.debug(QApplication.translate("uDisk","On n'ajoute pas le disque : partition à 0 octets.",None)+inspectData())
345 self.
logger.
debug(QApplication.translate(
"uDisk",
"On n'ajoute pas le disque : partition non-USB",
None)+
inspectData())
351 vendor=drive.get_cached_property(
'Vendor').get_string(),
352 model=drive.get_cached_property(
'Model').get_string(),
353 serial=block.get_cached_property(
'Drive').get_string().split(
'_')[-1],
354 uuid=block.get_cached_property(
'IdUUID').get_string(),
356 device=block.get_cached_property(
'Device').get_bytestring().decode(
'utf-8'),
362 def _device_changed(self, obj):
364 self.
logger.
debug(QApplication.translate(
"uDisk",
"Changement pour le disque %s",
None) % path+
inspectData())
373 def _udisks_obj_removed(self, obj):
375 logging.debug(QApplication.translate(
"uDisk",
"Disque débranché du système : %s",
None) % path)
410 def __init__(self, path, mp='', isUsb=False, vendor='', model='', parent=None,
411 fstype='', serial='', uuid='',
412 free=0, capacity=0, device='', firstFat=None, selected=True):
427 self.
rlock=threading.RLock()
434 "1mp":QApplication.translate(
"uDisk",
"point de montage",
None),
435 "2capacity":QApplication.translate(
"uDisk",
"taille",
None),
436 "3vendor":QApplication.translate(
"uDisk",
"marque",
None),
437 "4model":QApplication.translate(
"uDisk",
"modèle de disque",
None),
438 "5stickid":QApplication.translate(
"uDisk",
"numéro de série",
None),
444 _specialItems={
"0Check":QApplication.translate(
"uDisk",
"cocher",
None)}
449 _ItemPattern=re.compile(
"[0-9]?(.*)")
467 result= list(uDisk2._specialItems.keys())+ list(uDisk2._itemNames.keys())
468 return sorted(result)
470 headers = staticmethod(headers)
491 return self.
fstype==
"vfat"
504 prefix=
"\n"+
" "*indent
506 props=[
"mp",
"parent",
"fstype",
"stickid",
"uuid",
"vendor",
"model",
"devStuff",
"free",
"capacity"]
508 r+=prefix+
"%s = %s" %(prop, getattr(self,prop))
524 m=uDisk2._ItemPattern.match(self.
headers()[n])
526 return getattr(self, m.group(1))
541 elif n <= len(propListe):
550 if mount_paths==
None:
553 while len(mount_paths)==0
and leftTries >0:
554 leftTries = leftTries - 1
557 subprocess.call(
"udisks --mount %s > /dev/null" %path,shell=
True)
559 print(
"STILL TO DEBUG: is the mount OK? is self.mp updated?")
567 raise Exception (
"Could not mount the VFAT after 5 tries.")
592 def __init__(self, access="disk", diskClass=uDisk2):
593 UDisksBackend.__init__(self, diskClass=diskClass)
610 if self.
access==
"firstFat":
612 uDisk2(p,self).ensureMounted()
627 result=self.
summary()==other.summary()
674 r=
"Available USB disks\n"
675 r+=
"===================\n"
676 for d
in sorted(self.
disks()):
677 r+=
"%s\n" %(self.
targets[d].devStuff)
678 partlist=self.
parts(d)
681 for part
in partlist:
682 r+=
" %s\n" %(self.
targets[part].devStuff,)
690 r=
"Available USB disks\n"
691 r+=
"===================\n"
692 for d
in self.
disks():
694 partlist=self.
parts(d)
697 for part
in sorted(partlist):
698 r+=
" %s\n" %(self.
targets[part].devStuff)
699 r+=self.
targets[part].valuableProperties(12)+
"\n"
711 elif self.
access==
"firstFat":
723 elif self.
access==
"firstFat":
740 if self.
targets[p].fstype==
"vfat":
752 for p
in self.fatPaths:
753 if p.split(
"/")[-1]==s:
761 if __name__==
"__main__":
767 QMainWindow.__init__(self)
770 quitbutton = QPushButton(
'Examinez le terminal\nbranchez et débranchez des clés USB, puis\nQuittez', self)
771 quitbutton.clicked.connect(self.close)
772 self.setCentralWidget(quitbutton)
779 print([s.split(
"/")[-1]
for s
in machin.targets.keys()])
780 machin.modified=
False
781 machin.addHook(
'object-added', print_targets_if_modif)
782 machin.addHook(
'object-removed', print_targets_if_modif)
784 app = QApplication(sys.argv)
787 sys.exit(app.exec_())