Skip to content

NVMe

This section documents the NVMe (Non-Volatile Memory Express) device functionality.

sts.nvme

NVMe device management.

This module provides functionality for managing NVMe devices: - Device discovery - Device information - Device operations

NVMe (Non-Volatile Memory Express) is a protocol designed for: - High-performance SSDs - Low latency access - Parallel operations - Advanced management features

Key advantages over SCSI/SATA: - Higher queue depths - Lower protocol overhead - Better error handling - More detailed device information

NvmeDevice dataclass

Bases: StorageDevice

NVMe device representation.

NVMe devices are identified by: - Controller number (e.g. nvme0) - Namespace ID (e.g. n1) - Combined name (e.g. nvme0n1)

Device information includes: - Model and serial number - Firmware version - Capacity and block size - Health and error logs

Parameters:

Name Type Description Default
name str | None

Device name (optional, e.g. 'nvme0n1')

 None
path Path | str | None

Device path (optional, defaults to /dev/)

 None
size int | None

Device size in bytes (optional, discovered from device)

 None
model str | None

Device model (optional, discovered from device)

 None
serial str | None

Device serial number (optional, discovered from device)

 None
firmware str | None

Device firmware version (optional, discovered from device)

 None
Example 
device = NvmeDevice(name='nvme0n1')  # Discovers other values
device = NvmeDevice(model='Samsung SSD 970 EVO')  # Discovers device by model
Source code in sts_libs/src/sts/nvme.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
@dataclass
class NvmeDevice(StorageDevice):
    """NVMe device representation.

    NVMe devices are identified by:
    - Controller number (e.g. nvme0)
    - Namespace ID (e.g. n1)
    - Combined name (e.g. nvme0n1)

    Device information includes:
    - Model and serial number
    - Firmware version
    - Capacity and block size
    - Health and error logs

    Args:
        name: Device name (optional, e.g. 'nvme0n1')
        path: Device path (optional, defaults to /dev/<name>)
        size: Device size in bytes (optional, discovered from device)
        model: Device model (optional, discovered from device)
        serial: Device serial number (optional, discovered from device)
        firmware: Device firmware version (optional, discovered from device)

    Example:
        ```python
        device = NvmeDevice(name='nvme0n1')  # Discovers other values
        device = NvmeDevice(model='Samsung SSD 970 EVO')  # Discovers device by model
        ```
    """

    # Optional parameters from parent classes
    name: str | None = None
    path: Path | str | None = None
    size: int | None = None
    model: str | None = None

    # Optional parameters for this class
    serial: str | None = None  # Device serial number
    firmware: str | None = None  # Firmware version

    # Device nodes path
    NVME_PATH: ClassVar[Path] = Path('/dev/nvme')

    def __post_init__(self) -> None:
        """Initialize NVMe device.

        Discovery process:
        1. Set device path if needed
        2. Get controller information (model, serial, firmware)
        3. Get namespace information (size, block size)

        Raises:
            DeviceError: If device cannot be initialized
        """
        # Set path based on name if not provided
        if not self.path and self.name:
            self.path = f'/dev/{self.name}'

        # Initialize parent class
        super().__post_init__()

        # Discover device info if path is available
        if self.path:
            # Get controller information (nvme id-ctrl)
            result = run(f'nvme id-ctrl {self.path} -o json')
            if result.succeeded and result.stdout:
                try:
                    info = json.loads(result.stdout)
                    if not self.model:
                        self.model = info.get('mn', '').strip()
                    if not self.serial:
                        self.serial = info.get('sn', '').strip()
                    if not self.firmware:
                        self.firmware = info.get('fr', '').strip()
                except json.JSONDecodeError:
                    pass

            # Get namespace information (nvme id-ns)
            if not self.size:
                result = run(f'nvme id-ns {self.path} -o json')
                if result.succeeded and result.stdout:
                    try:
                        info = json.loads(result.stdout)
                        if 'nsze' in info:
                            # Calculate size: blocks * block size
                            self.size = int(info['nsze']) * int(info.get('lbaf', [{}])[0].get('ds', 512))
                    except (json.JSONDecodeError, IndexError, ValueError):
                        pass

    def format(self) -> bool:
        """Format device.

        Performs a low-level format:
        - Erases all data
        - Resets metadata
        - May take significant time
        - Requires admin privileges

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            device.format()
            True
            ```
        """
        if not self.path:
            logging.error('Device path not available')
            return False

        result = run(f'nvme format {self.path}')
        return result.succeeded

    def sanitize(self) -> bool:
        """Sanitize device.

        Performs secure data erasure:
        - More thorough than format
        - May use crypto erase
        - Takes longer than format
        - Not all devices support this

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            device.sanitize()
            True
            ```
        """
        if not self.path:
            logging.error('Device path not available')
            return False

        result = run(f'nvme sanitize {self.path}')
        return result.succeeded

    def get_smart_log(self) -> dict[str, str]:
        """Get SMART log.

        Retrieves device health information:
        - Critical warnings
        - Temperature
        - Available spare
        - Media errors
        - Read/write statistics

        Returns:
            Dictionary of SMART log entries

        Example:
            ```python
            device.get_smart_log()
            {'critical_warning': '0x0', 'temperature': '35 C', ...}
            ```
        """
        if not self.path:
            return {}

        result = run(f'nvme smart-log {self.path} -o json')
        if result.failed or not result.stdout:
            return {}

        try:
            return json.loads(result.stdout)
        except json.JSONDecodeError:
            return {}

    def get_error_log(self) -> dict[str, str]:
        """Get error log.

        Retrieves device error history:
        - Error count
        - Error types
        - Error locations
        - Timestamps

        Returns:
            Dictionary of error log entries

        Example:
            ```python
            device.get_error_log()
            {'error_count': '0', 'error_entries': [], ...}
            ```
        """
        if not self.path:
            return {}

        result = run(f'nvme error-log {self.path} -o json')
        if result.failed or not result.stdout:
            return {}

        try:
            return json.loads(result.stdout)
        except json.JSONDecodeError:
            return {}

    @classmethod
    def get_all(cls) -> list[NvmeDevice]:
        """Get list of all NVMe devices.

        Lists all NVMe devices in the system:
        - Controllers and namespaces
        - Both active and standby
        - Including hot-plugged devices

        Returns:
            List of NvmeDevice instances

        Example:
            ```python
            NvmeDevice.get_all()
            [NvmeDevice(name='nvme0n1', ...), NvmeDevice(name='nvme1n1', ...)]
            ```
        """
        # Get device list using nvme-cli
        result = run('nvme list -o json')
        if result.failed or not result.stdout:
            return []

        # Parse JSON output
        try:
            data = json.loads(result.stdout)
        except json.JSONDecodeError:
            logging.warning('Failed to parse nvme list output')
            return []

        # Extract device paths from JSON
        device_paths = [dev_info['DevicePath'] for dev_info in data.get('Devices', []) if 'DevicePath' in dev_info]

        # Create device objects
        devices = []
        for path in device_paths:
            name = Path(path).name
            try:
                device = cls(name=name)
                devices.append(device)
            except (ValueError, TypeError, DeviceError) as e:
                logging.warning(f'Failed to create device {name}: {e}')

        return devices

    @classmethod
    def get_by_model(cls, model: str) -> list[NvmeDevice]:
        """Get devices by model.

        Finds devices matching model string:
        - Case-sensitive match
        - Returns multiple if found
        - Empty list if none found

        Args:
            model: Device model (e.g. 'Samsung SSD 970 EVO')

        Returns:
            List of NvmeDevice instances

        Example:
            ```python
            NvmeDevice.get_by_model('Samsung SSD 970 EVO')
            [NvmeDevice(name='nvme0n1', ...), NvmeDevice(name='nvme1n1', ...)]
            ```
        """
        return [device for device in cls.get_all() if device.model == model]

