NVMe

NVMe devices (/dev/nvme*) are PCIe-attached storage with a controller → namespace hierarchy. Supports local NVMe and NVMe-oF (NVMe over Fabrics — RDMA or TCP transport) subsystems.

sts.nvme

NVMe device management.

Discovery uses nvme list -o json -v and handles both controller-attached and subsystem-level namespaces.

NvmeDevice pydantic-model

Bases: StorageDevice

NVMe device representation.

Note

The constructor only derives path from the device name. Call discover() or use get_all() to populate full metadata (model, serial, firmware, transport, etc.).

Example

device = NvmeDevice(name='nvme0n1').discover()

Show JSON schema:

{
  "additionalProperties": false,
  "description": "NVMe device representation.\n\nNote:\n    The constructor only derives ``path`` from the device name. Call\n    ``discover()`` or use ``get_all()`` to populate full metadata\n    (model, serial, firmware, transport, etc.).\n\nExample:\n    ```python\n    device = NvmeDevice(name='nvme0n1').discover()\n    ```",
  "properties": {
    "path": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "format": "path",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Path"
    },
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Name"
    },
    "size": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "model": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Model"
    },
    "serial": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Serial"
    },
    "firmware": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Firmware"
    },
    "nsid": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Nsid"
    },
    "transport": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Transport"
    },
    "address": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Address"
    },
    "sector_size": {
      "default": 512,
      "title": "Sector Size",
      "type": "integer"
    },
    "host_nqn": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Host Nqn"
    },
    "host_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Host Id"
    },
    "subsystem": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Subsystem"
    },
    "subsystem_nqn": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Subsystem Nqn"
    },
    "physical_size": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Physical Size"
    },
    "used_bytes": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Used Bytes"
    },
    "maximum_lba": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Maximum Lba"
    }
  },
  "title": "NvmeDevice",
  "type": "object"
}

Fields:

• name (str | None)
• path (PathOrStr | None)
• size (int | None)
• model (str | None)
• serial (str | None)
• firmware (str | None)
• nsid (int | None)
• transport (str | None)
• address (str | None)
• sector_size (int)
• host_nqn (str | None)
• host_id (str | None)
• subsystem (str | None)
• subsystem_nqn (str | None)
• physical_size (int | None)
• used_bytes (int | None)
• maximum_lba (int | None)

Validators:

• _derive_fields

Source code in sts_libs/src/sts/nvme.py

