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,
 35                 remain=None,
 36                 segment=None,
 37                 sense=None,
 38                 asc=None,
 39                 ascq=None,
 40                 nsid=None,
 41                 sct=None,
 42                 code=None):
 43
 44        self._format = format
 45        """
 46        **(str):** Indicates whether this entry was taken from an 'ata' or
 47        'scsi' self-test log. Used to display the content properly.
 48        """
 49        self.num: Optional[int] = num
 50        """
 51        **(int):** Entry's position in the log from 1 (most recent) to 21
 52        (least recent).  ATA logs save the last 21 entries while SCSI logs
 53        only save the last 20.
 54        """
 55        self.type = test_type
 56        """
 57        **(str):** Type of test run.  Generally short, long (extended), or
 58        conveyance, plus offline (background) or captive (foreground).
 59        """
 60        self.status = status
 61        """
 62        **(str):** Self-test's status message, for example 'Completed without
 63        error' or 'Completed: read failure'.
 64        """
 65        self.hours = hours
 66        """
 67        **(str):** The device's power-on hours at the time the self-test
 68        was initiated.
 69        """
 70        self.LBA = lba
 71        """
 72        **(str):** Indicates the first LBA at which an error was encountered
 73        during this self-test. Presented as a decimal value for ATA/SATA
 74        devices and in hexadecimal notation for SAS/SCSI devices.
 75        """
 76        self.remain = remain
 77        """
 78        **(str):** Percentage value indicating how much of the self-test is
 79        left to perform. '00%' indicates a complete test, while any other
 80        value could indicate a test in progress or one that failed prior to
 81        completion. Only reported by ATA devices.
 82        """
 83        self.segment = segment
 84        """
 85        **(str):** A manufacturer-specific self-test segment number reported
 86        by SCSI devices on self-test failure. Set to '-' otherwise.
 87        """
 88        self.sense = sense
 89        """
 90        **(str):** SCSI sense key reported on self-test failure. Set to '-'
 91        otherwise.
 92        """
 93        self.ASC = asc
 94        """
 95        **(str):** SCSI 'Additonal Sense Code' reported on self-test failure.
 96        Set to '-' otherwise.
 97        """
 98        self.ASCQ = ascq
 99        """
100        **(str):** SCSI 'Additonal Sense Code Quaifier' reported on self-test
101        failure. Set to '-' otherwise.
102        """
103        self.nsid = nsid
104        """
105        **(str):** NVMe 'Name Space Identifier' reported on self-test failure.
106        Set to '-' if no namespace is defined.
107        """
108        self.sct = sct
109        """
110        **(str):** NVMe 'Status Code Type' reported on self-test failure.
111        Set to '-' if undefined.
112        """
113        self.code = code
114        """
115        **(str):** NVMe 'Status Code' reported on self-test failure.
116        Set to '-' if undefined.
117        """
118
119    def __getstate__(self):
120        return {
121            'num': self.num,
122            'type': self.type,
123            'status': self.status,
124            'hours': self.hours,
125            'lba': self.LBA,
126            'remain': self.remain,
127            'segment': self.segment,
128            'sense': self.sense,
129            'asc': self.ASC,
130            'ascq': self.ASCQ,
131            'nsid': self.nsid,
132            'sct': self.sct,
133            'code': self.code
134        }
135
136    def __repr__(self):
137        """Define a basic representation of the class object."""
138        return "<SMART Self-test [%s|%s] hrs:%s LBA:%s>" % (
139            self.type, self.status, self.hours, self.LBA)
140
141    def __str__(self):
142        """
143        Define a formatted string representation of the object's content.
144        Looks nearly identical to the output of smartctl, without overflowing
145        80-character lines.
146        """
147        if self._format == 'ata':
148            return "{0:>2} {1:17}{2:30}{3:5}{4:7}{5:17}".format(
149                self.num, self.type, self.status, self.remain, self.hours,
150                self.LBA)
151        elif self._format == 'scsi':
152            # 'Segment' could not be fit on the 80-char line. It's of limited
153            # utility anyway due to it's manufacturer-proprietary nature...
154            return ("{0:>2} {1:17}{2:23}{3:7}{4:14}[{5:4}{6:5}{7:4}]".format(
155                self.num,
156                self.type,
157                self.status,
158                self.hours,
159                self.LBA,
160                self.sense,
161                self.ASC,
162                self.ASCQ
163            ))
164        elif self._format == 'nvme':
165            ## NVME FORMAT ##
166            # Example smartctl output
167            # Self-test Log (NVMe Log 0x06)
168            # Self-test status: Extended self-test in progress (28% completed)
169            # Num  Test_Description  Status                       Power_on_Hours  Failing_LBA  NSID Seg SCT Code
170            #  0   Extended          Completed without error                3441            -     -   -   -    -
171            return ("{0:^4} {1:<18}{2:<29}{3:>14}{4:>13}{5:>6}{6:>4}{7:>4}{8:>5}".format(
172                self.num,
173                self.type,
174                self.status,
175                self.hours,
176                self.LBA if self.LBA is not None else '-',
177                self.nsid if self.LBA is not None else '-',
178                self.segment if self.segment is not None else '-',
179                self.sct if self.LBA is not None else '-',
180                self.code if self.LBA is not None else '-'
181            ))
182        else:
183            return "Unknown test format: %s" % self._format

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

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