added diagnostic class to encapsulate SCSI/SAS data

old diag dict is now deprecated but still accsible as a property

Author: ralequi

.gitignore

@@ -6,6 +6,7 @@ pySMART.egg-info
 .vscode
 .pypirc
 *.pyc
+__pycache__
 build/
 dist/
 .coverage

pySMART/device.py

@@ -31,12 +31,12 @@
 import os
 import re
 import warnings
-from subprocess import Popen, PIPE
 from time import time, strptime, mktime, sleep
-from typing import Tuple, Union, List
+from typing import Tuple, Union, List, Dict
 
 # pySMART module imports
 from .attribute import Attribute
+from .diagnostics import Diagnostics
 from .testentry import TestEntry
 from .smartctl import Smartctl
 from .utils import smartctl_type, smartctl_isvalid_type, any_in, all_in
@@ -192,9 +192,9 @@ def __init__(self, name: str, interface: str = None, abridged: bool = False, sma
         """
         **(int):** Estimate progress percantage of the running SMART selftest.
         """
-        self.diags = {}
+        self.diagnostics: Diagnostics = Diagnostics()
         """
-        **(dict of str):** Contains parsed and processed diagnostic information
+        **Diagnostics** Contains parsed and processed diagnostic information
         extracted from the SMART information. Currently only populated for
         SAS and SCSI devices, since ATA/SATA SMART attributes are manufacturer
         proprietary.
@@ -273,6 +273,12 @@ def capacity(self) -> str:
         """
         return self._capacity
 
+    @property
+    def diags(self) -> Dict[str, str]:
+        """Gets the old/deprecated version of SCSI/SAS diags atribute.
+        """
+        return self.diagnostics.get_classic_format()
+
     @property
     def size_raw(self) -> str:
         """Returns the capacity in the raw smartctl format.
@@ -317,7 +323,7 @@ def __getstate__(self, all_info=True):
             'messages': self.messages,
             'test_capabilities': self.test_capabilities.copy(),
             'tests': [t.__getstate__() for t in self.tests] if self.tests else [],
-            'diagnostics': self.diags.copy(),
+            'diagnostics': self.diagnostics.__getstate__(),
             'temperature': self.temperature,
             'attributes': [attr.__getstate__() if attr else None for attr in self.attributes]
         }
@@ -972,69 +978,71 @@ class members, including the SMART attribute table and self-test log.
             # the place of similar ATA SMART information
             if 'used endurance' in line:
                 pct = int(line.split(':')[1].strip()[:-1])
-                self.diags['Life_Left'] = str(100 - pct) + '%'
+                self.diagnostics.Life_Left = 100 - pct
             if 'Specified cycle count' in line:
-                self.diags['Start_Stop_Spec'] = line.split(':')[1].strip()
-                if self.diags['Start_Stop_Spec'] == '0':
-                    self.diags['Start_Stop_Pct_Left'] = '-'
+                self.diagnostics.Start_Stop_Spec = int(
+                    line.split(':')[1].strip())
             if 'Accumulated start-stop cycles' in line:
-                self.diags['Start_Stop_Cycles'] = line.split(':')[1].strip()
-                if 'Start_Stop_Pct_Left' not in self.diags:
-                    self.diags['Start_Stop_Pct_Left'] = str(int(round(
-                        100 - (int(self.diags['Start_Stop_Cycles']) /
-                               int(self.diags['Start_Stop_Spec'])), 0))) + '%'
+                self.diagnostics.Start_Stop_Cycles = int(
+                    line.split(':')[1].strip())
+                if self.diagnostics.Start_Stop_Spec != 0:
+                    self.diagnostics.Start_Stop_Pct_Left = int(round(
+                        100 - (self.diagnostics.Start_Stop_Cycles /
+                               self.diagnostics.Start_Stop_Spec), 0))
             if 'Specified load-unload count' in line:
-                self.diags['Load_Cycle_Spec'] = line.split(':')[1].strip()
-                if self.diags['Load_Cycle_Spec'] == '0':
-                    self.diags['Load_Cycle_Pct_Left'] = '-'
+                self.diagnostics.Load_Cycle_Spec = int(
+                    line.split(':')[1].strip())
             if 'Accumulated load-unload cycles' in line:
-                self.diags['Load_Cycle_Count'] = line.split(':')[1].strip()
-                if 'Load_Cycle_Pct_Left' not in self.diags:
-                    self.diags['Load_Cycle_Pct_Left'] = str(int(round(
-                        100 - (int(self.diags['Load_Cycle_Count']) /
-                               int(self.diags['Load_Cycle_Spec'])), 0))) + '%'
+                self.diagnostics.Load_Cycle_Count = int(
+                    line.split(':')[1].strip())
+                if self.diagnostics.Load_Cycle_Spec != 0:
+                    self.diagnostics.Load_Cycle_Pct_Left = int(round(
+                        100 - (self.diagnostics.Load_Cycle_Count /
+                               self.diagnostics.Load_Cycle_Spec), 0))
             if 'Elements in grown defect list' in line:
-                self.diags['Reallocated_Sector_Ct'] = line.split(':')[
-                    1].strip()
+                self.diagnostics.Reallocated_Sector_Ct = int(
+                    line.split(':')[1].strip())
             if 'read:' in line:
                 line_ = ' '.join(line.split()).split(' ')
                 if line_[1] == '0' and line_[2] == '0' and line_[3] == '0' and line_[4] == '0':
-                    self.diags['Corrected_Reads'] = '0'
+                    self.diagnostics.Corrected_Reads = 0
                 elif line_[4] == '0':
-                    self.diags['Corrected_Reads'] = str(
-                        int(line_[1]) + int(line_[2]) + int(line_[3]))
+                    self.diagnostics.Corrected_Reads = int(
+                        line_[1]) + int(line_[2]) + int(line_[3])
                 else:
-                    self.diags['Corrected_Reads'] = line_[4]
-                self.diags['Reads_GB'] = line_[6]
-                self.diags['Uncorrected_Reads'] = line_[7]
+                    self.diagnostics.Corrected_Reads = int(line_[4])
+                self.diagnostics.Reads_GB = float(line_[6])
+                self.diagnostics.Uncorrected_Reads = int(line_[7])
             if 'write:' in line:
                 line_ = ' '.join(line.split()).split(' ')
                 if (line_[1] == '0' and line_[2] == '0' and
                         line_[3] == '0' and line_[4] == '0'):
-                    self.diags['Corrected_Writes'] = '0'
+                    self.diagnostics.Corrected_Writes = 0
                 elif line_[4] == '0':
-                    self.diags['Corrected_Writes'] = str(
-                        int(line_[1]) + int(line_[2]) + int(line_[3]))
+                    self.diagnostics.Corrected_Writes = int(
+                        line_[1]) + int(line_[2]) + int(line_[3])
                 else:
-                    self.diags['Corrected_Writes'] = line_[4]
-                self.diags['Writes_GB'] = line_[6]
-                self.diags['Uncorrected_Writes'] = line_[7]
+                    self.diagnostics.Corrected_Writes = int(line_[4])
+                self.diagnostics.Writes_GB = float(line_[6])
+                self.diagnostics.Uncorrected_Writes = int(line_[7])
             if 'verify:' in line:
                 line_ = ' '.join(line.split()).split(' ')
                 if (line_[1] == '0' and line_[2] == '0' and
                         line_[3] == '0' and line_[4] == '0'):
-                    self.diags['Corrected_Verifies'] = '0'
+                    self.diagnostics.Corrected_Verifies = 0
                 elif line_[4] == '0':
-                    self.diags['Corrected_Verifies'] = str(
-                        int(line_[1]) + int(line_[2]) + int(line_[3]))
+                    self.diagnostics.Corrected_Verifies = int(
+                        line_[1]) + int(line_[2]) + int(line_[3])
                 else:
-                    self.diags['Corrected_Verifies'] = line_[4]
-                self.diags['Verifies_GB'] = line_[6]
-                self.diags['Uncorrected_Verifies'] = line_[7]
+                    self.diagnostics.Corrected_Verifies = int(line_[4])
+                self.diagnostics.Verifies_GB = float(line_[6])
+                self.diagnostics.Uncorrected_Verifies = int(line_[7])
             if 'non-medium error count' in line:
-                self.diags['Non-Medium_Errors'] = line.split(':')[1].strip()
+                self.diagnostics.Non_Medium_Errors = int(
+                    line.split(':')[1].strip())
             if 'Accumulated power on time' in line:
-                self.diags['Power_On_Hours'] = line.split(':')[1].split(' ')[1]
+                self.diagnostics.Power_On_Hours = int(
+                    line.split(':')[1].split(' ')[1])
             if 'Current Drive Temperature' in line or ('Temperature:' in
                                                        line and interface == 'nvme'):
                 try:
@@ -1061,22 +1069,9 @@ class members, including the SMART attribute table and self-test log.
                 # corresponding warnings for non-SCSI disks
                 self._make_smart_warnings()
             else:
-                # For SCSI disks, any diagnostic attribute which was not captured
-                # above gets set to '-' to indicate unsupported/unavailable.
-                for diag in ['Corrected_Reads', 'Corrected_Writes',
-                             'Corrected_Verifies', 'Uncorrected_Reads',
-                             'Uncorrected_Writes', 'Uncorrected_Verifies',
-                             'Reallocated_Sector_Ct',
-                             'Start_Stop_Spec', 'Start_Stop_Cycles',
-                             'Load_Cycle_Spec', 'Load_Cycle_Count',
-                             'Start_Stop_Pct_Left', 'Load_Cycle_Pct_Left',
-                             'Power_On_Hours', 'Life_Left', 'Non-Medium_Errors',
-                             'Reads_GB', 'Writes_GB', 'Verifies_GB']:
-                    if diag not in self.diags:
-                        self.diags[diag] = '-'
-                # If not obtained above, make a direct attempt to extract power on
+                # If not obtained Power_On_Hours above, make a direct attempt to extract power on
                 # hours from the background scan results log.
-                if self.diags['Power_On_Hours'] == '-':
+                if self.diagnostics.Power_On_Hours is None:
                     raw, returncode = self.smartctl.generic_call(
                         [
                             '-d',
@@ -1088,8 +1083,8 @@ class members, including the SMART attribute table and self-test log.
 
                     for line in raw:
                         if 'power on time' in line:
-                            self.diags['Power_On_Hours'] = line.split(':')[
-                                1].split(' ')[1]
+                            self.diagnostics.Power_On_Hours = int(
+                                line.split(':')[1].split(' ')[1])
         # map temperature
         if self.temperature is None:
             # in this case the disk is probably ata

pySMART/diagnostics.py

@@ -0,0 +1,127 @@
+# Copyright (C) 2021 Rafael Leira, Naudit HPCN S.L.
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License,
+# version 2, as published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
+# MA  02110-1301, USA.
+#
+################################################################
+"""
+This module contains the definition of the `Diagnostics` class/structure, used to
+represent SMART SCSI attributes associated with a `Device`.
+"""
+
+import copy
+import re
+from typing import Dict
+from pySMART.utils import get_object_properties
+
+
+class Diagnostics(object):
+    """
+    Contains all of the information associated with every SMART SCSI/SAS attribute
+    in a `Device`'s SMART table. This class pretends to provide a better view
+    of the data recolected from smartctl and its types.
+
+    Values set to None reflects that disk does not support such info
+    """
+
+    def __init__(self):
+        """Initialize the structure with every field set to None
+        """
+
+        # Generic counters
+        self.Reallocated_Sector_Ct: int = None
+        """**(int):** Reallocated sector count."""
+
+        self.Start_Stop_Spec: int = None
+        self.Start_Stop_Cycles: int = None
+        self.Start_Stop_Pct_Left: int = None
+        """**(int):** Percent left of the life-time start-stop cycles."""
+
+        self.Load_Cycle_Spec: int = None
+        self.Load_Cycle_Count: int = None
+        self.Load_Cycle_Pct_Left: int = None
+        """**(int):** Percent left of the life-time load cycles."""
+
+        self.Power_On_Hours: int = None
+        """**(int):** Number of hours the device have been powered on."""
+        self.Life_Left: int = None
+        """**(int):** Percent left of the whole disk life."""
+
+        # Error counters
+        self.Corrected_Reads: int = None
+        """**(float):** Total number of read operations that had an error but were corrected."""
+        self.Corrected_Writes: int = None
+        """**(float):** Total number of write operations that had an error but were corrected."""
+        self.Corrected_Verifies: int = None
+
+        self.Uncorrected_Reads: int = None
+        """**(float):** Total number of read operations that had an uncorrectable error."""
+        self.Uncorrected_Writes: int = None
+        """**(float):** Total number of write operations that had an uncorrectable error."""
+        self.Uncorrected_Verifies: int = None
+
+        self.Reads_GB: float = None
+        """**(float):** Total number of GBs readed in the disk life."""
+        self.Writes_GB: float = None
+        """**(float):** Total number of GBs written in the disk life."""
+        self.Verifies_GB: float = None
+
+        self.Non_Medium_Errors: int = None
+        """**(int):** Other errors not caused by this disk."""
+
+    def get_classic_format(self) -> Dict[str, str]:
+        """This method pretends to generate the previously/depreceted diag dictionary structure
+
+        Returns:
+            Dict[str,str]: the <1.1.0 PySMART diags structure
+        """
+        ret_dict = copy.deepcopy(vars(self))
+
+        # replace Non_Medium_Errors -> Non-Medium_Errors
+        ret_dict['Non-Medium_Errors'] = ret_dict['Non_Medium_Errors']
+        del ret_dict['Non_Medium_Errors']
+
+        # replace None with '-'
+        for value in ret_dict:
+            if ret_dict[value] is None:
+                ret_dict[value] = '-'
+
+        # ensure everything is a string
+        for value in ret_dict:
+            ret_dict[value] = str(ret_dict[value])
+
+        # include percent %
+        percent_values = [
+            'Life_Left',
+            'Start_Stop_Pct_Left',
+            'Load_Cycle_Pct_Left'
+        ]
+        for pv in percent_values:
+            if ret_dict[pv] != '-':
+                ret_dict[pv] = ret_dict[pv] + '%'
+
+        return ret_dict
+
+    def __getstate__(self, all_info=True):
+        """
+        Allows us to send a pySMART diagnostics object over a serializable
+        medium which uses json (or the likes of json) payloads
+        """
+        return vars(self)
+
+    def __setstate__(self, state):
+        self.__dict__.update(state)
+
+
+__all__ = ['Diagnostics']

pySMART/utils.py

@@ -20,11 +20,13 @@
 by the other submodules of the `pySMART` package.
 """
 
+import copy
 import io
 import logging
 import logging.handlers
 import os
 import traceback
+from typing import Dict, Any
 from shutil import which
 
 _srcfile = __file__
@@ -155,4 +157,31 @@ def smartctl_type(interface_type: str) -> str:
         return None
 
 
-__all__ = ['smartctl_type', 'SMARTCTL_PATH', 'all_in', 'any_in']
+def get_object_properties(obj: Any, deep_copy: bool = True, remove_private: bool = False) -> Dict[str, Any]:
+    type_name = type(obj).__name__
+    prop_names = dir(obj)
+
+    if deep_copy:
+        ret = copy.deepcopy(vars(obj))
+    else:
+        ret = vars(obj)
+
+    available_types = ['dict', 'str', 'int', 'float', 'list', 'NoneType']
+
+    for prop_name in prop_names:
+        prop_val = getattr(obj, prop_name)
+        prop_val_type_name = type(prop_val).__name__
+
+        if (prop_name[0] != '_') and (prop_val_type_name in available_types) and (prop_name not in ret):
+            ret[prop_name] = prop_val
+
+    if remove_private:
+        for key in ret.keys():
+            if key[0] == '_':
+                del ret[key]
+
+    return ret
+
+
+__all__ = ['smartctl_type', 'SMARTCTL_PATH',
+           'all_in', 'any_in', 'get_object_properties']