class NvmeDevice(StorageDevice):
    """NVMe device representation.

    Note:
        The constructor only derives ``path`` from the device name. Call
        ``discover()`` or use ``get_all()`` to populate full metadata
        (model, serial, firmware, transport, etc.).

    Example:
        ```python
        device = NvmeDevice(name='nvme0n1').discover()
        ```
    """

    # Optional parameters from parent classes
    name: str | None = None
    path: PathOrStr | 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
    nsid: int | None = None  # Namespace ID
    transport: str | None = None  # Transport type (pcie, fc, rdma, tcp)
    address: str | None = None  # Device address
    sector_size: int = 512  # Sector size in bytes
    host_nqn: str | None = None  # Host NQN
    host_id: str | None = None  # Host ID
    subsystem: str | None = None  # Subsystem name
    subsystem_nqn: str | None = None  # Subsystem NQN
    physical_size: int | None = None  # Physical device size
    used_bytes: int | None = None  # Used space
    maximum_lba: int | None = None  # Maximum LBA count

    @model_validator(mode='after')
    def _derive_fields(self) -> Self:
        """Derive path from device name, and name from path.

        Sets ``path`` to ``/dev/<name>``. ``controller`` is derived
        automatically via a computed field. Call
        ``_discover_from_nvme_list()`` separately for full metadata.
        """
        if not self.path and self.name:
            self.path = Path(f'/dev/{self.name}')

        if self.path and not self.name:
            self.name = Path(self.path).name

        return self

    @computed_field  # type: ignore[prop-decorator]
    @property
    def controller(self) -> str | None:
        """Controller name (e.g. nvme0), derived from device name."""
        if self.name:
            match = re.match(r'(nvme\d+)n\d+', self.name)
            return match.group(1) if match else None
        return None

    def _discover_from_nvme_list(self) -> bool:
        """Discover device information from ``nvme list -o json -v``.

        Parses ``Devices[] -> Subsystems[] -> Controllers[] -> Namespaces[]``.
        Namespaces may appear under a Controller or directly under a Subsystem
        (shared / NVMe-oF).
        """
        result = run('nvme list -o json -v')
        if not result.succeeded or not result.stdout:
            logger.warning('Failed to run nvme list command')
            return False

        try:
            data: dict[str, Any] = json.loads(result.stdout)
        except json.JSONDecodeError:
            logger.warning('Failed to parse nvme list JSON output')
            return False

        # Search through the JSON structure
        for device_info in data.get('Devices', []):
            # Get host information
            if 'HostNQN' in device_info:
                self.host_nqn = device_info['HostNQN']
            if 'HostID' in device_info:
                self.host_id = device_info['HostID']

            # Search through subsystems
            for subsystem in device_info.get('Subsystems', []):
                # Search through controllers
                for controller in subsystem.get('Controllers', []):
                    # Check controller and its namespaces
                    if 'Controller' in controller and controller.get('Controller') == self.controller:
                        self._extract_controller_info(controller)
                        if 'Subsystem' in subsystem:
                            self.subsystem = subsystem['Subsystem']
                        if 'SubsystemNQN' in subsystem:
                            self.subsystem_nqn = subsystem['SubsystemNQN']
                        for namespace in controller.get('Namespaces', []):
                            self._extract_namespace_info(namespace)
                        break

                # Also check namespaces at subsystem level
                for namespace in subsystem.get('Namespaces', []):
                    if 'NameSpace' in namespace and namespace.get('NameSpace') == self.name:
                        self._extract_namespace_info(namespace)
                        break
        return True

    def discover(self) -> Self:
        """Populate metadata (model, serial, firmware, transport, size) from nvme-cli."""
        self._discover_from_nvme_list()
        return self

    def _extract_controller_info(self, controller: dict[str, Any]) -> None:
        """Extract controller-level metadata.

        Sets ``serial``, ``model``, ``firmware``, ``transport``, and ``address``.
        """
        if 'SerialNumber' in controller:
            self.serial = controller['SerialNumber']
        if 'ModelNumber' in controller:
            self.model = controller['ModelNumber']
        if 'Firmware' in controller:
            self.firmware = controller['Firmware']
        if 'Transport' in controller:
            self.transport = controller['Transport']
        if 'Address' in controller:
            self.address = controller['Address']

    def _extract_namespace_info(self, namespace: dict[str, Any]) -> None:
        """Extract namespace-level metadata.

        Sets ``nsid``, ``used_bytes``, ``maximum_lba``, ``physical_size``, and ``sector_size``.
        Also copies ``physical_size`` to ``size`` when present.
        """
        if 'NSID' in namespace:
            self.nsid = namespace['NSID']
        if 'UsedBytes' in namespace:
            self.used_bytes = namespace['UsedBytes']
        if 'MaximumLBA' in namespace:
            self.maximum_lba = namespace['MaximumLBA']
        if 'PhysicalSize' in namespace:
            self.physical_size = namespace['PhysicalSize']
        if 'SectorSize' in namespace:
            self.sector_size = namespace['SectorSize']
        if self.physical_size:
            self.size = self.physical_size

    def _run(
        self,
        command: str,
        device: str | Path | None = None,
        arguments: Mapping[str, str | int | None] | None = None,
    ) -> CommandResult:
        """Build and execute an ``nvme <command>`` invocation."""
        command_list: list[str] = ['nvme', command]

        if device:
            command_list.append(str(device))

        # Add arguments
        if arguments is not None:
            # Handle different argument formats
            for k, v in arguments.items():
                if v is not None:
                    # Already combined format (e.g. '--key=value')
                    if '=' in k:
                        command_list.append(k)
                    # Long option: use --key=value format
                    elif k.startswith('--') and len(k) > 3:
                        command_list.append(f'{k}={v}')
                    # Short option: use -k value (space-separated)
                    else:
                        command_list.extend([k, str(v)])
                # Flag-only argument (no value)
                elif k.startswith('-'):
                    command_list.append(k)

        command_str: str = ' '.join(command_list)
        return run(command_str)

    @classmethod
    def has_nvme(cls) -> bool:
        """Check if the system has any NVMe devices."""
        result = run('nvme list')
        if result.failed or not result.stdout:
            return False

        return any(line.strip().startswith('/dev/nvme') for line in result.stdout.splitlines())

    @classmethod
    def get_all(cls) -> list[NvmeDevice]:
        """Discover all NVMe namespace devices on the system."""
        result = run('nvme list')
        if result.failed:
            logger.warning(f'Running "nvme list" failed:\n{result.stderr}')
            return []

        devices: list[NvmeDevice] = []
        for line in result.stdout.splitlines():
            # Lines like: /dev/nvme0n1     238.47  GB / 238.47  GB    512   B +  0 B
            if line.strip().startswith('/dev/nvme') and 'n' in line:
                device_path = line.strip().split()[0]
                device_name = device_path.replace('/dev/', '')
                if device_name:
                    device = cls(name=device_name)
                    device.discover()
                    devices.append(device)

        return devices

    @classmethod
    def get_by_attribute(cls, attribute: str, value: str) -> list[NvmeDevice]:
        """Find devices where ``attribute`` equals ``value`` (case-sensitive).

        Args:
            attribute: Device attribute name (e.g. 'model', 'serial', 'transport')
            value: Attribute value to match
        """
        if not attribute or not value:
            return []

        devices: list[NvmeDevice] = []
        for device in cls.get_all():
            device_value = getattr(device, attribute, None)
            if device_value and str(device_value) == str(value):
                devices.append(device)

        return devices

    def format(self, **kwargs: str | None) -> bool:
        """Low-level format the NVMe namespace (erases all data)."""
        if not self.path:
            logger.error('Device path not available')
            return False

        # Targets the namespace (e.g. /dev/nvme0n1)
        return self._run('format', device=self.path, arguments=kwargs).succeeded

    def sanitize(self, **kwargs: str | None) -> bool:
        """Sanitize the controller (block erase, crypto erase, or overwrite).

        Targets the controller (e.g. ``/dev/nvme0``), not the namespace.

        Examples:
            ```python
            device.sanitize(**{'--sanact': '0x01'})
            device.sanitize(**{'--sanact': '0x01', '--nodas': None})
            ```
        """
        if not self.path:
            logger.error('Device path not available')
            return False

        # Targets the controller (e.g. /dev/nvme0), not the namespace — sanitize is controller-level
        return self._run('sanitize', device=f'/dev/{self.controller}', arguments=kwargs).succeeded

    def reset(self, **kwargs: str | None) -> bool:
        """Reset the NVMe controller."""
        if not self.path:
            return False

        return self._run('reset', device=f'/dev/{self.controller}', arguments=kwargs).succeeded

    def flush(self, **kwargs: str | None) -> bool:
        """Commit data and metadata to nonvolatile media.

        Examples:
            ```python
            device.flush()
            device.flush(**{'--namespace-id': '1'})
            ```
        """
        if not self.path:
            return False

        return self._run('flush', device=self.path, arguments=kwargs).succeeded

    def get_smart_log(self, **kwargs: str | None) -> dict[str, Any]:
        """Get SMART health log as a dict."""
        if not self.path:
            return {}

        arguments = {'--output-format': 'json'} | kwargs

        result = self._run('smart-log', device=self.path, arguments=arguments)
        if result.failed or not result.stdout:
            return {}

        try:
            return json.loads(result.stdout)
        except json.JSONDecodeError:
            logger.warning('Failed to parse smart-log JSON output')
            return {}

    def get_error_log(self, **kwargs: str | None) -> dict[str, Any]:
        """Get error log as a dict.

        Example:
            ```python
            device.get_error_log(**{'-e': '1'})
            {'errors': [{'error_count': 76, ...}]}
            ```
        """
        if not self.path:
            return {}

        arguments = {'--output-format': 'json'} | kwargs

        result = self._run('error-log', device=self.path, arguments=arguments)
        if result.failed or not result.stdout:
            return {}

        try:
            return json.loads(result.stdout)
        except json.JSONDecodeError:
            logger.warning('Failed to parse error-log JSON output')
            return {}

    def get_fw_log(self, **kwargs: str | None) -> dict[str, Any]:
        """Get firmware slot log as a dict."""
        if not self.path:
            return {}

        arguments = {'--output-format': 'json'} | kwargs

        result = self._run('fw-log', device=self.path, arguments=arguments)
        if result.failed or not result.stdout:
            return {}

        try:
            return json.loads(result.stdout)
        except json.JSONDecodeError:
            logger.warning('Failed to parse fw-log JSON output')
            return {}

    def get_id_ctrl(self, **kwargs: str | None) -> dict[str, Any]:
        """Get controller identify data as a dict."""
        if not self.path:
            return {}

        arguments = {'--output-format': 'json'} | kwargs

        result = self._run('id-ctrl', device=self.path, arguments=arguments)
        if result.failed or not result.stdout:
            return {}

        try:
            return json.loads(result.stdout)
        except json.JSONDecodeError:
            logger.warning('Failed to parse id-ctrl JSON output')
            return {}

    def get_id_ns(self, **kwargs: str | None) -> dict[str, Any]:
        """Get namespace identify data as a dict."""
        if not self.path:
            return {}

        arguments = {'--output-format': 'json'} | kwargs

        result = self._run('id-ns', device=self.path, arguments=arguments)
        if result.failed or not result.stdout:
            return {}

        try:
            return json.loads(result.stdout)
        except json.JSONDecodeError:
            logger.warning('Failed to parse id-ns JSON output')
            return {}

    def list_ns(self, **kwargs: str | None) -> dict[str, Any]:
        """List namespaces attached to the controller."""
        if not self.controller:
            return {}

        arguments = {'--output-format': 'json'} | kwargs

        result = self._run('list-ns', device=f'/dev/{self.controller}', arguments=arguments)
        if result.failed or not result.stdout:
            return {}

        try:
            return json.loads(result.stdout)
        except json.JSONDecodeError:
            logger.warning('Failed to parse list-ns JSON output')
            return {}

    def get_feature(self, feature_id: str, **kwargs: str | None) -> dict[str, Any]:
        """Get an NVMe feature by ID (e.g. ``'0x02'`` for power management)."""
        if not self.path:
            return {}

        arguments = {'--feature-id': feature_id, '--output-format': 'json'} | kwargs

        result = self._run('get-feature', device=self.path, arguments=arguments)
        if result.failed or not result.stdout:
            return {}

        try:
            return json.loads(result.stdout)
        except json.JSONDecodeError:
            logger.warning('Failed to parse get-feature JSON output')
            return {}

    def set_feature(self, feature_id: str, value: str, **kwargs: str | None) -> bool:
        """Set an NVMe feature by ID.

        Example:
            ```python
            device.set_feature('0x02', '0x00')  # power management
            ```
        """
        if not self.path:
            return False

        arguments = {'--feature-id': feature_id, '--value': value} | kwargs

        return self._run('set-feature', device=self.path, arguments=arguments).succeeded

    def device_self_test(self, test_code: str = '1', **kwargs: str | None) -> bool:
        """Initiate a device self-test.

        Args:
            test_code: ``'1'`` = short, ``'2'`` = extended, ``'15'`` = abort.
        """
        if not self.path:
            return False

        arguments = {'--self-test-code': test_code} | kwargs

        return self._run('device-self-test', device=self.path, arguments=arguments).succeeded

    def get_self_test_log(self, **kwargs: str | None) -> dict[str, Any]:
        """Get self-test result log as a dict."""
        if not self.path:
            return {}

        arguments = {'--output-format': 'json'} | kwargs

        result = self._run('self-test-log', device=self.path, arguments=arguments)
        if result.failed or not result.stdout:
            return {}

        try:
            return json.loads(result.stdout)
        except json.JSONDecodeError:
            logger.warning('Failed to parse self-test-log JSON output')
            return {}

    def fw_download(self, firmware_file: str, **kwargs: str | None) -> bool:
        """Download a firmware image to the device."""
        if not self.path:
            return False

        arguments = {'--fw': firmware_file} | kwargs

        return self._run('fw-download', device=self.path, arguments=arguments).succeeded

    def fw_commit(self, slot: str, action: str = '1', **kwargs: str | None) -> bool:
        """Commit/activate firmware in a slot.

        Args:
            slot: Firmware slot (``'0'``-``'7'``).
            action: ``'0'`` download, ``'1'`` commit+activate, ``'2'`` activate,
                ``'3'`` commit+activate+reset.
        """
        if not self.path:
            return False

        arguments = {'--slot': slot, '--action': action} | kwargs

        return self._run('fw-commit', device=self.path, arguments=arguments).succeeded

    def get_lba_status(self, start_lba: str, block_count: str, **kwargs: str | None) -> dict[str, Any]:
        """Get logical block allocation status."""
        if not self.path:
            return {}

        arguments = {'--start-lba': start_lba, '--block-count': block_count, '--output-format': 'json'} | kwargs

        result = self._run('get-lba-status', device=self.path, arguments=arguments)
        if result.failed or not result.stdout:
            return {}

        try:
            return json.loads(result.stdout)
        except json.JSONDecodeError:
            logger.warning('Failed to parse get-lba-status JSON output')
            return {}

    def dsm(self, range_list: str, **kwargs: str | None) -> bool:
        """TRIM / deallocate LBA ranges.

        Args:
            range_list: Comma-separated LBA ranges (e.g. ``'0,1024'``).
        """
        if not self.path:
            return False

        arguments = {'--ad': 1, '--range': range_list} | kwargs

        return self._run('dsm', device=self.path, arguments=arguments).succeeded

    def show_regs(self, **kwargs: str | None) -> dict[str, Any]:
        """Read controller registers as a dict."""
        if not self.controller:
            return {}

        arguments = {'--output-format': 'json'} | kwargs

        result = self._run('show-regs', device=f'/dev/{self.controller}', arguments=arguments)
        if result.failed or not result.stdout:
            return {}

        try:
            return json.loads(result.stdout)
        except json.JSONDecodeError:
            logger.warning('Failed to parse show-regs JSON output')
            return {}

    def ns_rescan(self, **kwargs: str | None) -> bool:
        """Rescan controller for namespace changes."""
        if not self.controller:
            return False

        return self._run('ns-rescan', device=f'/dev/{self.controller}', arguments=kwargs).succeeded

    def discover_subsystems(self, **kwargs: str | None) -> dict[str, Any]:
        """Discover NVMe-oF subsystems (TCP, RDMA, or FC).

        Uses ``host_nqn`` automatically if set and ``--hostnqn`` not provided.

        Example:
            ```python
            device.discover_subsystems(**{'--transport': 'tcp', '--traddr': '192.168.1.100', '--trsvcid': '4420'})
            ```
        """
        arguments = {'--output-format': 'json'} | kwargs

        # Use device's host NQN if available and not provided
        if self.host_nqn and '--hostnqn' not in arguments:
            arguments['--hostnqn'] = self.host_nqn

        result = self._run('discover', arguments=arguments)
        if result.failed or not result.stdout:
            return {}

        try:
            return json.loads(result.stdout)
        except json.JSONDecodeError:
            logger.warning('Failed to parse discover JSON output')
            return {}

    def connect_all(self, **kwargs: str | None) -> bool:
        """Discover and connect to all NVMe-oF subsystems.

        Example:
            ```python
            device.connect_all(**{'--transport': 'tcp', '--traddr': '192.168.1.100', '--trsvcid': '4420'})
            ```
        """
        arguments = kwargs.copy()

        # Use device's host NQN if available and not provided
        if self.host_nqn and '--hostnqn' not in arguments:
            arguments['--hostnqn'] = self.host_nqn

        return self._run('connect-all', arguments=arguments).succeeded

    def connect(self, **kwargs: str | None) -> bool:
        """Connect to a specific NVMe-oF subsystem.

        Example:
            ```python
            device.connect(
                **{
                    '--transport': 'tcp',
                    '--traddr': '192.168.1.100',
                    '--trsvcid': '4420',
                    '--nqn': 'nqn.2016-06.io.spdk:cnode1',
                }
            )
            ```
        """
        arguments = kwargs.copy()

        # Use device's host NQN if available and not provided
        if self.host_nqn and '--hostnqn' not in arguments:
            arguments['--hostnqn'] = self.host_nqn

        return self._run('connect', arguments=arguments).succeeded

    def disconnect(self, nqn: str, **kwargs: str | None) -> bool:
        """Disconnect from a specific NVMe-oF subsystem by NQN."""
        arguments = {'--nqn': nqn} | kwargs

        return self._run('disconnect', arguments=arguments).succeeded

    def disconnect_all(self, **kwargs: str | None) -> bool:
        """Disconnect from all NVMe-oF subsystems."""
        return self._run('disconnect-all', arguments=kwargs).succeeded