__post_init__()

Initialize NVMe device.

Discovery process: 1. Set device path if needed 2. Get controller information (model, serial, firmware) 3. Get namespace information (size, block size)

Raises:

Type Description
DeviceError

If device cannot be initialized
Source code in sts_libs/src/sts/nvme.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def __post_init__(self) -> None:
    """Initialize NVMe device.

    Discovery process:
    1. Set device path if needed
    2. Get controller information (model, serial, firmware)
    3. Get namespace information (size, block size)

    Raises:
        DeviceError: If device cannot be initialized
    """
    # Set path based on name if not provided
    if not self.path and self.name:
        self.path = f'/dev/{self.name}'

    # Initialize parent class
    super().__post_init__()

    # Discover device info if path is available
    if self.path:
        # Get controller information (nvme id-ctrl)
        result = run(f'nvme id-ctrl {self.path} -o json')
        if result.succeeded and result.stdout:
            try:
                info = json.loads(result.stdout)
                if not self.model:
                    self.model = info.get('mn', '').strip()
                if not self.serial:
                    self.serial = info.get('sn', '').strip()
                if not self.firmware:
                    self.firmware = info.get('fr', '').strip()
            except json.JSONDecodeError:
                pass

        # Get namespace information (nvme id-ns)
        if not self.size:
            result = run(f'nvme id-ns {self.path} -o json')
            if result.succeeded and result.stdout:
                try:
                    info = json.loads(result.stdout)
                    if 'nsze' in info:
                        # Calculate size: blocks * block size
                        self.size = int(info['nsze']) * int(info.get('lbaf', [{}])[0].get('ds', 512))
                except (json.JSONDecodeError, IndexError, ValueError):
                    pass

