def _load_fsstat_data(self, timeout=3):
        """Using :command:`fsstat`, adds some additional information of the volume to the Volume."""

        def stats_thread():
            try:
                cmd = ['fsstat', self.get_raw_path(), '-o', str(self.offset // self.disk.block_size)]

                # Setting the fstype explicitly makes fsstat much faster and more reliable
                # In some versions, the auto-detect yaffs2 check takes ages for large images
                fstype = {
                    "ntfs": "ntfs", "fat": "fat", "ext": "ext", "iso": "iso9660", "hfs+": "hfs",
                    "ufs": "ufs", "swap": "swap", "exfat": "exfat",
                }.get(self.fstype, None)
                if fstype:
                    cmd.extend(["-f", fstype])

                logger.debug('$ {0}'.format(' '.join(cmd)))
                stats_thread.process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)

                for line in iter(stats_thread.process.stdout.readline, b''):
                    line = line.decode('utf-8')
                    logger.debug('< {0}'.format(line))
                    if line.startswith("File System Type:"):
                        self.info['statfstype'] = line[line.index(':') + 2:].strip()
                    elif line.startswith("Last Mount Point:") or line.startswith("Last mounted on:"):
                        self.info['lastmountpoint'] = line[line.index(':') + 2:].strip().replace("//", "/")
                    elif line.startswith("Volume Name:") and not self.info.get('label'):
                        self.info['label'] = line[line.index(':') + 2:].strip()
                    elif line.startswith("Version:"):
                        self.info['version'] = line[line.index(':') + 2:].strip()
                    elif line.startswith("Source OS:"):
                        self.info['version'] = line[line.index(':') + 2:].strip()
                    elif 'CYLINDER GROUP INFORMATION' in line or 'BLOCK GROUP INFORMATION' in line:
                        # noinspection PyBroadException
                        try:
                            stats_thread.process.terminate()
                            logger.debug("Terminated fsstat at cylinder/block group information.")
                        except Exception:
                            pass
                        break

                if self.info.get('lastmountpoint') and self.info.get('label'):
                    self.info['label'] = "{0} ({1})".format(self.info['lastmountpoint'], self.info['label'])
                elif self.info.get('lastmountpoint') and not self.info.get('label'):
                    self.info['label'] = self.info['lastmountpoint']
                elif not self.info.get('lastmountpoint') and self.info.get('label') and \
                        self.info['label'].startswith("/"):  # e.g. /boot1
                    if self.info['label'].endswith("1"):
                        self.info['lastmountpoint'] = self.info['label'][:-1]
                    else:
                        self.info['lastmountpoint'] = self.info['label']

            except Exception:  # ignore any exceptions here.
                logger.exception("Error while obtaining stats.")

        stats_thread.process = None

        thread = threading.Thread(target=stats_thread)
        thread.start()
        thread.join(timeout)
        if thread.is_alive():
            # noinspection PyBroadException
            try:
                stats_thread.process.terminate()
            except Exception:
                pass
            thread.join()
            logger.debug("Killed fsstat after {0}s".format(timeout))
def _load_fsstat_data ( self , timeout = 3 ) : def stats_thread ( ) : try : cmd = [ 'fsstat' , self . get_raw_path ( ) , '-o' , str ( self . offset // self . disk . block_size ) ] fstype = { "ntfs" : "ntfs" , "fat" : "fat" , "ext" : "ext" , "iso" : "iso9660" , "hfs+" : "hfs" , "ufs" : "ufs" , "swap" : "swap" , "exfat" : "exfat" , } . get ( self . fstype , None ) if fstype : cmd . extend ( [ "-f" , fstype ] ) logger . debug ( '$ {0}' . format ( ' ' . join ( cmd ) ) ) stats_thread . process = subprocess . Popen ( cmd , stdout = subprocess . PIPE , stderr = subprocess . PIPE ) for line in iter ( stats_thread . process . stdout . readline , b'' ) : line = line . decode ( 'utf-8' ) logger . debug ( '< {0}' . format ( line ) ) if line . startswith ( "File System Type:" ) : self . info [ 'statfstype' ] = line [ line . index ( ':' ) + 2 : ] . strip ( ) elif line . startswith ( "Last Mount Point:" ) or line . startswith ( "Last mounted on:" ) : self . info [ 'lastmountpoint' ] = line [ line . index ( ':' ) + 2 : ] . strip ( ) . replace ( "//" , "/" ) elif line . startswith ( "Volume Name:" ) and not self . info . get ( 'label' ) : self . info [ 'label' ] = line [ line . index ( ':' ) + 2 : ] . strip ( ) elif line . startswith ( "Version:" ) : self . info [ 'version' ] = line [ line . index ( ':' ) + 2 : ] . strip ( ) elif line . startswith ( "Source OS:" ) : self . info [ 'version' ] = line [ line . index ( ':' ) + 2 : ] . strip ( ) elif 'CYLINDER GROUP INFORMATION' in line or 'BLOCK GROUP INFORMATION' in line : try : stats_thread . process . terminate ( ) logger . debug ( "Terminated fsstat at cylinder/block group information." ) except Exception : pass break if self . info . get ( 'lastmountpoint' ) and self . info . get ( 'label' ) : self . info [ 'label' ] = "{0} ({1})" . format ( self . info [ 'lastmountpoint' ] , self . info [ 'label' ] ) elif self . info . get ( 'lastmountpoint' ) and not self . info . get ( 'label' ) : self . info [ 'label' ] = self . info [ 'lastmountpoint' ] elif not self . info . get ( 'lastmountpoint' ) and self . info . get ( 'label' ) and self . info [ 'label' ] . startswith ( "/" ) : if self . info [ 'label' ] . endswith ( "1" ) : self . info [ 'lastmountpoint' ] = self . info [ 'label' ] [ : - 1 ] else : self . info [ 'lastmountpoint' ] = self . info [ 'label' ] except Exception : logger . exception ( "Error while obtaining stats." ) stats_thread . process = None thread = threading . Thread ( target = stats_thread ) thread . start ( ) thread . join ( timeout ) if thread . is_alive ( ) : try : stats_thread . process . terminate ( ) except Exception : pass thread . join ( ) logger . debug ( "Killed fsstat after {0}s" . format ( timeout ) )
Using :command:`fsstat`, adds some additional information of the volume to the Volume.