controller property

Controller name (e.g. nvme0), derived from device name.

connect(**kwargs)

Connect to a specific NVMe-oF subsystem.

Example

device.connect(
    **{
        '--transport': 'tcp',
        '--traddr': '192.168.1.100',
        '--trsvcid': '4420',
        '--nqn': 'nqn.2016-06.io.spdk:cnode1',
    }
)

Source code in sts_libs/src/sts/nvme.py

def connect(self, **kwargs: str | None) -> bool:
    """Connect to a specific NVMe-oF subsystem.

    Example:
        ```python
        device.connect(
            **{
                '--transport': 'tcp',
                '--traddr': '192.168.1.100',
                '--trsvcid': '4420',
                '--nqn': 'nqn.2016-06.io.spdk:cnode1',
            }
        )
        ```
    """
    arguments = kwargs.copy()

    # Use device's host NQN if available and not provided
    if self.host_nqn and '--hostnqn' not in arguments:
        arguments['--hostnqn'] = self.host_nqn

    return self._run('connect', arguments=arguments).succeeded

connect_all(**kwargs)

Discover and connect to all NVMe-oF subsystems.

Example

device.connect_all(**{'--transport': 'tcp', '--traddr': '192.168.1.100', '--trsvcid': '4420'})

Source code in sts_libs/src/sts/nvme.py

