Skip to content

SCSI

This section documents the SCSI device functionality.

sts.scsi

SCSI device management.

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

SCSI (Small Computer System Interface) is a standard for: - Storage device communication - Device addressing and identification - Command and data transfer - Error handling and recovery

Common SCSI devices include: - Hard drives (HDDs) - Solid State Drives (SSDs) - Tape drives - CD/DVD/Blu-ray drives

ScsiDevice dataclass

Bases: StorageDevice

SCSI device representation.

A SCSI device is identified by: - SCSI ID (H:C:T:L format) - H: Host adapter number - C: Channel/Bus number - T: Target ID - L: Logical Unit Number (LUN) - Device node (e.g. /dev/sda) - Vendor and model information

Parameters:

Name Type Description Default
name str | None

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

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)

None
scsi_id str | None

SCSI ID (optional, discovered from device)

None
host_id str | None

FC host ID (optional, discovered from device)

None
Example
device = ScsiDevice(name='sda')  # Discovers other values
device = ScsiDevice(scsi_id='0:0:0:0')  # Discovers device from SCSI ID
Source code in sts_libs/src/sts/scsi.py
 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
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
@dataclass
class ScsiDevice(StorageDevice):
    """SCSI device representation.

    A SCSI device is identified by:
    - SCSI ID (H:C:T:L format)
      - H: Host adapter number
      - C: Channel/Bus number
      - T: Target ID
      - L: Logical Unit Number (LUN)
    - Device node (e.g. /dev/sda)
    - Vendor and model information

    Args:
        name: Device name (optional, e.g. 'sda')
        path: Device path (optional, defaults to /dev/<name>)
        size: Device size in bytes (optional, discovered from device)
        model: Device model (optional)
        scsi_id: SCSI ID (optional, discovered from device)
        host_id: FC host ID (optional, discovered from device)

    Example:
        ```python
        device = ScsiDevice(name='sda')  # Discovers other values
        device = ScsiDevice(scsi_id='0:0:0:0')  # Discovers device from SCSI ID
        ```
    """

    # 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
    scsi_id: str | None = None  # SCSI address (H:C:T:L)
    host_id: str | None = None

    # Sysfs path for SCSI devices
    SCSI_PATH: ClassVar[Path] = Path('/sys/class/scsi_device')
    SCSI_HOST_PATH: ClassVar[Path] = Path('/sys/class/scsi_host')

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

        - Sets device path if not provided
        - Gets device information from lsscsi
        - Gets model information if not provided

        Raises:
            DeviceNotFoundError: If device does not exist
            DeviceError: If device cannot be accessed
        """
        # Ensure lsscsi is installed
        ensure_installed('lsscsi')

        # Initialize parent class
        super().__post_init__()

        # Get SCSI ID from lsscsi if not provided
        if not self.scsi_id and self.name:
            result = run(f'lsscsi | grep "{self.name} $"')
            if result.succeeded:
                with contextlib.suppress(IndexError):
                    # Extract [H:C:T:L] from lsscsi output
                    self.scsi_id = result.stdout.split()[0].strip('[]')

        if self.scsi_id:
            self.host_id = self.scsi_id.strip('[]').split(':')[0]

        if self.scsi_id and not self.name:
            self.name = self.device_name

        # Set path based on name if not provided
        if not self.path and self.name:
            self.path = f'/dev/{self.name}'

        # Get model from sysfs if not provided
        if not self.model:
            self.model = self.model_name

    @property
    def vendor(self) -> str | None:
        """Get device vendor.

        Reads vendor string from sysfs:
        - Common vendors: ATA, SCSI, USB
        - Helps identify device type and capabilities
        - Used for device-specific handling

        Returns:
            Device vendor or None if not available

        Example:
            ```python
            device.vendor
            'ATA'
            ```
        """
        if not self.scsi_id:
            return None

        try:
            vendor_path = self.SCSI_PATH / self.scsi_id / 'device/vendor'
            return vendor_path.read_text().strip()
        except OSError:
            return None

    @property
    def model_name(self) -> str | None:
        """Get device model name.

        Reads model string from sysfs:
        - Identifies specific device model
        - Contains manufacturer information
        - Used for device compatibility

        Returns:
            Device model name or None if not available

        Example:
            ```python
            device.model_name
            'Samsung SSD 970 EVO'
            ```
        """
        if not self.scsi_id:
            return None

        try:
            model_path = self.SCSI_PATH / self.scsi_id / 'device/model'
            return model_path.read_text().strip()
        except OSError:
            return None

    @property
    def revision(self) -> str | None:
        """Get device revision.

        Reads firmware revision from sysfs:
        - Indicates firmware version
        - Important for bug tracking
        - Used for feature compatibility

        Returns:
            Device revision or None if not available

        Example:
            ```python
            device.revision
            '1.0'
            ```
        """
        if not self.scsi_id:
            return None

        try:
            rev_path = self.SCSI_PATH / self.scsi_id / 'device/rev'
            return rev_path.read_text().strip()
        except OSError:
            return None

    @property
    def device_name(self) -> str | None:
        """Get device name.

        Returns:
            Device name or None if not available

        Example:
            ```python
            device.device_name
            'sdc'
            ```
        """
        if not self.scsi_id:
            return None

        try:
            block_path = Path(f'{self.SCSI_PATH}/{self.scsi_id}/device/block')
            if block_path.exists():
                return next(block_path.iterdir()).name
        except OSError:
            logging.warning(f'Failed to get the device name for {self.scsi_id}')
            return None
        else:
            return None

    @property
    def transport(self) -> str | None:
        """Get device transport.

        Returns:
            Device transport or None if not available

        Examples:
            ```Python
            device.transport
            'iSCSI'
            ```
        """
        if not self.scsi_id:
            return None

        try:
            result = run(f'lsscsi {self.scsi_id} --list -t | grep transport')
            if result.succeeded:
                for line in result.stdout.splitlines():
                    if 'transport' in line:
                        return line.split('=')[1].strip()
                    return None
        except (OSError, IndexError):
            return None

    @property
    def driver(self) -> str | None:
        """Get device driver.

        Returns:
            Device driver or None if not available

        Example:
            ```Python
            device.driver
            'qla2xxx'
            ```
        """
        if not self.host_id:
            return None

        try:
            result = run(f'lsscsi -H {self.host_id}')
            if result.succeeded and len(result.stdout.split()) > 1:
                return result.stdout.split()[1]
        except (OSError, IndexError):
            logging.warning(f'Failed to get the driver for host {self.host_id}.')
            return None
        else:
            return None

    @property
    def state(self) -> str | None:
        """Get device state.

        Returns:
            The state of the device or None if not available

        """
        state_path = Path(f'{self.BLOCK_PATH}/{self.name}/device/state')
        try:
            return state_path.read_text().strip()
        except OSError:
            logging.warning(f'Failed to read {state_path}')
            return None

    @property
    def pci_id(self) -> str | None:
        """Get the PCI ID associated with the SCSI host.

        Returns:
            PCI ID or None if not available

        Example:
            ```Python
            device.pci_id
            '0000:08:00.0'
            ```
        """
        if not self.host_id:
            return None

        scsi_host_path = Path(f'{self.SCSI_HOST_PATH}/host{self.host_id}')
        try:
            link = scsi_host_path.resolve().as_posix()
        except OSError as e:
            logging.warning(f'Error resolving path for host{self.host_id}: {e}')
            return None
        else:
            regex_pci_id = r'([0-9a-f]{4}:[0-9a-f]{2}:[0-9a-f]{2}\.[0-9a-f])'
            pci_match = re.search(f'{regex_pci_id}/host{self.host_id}/scsi_host', link)
            return pci_match.group(1) if pci_match else None

    def rescan_host(self) -> bool:
        """Initiates a rescan operation for the specified host ID, to detect new devices.

        The scan file is part of the sysfs interface in Linux, which exposes kernel data structures to user space.
        Writing - - - to this file instructs the kernel to scan all channels, all targets, and all LUNs
        for the specified host adapter.

        Returns:
            True if the rescan operation is successful, False otherwise.
        """
        if not self.host_id:
            return False

        rescan_path = Path(f'{self.SCSI_HOST_PATH}/host{self.host_id}/scan')
        try:
            rescan_path.write_text('- - -')
        except OSError:
            logging.warning(f'Failed to write to {rescan_path}.')
            return False
        else:
            return True

    def rescan_disk(self) -> bool:
        """Rescan the disk.

        Returns:
            True if the rescan operation is successful, otherwise False.

        """
        rescan_path = Path(f'{self.BLOCK_PATH}/{self.name}/device/rescan')
        try:
            rescan_path.write_text('1')
        except OSError:
            logging.warning(f'Failed to write to {rescan_path}.')
            return False
        else:
            return True

    def delete_disk(self) -> bool:
        """Deletes the disk.

        Returns:
            True if the disk device is successfully deleted, otherwise False.

        """
        del_path = Path(f'{self.BLOCK_PATH}/{self.name}/device/delete')
        try:
            del_path.write_text('1')
        except OSError:
            logging.warning(f'Failed to write to {del_path}.')
            return False
        else:
            return True

    def up_or_down_disk(self, action: str) -> bool:
        """Change the state of the disk.

        Args:
            action: offline or running.

        Returns:
            True if the operation is successful, otherwise returns False.

        """
        state_path = Path(f'{self.BLOCK_PATH}/{self.name}/device/state')
        try:
            # Need a newline character at the end of the action, or else it cannot be written into
            action = f'{action}\n'
            state_path.write_text(action)
        except OSError:
            logging.warning(f'Failed to write to {state_path}')
            return False
        else:
            return True

    @classmethod
    def get_all_scsi_device_ids(cls) -> list[str]:
        """Get the list of all SCSI device IDs.

        Returns:
            List of SCSI device IDs

        Example:
            '''Python
            ScsiDevice.get_all_scsi_device_ids()
            ['7:0:0:0', '8:0:0:0', '7:0:0:1',...]
            '''

        """
        try:
            path = cls.SCSI_PATH
            if path.exists():
                return [d.name for d in path.iterdir()]
        except OSError:
            logging.warning(f'Failed to list SCSI devices from {cls.SCSI_PATH}')
            return []
        else:
            return []

    @classmethod
    def get_all_scsi_devices(cls: type[T]) -> list[T]:
        """Retrieve all SCSI devices.

        Returns:
            A list of ScsiDevice objects, each representing a SCSI device.

        Example:
            ```Python
            ScsiDevice.get_all_scsi_devices()
            [ScsiDevice(path='/dev/sda', name='sda', ...), ScsiDevice(path='/dev/sdb', name='sdb',...),...]
            ```

        """
        scsi_ids = ScsiDevice.get_all_scsi_device_ids()
        devices: list[T] = [cls(scsi_id=scsi_id) for scsi_id in scsi_ids]
        return devices

    @classmethod
    def get_by_vendor(cls: type[T], vendor: str) -> list[T]:
        """Retrieve a list of ScsiDevice objects that match the specified vendor.

        Args:
            vendor (str): used to filter the SCSI devices.

        Returns:
            A list of ScsiDevice objects.

        Example:
            ```Python
            ScsiDevice.get_by_vendor('NETAPP')
            [ScsiDevice(path='/dev/sdb', name='sdb', ...), ScsiDevice(path='/dev/sdc', ...), ...]
            ```

        """
        all_devices = cls.get_all_scsi_devices()
        return [device for device in all_devices if device.vendor == vendor]

    @classmethod
    def get_by_attribute(cls: type[T], attribute: str, value: str) -> list[T]:
        """Retrieve a list of ScsiDevice objects that match the specified attribute and value.

        Args:
            attribute (str): The attribute used to filter the SCSI devices (e.g., 'transport', 'vendor').
            value (str): The value of the attribute used to filter the SCSI devices.

        Returns:
            A list of ScsiDevice objects.

        Example:
            ```Python
            ScsiDevice.get_by_attribute('transport', 'fc:')
            [ScsiDevice(path='/dev/sdb', name='sdb', ...), ScsiDevice(path='/dev/sdc', ...), ...]

            ScsiDevice.get_by_attribute('vendor', 'NETAPP')
            [ScsiDevice(path='/dev/sdb', name='sdb', ...), ScsiDevice(path='/dev/sdc', ...), ...]
        """
        all_devices = cls.get_all_scsi_devices()
        return [device for device in all_devices if getattr(device, attribute, None) == value]

device_name: str | None property

Get device name.

Returns:

Type Description
str | None

Device name or None if not available

Example
device.device_name
'sdc'

driver: str | None property

Get device driver.

Returns:

Type Description
str | None

Device driver or None if not available

Example
device.driver
'qla2xxx'

model_name: str | None property

Get device model name.

Reads model string from sysfs: - Identifies specific device model - Contains manufacturer information - Used for device compatibility

Returns:

Type Description
str | None

Device model name or None if not available

Example
device.model_name
'Samsung SSD 970 EVO'

pci_id: str | None property

Get the PCI ID associated with the SCSI host.

Returns:

Type Description
str | None

PCI ID or None if not available

Example
device.pci_id
'0000:08:00.0'

revision: str | None property

Get device revision.

Reads firmware revision from sysfs: - Indicates firmware version - Important for bug tracking - Used for feature compatibility

Returns:

Type Description
str | None

Device revision or None if not available

Example
device.revision
'1.0'

state: str | None property

Get device state.

Returns:

Type Description
str | None

The state of the device or None if not available

transport: str | None property

Get device transport.

Returns:

Type Description
str | None

Device transport or None if not available

Examples:

device.transport
'iSCSI'

vendor: str | None property

Get device vendor.

Reads vendor string from sysfs: - Common vendors: ATA, SCSI, USB - Helps identify device type and capabilities - Used for device-specific handling

Returns:

Type Description
str | None

Device vendor or None if not available

Example
device.vendor
'ATA'

__post_init__()

Initialize SCSI device.

  • Sets device path if not provided
  • Gets device information from lsscsi
  • Gets model information if not provided

Raises:

Type Description
DeviceNotFoundError

If device does not exist

DeviceError

If device cannot be accessed

Source code in sts_libs/src/sts/scsi.py
 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
def __post_init__(self) -> None:
    """Initialize SCSI device.

    - Sets device path if not provided
    - Gets device information from lsscsi
    - Gets model information if not provided

    Raises:
        DeviceNotFoundError: If device does not exist
        DeviceError: If device cannot be accessed
    """
    # Ensure lsscsi is installed
    ensure_installed('lsscsi')

    # Initialize parent class
    super().__post_init__()

    # Get SCSI ID from lsscsi if not provided
    if not self.scsi_id and self.name:
        result = run(f'lsscsi | grep "{self.name} $"')
        if result.succeeded:
            with contextlib.suppress(IndexError):
                # Extract [H:C:T:L] from lsscsi output
                self.scsi_id = result.stdout.split()[0].strip('[]')

    if self.scsi_id:
        self.host_id = self.scsi_id.strip('[]').split(':')[0]

    if self.scsi_id and not self.name:
        self.name = self.device_name

    # Set path based on name if not provided
    if not self.path and self.name:
        self.path = f'/dev/{self.name}'

    # Get model from sysfs if not provided
    if not self.model:
        self.model = self.model_name

delete_disk()

Deletes the disk.

Returns:

Type Description
bool

True if the disk device is successfully deleted, otherwise False.

Source code in sts_libs/src/sts/scsi.py
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
def delete_disk(self) -> bool:
    """Deletes the disk.

    Returns:
        True if the disk device is successfully deleted, otherwise False.

    """
    del_path = Path(f'{self.BLOCK_PATH}/{self.name}/device/delete')
    try:
        del_path.write_text('1')
    except OSError:
        logging.warning(f'Failed to write to {del_path}.')
        return False
    else:
        return True

get_all_scsi_device_ids() classmethod

Get the list of all SCSI device IDs.

Returns:

Type Description
list[str]

List of SCSI device IDs

Example

'''Python ScsiDevice.get_all_scsi_device_ids() ['7:0:0:0', '8:0:0:0', '7:0:0:1',...] '''

Source code in sts_libs/src/sts/scsi.py
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
@classmethod
def get_all_scsi_device_ids(cls) -> list[str]:
    """Get the list of all SCSI device IDs.

    Returns:
        List of SCSI device IDs

    Example:
        '''Python
        ScsiDevice.get_all_scsi_device_ids()
        ['7:0:0:0', '8:0:0:0', '7:0:0:1',...]
        '''

    """
    try:
        path = cls.SCSI_PATH
        if path.exists():
            return [d.name for d in path.iterdir()]
    except OSError:
        logging.warning(f'Failed to list SCSI devices from {cls.SCSI_PATH}')
        return []
    else:
        return []

get_all_scsi_devices() classmethod

Retrieve all SCSI devices.

Returns:

Type Description
list[T]

A list of ScsiDevice objects, each representing a SCSI device.

Example
ScsiDevice.get_all_scsi_devices()
[ScsiDevice(path='/dev/sda', name='sda', ...), ScsiDevice(path='/dev/sdb', name='sdb',...),...]
Source code in sts_libs/src/sts/scsi.py
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
@classmethod
def get_all_scsi_devices(cls: type[T]) -> list[T]:
    """Retrieve all SCSI devices.

    Returns:
        A list of ScsiDevice objects, each representing a SCSI device.

    Example:
        ```Python
        ScsiDevice.get_all_scsi_devices()
        [ScsiDevice(path='/dev/sda', name='sda', ...), ScsiDevice(path='/dev/sdb', name='sdb',...),...]
        ```

    """
    scsi_ids = ScsiDevice.get_all_scsi_device_ids()
    devices: list[T] = [cls(scsi_id=scsi_id) for scsi_id in scsi_ids]
    return devices

get_by_attribute(attribute, value) classmethod

Retrieve a list of ScsiDevice objects that match the specified attribute and value.

Parameters:

Name Type Description Default
attribute str

The attribute used to filter the SCSI devices (e.g., 'transport', 'vendor').

required
value str

The value of the attribute used to filter the SCSI devices.

required

Returns:

Type Description
list[T]

A list of ScsiDevice objects.

Example

```Python ScsiDevice.get_by_attribute('transport', 'fc:') [ScsiDevice(path='/dev/sdb', name='sdb', ...), ScsiDevice(path='/dev/sdc', ...), ...]