format()

Format device.

Performs a low-level format: - Erases all data - Resets metadata - May take significant time - Requires admin privileges

Returns:

Type Description
bool

True if successful, False otherwise
Example 
device.format()
True
Source code in sts_libs/src/sts/nvme.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def format(self) -> bool:
    """Format device.

    Performs a low-level format:
    - Erases all data
    - Resets metadata
    - May take significant time
    - Requires admin privileges

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        device.format()
        True
        ```
    """
    if not self.path:
        logging.error('Device path not available')
        return False

    result = run(f'nvme format {self.path}')
    return result.succeeded

get_all() classmethod

Get list of all NVMe devices.

Lists all NVMe devices in the system: - Controllers and namespaces - Both active and standby - Including hot-plugged devices

Returns:

Type Description
list[NvmeDevice]

List of NvmeDevice instances
Example 
NvmeDevice.get_all()
[NvmeDevice(name='nvme0n1', ...), NvmeDevice(name='nvme1n1', ...)]
Source code in sts_libs/src/sts/nvme.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
@classmethod
def get_all(cls) -> list[NvmeDevice]:
    """Get list of all NVMe devices.

    Lists all NVMe devices in the system:
    - Controllers and namespaces
    - Both active and standby
    - Including hot-plugged devices

    Returns:
        List of NvmeDevice instances

    Example:
        ```python
        NvmeDevice.get_all()
        [NvmeDevice(name='nvme0n1', ...), NvmeDevice(name='nvme1n1', ...)]
        ```
    """
    # Get device list using nvme-cli
    result = run('nvme list -o json')
    if result.failed or not result.stdout:
        return []

    # Parse JSON output
    try:
        data = json.loads(result.stdout)
    except json.JSONDecodeError:
        logging.warning('Failed to parse nvme list output')
        return []

    # Extract device paths from JSON
    device_paths = [dev_info['DevicePath'] for dev_info in data.get('Devices', []) if 'DevicePath' in dev_info]

    # Create device objects
    devices = []
    for path in device_paths:
        name = Path(path).name
        try:
            device = cls(name=name)
            devices.append(device)
        except (ValueError, TypeError, DeviceError) as e:
            logging.warning(f'Failed to create device {name}: {e}')

    return devices