def connect_all(self, **kwargs: str | None) -> bool:
    """Discover and connect to all NVMe-oF subsystems.

    Example:
        ```python
        device.connect_all(**{'--transport': 'tcp', '--traddr': '192.168.1.100', '--trsvcid': '4420'})
        ```
    """
    arguments = kwargs.copy()

    # Use device's host NQN if available and not provided
    if self.host_nqn and '--hostnqn' not in arguments:
        arguments['--hostnqn'] = self.host_nqn

    return self._run('connect-all', arguments=arguments).succeeded

device_self_test(test_code='1', **kwargs)

Initiate a device self-test.

Parameters:

Name	Type	Description	Default
test_code	str	'1' = short, '2' = extended, '15' = abort.	'1'

Source code in sts_libs/src/sts/nvme.py

def device_self_test(self, test_code: str = '1', **kwargs: str | None) -> bool:
    """Initiate a device self-test.

    Args:
        test_code: ``'1'`` = short, ``'2'`` = extended, ``'15'`` = abort.
    """
    if not self.path:
        return False

    arguments = {'--self-test-code': test_code} | kwargs

    return self._run('device-self-test', device=self.path, arguments=arguments).succeeded

disconnect(nqn, **kwargs)

Disconnect from a specific NVMe-oF subsystem by NQN.

