27class TestEntry(object):
 28    """
 29    Contains all of the information associated with a single SMART Self-test
 30    log entry. This data is intended to exactly mirror that obtained through
 31    smartctl.
 32    """
 33
 34    def __init__(self, format, num: Optional[int], test_type, status, hours, lba, remain=None, segment=None, sense=None, asc=None,
 35                 ascq=None):
 36        self._format = format
 37        """
 38        **(str):** Indicates whether this entry was taken from an 'ata' or
 39        'scsi' self-test log. Used to display the content properly.
 40        """
 41        self.num: Optional[int] = num
 42        """
 43        **(int):** Entry's position in the log from 1 (most recent) to 21
 44        (least recent).  ATA logs save the last 21 entries while SCSI logs
 45        only save the last 20.
 46        """
 47        self.type = test_type
 48        """
 49        **(str):** Type of test run.  Generally short, long (extended), or
 50        conveyance, plus offline (background) or captive (foreground).
 51        """
 52        self.status = status
 53        """
 54        **(str):** Self-test's status message, for example 'Completed without
 55        error' or 'Completed: read failure'.
 56        """
 57        self.hours = hours
 58        """
 59        **(str):** The device's power-on hours at the time the self-test
 60        was initiated.
 61        """
 62        self.LBA = lba
 63        """
 64        **(str):** Indicates the first LBA at which an error was encountered
 65        during this self-test. Presented as a decimal value for ATA/SATA
 66        devices and in hexadecimal notation for SAS/SCSI devices.
 67        """
 68        self.remain = remain
 69        """
 70        **(str):** Percentage value indicating how much of the self-test is
 71        left to perform. '00%' indicates a complete test, while any other
 72        value could indicate a test in progress or one that failed prior to
 73        completion. Only reported by ATA devices.
 74        """
 75        self.segment = segment
 76        """
 77        **(str):** A manufacturer-specific self-test segment number reported
 78        by SCSI devices on self-test failure. Set to '-' otherwise.
 79        """
 80        self.sense = sense
 81        """
 82        **(str):** SCSI sense key reported on self-test failure. Set to '-'
 83        otherwise.
 84        """
 85        self.ASC = asc
 86        """
 87        **(str):** SCSI 'Additonal Sense Code' reported on self-test failure.
 88        Set to '-' otherwise.
 89        """
 90        self.ASCQ = ascq
 91        """
 92        **(str):** SCSI 'Additonal Sense Code Quaifier' reported on self-test
 93        failure. Set to '-' otherwise.
 94        """
 95
 96    def __getstate__(self):
 97        return {
 98            'num': self.num,
 99            'type': self.type,
100            'status': self.status,
101            'hours': self.hours,
102            'lba': self.LBA,
103            'remain': self.remain,
104            'segment': self.segment,
105            'sense': self.sense,
106            'asc': self.ASC,
107            'ascq': self.ASCQ
108        }
109
110    def __repr__(self):
111        """Define a basic representation of the class object."""
112        return "<SMART Self-test [%s|%s] hrs:%s LBA:%s>" % (
113            self.type, self.status, self.hours, self.LBA)
114
115    def __str__(self):
116        """
117        Define a formatted string representation of the object's content.
118        Looks nearly identical to the output of smartctl, without overflowing
119        80-character lines.
120        """
121        if self._format == 'ata':
122            return "{0:>2} {1:17}{2:30}{3:5}{4:7}{5:17}".format(
123                self.num, self.type, self.status, self.remain, self.hours,
124                self.LBA)
125        else:
126            # 'Segment' could not be fit on the 80-char line. It's of limited
127            # utility anyway due to it's manufacturer-proprietary nature...
128            return ("{0:>2} {1:17}{2:23}{3:7}{4:14}[{5:4}{6:5}{7:4}]".format(
129                self.num,
130                self.type,
131                self.status,
132                self.hours,
133                self.LBA,
134                self.sense,
135                self.ASC,
136                self.ASCQ
137            ))

 28class Attribute(object):
 29    """
 30    Contains all of the information associated with a single SMART attribute
 31    in a `Device`'s SMART table. This data is intended to exactly mirror that
 32    obtained through smartctl.
 33    """
 34
 35    def __init__(self, num: int, name, flags: int, value, worst, thresh, attr_type, updated, when_failed, raw):
 36        self.num: int = num
 37        """**(int):** Attribute's ID as a decimal value (1-255)."""
 38        self.name: str = name
 39        """
 40        **(str):** Attribute's name, as reported by smartmontools' drive.db.
 41        """
 42        self.flags: int = flags
 43        """**(int):** Attribute flags as a bit value (ie: 0x0032)."""
 44        self._value: str = value
 45        """**(str):** Attribute's current normalized value."""
 46        self._worst: str = worst
 47        """**(str):** Worst recorded normalized value for this attribute."""
 48        self._thresh: str = thresh
 49        """**(str):** Attribute's failure threshold."""
 50        self.type: str = attr_type
 51        """**(str):** Attribute's type, generally 'pre-fail' or 'old-age'."""
 52        self.updated: str = updated
 53        """
 54        **(str):** When is this attribute updated? Generally 'Always' or
 55        'Offline'
 56        """
 57        self.when_failed: str = when_failed
 58        """
 59        **(str):** When did this attribute cross below
 60        `pySMART.attribute.Attribute.thresh`? Reads '-' when not failed.
 61        Generally either 'FAILING_NOW' or 'In_the_Past' otherwise.
 62        """
 63        self.raw = raw
 64        """**(str):** Attribute's current raw (non-normalized) value."""
 65
 66    @property
 67    def value_str(self) -> str:
 68        """Gets the attribute value
 69
 70        Returns:
 71            str: The attribute value in string format
 72        """
 73        return self._value
 74
 75    @property
 76    def value_int(self) -> int:
 77        """Gets the attribute value
 78
 79        Returns:
 80            int: The attribute value in integer format.
 81        """
 82        return int(self._value)
 83
 84    @property
 85    def value(self) -> str:
 86        """Gets the attribue value
 87
 88        Returns:
 89            str: The attribute value in string format
 90        """
 91        return self.value_str
 92
 93    @property
 94    def worst(self) -> int:
 95        """Gets the worst value
 96
 97        Returns:
 98            int: The attribute worst field in integer format
 99        """
