usbDisk.py

Aller à la documentation de ce fichier.

# -*- coding: utf-8 -*-    
# $Id: usbDisk.py 36 2011-01-15 19:37:27Z georgesk $    

licence={}
licence_en="""
    file usbDisk.py
    this file is part of the project scolasync
    
    Copyright (C) 2010 Georges Khaznadar <georgesk@ofset.org>

    This program is free software: you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation, either version3 of the License, or
    (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with this program.  If not, see <http://www.gnu.org/licenses/>.
"""

licence['en']=licence_en
dependences="python3-dbus python3-dbus.mainloop.qt"
python3safe="True"

import dbus, subprocess, os, os.path, re, time, threading
from PyQt4.QtGui import *


##
# 
#     une classe pour représenter un disque ou une partition.
# 
#     les attributs publics sont :
#     - \b path  le chemin dans le système dbus
#     - \b device l'objet dbus qui correspond à l'instance
#     - \b device_prop un proxy pour questionner cet objet dbus
#     - \b selected booléen vrai si on doit considérer cette instance comme sélectionnée. Vrai à l'initialisation
#     - \b rlock un verrou récursif permettant de réserver l'usage du media pour un seul thread
#     
class uDisk:

    ##
    # 
    #         Le constructeur
    #         @param path un chemin dans le système dbus
    #         @param bus un objet dbus.BusSystem
    #         
    def __init__(self, path, bus):
        self.path=path
        self.mp=None # a variable to cache the result of self.mountPoint()
        self.device = bus.get_object("org.freedesktop.UDisks", self.path)
        self.device_prop = dbus.Interface(self.device, "org.freedesktop.DBus.Properties")
        self.selected=True
        self.rlock=threading.RLock()
        self.stickid=self.getProp("drive-serial")
        self.uuid=self.getProp("id-uuid")
        self.fatuuid=None  # pour l'uuid de la première partion vfat
        self.firstFat=None # poignée de la première partition vfat
        p=self.file()
        # self.devStuff is the name of device which is usable to umount safely this object
        if p:
            self.devStuff=os.path.abspath(os.path.join(os.path.dirname(p), os.readlink(p)))
        else:
            self.devStuff=None
        #

            
    _itemNames={
        "1device-mount-paths":QApplication.translate("uDisk","point de montage",None, QApplication.UnicodeUTF8),
        "2device-size":QApplication.translate("uDisk","taille",None, QApplication.UnicodeUTF8),
        "3drive-vendor":QApplication.translate("uDisk","marque",None, QApplication.UnicodeUTF8),
        "4drive-model":QApplication.translate("uDisk","modèle de disque",None, QApplication.UnicodeUTF8),
        "5drive-serial":QApplication.translate("uDisk","numéro de série",None, QApplication.UnicodeUTF8),
        }

    _specialItems={"0Check":QApplication.translate("uDisk","cocher",None, QApplication.UnicodeUTF8)}

    _ItemPattern=re.compile("[0-9]?(.*)")
    
    ##
    # 
    #         renvoie l'uuid de la première partition FAT après que celle-ci aura été
    #         identifiée (utile pour les disques partitionnés)
    #         @return un uuid
    #         
    def getFatUuid(self):
        return "%s" %self.fatuuid
        
    ##
    # 
    #         renvoie un identifiant unique. Dans cette classe, cette fonction
    #         est synonyme de file()
    #         @return un identifiant unique, garanti par le système de fichiers
    #         
    def uniqueId(self):
        return self.file()
        
    ##
    # 
    #         Méthode statique, pour avoir des titres de colonne.
    #         renvoie des titres pour les items obtenus par __getitem__. 
    #         @param locale la locale, pour traduire les titres éventuellement.
    #         Valeur par défaut : "C"
    #         @return une liste de titres de colonnes
    #         
    def headers(locale="C"):
        result= list(uDisk._specialItems.keys())+ list(uDisk._itemNames.keys())
        return sorted(result)
        
    headers = staticmethod(headers)
    
    ##
    # 
    #         renvoie un proxy vers un navigateur de propriétés
    #         @param bus une instace de dbus.SystemBus
    #         @return l'objet proxy
    #         
    def devicePropProxy(self, bus):
        return self.device_prop

    ##
    # 
    #         Renvoie la valeur de vérité d'une propriété
    #         @param prop une propriété
    #         @param value
    #         @return vrai si la propriété est vraie (cas où value==None) ou vrai si la propriété a exactement la valeur value.
    #         
    def isTrue(self,prop, value=None):
        if value==None:
            return  bool(self.getProp(prop))
        else:
            return self.getProp(prop)==value
    
    ##
    # 
    #         Facilite le réprage des disques USB USB
    #         @return vrai dans le cas d'un disque USB
    #         
    def isUsbDisk(self):
        return self.isTrue("device-is-removable") and self.isTrue("drive-connection-interface","usb") and self.isTrue("device-size")

    ##
    # 
    #         Fournit une représentation imprimable
    #         @return une représentation imprimable de l'instance
    #         
    def __str__(self):
        return self.title()+self.valuableProperties()

    ##
    # 
    #         Permet d'obtenir un identifiant unique de disque
    #         @return le chemin dbus de l'instance
    #         
    def title(self):
        return self.path

    ##
    # 
    #         Permet d'accèder à l'instance par un nom de fichier
    #         @return un nom valide dans le système de fichiers, pour accéder
    #         à l'instance.
    #         
    def file(self):
        fileByPath=self.getProp("device-file-by-path")
        if isinstance(fileByPath, dbus.Array) and len(fileByPath)>0:
            fileByPath=fileByPath[0]
            return fileByPath
        else:
            return None
    
    ##
    # 
    #         Permet d'accèder à l'instance par un point de montage
    #         @return un point de montage, s'il en existe, sinon None
    #         
    def mountPoint(self):
        if self.mp==None:
            paths=self.getProp("device-mount-paths")
            if isinstance(paths, dbus.Array) and len(paths)>0:
                self.mp=paths[0]
                return paths[0]
            else:
                return None
        else:
            return self.mp
    
    ##
    # 
    #         Facilite l'accès aux propriétés à l'aide des mots clés du module udisks
    #         @param name le nom d'une propriété
    #         @return une propriété dbus du disque ou de la partition, sinon None si le nom name est illégal
    #         
    def getProp(self, name):
        try:
            return self.device_prop.Get("org.freedesktop.UDisks", name)
        except:
            return None

    ##
    # 
    #         Permet de reconnaitre les partitions DOS-FAT
    #         @return True dans le cas d'une partition FAT16 ou FAT32
    #         
    def isDosFat(self):
        return self.getProp("id-type")=="vfat"

    ##
    # 
    #         @return True si le disque ou la partion est montée
    #         
    def isMounted(self):
        return bool(self.getProp("device-is-mounted"))
        
    ##
    # 
    #         Facilite l'accès aux propriétés intéressantes d'une instance
    #         @return une chaîne indentée avec les propriétés intéressantes, une par ligne
    #         
    def valuableProperties(self,indent=4):
        prefix="\n"+" "*indent
        r=""
        props=["device-file-by-id",
               "device-file-by-path",
               "device-mount-paths",
               "device-is-partition-table",
               "partition-table-count",
               "device-is-read-only",
               "device-is-drive",
               "device-is-optical-disc",
               "device-is-mounted",
               "drive-vendor",
               "drive-model",
               "drive-serial",
               "id-uuid",
               "partition-slave",
               "partition-type",
               "device-size",
               "id-type"]
        for prop in props:
            p=self.getProp(prop)
            if isinstance(p,dbus.Array):
                if len(p)>0:
                    r+=prefix+"%s = array:" %(prop)
                    for s in p:
                        r+=prefix+" "*indent+s
            elif isinstance(p,dbus.Boolean):
                r+=prefix+"%s = %s" %(prop, bool(p))
            elif isinstance(p,dbus.Int16) or isinstance(p,dbus.Int32) or isinstance(p,dbus.Int64) or isinstance(p,dbus.UInt16) or isinstance(p,dbus.UInt32) or isinstance(p,dbus.UInt64) or isinstance(p,int):
                if p < 10*1024:
                    r+=prefix+"%s = %s" %(prop,p)
                elif p < 10*1024*1024:
                    r+=prefix+"%s = %s k" %(prop,p/1024)
                elif p < 10*1024*1024*1024:
                    r+=prefix+"%s = %s M" %(prop,p/1024/1024)
                else:
                    r+=prefix+"%s = %s G" %(prop,p/1024/1024/1024)
            else:
                r+=prefix+"%s = %s" %(prop,p)
        r+=prefix+"%s = %s" %('devStuff', self.devStuff)
        return r

    ##
    # 
    #         renvoie le chemin du disque, dans le cas où self est une partition
    #         @return le chemin dbus du disque maître, sinon "/"
    #         
    def master(self):
        return self.getProp("partition-slave")

    ##
    # 
    #         retire le numéro des en-têtes pour en faire un nom de propriété
    #         valide pour interroger dbus
    #         @param n un numéro de propriété qui se réfère aux headers
    #         @return une propriété renvoyée par dbus, dans un format imprimable
    #         
    def unNumberProp(self,n):
        m=uDisk._ItemPattern.match(self.headers()[n])
        try:
            prop=m.group(1)
            result=self.showableProp(prop)
            return result
        except:
            return ""
        
    ##
    # 
    #         Renvoie un élément de listage de données internes au disque
    #         @param n un nombre
    #         @return un élément si n>0, et le drapeau self.selected si n==0. 
    #         Les noms des éléments sont dans la liste itemNames utilisée dans 
    #         la fonction statique headers
    #         
    def __getitem__(self,n):
        propListe=self.headers()
        if n==0:
            return self.selected
        elif n <= len(propListe):
            return self.unNumberProp(n-1)

    ##
    # 
    #         Renvoie une propriété dans un type "montrable" par QT.
    #         les propriétés que renvoie dbus ont des types inconnus de Qt4,
    #         cette fonction les transtype pour que QVariant arrive à les
    #         prendre en compte.
    #         @param name le nom de la propriété
    #         @return une nombre ou une chaîne selon le type de propriété
    #         
    def showableProp(self, name):
        p=self.getProp(name)
        if isinstance(p,dbus.Array):
            if len(p)>0: return str(p[0])
            else: return ""
        elif isinstance(p,dbus.Boolean):
            return "%s" %bool(p)
        elif  isinstance(p,dbus.Int16) or isinstance(p,dbus.Int32) or isinstance(p,dbus.Int64) or isinstance(p,dbus.UInt16) or isinstance(p,dbus.UInt32) or isinstance(p,dbus.UInt64) or isinstance(p,int):
            return int(p)
        else:
            return "%s" %p

    ##
    # 
    #         Renvoie la première partition VFAT
    #         @result la première partition VFAT ou None s'il n'y en a pas
    #         
    def getFirstFat(self):
        if self.isDosFat(): return self
        return self.firstFat

    ##
    # 
    #         Permet de s'assurer qu'une partition ou un disque sera bien monté
    #         @result le chemin du point de montage
    #         
    def ensureMounted(self):
        mount_paths=self.getProp("device-mount-paths")
        if mount_paths==None: # le cas où la notion de montage est hors-sujet
            return ""
        leftTries=5
        while len(mount_paths)==0 and leftTries >0:
            leftTries = leftTries - 1
            path=self.getProp("device-file-by-path")
            if isinstance(path,dbus.Array) and len(path)>0:
                path=path[0]
                subprocess.call("udisks --mount %s > /dev/null" %path,shell=True)
                paths=self.getProp("device-mount-paths")
                if paths:
                    return self.getProp("device-mount-paths")[0]
                else:
                    time.sleep(0.5)
            else:
                time.sleep(0.5)
        if leftTries==0:
            raise Exception ("Could not mount the VFAT after 5 tries.")
        else:
            return mount_paths[0]

            
        