Source code in sts_libs/src/sts/nvme.py

def disconnect(self, nqn: str, **kwargs: str | None) -> bool:
    """Disconnect from a specific NVMe-oF subsystem by NQN."""
    arguments = {'--nqn': nqn} | kwargs

    return self._run('disconnect', arguments=arguments).succeeded

disconnect_all(**kwargs)

Disconnect from all NVMe-oF subsystems.

Source code in sts_libs/src/sts/nvme.py

def disconnect_all(self, **kwargs: str | None) -> bool:
    """Disconnect from all NVMe-oF subsystems."""
    return self._run('disconnect-all', arguments=kwargs).succeeded

discover()

Populate metadata (model, serial, firmware, transport, size) from nvme-cli.

Source code in sts_libs/src/sts/nvme.py

def discover(self) -> Self:
    """Populate metadata (model, serial, firmware, transport, size) from nvme-cli."""
    self._discover_from_nvme_list()
    return self

discover_subsystems(**kwargs)

Discover NVMe-oF subsystems (TCP, RDMA, or FC).

Uses host_nqn automatically if set and --hostnqn not provided.

Example

device.discover_subsystems(**{'--transport': 'tcp', '--traddr': '192.168.1.100', '--trsvcid': '4420'})

Source code in sts_libs/src/sts/nvme.py