100        return int(self._worst)
101
102    @property
103    def thresh(self) -> Optional[int]:
104        """Gets the threshold value
105
106        Returns:
107            int: The attribute threshold field in integer format
108        """
109        return None if self._thresh == '---' else int(self._thresh)
110
111    @property
112    def raw_int(self) -> int:
113        """Gets the raw value converted to int
114        NOTE: Some values may not be correctly converted!
115
116        Returns:
117            int: The attribute raw-value field in integer format.
118            None: In case the raw string failed to be parsed
119        """
120        try:
121            return int(re.search(r'\d+', self.raw).group())
122        except:
123            return None
124
125    def __repr__(self):
126        """Define a basic representation of the class object."""
127        return "<SMART Attribute %r %s/%s raw:%s>" % (
128            self.name, self.value, self.thresh, self.raw)
129
130    def __str__(self):
131        """
132        Define a formatted string representation of the object's content.
133        In the interest of not overflowing 80-character lines this does not
134        print the value of `pySMART.attribute.Attribute.flags_hex`.
135        """
136        return "{0:>3} {1:23}{2:>4}{3:>4}{4:>4} {5:9}{6:8}{7:12}{8}".format(
137            self.num,
138            self.name,
139            self.value,
140            self.worst,
141            self.thresh,
142            self.type,
143            self.updated,
144            self.when_failed,
145            self.raw
146        )
147
148    def __getstate__(self):
149        return {
150            'num': self.num,
151            'flags': self.flags,
152            'raw': self.raw,
153            'value': self.value,
154            'worst': self.worst,
155            'threshold': self.thresh,
156            'type': self.type,
157            'updated': self.updated,
158            'when_failed': self.when_failed,
159        }

 37class DeviceList(object):
 38    """
 39    Represents a list of all the storage devices connected to this computer.
 40    """
 41
 42    def __init__(self, init: bool = True, smartctl=SMARTCTL, catch_errors: bool = False):
 43        """Instantiates and optionally initializes the `DeviceList`.
 44
 45        Args:
 46            init (bool, optional): By default, `pySMART.device_list.DeviceList.devices`
 47                is populated with `Device` objects during instantiation. Setting init
 48                to False will skip initialization and create an empty
 49                `pySMART.device_list.DeviceList` object instead. Defaults to True.
 50            smartctl ([type], optional): This stablish the smartctl wrapper.
 51                Defaults the global `SMARTCTL` object and should be only
 52                overwritten on tests.
 53            catch_errors (bool, optional): If True, individual device-parsing errors will be caught
 54        """
 55
 56        self.devices: List[Device] = []
 57        """
 58        **(list of `Device`):** Contains all storage devices detected during
 59        instantiation, as `Device` objects.
 60        """
 61        self.smartctl: Smartctl = smartctl
 62        """The smartctl wrapper
 63        """
 64        if init:
 65            self.initialize(catch_errors)
 66
 67    def __repr__(self):
 68        """Define a basic representation of the class object."""
 69        rep = "<DeviceList contents:\n"
 70        for device in self.devices:
 71            rep += str(device) + '\n'
 72        return rep + '>'
 73        # return "<DeviceList contents:%r>" % (self.devices)
 74
 75    def _cleanup(self):
 76        """
 77        Removes duplicate ATA devices that correspond to an existing CSMI
 78        device. Also removes any device with no capacity value, as this
 79        indicates removable storage, ie: CD/DVD-ROM, ZIP, etc.
 80        """
 81        # We can't operate directly on the list while we're iterating
 82        # over it, so we collect indeces to delete and remove them later
 83        to_delete = []
 84        # Enumerate the list to get tuples containing indeces and values
 85        for index, device in enumerate(self.devices):
 86            if device.interface == 'csmi':
 87                for otherindex, otherdevice in enumerate(self.devices):
 88                    if (otherdevice.interface == 'ata' or
 89                            otherdevice.interface == 'sata'):
 90                        if device.serial == otherdevice.serial:
 91                            to_delete.append(otherindex)
 92                            device._sd_name = otherdevice.name
 93            if device.capacity is None and index not in to_delete:
 94                to_delete.append(index)
 95        # Recreate the self.devices list without the marked indeces
 96        self.devices[:] = [v for i, v in enumerate(self.devices)
 97                           if i not in to_delete]
 98
 99    def initialize(self, catch_errors: bool = False):