ScsiDevice.get_by_attribute('vendor', 'NETAPP') [ScsiDevice(path='/dev/sdb', name='sdb', ...), ScsiDevice(path='/dev/sdc', ...), ...]

Source code in sts_libs/src/sts/scsi.py
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
@classmethod
def get_by_attribute(cls: type[T], attribute: str, value: str) -> list[T]:
    """Retrieve a list of ScsiDevice objects that match the specified attribute and value.

    Args:
        attribute (str): The attribute used to filter the SCSI devices (e.g., 'transport', 'vendor').
        value (str): The value of the attribute used to filter the SCSI devices.

    Returns:
        A list of ScsiDevice objects.

    Example:
        ```Python
        ScsiDevice.get_by_attribute('transport', 'fc:')
        [ScsiDevice(path='/dev/sdb', name='sdb', ...), ScsiDevice(path='/dev/sdc', ...), ...]

        ScsiDevice.get_by_attribute('vendor', 'NETAPP')
        [ScsiDevice(path='/dev/sdb', name='sdb', ...), ScsiDevice(path='/dev/sdc', ...), ...]
    """
    all_devices = cls.get_all_scsi_devices()
    return [device for device in all_devices if getattr(device, attribute, None) == value]

get_by_vendor(vendor) classmethod

Retrieve a list of ScsiDevice objects that match the specified vendor.