def discover_subsystems(self, **kwargs: str | None) -> dict[str, Any]:
    """Discover NVMe-oF subsystems (TCP, RDMA, or FC).

    Uses ``host_nqn`` automatically if set and ``--hostnqn`` not provided.

    Example:
        ```python
        device.discover_subsystems(**{'--transport': 'tcp', '--traddr': '192.168.1.100', '--trsvcid': '4420'})
        ```
    """
    arguments = {'--output-format': 'json'} | kwargs

    # Use device's host NQN if available and not provided
    if self.host_nqn and '--hostnqn' not in arguments:
        arguments['--hostnqn'] = self.host_nqn

    result = self._run('discover', arguments=arguments)
    if result.failed or not result.stdout:
        return {}

    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError:
        logger.warning('Failed to parse discover JSON output')
        return {}

dsm(range_list, **kwargs)

TRIM / deallocate LBA ranges.

Parameters:

Name	Type	Description	Default
range_list	str	Comma-separated LBA ranges (e.g. '0,1024').	required

Source code in sts_libs/src/sts/nvme.py

def dsm(self, range_list: str, **kwargs: str | None) -> bool:
    """TRIM / deallocate LBA ranges.

    Args:
        range_list: Comma-separated LBA ranges (e.g. ``'0,1024'``).
    """
    if not self.path:
        return False

    arguments = {'--ad': 1, '--range': range_list} | kwargs

    return self._run('dsm', device=self.path, arguments=arguments).succeeded

flush(**kwargs)

Commit data and metadata to nonvolatile media.

Examples:

device.flush()
device.flush(**{'--namespace-id': '1'})

Source code in sts_libs/src/sts/nvme.py

def flush(self, **kwargs: str | None) -> bool:
    """Commit data and metadata to nonvolatile media.

    Examples:
        ```python
        device.flush()
        device.flush(**{'--namespace-id': '1'})
        ```
    """
    if not self.path:
        return False

    return self._run('flush', device=self.path, arguments=kwargs).succeeded

format(**kwargs)

Low-level format the NVMe namespace (erases all data).

Source code in sts_libs/src/sts/nvme.py

def format(self, **kwargs: str | None) -> bool:
    """Low-level format the NVMe namespace (erases all data)."""
    if not self.path:
        logger.error('Device path not available')
        return False

    # Targets the namespace (e.g. /dev/nvme0n1)
    return self._run('format', device=self.path, arguments=kwargs).succeeded

fw_commit(slot, action='1', **kwargs)

Commit/activate firmware in a slot.

Parameters:

Name	Type	Description	Default
slot	str	Firmware slot ('0'-'7').	required
action	str	'0' download, '1' commit+activate, '2' activate, '3' commit+activate+reset.	'1'

Source code in sts_libs/src/sts/nvme.py

def fw_commit(self, slot: str, action: str = '1', **kwargs: str | None) -> bool:
    """Commit/activate firmware in a slot.

    Args:
        slot: Firmware slot (``'0'``-``'7'``).
        action: ``'0'`` download, ``'1'`` commit+activate, ``'2'`` activate,
            ``'3'`` commit+activate+reset.
    """
    if not self.path:
        return False

    arguments = {'--slot': slot, '--action': action} | kwargs

    return self._run('fw-commit', device=self.path, arguments=arguments).succeeded

fw_download(firmware_file, **kwargs)

Download a firmware image to the device.

Source code in sts_libs/src/sts/nvme.py

def fw_download(self, firmware_file: str, **kwargs: str | None) -> bool:
    """Download a firmware image to the device."""
    if not self.path:
        return False

    arguments = {'--fw': firmware_file} | kwargs

    return self._run('fw-download', device=self.path, arguments=arguments).succeeded

get_all() classmethod

Discover all NVMe namespace devices on the system.

Source code in sts_libs/src/sts/nvme.py

@classmethod
def get_all(cls) -> list[NvmeDevice]:
    """Discover all NVMe namespace devices on the system."""
    result = run('nvme list')
    if result.failed:
        logger.warning(f'Running "nvme list" failed:\n{result.stderr}')
        return []

    devices: list[NvmeDevice] = []
    for line in result.stdout.splitlines():
        # Lines like: /dev/nvme0n1     238.47  GB / 238.47  GB    512   B +  0 B
        if line.strip().startswith('/dev/nvme') and 'n' in line:
            device_path = line.strip().split()[0]
            device_name = device_path.replace('/dev/', '')
            if device_name:
                device = cls(name=device_name)
                device.discover()
                devices.append(device)

    return devices