##
# 
#     une classe pour représenter la collection des disques USB connectés
# 
#     les attributs publics sont :
#     - \b access le type d'accès qu'on veut pour les items
#     - \b bus une instance de dbus.SystemBus
#     - \b disks la collection de disques USB, organisée en un dictionnaire
#        de disques : les clés sont les disques, qui renvoient à un ensemble
#        de partitions du disque
#     - \b enumdev une liste de chemins dbus vers les disques trouvés
#     - \b firstFats une liste composée de la première partion DOS-FAT de chaque disque USB.
#     
class Available:

    ##
    # 
    #         Le constructeur
    #         @param access définit le type d'accès souhaité. Par défaut, c'est "disk"
    #           c'est à dire qu'on veut la liste des disques USB. Autres valeurs
    #           possibles : "firstFat" pour les premières partitions vfat.
    #         @param diskClass la classe de disques à créer
    #         @param diskDict un dictionnaire des disque maintenu par deviceListener
    #         
    def __init__(self,access="disk", diskClass=uDisk, diskDict=None):
        self.access=access
        self.bus = dbus.SystemBus()
        proxy = self.bus.get_object("org.freedesktop.UDisks", 
                                    "/org/freedesktop/UDisks")
        iface = dbus.Interface(proxy, "org.freedesktop.UDisks")
        self.disks={}
        self.enumDev=iface.EnumerateDevices()
        ### récupération des disques usb dans le dictionnaire self.disks
        for path in self.enumDev:
            ud=diskClass(path, self.bus)
            if ud.isUsbDisk():
                self.disks[ud]=[]
                # cas des disques sans partitions
                if bool(ud.getProp("device-is-partition-table")) == False:
                    # la propriété "device-is-partition-table" est fausse,
                    # probablement qu'il y a un système de fichiers
                    self.disks[ud].append(ud)
        ### une deuxième passe pour récupérer et associer les partitions
        for path in self.enumDev:
            ud=diskClass(path, self.bus)
            for d in self.disks.keys():
                if ud.master() == d.path:
                    self.disks[d].append(ud)
        self.finishInit()

    ##
    # 
    #         Fin de l'initialisation
    #         
    def finishInit(self):
        self.mountFirstFats()

    ##
    # 
    #         fabrique la liste des partitions FAT,
    #         monte les partitions FAT si elles ne le sont pas
    #         
    def mountFirstFats(self):
        self.firstFats = self.getFirstFats()
        if self.access=="firstFat":
            for p in self.firstFats:
                p.ensureMounted()

    ##
    # 
    #         @return le nombre de medias connectés
    #         
    def __trunc__(self):
        return len(self.firstFats)

    ##
    # 
    #         Sert à comparer deux collections de disques, par exemple
    #         une collection passée et une collection présente.
    #         @param other une instance de Available
    #         @return vrai si other semble être la même collection de disques USB
    #         
    def compare(self, other):
        result=self.summary()==other.summary()
        return result

    ##
    # 
    #         Permet de déterminer si un disque est dans la collection
    #         @param ud une instance de uDisk
    #         @return vrai si le uDisk ud est dans la collection
    #         
    def contains(self, ud):
        for k in self.disks.keys():
            if k.getProp("device-file-by-path")==ud.getProp("device-file-by-path"): return True
        return False
    
    ##
    # 
    #         Fournit une représentation imprimable d'un résumé
    #         @return une représentation imprimable d'un résumé de la collection
    #         
    def summary(self):
        r=  "Available USB discs\n"
        r+= "===================\n"
        for d in sorted(self.disks.keys(), key=lambda disk: disk.getFatUuid()):
            r+="%s\n" %(d.title(),)
            if len(self.disks[d])>0:
                r+="    Partitions :\n"
                for part in sorted(self.disks[d], key=lambda disk: disk.getFatUuid()):
                    r+="        %s\n" %(part.path,)
        return r

    ##
    # 
    #         Fournit une représentation imprimable
    #         @return une représentation imprimable de la collection
    #         
    def __str__(self):
        r=  "Available USB discs\n"
        r+= "===================\n"
        for d in self.disks.keys():
            r+="%s\n" %d
            if len(self.disks[d])>0:
                r+="    Partitions :\n"
                for part in self.disks[d]:
                    r+="        %s\n" %(part.path)
                    r+=part.valuableProperties(12)+"\n"
        return r

    ##
    # 
    #         Renvoye le nième disque. Le fonctionnement dépend du paramètre
    #         self.access
    #         @param n un numéro
    #         @return le nième disque USB connecté
    #         
    def __getitem__(self, n):
        if self.access=="disk":
            return self.disks.keys()[n]
        elif self.access=="firstFat":
            return self.firstFats[n]

    ##
    # 
    #         Renseigne sur la longueur de la collection. Le fonctionnement
    #         dépend du paramètre self.access
    #         @return la longueur de la collection de disques renvoyée
    #         
    def __len__(self):
        if self.access=="disk":
            return len(self.disks)
        elif self.access=="firstFat":
            return len(self.firstFats)

    ##
    # 
    #         Facilite l'accès aux partitions de type DOS-FAT, et a des effets
    #         de bord :
    #           * marque le disque avec l'uuid de la première partition FAT.
    #           * construit une liste des chemins uDisk des FATs
    #         @param setOwners si égale à True,
    #           signale que la liste devra comporter des attributs de propriétaire
    #           de medias.
    #         @return une liste de partitions, constituée de la première
    #           partition de type FAT de chaque disque USB connecté
    #         
    def getFirstFats(self, setOwners=False):
        result=[]
        self.fatPaths=[]
        for d in self.disks.keys():
            for p in self.disks[d]:
                if p.isDosFat() or p==d :
                    # le cas p == d correspond aux disques non partitionnés
                    # on va supposer que dans ce cas la partition ne peut
                    # être que de type DOS !!!
                    result.append(p)
                    self.fatPaths.append(p.title())
                    # on marque le disque père et la partition elle-même
                    d.fatuuid=p.uuid
                    d.firstFat=p
                    p.fatuuid=p.uuid
                    if setOwners:
                        p.owner=d.owner
                    break
        return result

    ##
    # 
    #         @param dev un chemin comme /org/freedesktop/UDisks/devices/sdb3
    #         @return True si la partition est dans la liste des partions disponibles
    #         
    def hasDev(self, dev):
        s="%s" %dev
        s=s.replace("/org/freedesktop/UDisks/devices/","")
        for p in self.fatPaths:
            if p.split("/")[-1]==s:
                return True
        return False
    

if __name__=="__main__":
    machin=Available()
    print (machin)