Parameters:

Name Type Description Default
vendor str

used to filter the SCSI devices.

required

Returns:

Type Description
list[T]

A list of ScsiDevice objects.

Example
ScsiDevice.get_by_vendor('NETAPP')
[ScsiDevice(path='/dev/sdb', name='sdb', ...), ScsiDevice(path='/dev/sdc', ...), ...]
Source code in sts_libs/src/sts/scsi.py
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
@classmethod
def get_by_vendor(cls: type[T], vendor: str) -> list[T]:
    """Retrieve a list of ScsiDevice objects that match the specified vendor.

    Args:
        vendor (str): used to filter the SCSI devices.

    Returns:
        A list of ScsiDevice objects.

    Example:
        ```Python
        ScsiDevice.get_by_vendor('NETAPP')
        [ScsiDevice(path='/dev/sdb', name='sdb', ...), ScsiDevice(path='/dev/sdc', ...), ...]
        ```

    """
    all_devices = cls.get_all_scsi_devices()
    return [device for device in all_devices if device.vendor == vendor]

rescan_disk()

Rescan the disk.

Returns:

Type Description
bool

True if the rescan operation is successful, otherwise False.

Source code in sts_libs/src/sts/scsi.py
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
def rescan_disk(self) -> bool:
    """Rescan the disk.

    Returns:
        True if the rescan operation is successful, otherwise False.

    """
    rescan_path = Path(f'{self.BLOCK_PATH}/{self.name}/device/rescan')
    try:
        rescan_path.write_text('1')
    except OSError:
        logging.warning(f'Failed to write to {rescan_path}.')
        return False
    else:
        return True