100        """
101        Scans system busses for attached devices and add them to the
102        `DeviceList` as `Device` objects.
103        If device list is already populated, it will be cleared first.
104
105        Args:
106            catch_errors (bool, optional): If True, individual device-parsing errors will be caught
107        """
108
109        # Clear the list if it's already populated
110        if len(self.devices):
111            self.devices = []
112
113        # Scan for devices
114        for line in self.smartctl.scan():
115            if not ('failed:' in line or line == ''):
116                groups = re.compile(
117                    '^(\S+)\s+-d\s+(\S+)').match(line).groups()
118                name = groups[0]
119                interface = groups[1]
120
121                try:
122                    # Add the device to the list
123                    self.devices.append(
124                        Device(name, interface=interface, smartctl=self.smartctl))
125
126                except Exception as e:
127                    if catch_errors:
128                        # Print the exception
129                        import logging
130
131                        logging.exception(f"Error parsing device {name}")
132
133                    else:
134                        # Reraise the exception
135                        raise e
136
137        # Remove duplicates and unwanted devices (optical, etc.) from the list
138        self._cleanup()
139        # Sort the list alphabetically by device name
140        self.devices.sort(key=lambda device: device.name)
141
142    def __getitem__(self, index: int) -> Device:
143        """Returns an element from self.devices
144
145        Args:
146            index (int): An index of self.devices
147
148        Returns:
149            Device: Returns a Device that is located on the asked index
150        """
151        return self.devices[index]

  81class Device(object):
  82    """
  83    Represents any device attached to an internal storage interface, such as a
  84    hard drive or DVD-ROM, and detected by smartmontools. Includes eSATA
  85    (considered SATA) but excludes other external devices (USB, Firewire).
  86    """
  87
  88    def __init__(self, name: str, interface: Optional[str] = None, abridged: bool = False, smart_options: Union[str, List[str], None] = None, smartctl: Smartctl = SMARTCTL):
  89        """Instantiates and initializes the `pySMART.device.Device`."""
  90        if not (
  91                interface is None or
  92                smartctl_isvalid_type(interface.lower())
  93        ):
  94            raise ValueError(
  95                'Unknown interface: {0} specified for {1}'.format(interface, name))
  96        self.abridged = abridged or interface == 'UNKNOWN INTERFACE'
  97        if smart_options is not None:
  98            if isinstance(smart_options,  str):
  99                smart_options = smart_options.split(' ')
 100            smartctl.add_options(smart_options)
 101        self.smartctl = smartctl
 102        """
 103        """
 104        self.name: str = name.replace('/dev/', '').replace('nvd', 'nvme')
 105        """
 106        **(str):** Device's hardware ID, without the '/dev/' prefix.
 107        (ie: sda (Linux), pd0 (Windows))
 108        """
 109        self.family: Optional[str] = None
 110        """**(str):** Device's family (if any)."""
 111        self.model: Optional[str] = None
 112        """**(str):** Device's model number (if any)."""
 113        self.serial: Optional[str] = None
 114        """**(str):** Device's serial number (if any)."""
 115        self._vendor: Optional[str] = None
 116        """**(str):** Device's vendor (if any)."""
 117        self._interface: Optional[str] = None if interface == 'UNKNOWN INTERFACE' else interface
 118        """
 119        **(str):** Device's interface type. Must be one of:
 120            * **ATA** - Advanced Technology Attachment
 121            * **SATA** - Serial ATA
 122            * **SCSI** - Small Computer Systems Interface
 123            * **SAS** - Serial Attached SCSI
 124            * **SAT** - SCSI-to-ATA Translation (SATA device plugged into a
 125            SAS port)
 126            * **CSMI** - Common Storage Management Interface (Intel ICH /
 127            Matrix RAID)
 128        Generally this should not be specified to allow auto-detection to
 129        occur. Otherwise, this value overrides the auto-detected type and could
 130        produce unexpected or no data.
 131        """
 132        self._capacity: Optional[int] = None
 133        """**(str):** Device's user capacity as reported directly by smartctl (RAW)."""
 134        self._capacity_human: Optional[str] = None
 135        """**(str):** Device's user capacity (human readable) as reported directly by smartctl (RAW)."""
 136        self.firmware: Optional[str] = None
 137        """**(str):** Device's firmware version."""
 138        self.smart_capable: bool = 'nvme' in self.name
 139        """
 140        **(bool):** True if the device has SMART Support Available.
 141        False otherwise. This is useful for VMs amongst other things.
 142        """
 143        self.smart_enabled: bool = 'nvme' in self.name
 144        """
 145        **(bool):** True if the device supports SMART (or SCSI equivalent) and
 146        has the feature set enabled. False otherwise.
 147        """
 148        self.assessment: Optional[str] = None
 149        """
 150        **(str):** SMART health self-assessment as reported by the device.
 151        """
 152        self.messages: List[str] = []
 153        """
 154        **(list of str):** Contains any SMART warnings or other error messages
 155        reported by the device (ie: ascq codes).
 156        """
 157        self.is_ssd: bool = True if 'nvme' in self.name else False
 158        """
 159        **(bool):** True if this device is a Solid State Drive.
 160        False otherwise.
 161        """
 162        self.rotation_rate: Optional[int] = None
 163        """
 164        **(int):** The Roatation Rate of the Drive if it is not a SSD.
 165        The Metric is RPM.
 166        """
 167        self.attributes: List[Optional[Attribute]] = [None] * 256
 168        """
 169        **(list of `Attribute`):** Contains the complete SMART table
 170        information for this device, as provided by smartctl. Indexed by
 171        attribute #, values are set to 'None' for attributes not suported by
 172        this device.
 173        """
 174        self.test_capabilities = {
 175            'offline': False,  # SMART execute Offline immediate (ATA only)
 176            'short': 'nvme' not in self.name,  # SMART short Self-test
 177            'long': 'nvme' not in self.name,  # SMART long Self-test
 178            'conveyance': False,  # SMART Conveyance Self-Test (ATA only)
 179            'selective': False,  # SMART Selective Self-Test (ATA only)
 180        }
 181        # Note have not included 'offline' test for scsi as it runs in the foregorund
 182        # mode. While this may be beneficial to us in someways it is against the
 183        # general layout and pattern that the other tests issued using pySMART are
 184        # followed hence not doing it currently
 185        """
 186        **(dict): ** This dictionary contains key == 'Test Name' and
 187        value == 'True/False' of self-tests that this device is capable of.
 188        """
 189        # Note: The above are just default values and can/will be changed
 190        # upon update() when the attributes and type of the disk is actually
 191        # determined.
 192        self.tests: List[TestEntry] = []
 193        """
 194        **(list of `TestEntry`):** Contains the complete SMART self-test log
 195        for this device, as provided by smartctl.
 196        """
 197        self._test_running = False
 198        """
 199        **(bool):** True if a self-test is currently being run.
 200        False otherwise.
 201        """
 202        self._test_ECD = None
 203        """
 204        **(str):** Estimated completion time of the running SMART selftest.
 205        Not provided by SAS/SCSI devices.
 206        """
 207        self._test_progress = None
 208        """
 209        **(int):** Estimate progress percantage of the running SMART selftest.
 210        """
 211        self.diagnostics: Diagnostics = Diagnostics()
 212        """
 213        **Diagnostics** Contains parsed and processed diagnostic information
 214        extracted from the SMART information. Currently only populated for
 215        SAS and SCSI devices, since ATA/SATA SMART attributes are manufacturer
 216        proprietary.
 217        """
 218        self.temperature: Optional[int] = None
 219        """
 220        **(int or None): Since SCSI disks do not report attributes like ATA ones
 221        we need to grep/regex the shit outta the normal "smartctl -a" output.
 222        In case the device have more than one temperature sensor the first value
 223        will be stored here too.
 224        Note: Temperatures are always in Celsius (if possible).
 225        """
 226        self.temperatures: Dict[int, int] = {}
 227        """
 228        **(dict of int): NVMe disks usually report multiple temperatures, which
 229        will be stored here if available. Keys are sensor numbers as reported in
 230        output data.
 231        Note: Temperatures are always in Celsius (if possible).
 232        """
 233        self.logical_sector_size: Optional[int] = None
 234        """
 235        **(int):** The logical sector size of the device (or LBA).
 236        """
 237        self.physical_sector_size: Optional[int] = None
 238        """
 239        **(int):** The physical sector size of the device.
 240        """
 241        self.if_attributes: Union[None, NvmeAttributes] = None
 242        """
 243        **(NvmeAttributes):** This object may vary for each device interface attributes.
 244        It will store all data obtained from smartctl
 245        """
 246
 247        if self.name is None:
 248            warnings.warn(
 249                "\nDevice '{0}' does not exist! This object should be destroyed.".format(
 250                    name)
 251            )
 252            return
 253        # If no interface type was provided, scan for the device
 254        # Lets do this only for the non-abridged case
 255        # (we can work with no interface for abridged case)
 256        elif self._interface is None and not self.abridged:
 257            logger.trace(
 258                "Determining interface of disk: {0}".format(self.name))
 259            raw, returncode = self.smartctl.generic_call(
 260                ['-d', 'test', self.dev_reference])
 261
 262            if len(raw) > 0:
 263                # I do not like this parsing logic but it works for now!
 264                # just for reference _stdout.split('\n') gets us
 265                # something like
 266                # [
 267                #     ...copyright string...,
 268                #     '',
 269                #     "/dev/ada2: Device of type 'atacam' [ATA] detected",
 270                #     "/dev/ada2: Device of type 'atacam' [ATA] opened",
 271                #     ''
 272                # ]
 273                # The above example should be enough for anyone to understand the line below
 274                try:
 275                    self._interface = raw[-2].split("'")[1]
 276                    if self._interface == "nvme":  # if nvme set SMART to true
 277                        self.smart_capable = True
 278                        self.smart_enabled = True
 279                except:
 280                    # for whatever reason we could not get the interface type
 281                    # we should mark this as an `abbridged` case and move on
 282                    self._interface = None
 283                    self.abbridged = True
 284                # TODO: Uncomment the classify call if we ever find out that we need it
 285                # Disambiguate the generic interface to a specific type
 286                # self._classify()
 287            else:
 288                warnings.warn(
 289                    "\nDevice '{0}' does not exist! This object should be destroyed.".format(
 290                        name)
 291                )
 292                return
 293        # If a valid device was detected, populate its information
 294        # OR if in unabridged mode, then do it even without interface info
 295        if self._interface is not None or self.abridged:
 296            self.update()
 297
 298    @property
 299    def dev_interface(self) -> Optional[str]:
 300        """Returns the internal interface type of the device.
 301           It may not be the same as the interface type as used by smartctl.
 302
 303        Returns:
 304            str: The interface type of the device. (example: ata, scsi, nvme)
 305                 None if the interface type could not be determined.
 306        """
 307        # Try to get the fine-tuned interface type
 308        fineType = self._classify()
 309
 310        # If return still contains a megaraid, just asume it's type
 311        if 'megaraid' in fineType:
 312            # If any attributes is not None and has at least non None value, then it is a sat+megaraid device
 313            if self.attributes and any(self.attributes):
 314                return 'ata'
 315            else:
 316                return 'sas'
 317
 318        return fineType
 319
 320    @property
 321    def smartctl_interface(self) -> Optional[str]:
 322        """Returns the interface type of the device as it is used in smartctl.
 323
 324        Returns:
 325            str: The interface type of the device. (example: ata, scsi, nvme)
 326                 None if the interface type could not be determined.
 327        """
 328        return self._interface
 329
 330    @property
 331    def interface(self) -> Optional[str]:
 332        """Returns the interface type of the device as it is used in smartctl.
 333
 334        Returns:
 335            str: The interface type of the device. (example: ata, scsi, nvme)
 336                 None if the interface type could not be determined.
 337        """
 338        return self.smartctl_interface
 339
 340    @property
 341    def dev_reference(self) -> str:
 342        """The reference to the device as provided by smartctl.
 343           - On unix-like systems, this is the path to the device. (example /dev/<name>)
 344           - On MacOS, this is the name of the device. (example <name>)
 345           - On Windows, this is the drive letter of the device. (example <drive letter>)
 346
 347        Returns:
 348            str: The reference to the device as provided by smartctl.
 349        """
 350
 351        # detect if we are on MacOS
 352        if 'IOService' in self.name:
 353            return self.name
 354
 355        # otherwise asume we are on unix-like systems
 356        return os.path.join('/dev/', self.name)
 357
 358    @property
 359    def vendor(self) -> Optional[str]:
 360        """Returns the vendor of the device.
 361
 362        Returns:
 363            str: The vendor of the device.
 364        """
 365        if self._vendor:
 366            return self._vendor
 367
 368        # If family is present, try to stract from family. Skip anything but letters.
 369        elif self.family:
 370            filter = re.search(r'^[a-zA-Z]+', self.family.strip())
 371            if filter:
 372                return filter.group(0)
 373
 374        # If model is present, try to stract from model. Skip anything but letters.
 375        elif self.model:
 376            filter = re.search(r'^[a-zA-Z]+', self.model.strip())
 377            if filter:
 378                return filter.group(0)
 379
 380        # If all else fails, return None
 381        return None
 382
 383    @property
 384    def capacity(self) -> Optional[str]:
 385        """Returns the capacity in the raw smartctl format.
 386        This may be deprecated in the future and its only retained for compatibility.
 387
 388        Returns:
 389            str: The capacity in the raw smartctl format
 390        """
 391        return self._capacity_human
 392
 393    @property
 394    def diags(self) -> Dict[str, str]:
 395        """Gets the old/deprecated version of SCSI/SAS diags atribute.
 396        """
 397        return self.diagnostics.get_classic_format()
 398
 399    @property
 400    def size_raw(self) -> Optional[str]:
 401        """Returns the capacity in the raw smartctl format.
 402
 403        Returns:
 404            str: The capacity in the raw smartctl format
 405        """
 406        return self._capacity_human
 407
 408    @property
 409    def size(self) -> int:
 410        """Returns the capacity in bytes
 411
 412        Returns:
 413            int: The capacity in bytes
 414        """
 415        import humanfriendly
 416
 417        if self._capacity is not None:
 418            return self._capacity
 419        elif self._capacity_human is not None:
 420            return humanfriendly.parse_size(self._capacity_human)
 421        else:
 422            return 0
 423
 424    @property
 425    def sector_size(self) -> int:
 426        """Returns the sector size of the device.
 427
 428        Returns:
 429            int: The sector size of the device in Bytes. If undefined, we'll assume 512B
 430        """
 431        if self.logical_sector_size is not None:
 432            return self.logical_sector_size
 433        elif self.physical_sector_size is not None:
 434            return self.physical_sector_size
 435        else:
 436            return 512
 437
 438    def __repr__(self):
 439        """Define a basic representation of the class object."""
 440        return "<{0} device on /dev/{1} mod:{2} sn:{3}>".format(
 441            self._interface.upper() if self._interface else 'UNKNOWN INTERFACE',
 442            self.name,
 443            self.model,
 444            self.serial
 445        )
 446
 447    def __getstate__(self, all_info=True):
 448        """
 449        Allows us to send a pySMART Device object over a serializable
 450        medium which uses json (or the likes of json) payloads
 451        """
 452        state_dict = {
 453            'interface': self._interface if self._interface else 'UNKNOWN INTERFACE',
 454            'model': self.model,
 455            'firmware': self.firmware,
 456            'smart_capable': self.smart_capable,
 457            'smart_enabled': self.smart_enabled,
 458            'smart_status': self.assessment,
 459            'messages': self.messages,
 460            'test_capabilities': self.test_capabilities.copy(),
 461            'tests': [t.__getstate__() for t in self.tests] if self.tests else [],
 462            'diagnostics': self.diagnostics.__getstate__(),
 463            'temperature': self.temperature,
 464            'attributes': [attr.__getstate__() if attr else None for attr in self.attributes]
 465        }
 466        if all_info:
 467            state_dict.update({
 468                'name': self.name,
 469                'path': self.dev_reference,
 470                'serial': self.serial,
 471                'is_ssd': self.is_ssd,
 472                'rotation_rate': self.rotation_rate,
 473                'capacity': self._capacity_human
 474            })
 475        return state_dict
 476
 477    def __setstate__(self, state):
 478        state['assessment'] = state['smart_status']
 479        del state['smart_status']
 480        self.__dict__.update(state)
 481
 482    def smart_toggle(self, action: str) -> Tuple[bool, List[str]]:
 483        """
 484        A basic function to enable/disable SMART on device.
 485
 486        # Args:
 487        * **action (str):** Can be either 'on'(for enabling) or 'off'(for disabling).
 488
 489        # Returns"
 490        * **(bool):** Return True (if action succeded) else False
 491        * **(List[str]):** None if option succeded else contains the error message.
 492        """
 493        # Lets make the action verb all lower case
 494        if self._interface == 'nvme':
 495            return False, ['NVME devices do not currently support toggling SMART enabled']
 496        action_lower = action.lower()
 497        if action_lower not in ['on', 'off']:
 498            return False, ['Unsupported action {0}'.format(action)]
 499        # Now lets check if the device's smart enabled status is already that of what
 500        # the supplied action is intending it to be. If so then just return successfully
 501        if self.smart_enabled:
 502            if action_lower == 'on':
 503                return True, []
 504        else:
 505            if action_lower == 'off':
 506                return True, []
 507        if self._interface is not None:
 508            raw, returncode = self.smartctl.generic_call(
 509                ['-s', action_lower, '-d', self._interface, self.dev_reference])
 510        else:
 511            raw, returncode = self.smartctl.generic_call(
 512                ['-s', action_lower, self.dev_reference])
 513
 514        if returncode != 0:
 515            return False, raw
 516        # if everything worked out so far lets perform an update() and check the result
 517        self.update()
 518        if action_lower == 'off' and self.smart_enabled:
 519            return False, ['Failed to turn SMART off.']
 520        if action_lower == 'on' and not self.smart_enabled:
 521            return False, ['Failed to turn SMART on.']
 522        return True, []
 523
 524    def all_attributes(self, print_fn=print):
 525        """
 526        Prints the entire SMART attribute table, in a format similar to
 527        the output of smartctl.
 528        allows usage of custom print function via parameter print_fn by default uses print
 529        """
 530        header_printed = False
 531        for attr in self.attributes:
 532            if attr is not None:
 533                if not header_printed:
 534                    print_fn("{0:>3} {1:24}{2:4}{3:4}{4:4}{5:9}{6:8}{7:12}{8}"
 535                             .format('ID#', 'ATTRIBUTE_NAME', 'CUR', 'WST', 'THR', 'TYPE', 'UPDATED', 'WHEN_FAIL',
 536                                     'RAW'))
 537                    header_printed = True
 538                print_fn(attr)
 539        if not header_printed:
 540            print_fn('This device does not support SMART attributes.')
 541
 542    def all_selftests(self):
 543        """
 544        Prints the entire SMART self-test log, in a format similar to
 545        the output of smartctl.
 546        """
 547        if self.tests:
 548            all_tests = []
 549            if smartctl_type(self._interface) == 'scsi':
 550                header = "{0:3}{1:17}{2:23}{3:7}{4:14}{5:15}".format(
 551                    'ID',
 552                    'Test Description',
 553                    'Status',
 554                    'Hours',
 555                    '1st_Error@LBA',
 556                    '[SK  ASC  ASCQ]'
 557                )
 558            else:
 559                header = ("{0:3}{1:17}{2:30}{3:5}{4:7}{5:17}".format(
 560                    'ID',
 561                    'Test_Description',
 562                    'Status',
 563                    'Left',
 564                    'Hours',
 565                    '1st_Error@LBA'))
 566            all_tests.append(header)
 567            for test in self.tests:
 568                all_tests.append(str(test))
 569
 570            return all_tests
 571        else:
 572            no_tests = 'No self-tests have been logged for this device.'
 573            return no_tests
 574
 575    def _classify(self) -> str:
 576        """
 577        Disambiguates generic device types ATA and SCSI into more specific
 578        ATA, SATA, SAS, SAT and SCSI.
 579        """
 580
 581        fine_interface = self._interface or ''
 582        # SCSI devices might be SCSI, SAS or SAT
 583        # ATA device might be ATA or SATA
 584        if fine_interface in ['scsi', 'ata'] or 'megaraid' in fine_interface:
 585            if 'megaraid' in fine_interface:
 586                if not 'sat+' in fine_interface:
 587                    test = 'sat'+fine_interface
 588                else:
 589                    test = fine_interface
 590            else:
 591                test = 'sat' if fine_interface == 'scsi' else 'sata'
 592            # Look for a SATA PHY to detect SAT and SATA
 593            raw, returncode = self.smartctl.try_generic_call([
 594                '-d',
 595                smartctl_type(test),
 596                '-l',
 597                'sataphy',
 598                self.dev_reference])
 599
 600            if returncode == 0 and 'GP Log 0x11' in raw[3]:
 601                fine_interface = test
 602        # If device type is still SCSI (not changed to SAT above), then
 603        # check for a SAS PHY
 604        if fine_interface in ['scsi'] or 'megaraid' in fine_interface:
 605            raw, returncode = self.smartctl.try_generic_call([
 606                '-d',
 607                smartctl_type(fine_interface),
 608                '-l',
 609                'sasphy',
 610                self.dev_reference])
 611            if returncode == 0 and 'SAS SSP' in raw[4]:
 612                fine_interface = 'sas'
 613            # Some older SAS devices do not support the SAS PHY log command.
 614            # For these, see if smartmontools reports a transport protocol.
 615            else:
 616                raw = self.smartctl.all(self.dev_reference, fine_interface)
 617
 618                for line in raw:
 619                    if 'Transport protocol' in line and 'SAS' in line:
 620                        fine_interface = 'sas'
 621
 622        return fine_interface
 623
 624    def _guess_smart_type(self, line):
 625        """
 626        This function is not used in the generic wrapper, however the header
 627        is defined so that it can be monkey-patched by another application.
 628        """
 629        pass
 630
 631    def _make_smart_warnings(self):
 632        """
 633        Parses an ATA/SATA SMART table for attributes with the 'when_failed'
 634        value set. Generates an warning message for any such attributes and
 635        updates the self-assessment value if necessary.
 636        """
 637        if smartctl_type(self._interface) == 'scsi':
 638            return
 639        for attr in self.attributes:
 640            if attr is not None:
 641                if attr.when_failed == 'In_the_past':
 642                    warn_str = "{0} failed in the past with value {1}. [Threshold: {2}]".format(
 643                        attr.name, attr.worst, attr.thresh)
 644                    self.messages.append(warn_str)
 645                    if not self.assessment == 'FAIL':
 646                        self.assessment = 'WARN'
 647                elif attr.when_failed == 'FAILING_NOW':
 648                    warn_str = "{0} is failing now with value {1}. [Threshold: {2}]".format(
 649                        attr.name, attr.value, attr.thresh)
 650                    self.assessment = 'FAIL'
 651                    self.messages.append(warn_str)
 652                elif not attr.when_failed == '-':
 653                    warn_str = "{0} says it failed '{1}'. [V={2},W={3},T={4}]".format(
 654                        attr.name, attr.when_failed, attr.value, attr.worst, attr.thresh)
 655                    self.messages.append(warn_str)
 656                    if not self.assessment == 'FAIL':
 657                        self.assessment = 'WARN'
 658
 659    def get_selftest_result(self, output=None):
 660        """
 661        Refreshes a device's `pySMART.device.Device.tests` attribute to obtain
 662        the latest test results. If a new test result is obtained, its content
 663        is returned.
 664
 665        # Args:
 666        * **output (str, optional):** If set to 'str', the string
 667        representation of the most recent test result will be returned, instead
 668        of a `Test_Entry` object.
 669
 670        # Returns:
 671        * **(int):** Return status code. One of the following:
 672            * 0 - Success. Object (or optionally, string rep) is attached.
 673            * 1 - Self-test in progress. Must wait for it to finish.
 674            * 2 - No new test results.
 675            * 3 - The Self-test was Aborted by host
 676        * **(`Test_Entry` or str):** Most recent `Test_Entry` object (or
 677        optionally it's string representation) if new data exists.  Status
 678        message string on failure.
 679        * **(int):** Estimate progress percantage of the running SMART selftest, if known.
 680        Otherwise 'None'.
 681        """
 682        # SCSI self-test logs hold 20 entries while ATA logs hold 21
 683        if smartctl_type(self._interface) == 'scsi':
 684            maxlog = 20
 685        else:
 686            maxlog = 21
 687        # If we looked only at the most recent test result we could be fooled
 688        # by two short tests run close together (within the same hour)
 689        # appearing identical. Comparing the length of the log adds some
 690        # confidence until it maxes, as above. Comparing the least-recent test
 691        # result greatly diminishes the chances that two sets of two tests each
 692        # were run within an hour of themselves, but with 16-17 other tests run
 693        # in between them.
 694        if self.tests:
 695            _first_entry = self.tests[0]
 696            _len = len(self.tests)
 697            _last_entry = self.tests[_len - 1]
 698        else:
 699            _len = 0
 700        self.update()
 701        # Since I have changed the update() parsing to DTRT to pickup currently
 702        # running selftests we can now purely rely on that for self._test_running
 703        # Thus check for that variable first and return if it is True with appropos message.
 704        if self._test_running is True:
 705            return 1, 'Self-test in progress. Please wait.', self._test_progress
 706        # Check whether the list got longer (ie: new entry)
 707        # If so return the newest test result
 708        # If not, because it's max size already, check for new entries
 709        if (
 710                (len(self.tests) != _len) or
 711                (
 712                    len == maxlog and
 713                    (
 714                        _first_entry.type != self.tests[0].type or
 715                        _first_entry.hours != self.tests[0].hours or
 716                        _last_entry.type != self.tests[len(self.tests) - 1].type or
 717                        _last_entry.hours != self.tests[len(
 718                            self.tests) - 1].hours
 719                    )
 720                )
 721        ):
 722            return (
 723                0 if 'Aborted' not in self.tests[0].status else 3,
 724                str(self.tests[0]) if output == 'str' else self.tests[0],
 725                None
 726            )
 727        else:
 728            return 2, 'No new self-test results found.', None
 729
 730    def abort_selftest(self):
 731        """
 732        Aborts non-captive SMART Self Tests.   Note that this command
 733        will  abort the Offline Immediate Test routine only if your disk
 734        has the "Abort Offline collection upon new command"  capability.
 735
 736        # Args: Nothing (just aborts directly)
 737
 738        # Returns:
 739        * **(int):** The returncode of calling `smartctl -X device_path`
 740        """
 741        return self.smartctl.test_stop(smartctl_type(self._interface), self.dev_reference)
 742
 743    def run_selftest(self, test_type, ETA_type='date'):
 744        """
 745        Instructs a device to begin a SMART self-test. All tests are run in
 746        'offline' / 'background' mode, allowing normal use of the device while
 747        it is being tested.
 748
 749        # Args:
 750        * **test_type (str):** The type of test to run. Accepts the following
 751        (not case sensitive):
 752            * **short** - Brief electo-mechanical functionality check.
 753            Generally takes 2 minutes or less.
 754            * **long** - Thorough electro-mechanical functionality check,
 755            including complete recording media scan. Generally takes several
 756            hours.
 757            * **conveyance** - Brief test used to identify damage incurred in
 758            shipping. Generally takes 5 minutes or less. **This test is not
 759            supported by SAS or SCSI devices.**
 760            * **offline** - Runs SMART Immediate Offline Test. The effects of
 761            this test are visible only in that it updates the SMART Attribute
 762            values, and if errors are found they will appear in the SMART error
 763            log, visible with the '-l error' option to smartctl. **This test is
 764            not supported by SAS or SCSI devices in pySMART use cli smartctl for
 765            running 'offline' selftest (runs in foreground) on scsi devices.**
 766            * **ETA_type** - Format to return the estimated completion time/date
 767            in. Default is 'date'. One could otherwise specidy 'seconds'.
 768            Again only for ATA devices.
 769
 770        # Returns:
 771        * **(int):** Return status code.  One of the following:
 772            * 0 - Self-test initiated successfully
 773            * 1 - Previous self-test running. Must wait for it to finish.
 774            * 2 - Unknown or unsupported (by the device) test type requested.
 775            * 3 - Unspecified smartctl error. Self-test not initiated.
 776        * **(str):** Return status message.
 777        * **(str)/(float):** Estimated self-test completion time if a test is started.
 778        The optional argument of 'ETA_type' (see above) controls the return type.
 779        if 'ETA_type' == 'date' then a date string is returned else seconds(float)
 780        is returned.
 781        Note: The self-test completion time can only be obtained for ata devices.
 782        Otherwise 'None'.
 783        """
 784        # Lets call get_selftest_result() here since it does an update() and
 785        # checks for an existing selftest is running or not, this way the user
 786        # can issue a test from the cli and this can still pick that up
 787        # Also note that we do not need to obtain the results from this as the
 788        # data is already stored in the Device class object's variables
 789        self.get_selftest_result()
 790        if self._test_running:
 791            return 1, 'Self-test in progress. Please wait.', self._test_ECD
 792        test_type = test_type.lower()
 793        interface = smartctl_type(self._interface)
 794        try:
 795            if not self.test_capabilities[test_type]:
 796                return (
 797                    2,
 798                    "Device {0} does not support the '{1}' test ".format(
 799                        self.name, test_type),
 800                    None
 801                )
 802        except KeyError:
 803            return 2, "Unknown test type '{0}' requested.".format(test_type), None
 804
 805        raw, rc = self.smartctl.test_start(
 806            interface, test_type, self.dev_reference)
 807        _success = False
 808        _running = False
 809        for line in raw:
 810            if 'has begun' in line:
 811                _success = True
 812                self._test_running = True
 813            if 'aborting current test' in line:
 814                _running = True
 815                try:
 816                    self._test_progress = 100 - \
 817                        int(line.split('(')[-1].split('%')[0])
 818                except ValueError:
 819                    pass
 820
 821            if _success and 'complete after' in line:
 822                self._test_ECD = line[25:].rstrip()
 823                if ETA_type == 'seconds':
 824                    self._test_ECD = mktime(
 825                        strptime(self._test_ECD, '%a %b %d %H:%M:%S %Y')) - time()
 826                self._test_progress = 0
 827        if _success:
 828            return 0, 'Self-test started successfully', self._test_ECD
 829        else:
 830            if _running:
 831                return 1, 'Self-test already in progress. Please wait.', self._test_ECD
 832            else:
 833                return 3, 'Unspecified Error. Self-test not started.', None
 834
 835    def run_selftest_and_wait(self, test_type, output=None, polling=5, progress_handler=None):
 836        """
 837        This is essentially a wrapper around run_selftest() such that we
 838        call self.run_selftest() and wait on the running selftest till
 839        it finished before returning.
 840        The above holds true for all pySMART supported tests with the
 841        exception of the 'offline' test (ATA only) as it immediately
 842        returns, since the entire test only affects the smart error log
 843        (if any errors found) and updates the SMART attributes. Other
 844        than that it is not visibile anywhere else, so we start it and
 845        simply return.
 846        # Args:
 847        * **test_type (str):** The type of test to run. Accepts the following
 848        (not case sensitive):
 849            * **short** - Brief electo-mechanical functionality check.
 850            Generally takes 2 minutes or less.
 851            * **long** - Thorough electro-mechanical functionality check,
 852            including complete recording media scan. Generally takes several
 853            hours.
 854            * **conveyance** - Brief test used to identify damage incurred in
 855            shipping. Generally takes 5 minutes or less. **This test is not
 856            supported by SAS or SCSI devices.**
 857            * **offline** - Runs SMART Immediate Offline Test. The effects of
 858            this test are visible only in that it updates the SMART Attribute
 859            values, and if errors are found they will appear in the SMART error
 860            log, visible with the '-l error' option to smartctl. **This test is
 861            not supported by SAS or SCSI devices in pySMART use cli smartctl for
 862            running 'offline' selftest (runs in foreground) on scsi devices.**
 863        * **output (str, optional):** If set to 'str', the string
 864            representation of the most recent test result will be returned,
 865            instead of a `Test_Entry` object.
 866        * **polling (int, default=5):** The time duration to sleep for between
 867            checking for test_results and progress.
 868        * **progress_handler (function, optional):** This if provided is called
 869            with self._test_progress as the supplied argument everytime a poll to
 870            check the status of the selftest is done.
 871        # Returns:
 872        * **(int):** Return status code.  One of the following:
 873            * 0 - Self-test executed and finished successfully
 874            * 1 - Previous self-test running. Must wait for it to finish.
 875            * 2 - Unknown or illegal test type requested.
 876            * 3 - The Self-test was Aborted by host
 877            * 4 - Unspecified smartctl error. Self-test not initiated.
 878        * **(`Test_Entry` or str):** Most recent `Test_Entry` object (or
 879        optionally it's string representation) if new data exists.  Status
 880        message string on failure.
 881        """
 882        test_initiation_result = self.run_selftest(test_type)
 883        if test_initiation_result[0] != 0:
 884            return test_initiation_result[:2]
 885        if test_type == 'offline':
 886            self._test_running = False
 887        # if not then the test initiated correctly and we can start the polling.
 888        # For now default 'polling' value is 5 seconds if not specified by the user
 889
 890        # Do an initial check, for good measure.
 891        # In the probably impossible case that self._test_running is instantly False...
 892        selftest_results = self.get_selftest_result(output=output)
 893        while self._test_running:
 894            if selftest_results[0] != 1:
 895                # the selftest is run and finished lets return with the results
 896                break
 897            # Otherwise see if we are provided with the progress_handler to update progress
 898            if progress_handler is not None:
 899                progress_handler(
 900                    selftest_results[2] if selftest_results[2] is not None else 50)
 901            # Now sleep 'polling' seconds before checking the progress again
 902            sleep(polling)
 903
 904            # Check after the sleep to ensure we return the right result, and not an old one.
 905            selftest_results = self.get_selftest_result(output=output)
 906
 907        # Now if (selftes_results[0] == 2) i.e No new selftest (because the same
 908        # selftest was run twice within the last hour) but we know for a fact that
 909        # we just ran a new selftest then just return the latest entry in self.tests
 910        if selftest_results[0] == 2:
 911            selftest_return_value = 0 if 'Aborted' not in self.tests[0].status else 3
 912            return selftest_return_value, str(self.tests[0]) if output == 'str' else self.tests[0]
 913        return selftest_results[:2]
 914
 915    def update(self):
 916        """
 917        Queries for device information using smartctl and updates all
 918        class members, including the SMART attribute table and self-test log.
 919        Can be called at any time to refresh the `pySMART.device.Device`
 920        object's data content.
 921        """
 922        # set temperature back to None so that if update() is called more than once
 923        # any logic that relies on self.temperature to be None to rescan it works.it
 924        self.temperature = None
 925        # same for temperatures
 926        self.temperatures = {}
 927        if self.abridged:
 928            interface = None
 929            raw = self.smartctl.info(self.dev_reference)
 930
 931        else:
 932            interface = smartctl_type(self._interface)
 933            raw = self.smartctl.all(
 934                self.dev_reference, interface)
 935
 936        parse_self_tests = False
 937        parse_running_test = False
 938        parse_ascq = False
 939        message = ''
 940        self.tests = []
 941        self._test_running = False
 942        self._test_progress = None
 943        # Lets skip the first couple of non-useful lines
 944        _stdout = raw[4:]
 945
 946        #######################################
 947        #           Encoding fixing           #
 948        #######################################
 949        # In some scenarios, smartctl returns some lines with a different/strange encoding
 950        # This is a workaround to fix that
 951        for i, line in enumerate(_stdout):
 952            # character ' ' (U+202F) should be removed
 953            _stdout[i] = line.replace('\u202f', '')
 954
 955        #######################################
 956        #   Dedicated interface attributes    #
 957        #######################################
 958
 959        if interface == 'nvme':
 960            self.if_attributes = NvmeAttributes(iter(_stdout))
 961        else:
 962            self.if_attributes = None
 963
 964        #######################################
 965        #    Global / generic  attributes     #
 966        #######################################
 967        stdout_iter = iter(_stdout)
 968        for line in stdout_iter:
 969            if line.strip() == '':  # Blank line stops sub-captures
 970                if parse_self_tests is True:
 971                    parse_self_tests = False
 972                if parse_ascq:
 973                    parse_ascq = False
 974                    self.messages.append(message)
 975            if parse_ascq:
 976                message += ' ' + line.lstrip().rstrip()
 977            if parse_self_tests:
 978                num = line[0:3]
 979                if '#' not in num:
 980                    continue
 981
 982                # Detect Test Format
 983
 984                ## SCSI/SAS FORMAT ##
 985                # Example smartctl output
 986                # SMART Self-test log
 987                # Num  Test              Status                 segment  LifeTime  LBA_first_err [SK ASC ASQ]
 988                #      Description                              number   (hours)
 989                # # 1  Background short  Completed                   -   33124                 - [-   -    -]
 990                format_scsi = re.compile(
 991                    r'^[#\s]*([^\s]+)\s{2,}(.*[^\s])\s{2,}(.*[^\s])\s{2,}(.*[^\s])\s{2,}(.*[^\s])\s{2,}(.*[^\s])\s+\[([^\s]+)\s+([^\s]+)\s+([^\s]+)\]$').match(line)
 992
 993                if format_scsi is not None:
 994                    format = 'scsi'
 995                    parsed = format_scsi.groups()
 996                    num = int(parsed[0])
 997                    test_type = parsed[1]
 998                    status = parsed[2]
 999                    segment = parsed[3]