get_by_attribute(attribute, value) classmethod

Find devices where attribute equals value (case-sensitive).

Parameters:

Name	Type	Description	Default
attribute	str	Device attribute name (e.g. 'model', 'serial', 'transport')	required
value	str	Attribute value to match	required

Source code in sts_libs/src/sts/nvme.py

@classmethod
def get_by_attribute(cls, attribute: str, value: str) -> list[NvmeDevice]:
    """Find devices where ``attribute`` equals ``value`` (case-sensitive).

    Args:
        attribute: Device attribute name (e.g. 'model', 'serial', 'transport')
        value: Attribute value to match
    """
    if not attribute or not value:
        return []

    devices: list[NvmeDevice] = []
    for device in cls.get_all():
        device_value = getattr(device, attribute, None)
        if device_value and str(device_value) == str(value):
            devices.append(device)

    return devices

get_error_log(**kwargs)

Get error log as a dict.

Example

device.get_error_log(**{'-e': '1'})
{'errors': [{'error_count': 76, ...}]}

Source code in sts_libs/src/sts/nvme.py

def get_error_log(self, **kwargs: str | None) -> dict[str, Any]:
    """Get error log as a dict.

    Example:
        ```python
        device.get_error_log(**{'-e': '1'})
        {'errors': [{'error_count': 76, ...}]}
        ```
    """
    if not self.path:
        return {}

    arguments = {'--output-format': 'json'} | kwargs

    result = self._run('error-log', device=self.path, arguments=arguments)
    if result.failed or not result.stdout:
        return {}

    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError:
        logger.warning('Failed to parse error-log JSON output')
        return {}

get_feature(feature_id, **kwargs)

Get an NVMe feature by ID (e.g. '0x02' for power management).

Source code in sts_libs/src/sts/nvme.py

def get_feature(self, feature_id: str, **kwargs: str | None) -> dict[str, Any]:
    """Get an NVMe feature by ID (e.g. ``'0x02'`` for power management)."""
    if not self.path:
        return {}

    arguments = {'--feature-id': feature_id, '--output-format': 'json'} | kwargs

    result = self._run('get-feature', device=self.path, arguments=arguments)
    if result.failed or not result.stdout:
        return {}

    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError:
        logger.warning('Failed to parse get-feature JSON output')
        return {}

get_fw_log(**kwargs)

Get firmware slot log as a dict.

Source code in sts_libs/src/sts/nvme.py

def get_fw_log(self, **kwargs: str | None) -> dict[str, Any]:
    """Get firmware slot log as a dict."""
    if not self.path:
        return {}

    arguments = {'--output-format': 'json'} | kwargs

    result = self._run('fw-log', device=self.path, arguments=arguments)
    if result.failed or not result.stdout:
        return {}

    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError:
        logger.warning('Failed to parse fw-log JSON output')
        return {}

get_id_ctrl(**kwargs)

Get controller identify data as a dict.

Source code in sts_libs/src/sts/nvme.py

def get_id_ctrl(self, **kwargs: str | None) -> dict[str, Any]:
    """Get controller identify data as a dict."""
    if not self.path:
        return {}

    arguments = {'--output-format': 'json'} | kwargs

    result = self._run('id-ctrl', device=self.path, arguments=arguments)
    if result.failed or not result.stdout:
        return {}

    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError:
        logger.warning('Failed to parse id-ctrl JSON output')
        return {}

get_id_ns(**kwargs)

Get namespace identify data as a dict.

Source code in sts_libs/src/sts/nvme.py

def get_id_ns(self, **kwargs: str | None) -> dict[str, Any]:
    """Get namespace identify data as a dict."""
    if not self.path:
        return {}

    arguments = {'--output-format': 'json'} | kwargs

    result = self._run('id-ns', device=self.path, arguments=arguments)
    if result.failed or not result.stdout:
        return {}

    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError:
        logger.warning('Failed to parse id-ns JSON output')
        return {}

get_lba_status(start_lba, block_count, **kwargs)

Get logical block allocation status.

Source code in sts_libs/src/sts/nvme.py

def get_lba_status(self, start_lba: str, block_count: str, **kwargs: str | None) -> dict[str, Any]:
    """Get logical block allocation status."""
    if not self.path:
        return {}

    arguments = {'--start-lba': start_lba, '--block-count': block_count, '--output-format': 'json'} | kwargs

    result = self._run('get-lba-status', device=self.path, arguments=arguments)
    if result.failed or not result.stdout:
        return {}

    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError:
        logger.warning('Failed to parse get-lba-status JSON output')
        return {}

get_self_test_log(**kwargs)

Get self-test result log as a dict.

Source code in sts_libs/src/sts/nvme.py