rescan_host()

Initiates a rescan operation for the specified host ID, to detect new devices.

The scan file is part of the sysfs interface in Linux, which exposes kernel data structures to user space. Writing - - - to this file instructs the kernel to scan all channels, all targets, and all LUNs for the specified host adapter.

Returns:

Type Description
bool

True if the rescan operation is successful, False otherwise.

Source code in sts_libs/src/sts/scsi.py
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
def rescan_host(self) -> bool:
    """Initiates a rescan operation for the specified host ID, to detect new devices.

    The scan file is part of the sysfs interface in Linux, which exposes kernel data structures to user space.
    Writing - - - to this file instructs the kernel to scan all channels, all targets, and all LUNs
    for the specified host adapter.

    Returns:
        True if the rescan operation is successful, False otherwise.
    """
    if not self.host_id:
        return False

    rescan_path = Path(f'{self.SCSI_HOST_PATH}/host{self.host_id}/scan')
    try:
        rescan_path.write_text('- - -')
    except OSError:
        logging.warning(f'Failed to write to {rescan_path}.')
        return False
    else:
        return True

up_or_down_disk(action)

Change the state of the disk.

Parameters:

Name Type Description Default
action str

offline or running.

required

Returns:

Type Description
bool

True if the operation is successful, otherwise returns False.

Source code in sts_libs/src/sts/scsi.py
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
def up_or_down_disk(self, action: str) -> bool:
    """Change the state of the disk.

    Args:
        action: offline or running.

    Returns:
        True if the operation is successful, otherwise returns False.

    """
    state_path = Path(f'{self.BLOCK_PATH}/{self.name}/device/state')
    try:
        # Need a newline character at the end of the action, or else it cannot be written into
        action = f'{action}\n'
        state_path.write_text(action)
    except OSError:
        logging.warning(f'Failed to write to {state_path}')
        return False
    else:
        return True