1000                    hours = parsed[4]
1001                    lba = parsed[5]
1002                    sense = parsed[6]
1003                    asc = parsed[7]
1004                    ascq = parsed[8]
1005                    self.tests.append(TestEntry(
1006                        format,
1007                        num,
1008                        test_type,
1009                        status,
1010                        hours,
1011                        lba,
1012                        segment=segment,
1013                        sense=sense,
1014                        asc=asc,
1015                        ascq=ascq
1016                    ))
1017                else:
1018                    ## ATA FORMAT ##
1019                    # Example smartctl output:
1020                    # SMART Self-test log structure revision number 1
1021                    # Num  Test_Description    Status                  Remaining  LifeTime(hours)  LBA_of_first_error
1022                    # # 1  Extended offline    Completed without error       00%     46660         -
1023                    format = 'ata'
1024                    parsed = re.compile(
1025                        r'^[#\s]*([^\s]+)\s{2,}(.*[^\s])\s{2,}(.*[^\s])\s{1,}(.*[^\s])\s{2,}(.*[^\s])\s{2,}(.*[^\s])$').match(line).groups()
1026                    num = parsed[0]
1027                    test_type = parsed[1]
1028                    status = parsed[2]
1029                    remain = parsed[3]
1030                    hours = parsed[4]
1031                    lba = parsed[5]
1032
1033                    try:
1034                        num = int(num)
1035                    except:
1036                        num = None
1037
1038                    self.tests.append(
1039                        TestEntry(format, num, test_type, status,
1040                                  hours, lba, remain=remain)
1041                    )
1042            # Basic device information parsing
1043            if any_in(line, 'Device Model', 'Product', 'Model Number'):
1044                self.model = line.split(':')[1].lstrip().rstrip()
1045                self._guess_smart_type(line.lower())
1046                continue
1047
1048            if 'Model Family' in line:
1049                self.family = line.split(':')[1].strip()
1050                self._guess_smart_type(line.lower())
1051                continue
1052
1053            if 'LU WWN' in line:
1054                self._guess_smart_type(line.lower())
1055                continue
1056
1057            if any_in(line, 'Serial Number', 'Serial number'):
1058                self.serial = line.split(':')[1].split()[0].rstrip()
1059                continue
1060
1061            vendor = re.compile(r'^Vendor:\s+(\w+)').match(line)
1062            if vendor is not None:
1063                self._vendor = vendor.groups()[0]
1064
1065            if any_in(line, 'Firmware Version', 'Revision'):
1066                self.firmware = line.split(':')[1].strip()
1067
1068            if any_in(line, 'User Capacity', 'Total NVM Capacity', 'Namespace 1 Size/Capacity'):
1069                # TODO: support for multiple NVMe namespaces
1070                m = re.match(
1071                    r'.*:\s+([\d,.]+)\s\D*\[?([^\]]+)?\]?', line.strip())
1072
1073                if m is not None:
1074                    tmp = m.groups()
1075                    self._capacity = int(
1076                        tmp[0].strip().replace(',', '').replace('.', ''))
1077
1078                    if len(tmp) == 2 and tmp[1] is not None:
1079                        self._capacity_human = tmp[1].strip().replace(',', '.')
1080
1081            if 'SMART support' in line:
1082                # self.smart_capable = 'Available' in line
1083                # self.smart_enabled = 'Enabled' in line
1084                # Since this line repeats twice the above method is flawed
1085                # Lets try the following instead, it is a bit redundant but
1086                # more robust.
1087                if any_in(line, 'Unavailable', 'device lacks SMART capability'):
1088                    self.smart_capable = False
1089                    self.smart_enabled = False
1090                elif 'Enabled' in line:
1091                    self.smart_enabled = True
1092                elif 'Disabled' in line:
1093                    self.smart_enabled = False
1094                elif any_in(line, 'Available', 'device has SMART capability'):
1095                    self.smart_capable = True
1096                continue
1097
1098            if 'does not support SMART' in line:
1099                self.smart_capable = False
1100                self.smart_enabled = False
1101                continue
1102
1103            if 'Rotation Rate' in line:
1104                if 'Solid State Device' in line:
1105                    self.is_ssd = True
1106                elif 'rpm' in line:
1107                    self.is_ssd = False
1108                    try:
1109                        self.rotation_rate = int(
1110                            line.split(':')[1].lstrip().rstrip()[:-4])
1111                    except ValueError:
1112                        # Cannot parse the RPM? Assigning None instead
1113                        self.rotation_rate = None
1114                continue
1115
1116            if 'SMART overall-health self-assessment' in line:  # ATA devices
1117                if line.split(':')[1].strip() == 'PASSED':
1118                    self.assessment = 'PASS'
1119                else:
1120                    self.assessment = 'FAIL'
1121                continue
1122
1123            if 'SMART Health Status' in line:  # SCSI devices
1124                if line.split(':')[1].strip() == 'OK':
1125                    self.assessment = 'PASS'
1126                else:
1127                    self.assessment = 'FAIL'
1128                    parse_ascq = True  # Set flag to capture status message
1129                    message = line.split(':')[1].lstrip().rstrip()
1130                continue
1131
1132            # Parse SMART test capabilities (ATA only)
1133            # Note: SCSI does not list this but and allows for only 'offline', 'short' and 'long'
1134            if 'SMART execute Offline immediate' in line:
1135                self.test_capabilities['offline'] = 'No' not in line
1136                continue
1137
1138            if 'Conveyance Self-test supported' in line:
1139                self.test_capabilities['conveyance'] = 'No' not in line
1140                continue
1141
1142            if 'Selective Self-test supported' in line:
1143                self.test_capabilities['selective'] = 'No' not in line
1144                continue
1145
1146            if 'Self-test supported' in line:
1147                self.test_capabilities['short'] = 'No' not in line
1148                self.test_capabilities['long'] = 'No' not in line
1149                continue
1150
1151            # SMART Attribute table parsing
1152            if all_in(line, '0x0', '_') and not interface == 'nvme':
1153                # Replace multiple space separators with a single space, then
1154                # tokenize the string on space delimiters
1155                line_ = ' '.join(line.split()).split(' ')
1156                if '' not in line_:
1157                    self.attributes[int(line_[0])] = Attribute(
1158                        int(line_[0]), line_[1], int(line[2], base=16), line_[3], line_[4], line_[5], line_[6], line_[7], line_[8], line_[9])
1159            # For some reason smartctl does not show a currently running test
1160            # for 'ATA' in the Test log so I just have to catch it this way i guess!
1161            # For 'scsi' I still do it since it is the only place I get % remaining in scsi
1162            if 'Self-test execution status' in line:
1163                if 'progress' in line:
1164                    self._test_running = True
1165                    # for ATA the "%" remaining is on the next line
1166                    # thus set the parse_running_test flag and move on
1167                    parse_running_test = True
1168                elif '%' in line:
1169                    # for scsi the progress is on the same line
1170                    # so we can just parse it and move on
1171                    self._test_running = True
1172                    try:
1173                        self._test_progress = 100 - \
1174                            int(line.split('%')[0][-3:].strip())
1175                    except ValueError:
1176                        pass
1177                continue
1178            if parse_running_test is True:
1179                try:
1180                    self._test_progress = 100 - \
1181                        int(line.split('%')[0][-3:].strip())
1182                except ValueError:
1183                    pass
1184                parse_running_test = False
1185
1186            if all_in(line, 'Description', '(hours)'):
1187                parse_self_tests = True  # Set flag to capture test entries
1188
1189            #######################################
1190            #              SCSI only              #
1191            #######################################
1192            #
1193            # Everything from here on is parsing SCSI information that takes
1194            # the place of similar ATA SMART information
1195            if 'used endurance' in line:
1196                pct = int(line.split(':')[1].strip()[:-1])
1197                self.diagnostics.Life_Left = 100 - pct
1198                continue
1199
1200            if 'Specified cycle count' in line:
1201                self.diagnostics.Start_Stop_Spec = int(
1202                    line.split(':')[1].strip())
1203                continue
1204
1205            if 'Accumulated start-stop cycles' in line:
1206                self.diagnostics.Start_Stop_Cycles = int(
1207                    line.split(':')[1].strip())
1208                if self.diagnostics.Start_Stop_Spec and self.diagnostics.Start_Stop_Spec != 0:
1209                    self.diagnostics.Start_Stop_Pct_Left = int(round(
1210                        100 - (self.diagnostics.Start_Stop_Cycles /
1211                               self.diagnostics.Start_Stop_Spec), 0))
1212                continue
1213
1214            if 'Specified load-unload count' in line:
1215                self.diagnostics.Load_Cycle_Spec = int(
1216                    line.split(':')[1].strip())
1217                continue
1218
1219            if 'Accumulated load-unload cycles' in line:
1220                self.diagnostics.Load_Cycle_Count = int(
1221                    line.split(':')[1].strip())
1222                if self.diagnostics.Load_Cycle_Spec and self.diagnostics.Load_Cycle_Spec != 0:
1223                    self.diagnostics.Load_Cycle_Pct_Left = int(round(
1224                        100 - (self.diagnostics.Load_Cycle_Count /
1225                               self.diagnostics.Load_Cycle_Spec), 0))
1226                continue
1227
1228            if 'Elements in grown defect list' in line:
1229                self.diagnostics.Reallocated_Sector_Ct = int(
1230                    line.split(':')[1].strip())
1231                continue
1232
1233            if 'read:' in line:
1234                line_ = ' '.join(line.split()).split(' ')
1235                if line_[1] == '0' and line_[2] == '0' and line_[3] == '0' and line_[4] == '0':
1236                    self.diagnostics.Corrected_Reads = 0
1237                elif line_[4] == '0':
1238                    self.diagnostics.Corrected_Reads = int(
1239                        line_[1]) + int(line_[2]) + int(line_[3])
1240                else:
1241                    self.diagnostics.Corrected_Reads = int(line_[4])
1242                self.diagnostics._Reads_GB = float(line_[6].replace(',', '.'))
1243                self.diagnostics._Uncorrected_Reads = int(line_[7])
1244                continue
1245
1246            if 'write:' in line:
1247                line_ = ' '.join(line.split()).split(' ')
1248                if (line_[1] == '0' and line_[2] == '0' and
1249                        line_[3] == '0' and line_[4] == '0'):
1250                    self.diagnostics.Corrected_Writes = 0
1251                elif line_[4] == '0':
1252                    self.diagnostics.Corrected_Writes = int(
1253                        line_[1]) + int(line_[2]) + int(line_[3])
1254                else:
1255                    self.diagnostics.Corrected_Writes = int(line_[4])
1256                self.diagnostics._Writes_GB = float(line_[6].replace(',', '.'))
1257                self.diagnostics._Uncorrected_Writes = int(line_[7])
1258                continue
1259
1260            if 'verify:' in line:
1261                line_ = ' '.join(line.split()).split(' ')
1262                if (line_[1] == '0' and line_[2] == '0' and
1263                        line_[3] == '0' and line_[4] == '0'):
1264                    self.diagnostics.Corrected_Verifies = 0
1265                elif line_[4] == '0':
1266                    self.diagnostics.Corrected_Verifies = int(
1267                        line_[1]) + int(line_[2]) + int(line_[3])
1268                else:
1269                    self.diagnostics.Corrected_Verifies = int(line_[4])
1270                self.diagnostics._Verifies_GB = float(
1271                    line_[6].replace(',', '.'))
1272                self.diagnostics._Uncorrected_Verifies = int(line_[7])
1273                continue
1274
1275            if 'non-medium error count' in line:
1276                self.diagnostics.Non_Medium_Errors = int(
1277                    line.split(':')[1].strip())
1278                continue
1279
1280            if 'Accumulated power on time' in line:
1281                self.diagnostics.Power_On_Hours = int(
1282                    line.split(':')[1].split(' ')[1])
1283                continue
1284
1285            if 'Current Drive Temperature' in line or ('Temperature:' in
1286                                                       line and interface == 'nvme'):
1287                try:
1288                    self.temperature = int(
1289                        line.split(':')[-1].strip().split()[0])
1290
1291                    if 'fahrenheit' in line.lower():
1292                        self.temperature = int((self.temperature - 32) * 5 / 9)
1293
1294                except ValueError:
1295                    pass
1296
1297                continue
1298
1299            if 'Temperature Sensor ' in line:
1300                try:
1301                    match = re.search(
1302                        r'Temperature\sSensor\s([0-9]+):\s+(-?[0-9]+)', line)
1303                    if match:
1304                        (tempsensor_number_s, tempsensor_value_s) = match.group(1, 2)
1305                        tempsensor_number = int(tempsensor_number_s)
1306                        tempsensor_value = int(tempsensor_value_s)
1307
1308                        if 'fahrenheit' in line.lower():
1309                            tempsensor_value = int(
1310                                (tempsensor_value - 32) * 5 / 9)
1311
1312                        self.temperatures[tempsensor_number] = tempsensor_value
1313                        if self.temperature is None or tempsensor_number == 0:
1314                            self.temperature = tempsensor_value
1315                except ValueError:
1316                    pass
1317
1318                continue
1319
1320            #######################################
1321            #            Common values            #
1322            #######################################
1323
1324            # Sector sizes
1325            if 'Sector Sizes' in line:  # ATA
1326                m = re.match(
1327                    r'.* (\d+) bytes logical,\s*(\d+) bytes physical', line)
1328                if m:
1329                    self.logical_sector_size = int(m.group(1))
1330                    self.physical_sector_size = int(m.group(2))
1331                    # set diagnostics block size to physical sector size
1332                    self.diagnostics._block_size = self.physical_sector_size
1333                continue
1334            if 'Logical block size:' in line:  # SCSI 1/2
1335                self.logical_sector_size = int(
1336                    line.split(':')[1].strip().split(' ')[0])
1337                # set diagnostics block size to logical sector size
1338                self.diagnostics._block_size = self.logical_sector_size
1339                continue
1340            if 'Physical block size:' in line:  # SCSI 2/2
1341                self.physical_sector_size = int(
1342                    line.split(':')[1].strip().split(' ')[0])
1343                continue
1344            if 'Namespace 1 Formatted LBA Size' in line:  # NVMe
1345                # Note: we will assume that there is only one namespace
1346                self.logical_sector_size = int(
1347                    line.split(':')[1].strip().split(' ')[0])
1348                continue
1349
1350        if not self.abridged:
1351            if not interface == 'scsi':
1352                # Parse the SMART table for below-threshold attributes and create
1353                # corresponding warnings for non-SCSI disks
1354                self._make_smart_warnings()
1355            else:
1356                # If not obtained Power_On_Hours above, make a direct attempt to extract power on
1357                # hours from the background scan results log.
1358                if self.diagnostics.Power_On_Hours is None:
1359                    raw, returncode = self.smartctl.generic_call(
1360                        [
1361                            '-d',
1362                            'scsi',
1363                            '-l',
1364                            'background',
1365                            self.dev_reference
1366                        ])
1367
1368                    for line in raw:
1369                        if 'power on time' in line:
1370                            self.diagnostics.Power_On_Hours = int(
1371                                line.split(':')[1].split(' ')[1])
1372        # map temperature
1373        if self.temperature is None:
1374            # in this case the disk is probably ata
1375            try:
1376                # Some disks report temperature to attribute number 190 ('Airflow_Temperature_Cel')
1377                # see https://bugs.freenas.org/issues/20860
1378                temp_attr = self.attributes[194] or self.attributes[190]
1379                self.temperature = int(temp_attr.raw)
1380            except (ValueError, AttributeError):
1381                pass
1382        # Now that we have finished the update routine, if we did not find a runnning selftest
1383        # nuke the self._test_ECD and self._test_progress
1384        if self._test_running is False:
1385            self._test_ECD = None
1386            self._test_progress = None