def get_self_test_log(self, **kwargs: str | None) -> dict[str, Any]:
    """Get self-test result log as a dict."""
    if not self.path:
        return {}

    arguments = {'--output-format': 'json'} | kwargs

    result = self._run('self-test-log', device=self.path, arguments=arguments)
    if result.failed or not result.stdout:
        return {}

    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError:
        logger.warning('Failed to parse self-test-log JSON output')
        return {}

get_smart_log(**kwargs)

Get SMART health log as a dict.

Source code in sts_libs/src/sts/nvme.py

def get_smart_log(self, **kwargs: str | None) -> dict[str, Any]:
    """Get SMART health log as a dict."""
    if not self.path:
        return {}

    arguments = {'--output-format': 'json'} | kwargs

    result = self._run('smart-log', device=self.path, arguments=arguments)
    if result.failed or not result.stdout:
        return {}

    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError:
        logger.warning('Failed to parse smart-log JSON output')
        return {}

has_nvme() classmethod

Check if the system has any NVMe devices.

Source code in sts_libs/src/sts/nvme.py

@classmethod
def has_nvme(cls) -> bool:
    """Check if the system has any NVMe devices."""
    result = run('nvme list')
    if result.failed or not result.stdout:
        return False

    return any(line.strip().startswith('/dev/nvme') for line in result.stdout.splitlines())

list_ns(**kwargs)

List namespaces attached to the controller.

Source code in sts_libs/src/sts/nvme.py

def list_ns(self, **kwargs: str | None) -> dict[str, Any]:
    """List namespaces attached to the controller."""
    if not self.controller:
        return {}

    arguments = {'--output-format': 'json'} | kwargs

    result = self._run('list-ns', device=f'/dev/{self.controller}', arguments=arguments)
    if result.failed or not result.stdout:
        return {}

    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError:
        logger.warning('Failed to parse list-ns JSON output')
        return {}

ns_rescan(**kwargs)

Rescan controller for namespace changes.

Source code in sts_libs/src/sts/nvme.py

def ns_rescan(self, **kwargs: str | None) -> bool:
    """Rescan controller for namespace changes."""
    if not self.controller:
        return False

    return self._run('ns-rescan', device=f'/dev/{self.controller}', arguments=kwargs).succeeded

reset(**kwargs)

Reset the NVMe controller.

Source code in sts_libs/src/sts/nvme.py

def reset(self, **kwargs: str | None) -> bool:
    """Reset the NVMe controller."""
    if not self.path:
        return False

    return self._run('reset', device=f'/dev/{self.controller}', arguments=kwargs).succeeded

sanitize(**kwargs)

Sanitize the controller (block erase, crypto erase, or overwrite).

Targets the controller (e.g. /dev/nvme0), not the namespace.

Examples:

device.sanitize(**{'--sanact': '0x01'})
device.sanitize(**{'--sanact': '0x01', '--nodas': None})

Source code in sts_libs/src/sts/nvme.py

def sanitize(self, **kwargs: str | None) -> bool:
    """Sanitize the controller (block erase, crypto erase, or overwrite).

    Targets the controller (e.g. ``/dev/nvme0``), not the namespace.

    Examples:
        ```python
        device.sanitize(**{'--sanact': '0x01'})
        device.sanitize(**{'--sanact': '0x01', '--nodas': None})
        ```
    """
    if not self.path:
        logger.error('Device path not available')
        return False

    # Targets the controller (e.g. /dev/nvme0), not the namespace — sanitize is controller-level
    return self._run('sanitize', device=f'/dev/{self.controller}', arguments=kwargs).succeeded

set_feature(feature_id, value, **kwargs)

Set an NVMe feature by ID.

Example

device.set_feature('0x02', '0x00')  # power management

Source code in sts_libs/src/sts/nvme.py

def set_feature(self, feature_id: str, value: str, **kwargs: str | None) -> bool:
    """Set an NVMe feature by ID.

    Example:
        ```python
        device.set_feature('0x02', '0x00')  # power management
        ```
    """
    if not self.path:
        return False

    arguments = {'--feature-id': feature_id, '--value': value} | kwargs

    return self._run('set-feature', device=self.path, arguments=arguments).succeeded

show_regs(**kwargs)

Read controller registers as a dict.

Source code in sts_libs/src/sts/nvme.py

def show_regs(self, **kwargs: str | None) -> dict[str, Any]:
    """Read controller registers as a dict."""
    if not self.controller:
        return {}

    arguments = {'--output-format': 'json'} | kwargs

    result = self._run('show-regs', device=f'/dev/{self.controller}', arguments=arguments)
    if result.failed or not result.stdout:
        return {}

    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError:
        logger.warning('Failed to parse show-regs JSON output')
        return {}

NvmeError

Bases: DeviceError

Base class for NVMe-related errors.

Source code in sts_libs/src/sts/nvme.py

class NvmeError(DeviceError):
    """Base class for NVMe-related errors."""