get_by_model(model) classmethod

Get devices by model.

Finds devices matching model string: - Case-sensitive match - Returns multiple if found - Empty list if none found

Parameters:

Name Type Description Default
model str

Device model (e.g. 'Samsung SSD 970 EVO')

 required

Returns:

Type Description
list[NvmeDevice]

List of NvmeDevice instances
Example 
NvmeDevice.get_by_model('Samsung SSD 970 EVO')
[NvmeDevice(name='nvme0n1', ...), NvmeDevice(name='nvme1n1', ...)]
Source code in sts_libs/src/sts/nvme.py 
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
@classmethod
def get_by_model(cls, model: str) -> list[NvmeDevice]:
    """Get devices by model.

    Finds devices matching model string:
    - Case-sensitive match
    - Returns multiple if found
    - Empty list if none found

    Args:
        model: Device model (e.g. 'Samsung SSD 970 EVO')

    Returns:
        List of NvmeDevice instances

    Example:
        ```python
        NvmeDevice.get_by_model('Samsung SSD 970 EVO')
        [NvmeDevice(name='nvme0n1', ...), NvmeDevice(name='nvme1n1', ...)]
        ```
    """
    return [device for device in cls.get_all() if device.model == model]

get_error_log()

Get error log.

Retrieves device error history: - Error count - Error types - Error locations - Timestamps

Returns:

Type Description
dict[str, str]

Dictionary of error log entries
Example 
device.get_error_log()
{'error_count': '0', 'error_entries': [], ...}
Source code in sts_libs/src/sts/nvme.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
def get_error_log(self) -> dict[str, str]:
    """Get error log.

    Retrieves device error history:
    - Error count
    - Error types
    - Error locations
    - Timestamps

    Returns:
        Dictionary of error log entries

    Example:
        ```python
        device.get_error_log()
        {'error_count': '0', 'error_entries': [], ...}
        ```
    """
    if not self.path:
        return {}

    result = run(f'nvme error-log {self.path} -o json')
    if result.failed or not result.stdout:
        return {}

    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError:
        return {}

get_smart_log()

Get SMART log.

Retrieves device health information: - Critical warnings - Temperature - Available spare - Media errors - Read/write statistics

Returns:

Type Description
dict[str, str]

Dictionary of SMART log entries
Example 
device.get_smart_log()
{'critical_warning': '0x0', 'temperature': '35 C', ...}
Source code in sts_libs/src/sts/nvme.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
def get_smart_log(self) -> dict[str, str]:
    """Get SMART log.

    Retrieves device health information:
    - Critical warnings
    - Temperature
    - Available spare
    - Media errors
    - Read/write statistics

    Returns:
        Dictionary of SMART log entries

    Example:
        ```python
        device.get_smart_log()
        {'critical_warning': '0x0', 'temperature': '35 C', ...}
        ```
    """
    if not self.path:
        return {}

    result = run(f'nvme smart-log {self.path} -o json')
    if result.failed or not result.stdout:
        return {}

    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError:
        return {}

sanitize()

Sanitize device.

Performs secure data erasure: - More thorough than format - May use crypto erase - Takes longer than format - Not all devices support this

Returns:

Type Description
bool

True if successful, False otherwise
Example 
device.sanitize()
True
Source code in sts_libs/src/sts/nvme.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def sanitize(self) -> bool:
    """Sanitize device.

    Performs secure data erasure:
    - More thorough than format
    - May use crypto erase
    - Takes longer than format
    - Not all devices support this

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        device.sanitize()
        True
        ```
    """
    if not self.path:
        logging.error('Device path not available')
        return False

    result = run(f'nvme sanitize {self.path}')
    return result.succeeded