Skip to content

SG3 Utils

sg3_utils — low-level SCSI command utilities (sg_inq, sg_readcap, sg_vpd, sg_luns, etc.) for querying and manipulating SCSI devices below the block layer.

sts.sg3_utils

Python wrappers for sg3_utils SCSI commands.

Design decision: sg3_utils classes remain plain Python (not Pydantic). They are stateless command runners -- no fields to validate, no state to manage. See docs/superpowers/plans/2026-07-27-master-modernization-roadmap.md Decision Log.

RescanScsiBus

Bases: Sg3UtilsCommand

SCSI bus rescanning using rescan-scsi-bus.sh.

Source code in sts_libs/src/sts/sg3_utils.py 
795
796
797
798
class RescanScsiBus(Sg3UtilsCommand):
    """SCSI bus rescanning using rescan-scsi-bus.sh."""

    COMMAND: ClassVar[str] = 'rescan-scsi-bus.sh'

ScsiRescan

Bases: Sg3UtilsCommand

SCSI rescanning using scsi-rescan.

Source code in sts_libs/src/sts/sg3_utils.py 
801
802
803
804
class ScsiRescan(Sg3UtilsCommand):
    """SCSI rescanning using scsi-rescan."""

    COMMAND: ClassVar[str] = 'scsi-rescan'

Sg3UtilsCommand

Base class for sg3_utils commands with shared execution and output parsing.

Source code in sts_libs/src/sts/sg3_utils.py 
 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
class Sg3UtilsCommand:
    """Base class for sg3_utils commands with shared execution and output parsing."""

    PACKAGE_NAME: ClassVar[str] = PACKAGE_NAME
    COMMAND: ClassVar[str] = ''

    def __init__(self) -> None:
        pass

    def _run_command(self, *args: str | None, **kwargs: str | None) -> CommandResult:
        """Run sg3_utils command, filtering out None args and kwargs."""
        # Filter out None values from positional arguments
        filtered_args = [arg for arg in args if arg is not None]

        # Build base command string
        command_str = ' '.join(filtered_args)

        # Handle kwargs by converting to key=value arguments
        if kwargs:
            # Filter out None values from kwargs
            filtered_kwargs = {k: v for k, v in kwargs.items() if v is not None}
            if filtered_kwargs:
                kwargs_str = ' '.join([f'{key}={value}' for key, value in filtered_kwargs.items()])
                command_str = f'{command_str} {kwargs_str}'

        logger.debug(f'Running sg3_utils command: {command_str}')
        result = _run(command_str)

        if result.failed:
            logger.warning(f'Command failed: {command_str}, stderr: {result.stderr}')

        return result

    def run(self, *args: str, device: str | Path | None = None, **options: bool | str | float | None) -> CommandResult:
        """Run the command with arbitrary flags and options.

        Kwargs are translated to CLI flags (underscores to hyphens,
        bools to bare flags, values to ``--key=value``).
        """
        parts = [self.COMMAND]
        parts.extend(args)
        for key, value in options.items():
            if value is None:
                continue
            cli_key = key.replace('_', '-')
            if isinstance(value, bool):
                if value:
                    parts.append(f'--{cli_key}')
            else:
                parts.append(f'--{cli_key}={value}')
        if device:
            parts.append(str(device))
        return _run(' '.join(parts))

    def version(self) -> CommandResult:
        """Get version information for the tool."""
        return _run(f'{self.COMMAND} --version')

    def help(self) -> CommandResult:
        """Get help information for the tool."""
        return _run(f'{self.COMMAND} --help')

    # ==========================================================================
    # Common Output Parsing Utilities
    # ==========================================================================

    @staticmethod
    def parse_key_value_pairs(output: str, separators: str = '=:') -> dict[str, str]:
        r"""Parse 'key=value' or 'key: value' lines from output.

        Example:
            ```python
            output = 'Vendor identification: Samsung\nKey=0x1234'
            pairs = Sg3UtilsCommand.parse_key_value_pairs(output)
            # {'Vendor identification': 'Samsung', 'Key': '0x1234'}
            ```
        """
        pairs: dict[str, str] = {}
        for line in output.split('\n'):
            stripped_line = line.strip()
            if not stripped_line:
                continue

            # Find first separator in the line
            separator_pos = -1
            for sep in separators:
                pos = stripped_line.find(sep)
                if pos != -1 and (separator_pos == -1 or pos < separator_pos):
                    separator_pos = pos

            if separator_pos != -1:
                key = stripped_line[:separator_pos].strip()
                value = stripped_line[separator_pos + 1 :].strip()
                if key and value:
                    pairs[key] = value

        return pairs

    @staticmethod
    def extract_section_lines(output: str, section_header: str) -> list[str]:
        r"""Extract lines after a section header until the next section or end.

        Example:
            ```python
            output = 'PR generation=0x6, Reservation follows:\n  Key=0x1234\n  scope: LU_SCOPE'
            lines = Sg3UtilsCommand.extract_section_lines(output, 'Reservation follows:')
            # ['Key=0x1234', 'scope: LU_SCOPE']
            ```
        """
        lines = output.split('\n')
        section_lines: list[str] = []
        in_section = False

        for line in lines:
            line_stripped = line.strip()

            # Check if this line contains the section header
            if section_header.lower() in line.lower():
                in_section = True
                continue

            # If we're in the section, collect non-empty lines
            if in_section:
                # Stop at next section (lines ending with ':') or empty lines that might indicate new section
                if line_stripped.endswith(':') and not line_stripped.startswith(' '):
                    break
                if line_stripped:  # Only add non-empty lines
                    section_lines.append(line_stripped)

        return section_lines

    @staticmethod
    def extract_hex_keys(output: str) -> list[str]:
        r"""Extract all hex keys (0x... format) from output.

        Example:
            ```python
            output = 'registered reservation key follows:\n  0x1234\n  0xabcd'
            keys = Sg3UtilsCommand.extract_hex_keys(output)
            # ['0x1234', '0xabcd']
            ```
        """
        hex_pattern = re.compile(r'0x[0-9a-fA-F]+')
        return hex_pattern.findall(output)

    @staticmethod
    def clean_line(line: str) -> str:
        """Strip and normalize whitespace in a line."""
        return ' '.join(line.strip().split())

    @staticmethod
    def find_line_containing(output: str, pattern: str, *, case_sensitive: bool = False) -> str | None:
        """Find first line containing a specific pattern."""
        search_pattern = pattern if case_sensitive else pattern.lower()

        for line in output.split('\n'):
            search_line = line if case_sensitive else line.lower()
            if search_pattern in search_line:
                return line.strip()

        return None

clean_line(line) staticmethod

Strip and normalize whitespace in a line.

Source code in sts_libs/src/sts/sg3_utils.py 
201
202
203
204
@staticmethod
def clean_line(line: str) -> str:
    """Strip and normalize whitespace in a line."""
    return ' '.join(line.strip().split())

extract_hex_keys(output) staticmethod

Extract all hex keys (0x... format) from output.

Example 
output = 'registered reservation key follows:\n  0x1234\n  0xabcd'
keys = Sg3UtilsCommand.extract_hex_keys(output)
# ['0x1234', '0xabcd']
Source code in sts_libs/src/sts/sg3_utils.py 
187
188
189
190
191
192
193
194
195
196
197
198
199
@staticmethod
def extract_hex_keys(output: str) -> list[str]:
    r"""Extract all hex keys (0x... format) from output.

    Example:
        ```python
        output = 'registered reservation key follows:\n  0x1234\n  0xabcd'
        keys = Sg3UtilsCommand.extract_hex_keys(output)
        # ['0x1234', '0xabcd']
        ```
    """
    hex_pattern = re.compile(r'0x[0-9a-fA-F]+')
    return hex_pattern.findall(output)

extract_section_lines(output, section_header) staticmethod

Extract lines after a section header until the next section or end.

Example 
output = 'PR generation=0x6, Reservation follows:\n  Key=0x1234\n  scope: LU_SCOPE'
lines = Sg3UtilsCommand.extract_section_lines(output, 'Reservation follows:')
# ['Key=0x1234', 'scope: LU_SCOPE']
Source code in sts_libs/src/sts/sg3_utils.py 
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
@staticmethod
def extract_section_lines(output: str, section_header: str) -> list[str]:
    r"""Extract lines after a section header until the next section or end.

    Example:
        ```python
        output = 'PR generation=0x6, Reservation follows:\n  Key=0x1234\n  scope: LU_SCOPE'
        lines = Sg3UtilsCommand.extract_section_lines(output, 'Reservation follows:')
        # ['Key=0x1234', 'scope: LU_SCOPE']
        ```
    """
    lines = output.split('\n')
    section_lines: list[str] = []
    in_section = False

    for line in lines:
        line_stripped = line.strip()

        # Check if this line contains the section header
        if section_header.lower() in line.lower():
            in_section = True
            continue

        # If we're in the section, collect non-empty lines
        if in_section:
            # Stop at next section (lines ending with ':') or empty lines that might indicate new section
            if line_stripped.endswith(':') and not line_stripped.startswith(' '):
                break
            if line_stripped:  # Only add non-empty lines
                section_lines.append(line_stripped)

    return section_lines

find_line_containing(output, pattern, *, case_sensitive=False) staticmethod

Find first line containing a specific pattern.

Source code in sts_libs/src/sts/sg3_utils.py 
206
207
208
209
210
211
212
213
214
215
216
@staticmethod
def find_line_containing(output: str, pattern: str, *, case_sensitive: bool = False) -> str | None:
    """Find first line containing a specific pattern."""
    search_pattern = pattern if case_sensitive else pattern.lower()

    for line in output.split('\n'):
        search_line = line if case_sensitive else line.lower()
        if search_pattern in search_line:
            return line.strip()

    return None

help()

Get help information for the tool.

Source code in sts_libs/src/sts/sg3_utils.py 
114
115
116
def help(self) -> CommandResult:
    """Get help information for the tool."""
    return _run(f'{self.COMMAND} --help')

parse_key_value_pairs(output, separators='=:') staticmethod

Parse 'key=value' or 'key: value' lines from output.

Example 
output = 'Vendor identification: Samsung\nKey=0x1234'
pairs = Sg3UtilsCommand.parse_key_value_pairs(output)
# {'Vendor identification': 'Samsung', 'Key': '0x1234'}
Source code in sts_libs/src/sts/sg3_utils.py 
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
@staticmethod
def parse_key_value_pairs(output: str, separators: str = '=:') -> dict[str, str]:
    r"""Parse 'key=value' or 'key: value' lines from output.

    Example:
        ```python
        output = 'Vendor identification: Samsung\nKey=0x1234'
        pairs = Sg3UtilsCommand.parse_key_value_pairs(output)
        # {'Vendor identification': 'Samsung', 'Key': '0x1234'}
        ```
    """
    pairs: dict[str, str] = {}
    for line in output.split('\n'):
        stripped_line = line.strip()
        if not stripped_line:
            continue

        # Find first separator in the line
        separator_pos = -1
        for sep in separators:
            pos = stripped_line.find(sep)
            if pos != -1 and (separator_pos == -1 or pos < separator_pos):
                separator_pos = pos

        if separator_pos != -1:
            key = stripped_line[:separator_pos].strip()
            value = stripped_line[separator_pos + 1 :].strip()
            if key and value:
                pairs[key] = value

    return pairs

run(*args, device=None, **options)

Run the command with arbitrary flags and options.

Kwargs are translated to CLI flags (underscores to hyphens, bools to bare flags, values to --key=value).

Source code in sts_libs/src/sts/sg3_utils.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def run(self, *args: str, device: str | Path | None = None, **options: bool | str | float | None) -> CommandResult:
    """Run the command with arbitrary flags and options.

    Kwargs are translated to CLI flags (underscores to hyphens,
    bools to bare flags, values to ``--key=value``).
    """
    parts = [self.COMMAND]
    parts.extend(args)
    for key, value in options.items():
        if value is None:
            continue
        cli_key = key.replace('_', '-')
        if isinstance(value, bool):
            if value:
                parts.append(f'--{cli_key}')
        else:
            parts.append(f'--{cli_key}={value}')
    if device:
        parts.append(str(device))
    return _run(' '.join(parts))

version()

Get version information for the tool.

Source code in sts_libs/src/sts/sg3_utils.py 
110
111
112
def version(self) -> CommandResult:
    """Get version information for the tool."""
    return _run(f'{self.COMMAND} --version')

SgDd

Bases: Sg3UtilsCommand

SCSI dd utility using sg_dd.

Source code in sts_libs/src/sts/sg3_utils.py 
693
694
695
696
class SgDd(Sg3UtilsCommand):
    """SCSI dd utility using sg_dd."""

    COMMAND: ClassVar[str] = 'sg_dd'

SgFormat

Bases: Sg3UtilsCommand

SCSI FORMAT UNIT commands using sg_format.

Source code in sts_libs/src/sts/sg3_utils.py 
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
class SgFormat(Sg3UtilsCommand):
    """SCSI FORMAT UNIT commands using sg_format."""

    COMMAND: ClassVar[str] = 'sg_format'

    def format_device(
        self, device: str | Path, block_size: int | None = None, *, dry_run: bool = True
    ) -> CommandResult:
        """Format a SCSI device.

        Args:
            device: Device path
            block_size: Block size for formatting
            dry_run: Perform dry run without actual formatting (default: True for safety)
        """
        return self._run_command(
            'sg_format',
            '--dry-run' if dry_run else None,
            '--size' if block_size else None,
            str(block_size) if block_size else None,
            str(device),
        )

format_device(device, block_size=None, *, dry_run=True)

Format a SCSI device.

Parameters:

Name Type Description Default
device str | Path

Device path

 required
block_size int | None

Block size for formatting

 None
dry_run bool

Perform dry run without actual formatting (default: True for safety)

 True
Source code in sts_libs/src/sts/sg3_utils.py 
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
def format_device(
    self, device: str | Path, block_size: int | None = None, *, dry_run: bool = True
) -> CommandResult:
    """Format a SCSI device.

    Args:
        device: Device path
        block_size: Block size for formatting
        dry_run: Perform dry run without actual formatting (default: True for safety)
    """
    return self._run_command(
        'sg_format',
        '--dry-run' if dry_run else None,
        '--size' if block_size else None,
        str(block_size) if block_size else None,
        str(device),
    )

SgGetConfig

Bases: Sg3UtilsCommand

SCSI GET CONFIGURATION commands using sg_get_config.

Source code in sts_libs/src/sts/sg3_utils.py 
717
718
719
720
class SgGetConfig(Sg3UtilsCommand):
    """SCSI GET CONFIGURATION commands using sg_get_config."""

    COMMAND: ClassVar[str] = 'sg_get_config'

SgIdent

Bases: Sg3UtilsCommand

SCSI device identification using sg_ident.

Source code in sts_libs/src/sts/sg3_utils.py 
723
724
725
726
class SgIdent(Sg3UtilsCommand):
    """SCSI device identification using sg_ident."""

    COMMAND: ClassVar[str] = 'sg_ident'

SgInq

Bases: Sg3UtilsCommand

SCSI INQUIRY command using sg_inq.

Source code in sts_libs/src/sts/sg3_utils.py 
479
480
481
482
483
484
485
486
487
488
489
490
491
class SgInq(Sg3UtilsCommand):
    """SCSI INQUIRY command using sg_inq."""

    COMMAND: ClassVar[str] = 'sg_inq'

    def inquiry(self, device: str | Path, *, vpd: bool = False) -> CommandResult:
        """Perform SCSI INQUIRY on a device.

        Args:
            device: Device path
            vpd: Request Vital Product Data pages
        """
        return self._run_command(self.COMMAND, '--vpd' if vpd else None, str(device))

inquiry(device, *, vpd=False)

Perform SCSI INQUIRY on a device.

Parameters:

Name Type Description Default
device str | Path

Device path

 required
vpd bool

Request Vital Product Data pages

 False
Source code in sts_libs/src/sts/sg3_utils.py 
484
485
486
487
488
489
490
491
def inquiry(self, device: str | Path, *, vpd: bool = False) -> CommandResult:
    """Perform SCSI INQUIRY on a device.

    Args:
        device: Device path
        vpd: Request Vital Product Data pages
    """
    return self._run_command(self.COMMAND, '--vpd' if vpd else None, str(device))

SgLogs

Bases: Sg3UtilsCommand

SCSI LOG SENSE commands using sg_logs.

Source code in sts_libs/src/sts/sg3_utils.py 
581
582
583
584
585
586
587
588
589
590
591
592
593
class SgLogs(Sg3UtilsCommand):
    """SCSI LOG SENSE commands using sg_logs."""

    COMMAND: ClassVar[str] = 'sg_logs'

    def get_log_page(self, device: str | Path, page: str) -> CommandResult:
        """Get a log page from a device.

        Args:
            device: Device path
            page: Log page identifier (e.g., 'temp', 'ie', 'sp')
        """
        return self._run_command('sg_logs', f'--page={page}', str(device))

get_log_page(device, page)

Get a log page from a device.

Parameters:

Name Type Description Default
device str | Path

Device path

 required
page str

Log page identifier (e.g., 'temp', 'ie', 'sp')

 required
Source code in sts_libs/src/sts/sg3_utils.py 
586
587
588
589
590
591
592
593
def get_log_page(self, device: str | Path, page: str) -> CommandResult:
    """Get a log page from a device.

    Args:
        device: Device path
        page: Log page identifier (e.g., 'temp', 'ie', 'sp')
    """
    return self._run_command('sg_logs', f'--page={page}', str(device))

SgLuns

Bases: Sg3UtilsCommand

SCSI REPORT LUNS commands using sg_luns.

Source code in sts_libs/src/sts/sg3_utils.py 
729
730
731
732
class SgLuns(Sg3UtilsCommand):
    """SCSI REPORT LUNS commands using sg_luns."""

    COMMAND: ClassVar[str] = 'sg_luns'

SgMap

Bases: Sg3UtilsCommand

SCSI device mapping using sg_map.

Source code in sts_libs/src/sts/sg3_utils.py 
665
666
667
668
669
670
671
672
class SgMap(Sg3UtilsCommand):
    """SCSI device mapping using sg_map."""

    COMMAND: ClassVar[str] = 'sg_map'

    def map_devices(self) -> CommandResult:
        """Map SCSI generic devices to block devices."""
        return self._run_command('sg_map')

map_devices()

Map SCSI generic devices to block devices.

Source code in sts_libs/src/sts/sg3_utils.py 
670
671
672
def map_devices(self) -> CommandResult:
    """Map SCSI generic devices to block devices."""
    return self._run_command('sg_map')

SgMap26

Bases: Sg3UtilsCommand

SCSI device mapping using sg_map26.

Source code in sts_libs/src/sts/sg3_utils.py 
789
790
791
792
class SgMap26(Sg3UtilsCommand):
    """SCSI device mapping using sg_map26."""

    COMMAND: ClassVar[str] = 'sg_map26'

SgModes

Bases: Sg3UtilsCommand

SCSI MODE SENSE commands using sg_modes.

Source code in sts_libs/src/sts/sg3_utils.py 
735
736
737
738
class SgModes(Sg3UtilsCommand):
    """SCSI MODE SENSE commands using sg_modes."""

    COMMAND: ClassVar[str] = 'sg_modes'

SgOpcodes

Bases: Sg3UtilsCommand

SCSI REPORT SUPPORTED OPERATION CODES using sg_opcodes.

Source code in sts_libs/src/sts/sg3_utils.py 
741
742
743
744
class SgOpcodes(Sg3UtilsCommand):
    """SCSI REPORT SUPPORTED OPERATION CODES using sg_opcodes."""

    COMMAND: ClassVar[str] = 'sg_opcodes'

SgPersist

Bases: Sg3UtilsCommand

SCSI Persistent Reservations using sg_persist.

Coordinates access to shared SCSI devices between multiple initiators via registration keys, reservations, and preemption.

Source code in sts_libs/src/sts/sg3_utils.py 
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
class SgPersist(Sg3UtilsCommand):
    """SCSI Persistent Reservations using sg_persist.

    Coordinates access to shared SCSI devices between multiple initiators
    via registration keys, reservations, and preemption.
    """

    COMMAND: ClassVar[str] = 'sg_persist'

    # ==========================================================================
    # Command-Specific Parser Methods
    # ==========================================================================

    def _parse_keys_from_output(self, output: str) -> list[str]:
        """Parse registered keys from sg_persist --read-keys output."""
        keys: list[str] = []
        lines = output.split('\n')
        found_keys_section = False

        for line in lines:
            stripped_line = line.strip()
            # Look for the keys section header
            if 'registered reservation key' in stripped_line.lower():
                found_keys_section = True
                continue
            # Parse keys (lines starting with 0x after the header)
            if found_keys_section and stripped_line.startswith('0x'):
                keys.append(stripped_line)

        # Return keys exactly as reported by the device (including duplicates)
        return keys

    def _parse_reservation_from_output(self, output: str) -> tuple[str | None, str | None]:
        """Parse (holder_key, reservation_type) from --read-reservation output."""
        holder = None
        res_type = None

        # Check if there's actually a reservation
        if self.find_line_containing(output, 'NO reservation held'):
            return None, None

        # Look for reservation section
        if 'Reservation follows:' in output:
            reservation_lines = self.extract_section_lines(output, 'Reservation follows:')

            for line in reservation_lines:
                # Parse Key=0x1234 format
                if line.startswith('Key='):
                    holder = line.split('=')[1].strip()

                # Parse type from scope/type line
                elif 'type:' in line:
                    type_part = line.split('type:')[1].strip()
                    res_type = self._parse_pr_type_from_text(type_part)

        return holder, res_type

    def _parse_pr_type_from_text(self, text: str) -> str | None:
        """Convert PR type text (e.g., 'Write Exclusive') to its type number string."""
        if not text:
            return None

        # Use centralized mapping for type names to numbers
        # Sort by length (longest first) to match most specific patterns first
        sorted_mappings = sorted(PR_NAME_TO_TYPE.items(), key=lambda x: len(x[0]), reverse=True)
        for type_name_lower, type_num in sorted_mappings:
            if type_name_lower in text.lower():
                return type_num

        # If no exact mapping found, return the raw type
        return text.split(',', maxsplit=1)[0].strip()

    def _parse_full_status_from_output(self, output: str) -> dict[str, str]:
        """Parse key-value pairs from --read-full-status output."""
        return self.parse_key_value_pairs(output)

    def _parse_transport_ids_from_output(self, output: str) -> list[dict[str, str]]:
        """Parse transport IDs (type/id/key dicts) from --read-full-status output."""
        transport_ids: list[dict[str, str]] = []
        lines = output.split('\n')
        current_key = None

        # Transport type keywords (order matters for specificity)
        transport_keywords = [
            ('iscsi', 'iscsi'),
            ('sas', 'sas address'),
            ('fc', 'fc n_port_id'),
            ('fc', 'fibre channel'),
        ]

        for i, line in enumerate(lines):
            stripped_line = line.strip()

            if stripped_line.startswith('Key='):
                current_key = stripped_line.split('=')[1].strip()
            elif 'Transport Id of initiator:' in stripped_line and i + 1 < len(lines):
                next_line = lines[i + 1].strip()
                if ':' not in next_line:
                    continue

                transport_id = next_line.split(':', 1)[1].strip()
                transport_type = 'unknown'

                # Find transport type by keyword matching
                for t_type, keyword in transport_keywords:
                    if keyword in next_line.lower():
                        transport_type = t_type
                        break

                transport_ids.append({'type': transport_type, 'id': transport_id, 'key': current_key or 'unknown'})

        return transport_ids

    # ==========================================================================
    # Public Methods
    # ==========================================================================

    def read_keys(self, device: str | Path) -> list[str]:
        """Read all registered reservation keys for a device."""
        result = self._run_command(self.COMMAND, '--in', '--read-keys', str(device))

        if result.succeeded:
            return self._parse_keys_from_output(result.stdout)
        logger.warning(f'Failed to read keys for {device}: {result.stderr}')
        return []

    def read_reservation(self, device: str | Path) -> tuple[str | None, str | None]:
        """Read current reservation holder for a device.

        Returns:
            (holder_key, reservation_type) or (None, None) if no reservation
        """
        result = self._run_command(self.COMMAND, '--in', '--read-reservation', str(device))

        if result.succeeded:
            return self._parse_reservation_from_output(result.stdout)
        logger.warning(f'Failed to read reservation for {device}: {result.stderr}')
        return (None, None)

    def read_full_status(self, device: str | Path) -> dict[str, str]:
        """Read full status (keys and reservation) for a device."""
        result = self._run_command(self.COMMAND, '--in', '--read-full-status', str(device))

        if result.succeeded:
            return self._parse_full_status_from_output(result.stdout)
        logger.warning(f'Failed to read full status for {device}: {result.stderr}')
        return {}

    def get_transport_ids(self, device: str | Path) -> list[dict[str, str]]:
        """Get transport IDs of all registered initiators.

        Returns:
            List of dicts with 'type' (iscsi/sas/fc/unknown), 'id', and 'key'.
        """
        result = self._run_command(self.COMMAND, '--in', '--read-full-status', str(device))

        if result.succeeded:
            return self._parse_transport_ids_from_output(result.stdout)
        logger.warning(f'Failed to read transport IDs for {device}: {result.stderr}')
        return []

    def register(self, device: str | Path, key: str, transport_id: str | None = None) -> bool:
        """Register a key for persistent reservations.

        Args:
            device: Device path
            key: Registration key (e.g., '0xaaaa')
            transport_id: Transport-specific identifier (e.g., 'sas,5001405f31c32fa2')

        Example:
            ```python
            persist.register('/dev/sdb', '0xaaaa')
            persist.register('/dev/sda', '0x2', transport_id='sas,5001405f31c32fa2')
            ```
        """
        result = self._run_command(
            self.COMMAND,
            '--out',
            '--register',
            '--param-sark',
            key,
            '-X' if transport_id else None,
            transport_id,
            str(device),
        )

        if not result.succeeded:
            logger.warning(f'Failed to register key {key} for {device}: {result.stderr}')

        return result.succeeded

    def unregister(self, device: str | Path, key_to_remove: str) -> bool:
        """Delete an existing registration (sets SARK to 0 for the given key)."""
        result = self._run_command(
            self.COMMAND, '--out', '--register', f'--param-rk={key_to_remove}', '--param-sark=0', str(device)
        )

        if not result.succeeded:
            logger.warning(f'Failed to unregister key {key_to_remove} for {device}: {result.stderr}')

        return result.succeeded

    def reserve(self, device: str | Path, key: str, prout_type: int | str = 1) -> bool:
        """Create a reservation on a device.

        Args:
            device: Device path
            key: Reservation key (must be previously registered)
            prout_type: PR type (default: 1 = Write Exclusive, see PR_TYPE_* constants)
        """
        result = self._run_command(
            self.COMMAND, '--out', '--reserve', f'--param-rk={key}', f'--prout-type={prout_type}', str(device)
        )

        if not result.succeeded:
            logger.warning(f'Failed to reserve {device} with key {key}: {result.stderr}')

        return result.succeeded

    def release(self, device: str | Path, key: str, prout_type: int | str = 1) -> bool:
        """Release a reservation on a device.

        Args:
            device: Device path
            key: Reservation key
            prout_type: PR type to release (default: 1 = Write Exclusive)
        """
        result = self._run_command(
            self.COMMAND, '--out', '--release', f'--param-rk={key}', f'--prout-type={prout_type}', str(device)
        )

        if not result.succeeded:
            logger.warning(f'Failed to release reservation on {device} with key {key}: {result.stderr}')

        return result.succeeded

    def report_capabilities(self, device: str | Path) -> bool:
        """Check whether a device supports persistent reservation operations."""
        return self.report_capabilities_result(device).succeeded

    def report_capabilities_result(self, device: str | Path) -> CommandResult:
        """Query persistent reservation capabilities and preserve failure details."""
        result = self._run_command(self.COMMAND, '--in', '--report-capabilities', str(device))

        if not result.succeeded:
            logger.warning(f'Failed to check PR capabilities for {device}: {result.stderr}')

        return result

get_transport_ids(device)

Get transport IDs of all registered initiators.

Returns:

Type Description
list[dict[str, str]]

List of dicts with 'type' (iscsi/sas/fc/unknown), 'id', and 'key'.
Source code in sts_libs/src/sts/sg3_utils.py 
372
373
374
375
376
377
378
379
380
381
382
383
def get_transport_ids(self, device: str | Path) -> list[dict[str, str]]:
    """Get transport IDs of all registered initiators.

    Returns:
        List of dicts with 'type' (iscsi/sas/fc/unknown), 'id', and 'key'.
    """
    result = self._run_command(self.COMMAND, '--in', '--read-full-status', str(device))

    if result.succeeded:
        return self._parse_transport_ids_from_output(result.stdout)
    logger.warning(f'Failed to read transport IDs for {device}: {result.stderr}')
    return []

read_full_status(device)

Read full status (keys and reservation) for a device.

Source code in sts_libs/src/sts/sg3_utils.py 
363
364
365
366
367
368
369
370
def read_full_status(self, device: str | Path) -> dict[str, str]:
    """Read full status (keys and reservation) for a device."""
    result = self._run_command(self.COMMAND, '--in', '--read-full-status', str(device))

    if result.succeeded:
        return self._parse_full_status_from_output(result.stdout)
    logger.warning(f'Failed to read full status for {device}: {result.stderr}')
    return {}

read_keys(device)

Read all registered reservation keys for a device.

Source code in sts_libs/src/sts/sg3_utils.py 
341
342
343
344
345
346
347
348
def read_keys(self, device: str | Path) -> list[str]:
    """Read all registered reservation keys for a device."""
    result = self._run_command(self.COMMAND, '--in', '--read-keys', str(device))

    if result.succeeded:
        return self._parse_keys_from_output(result.stdout)
    logger.warning(f'Failed to read keys for {device}: {result.stderr}')
    return []

read_reservation(device)

Read current reservation holder for a device.

Returns:

Type Description
tuple[str | None, str | None]

(holder_key, reservation_type) or (None, None) if no reservation
Source code in sts_libs/src/sts/sg3_utils.py 
350
351
352
353
354
355
356
357
358
359
360
361
def read_reservation(self, device: str | Path) -> tuple[str | None, str | None]:
    """Read current reservation holder for a device.

    Returns:
        (holder_key, reservation_type) or (None, None) if no reservation
    """
    result = self._run_command(self.COMMAND, '--in', '--read-reservation', str(device))

    if result.succeeded:
        return self._parse_reservation_from_output(result.stdout)
    logger.warning(f'Failed to read reservation for {device}: {result.stderr}')
    return (None, None)

register(device, key, transport_id=None)

Register a key for persistent reservations.

Parameters:

Name Type Description Default
device str | Path

Device path

 required
key str

Registration key (e.g., '0xaaaa')

 required
transport_id str | None

Transport-specific identifier (e.g., 'sas,5001405f31c32fa2')

 None
Example 
persist.register('/dev/sdb', '0xaaaa')
persist.register('/dev/sda', '0x2', transport_id='sas,5001405f31c32fa2')
Source code in sts_libs/src/sts/sg3_utils.py 
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
def register(self, device: str | Path, key: str, transport_id: str | None = None) -> bool:
    """Register a key for persistent reservations.

    Args:
        device: Device path
        key: Registration key (e.g., '0xaaaa')
        transport_id: Transport-specific identifier (e.g., 'sas,5001405f31c32fa2')

    Example:
        ```python
        persist.register('/dev/sdb', '0xaaaa')
        persist.register('/dev/sda', '0x2', transport_id='sas,5001405f31c32fa2')
        ```
    """
    result = self._run_command(
        self.COMMAND,
        '--out',
        '--register',
        '--param-sark',
        key,
        '-X' if transport_id else None,
        transport_id,
        str(device),
    )

    if not result.succeeded:
        logger.warning(f'Failed to register key {key} for {device}: {result.stderr}')

    return result.succeeded

release(device, key, prout_type=1)

Release a reservation on a device.

Parameters:

Name Type Description Default
device str | Path

Device path

 required
key str

Reservation key

 required
prout_type int | str

PR type to release (default: 1 = Write Exclusive)

 1
Source code in sts_libs/src/sts/sg3_utils.py 
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
def release(self, device: str | Path, key: str, prout_type: int | str = 1) -> bool:
    """Release a reservation on a device.

    Args:
        device: Device path
        key: Reservation key
        prout_type: PR type to release (default: 1 = Write Exclusive)
    """
    result = self._run_command(
        self.COMMAND, '--out', '--release', f'--param-rk={key}', f'--prout-type={prout_type}', str(device)
    )

    if not result.succeeded:
        logger.warning(f'Failed to release reservation on {device} with key {key}: {result.stderr}')

    return result.succeeded

report_capabilities(device)

Check whether a device supports persistent reservation operations.

Source code in sts_libs/src/sts/sg3_utils.py 
460
461
462
def report_capabilities(self, device: str | Path) -> bool:
    """Check whether a device supports persistent reservation operations."""
    return self.report_capabilities_result(device).succeeded

report_capabilities_result(device)

Query persistent reservation capabilities and preserve failure details.

Source code in sts_libs/src/sts/sg3_utils.py 
464
465
466
467
468
469
470
471
def report_capabilities_result(self, device: str | Path) -> CommandResult:
    """Query persistent reservation capabilities and preserve failure details."""
    result = self._run_command(self.COMMAND, '--in', '--report-capabilities', str(device))

    if not result.succeeded:
        logger.warning(f'Failed to check PR capabilities for {device}: {result.stderr}')

    return result

reserve(device, key, prout_type=1)

Create a reservation on a device.

Parameters:

Name Type Description Default
device str | Path

Device path

 required
key str

Reservation key (must be previously registered)

 required
prout_type int | str

PR type (default: 1 = Write Exclusive, see PR_TYPE_* constants)

 1
Source code in sts_libs/src/sts/sg3_utils.py 
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
def reserve(self, device: str | Path, key: str, prout_type: int | str = 1) -> bool:
    """Create a reservation on a device.

    Args:
        device: Device path
        key: Reservation key (must be previously registered)
        prout_type: PR type (default: 1 = Write Exclusive, see PR_TYPE_* constants)
    """
    result = self._run_command(
        self.COMMAND, '--out', '--reserve', f'--param-rk={key}', f'--prout-type={prout_type}', str(device)
    )

    if not result.succeeded:
        logger.warning(f'Failed to reserve {device} with key {key}: {result.stderr}')

    return result.succeeded

unregister(device, key_to_remove)

Delete an existing registration (sets SARK to 0 for the given key).

Source code in sts_libs/src/sts/sg3_utils.py 
415
416
417
418
419
420
421
422
423
424
def unregister(self, device: str | Path, key_to_remove: str) -> bool:
    """Delete an existing registration (sets SARK to 0 for the given key)."""
    result = self._run_command(
        self.COMMAND, '--out', '--register', f'--param-rk={key_to_remove}', '--param-sark=0', str(device)
    )

    if not result.succeeded:
        logger.warning(f'Failed to unregister key {key_to_remove} for {device}: {result.stderr}')

    return result.succeeded

SgRbuf

Bases: Sg3UtilsCommand

SCSI READ BUFFER commands using sg_rbuf.

Source code in sts_libs/src/sts/sg3_utils.py 
747
748
749
750
class SgRbuf(Sg3UtilsCommand):
    """SCSI READ BUFFER commands using sg_rbuf."""

    COMMAND: ClassVar[str] = 'sg_rbuf'

SgRead

Bases: Sg3UtilsCommand

SCSI READ commands using sg_read.

Source code in sts_libs/src/sts/sg3_utils.py 
514
515
516
517
518
519
520
521
class SgRead(Sg3UtilsCommand):
    """SCSI READ commands using sg_read."""

    COMMAND: ClassVar[str] = 'sg_read'

    def read_blocks(self, device: str | Path, lba: int, count: int, block_size: int = 512) -> CommandResult:
        """Read blocks from a SCSI device."""
        return self._run_command(self.COMMAND, f'--lba={lba}', f'--num={count}', f'--bs={block_size}', str(device))

read_blocks(device, lba, count, block_size=512)

Read blocks from a SCSI device.

Source code in sts_libs/src/sts/sg3_utils.py 
519
520
521
def read_blocks(self, device: str | Path, lba: int, count: int, block_size: int = 512) -> CommandResult:
    """Read blocks from a SCSI device."""
    return self._run_command(self.COMMAND, f'--lba={lba}', f'--num={count}', f'--bs={block_size}', str(device))

SgReadBuffer

Bases: Sg3UtilsCommand

SCSI READ BUFFER commands using sg_read_buffer.

Source code in sts_libs/src/sts/sg3_utils.py 
675
676
677
678
class SgReadBuffer(Sg3UtilsCommand):
    """SCSI READ BUFFER commands using sg_read_buffer."""

    COMMAND: ClassVar[str] = 'sg_read_buffer'

SgReadLong

Bases: Sg3UtilsCommand

SCSI READ LONG commands using sg_read_long.

Source code in sts_libs/src/sts/sg3_utils.py 
681
682
683
684
class SgReadLong(Sg3UtilsCommand):
    """SCSI READ LONG commands using sg_read_long."""

    COMMAND: ClassVar[str] = 'sg_read_long'

SgReadcap

Bases: Sg3UtilsCommand

SCSI READ CAPACITY commands using sg_readcap.

Source code in sts_libs/src/sts/sg3_utils.py 
524
525
526
527
528
529
530
531
532
533
534
535
536
class SgReadcap(Sg3UtilsCommand):
    """SCSI READ CAPACITY commands using sg_readcap."""

    COMMAND: ClassVar[str] = 'sg_readcap'

    def get_capacity(self, device: str | Path, *, long_format: bool = False) -> CommandResult:
        """Get device capacity information.

        Args:
            device: Device path
            long_format: Use READ CAPACITY(16) instead of READ CAPACITY(10)
        """
        return self._run_command(self.COMMAND, '--16' if long_format else None, str(device))

get_capacity(device, *, long_format=False)

Get device capacity information.

Parameters:

Name Type Description Default
device str | Path

Device path

 required
long_format bool

Use READ CAPACITY(16) instead of READ CAPACITY(10)

 False
Source code in sts_libs/src/sts/sg3_utils.py 
529
530
531
532
533
534
535
536
def get_capacity(self, device: str | Path, *, long_format: bool = False) -> CommandResult:
    """Get device capacity information.

    Args:
        device: Device path
        long_format: Use READ CAPACITY(16) instead of READ CAPACITY(10)
    """
    return self._run_command(self.COMMAND, '--16' if long_format else None, str(device))

SgReset

Bases: Sg3UtilsCommand

SCSI device/bus/host/target reset via sg_reset.

Source code in sts_libs/src/sts/sg3_utils.py 
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
class SgReset(Sg3UtilsCommand):
    """SCSI device/bus/host/target reset via sg_reset."""

    COMMAND: ClassVar[str] = 'sg_reset'

    def device(self, device: str | Path, *, no_esc: bool = False) -> CommandResult:
        """Device reset."""
        return self.run('--device', device=device, no_esc=no_esc or None)

    def bus(self, device: str | Path, *, no_esc: bool = False) -> CommandResult:
        """Bus reset."""
        return self.run('--bus', device=device, no_esc=no_esc or None)

    def target(self, device: str | Path, *, no_esc: bool = False) -> CommandResult:
        """Target reset."""
        return self.run('--target', device=device, no_esc=no_esc or None)

    def host(self, device: str | Path, *, no_esc: bool = False) -> CommandResult:
        """Host adapter reset."""
        return self.run('--host', device=device, no_esc=no_esc or None)

bus(device, *, no_esc=False)

Bus reset.

Source code in sts_libs/src/sts/sg3_utils.py 
563
564
565
def bus(self, device: str | Path, *, no_esc: bool = False) -> CommandResult:
    """Bus reset."""
    return self.run('--bus', device=device, no_esc=no_esc or None)

device(device, *, no_esc=False)

Device reset.

Source code in sts_libs/src/sts/sg3_utils.py 
559
560
561
def device(self, device: str | Path, *, no_esc: bool = False) -> CommandResult:
    """Device reset."""
    return self.run('--device', device=device, no_esc=no_esc or None)

host(device, *, no_esc=False)

Host adapter reset.

Source code in sts_libs/src/sts/sg3_utils.py 
571
572
573
def host(self, device: str | Path, *, no_esc: bool = False) -> CommandResult:
    """Host adapter reset."""
    return self.run('--host', device=device, no_esc=no_esc or None)

target(device, *, no_esc=False)

Target reset.

Source code in sts_libs/src/sts/sg3_utils.py 
567
568
569
def target(self, device: str | Path, *, no_esc: bool = False) -> CommandResult:
    """Target reset."""
    return self.run('--target', device=device, no_esc=no_esc or None)

SgSafte

Bases: Sg3UtilsCommand

SCSI SAF-TE enclosure management using sg_safte.

Source code in sts_libs/src/sts/sg3_utils.py 
753
754
755
756
class SgSafte(Sg3UtilsCommand):
    """SCSI SAF-TE enclosure management using sg_safte."""

    COMMAND: ClassVar[str] = 'sg_safte'

SgSatIdentify

Bases: Sg3UtilsCommand

SATA IDENTIFY DEVICE via SAT using sg_sat_identify.

Source code in sts_libs/src/sts/sg3_utils.py 
759
760
761
762
class SgSatIdentify(Sg3UtilsCommand):
    """SATA IDENTIFY DEVICE via SAT using sg_sat_identify."""

    COMMAND: ClassVar[str] = 'sg_sat_identify'

SgScan

Bases: Sg3UtilsCommand

SCSI device scanning using sg_scan.

Source code in sts_libs/src/sts/sg3_utils.py 
655
656
657
658
659
660
661
662
class SgScan(Sg3UtilsCommand):
    """SCSI device scanning using sg_scan."""

    COMMAND: ClassVar[str] = 'sg_scan'

    def scan_devices(self) -> CommandResult:
        """Scan for SCSI devices."""
        return self._run_command('sg_scan')

scan_devices()

Scan for SCSI devices.

Source code in sts_libs/src/sts/sg3_utils.py 
660
661
662
def scan_devices(self) -> CommandResult:
    """Scan for SCSI devices."""
    return self._run_command('sg_scan')

SgSenddiag

Bases: Sg3UtilsCommand

SCSI SEND DIAGNOSTIC commands using sg_senddiag.

Source code in sts_libs/src/sts/sg3_utils.py 
596
597
598
599
600
601
602
603
604
605
606
607
608
class SgSenddiag(Sg3UtilsCommand):
    """SCSI SEND DIAGNOSTIC commands using sg_senddiag."""

    COMMAND: ClassVar[str] = 'sg_senddiag'

    def send_diagnostic(self, device: str | Path, test_type: str = 'default') -> CommandResult:
        """Send diagnostic command to device."""
        return self._run_command(
            'sg_senddiag',
            '--test' if test_type != 'default' else None,
            test_type if test_type != 'default' else None,
            str(device),
        )

send_diagnostic(device, test_type='default')

Send diagnostic command to device.

Source code in sts_libs/src/sts/sg3_utils.py 
601
602
603
604
605
606
607
608
def send_diagnostic(self, device: str | Path, test_type: str = 'default') -> CommandResult:
    """Send diagnostic command to device."""
    return self._run_command(
        'sg_senddiag',
        '--test' if test_type != 'default' else None,
        test_type if test_type != 'default' else None,
        str(device),
    )

SgSes

Bases: Sg3UtilsCommand

SCSI Enclosure Services using sg_ses.

Source code in sts_libs/src/sts/sg3_utils.py 
765
766
767
768
class SgSes(Sg3UtilsCommand):
    """SCSI Enclosure Services using sg_ses."""

    COMMAND: ClassVar[str] = 'sg_ses'

SgStart

Bases: Sg3UtilsCommand

SCSI START STOP UNIT commands using sg_start.

Source code in sts_libs/src/sts/sg3_utils.py 
771
772
773
774
class SgStart(Sg3UtilsCommand):
    """SCSI START STOP UNIT commands using sg_start."""

    COMMAND: ClassVar[str] = 'sg_start'

SgSync

Bases: Sg3UtilsCommand

SCSI SYNCHRONIZE CACHE commands using sg_sync.

Source code in sts_libs/src/sts/sg3_utils.py 
777
778
779
780
class SgSync(Sg3UtilsCommand):
    """SCSI SYNCHRONIZE CACHE commands using sg_sync."""

    COMMAND: ClassVar[str] = 'sg_sync'

SgTurs

Bases: Sg3UtilsCommand

SCSI TEST UNIT READY commands using sg_turs.

Source code in sts_libs/src/sts/sg3_utils.py 
783
784
785
786
class SgTurs(Sg3UtilsCommand):
    """SCSI TEST UNIT READY commands using sg_turs."""

    COMMAND: ClassVar[str] = 'sg_turs'

SgVerify

Bases: Sg3UtilsCommand

SCSI VERIFY commands using sg_verify.

Source code in sts_libs/src/sts/sg3_utils.py 
640
641
642
643
644
645
646
647
class SgVerify(Sg3UtilsCommand):
    """SCSI VERIFY commands using sg_verify."""

    COMMAND: ClassVar[str] = 'sg_verify'

    def verify_blocks(self, device: str | Path, lba: int, count: int) -> CommandResult:
        """Verify blocks on a SCSI device."""
        return self._run_command('sg_verify', f'--lba={lba}', f'--num={count}', str(device))

verify_blocks(device, lba, count)

Verify blocks on a SCSI device.

Source code in sts_libs/src/sts/sg3_utils.py 
645
646
647
def verify_blocks(self, device: str | Path, lba: int, count: int) -> CommandResult:
    """Verify blocks on a SCSI device."""
    return self._run_command('sg_verify', f'--lba={lba}', f'--num={count}', str(device))

SgVpd

Bases: Sg3UtilsCommand

SCSI Vital Product Data using sg_vpd.

Source code in sts_libs/src/sts/sg3_utils.py 
494
495
496
497
498
499
500
501
502
503
504
505
506
class SgVpd(Sg3UtilsCommand):
    """SCSI Vital Product Data using sg_vpd."""

    COMMAND: ClassVar[str] = 'sg_vpd'

    def get_vpd_page(self, device: str | Path, page: str) -> CommandResult:
        """Get a specific VPD page from a device.

        Args:
            device: Device path
            page: VPD page identifier (e.g., 'sn', 'di', 'bl')
        """
        return self._run_command(self.COMMAND, f'--page={page}', str(device))

get_vpd_page(device, page)

Get a specific VPD page from a device.

Parameters:

Name Type Description Default
device str | Path

Device path

 required
page str

VPD page identifier (e.g., 'sn', 'di', 'bl')

 required
Source code in sts_libs/src/sts/sg3_utils.py 
499
500
501
502
503
504
505
506
def get_vpd_page(self, device: str | Path, page: str) -> CommandResult:
    """Get a specific VPD page from a device.

    Args:
        device: Device path
        page: VPD page identifier (e.g., 'sn', 'di', 'bl')
    """
    return self._run_command(self.COMMAND, f'--page={page}', str(device))

SgWrite

Bases: Sg3UtilsCommand

SCSI WRITE commands using sg_write_*.

Source code in sts_libs/src/sts/sg3_utils.py 
539
540
541
542
543
544
545
546
class SgWrite(Sg3UtilsCommand):
    """SCSI WRITE commands using sg_write_*."""

    COMMAND: ClassVar[str] = 'sg_write_same'

    def write_same(self, device: str | Path, lba: int, count: int, data_pattern: str = '0x00') -> CommandResult:
        """Write same data pattern to multiple blocks."""
        return self._run_command(self.COMMAND, f'--lba={lba}', f'--num={count}', f'--in={data_pattern}', str(device))

write_same(device, lba, count, data_pattern='0x00')

Write same data pattern to multiple blocks.

Source code in sts_libs/src/sts/sg3_utils.py 
544
545
546
def write_same(self, device: str | Path, lba: int, count: int, data_pattern: str = '0x00') -> CommandResult:
    """Write same data pattern to multiple blocks."""
    return self._run_command(self.COMMAND, f'--lba={lba}', f'--num={count}', f'--in={data_pattern}', str(device))

SgWriteBuffer

Bases: Sg3UtilsCommand

SCSI WRITE BUFFER commands using sg_write_buffer.

Source code in sts_libs/src/sts/sg3_utils.py 
687
688
689
690
class SgWriteBuffer(Sg3UtilsCommand):
    """SCSI WRITE BUFFER commands using sg_write_buffer."""

    COMMAND: ClassVar[str] = 'sg_write_buffer'

Sginfo

Bases: Sg3UtilsCommand

SCSI device information using sginfo.

Source code in sts_libs/src/sts/sg3_utils.py 
711
712
713
714
class Sginfo(Sg3UtilsCommand):
    """SCSI device information using sginfo."""

    COMMAND: ClassVar[str] = 'sginfo'

SgmDd

Bases: Sg3UtilsCommand

SCSI dd utility using sgm_dd (memory-mapped).

Source code in sts_libs/src/sts/sg3_utils.py 
699
700
701
702
class SgmDd(Sg3UtilsCommand):
    """SCSI dd utility using sgm_dd (memory-mapped)."""

    COMMAND: ClassVar[str] = 'sgm_dd'

SgpDd

Bases: Sg3UtilsCommand

SCSI dd utility using sgp_dd (POSIX threads).

Source code in sts_libs/src/sts/sg3_utils.py 
705
706
707
708
class SgpDd(Sg3UtilsCommand):
    """SCSI dd utility using sgp_dd (POSIX threads)."""

    COMMAND: ClassVar[str] = 'sgp_dd'

get_pr_type_name(pr_type)

Get human-readable name for a PR type number.

Source code in sts_libs/src/sts/sg3_utils.py 
51
52
53
def get_pr_type_name(pr_type: int) -> str:
    """Get human-readable name for a PR type number."""
    return PR_TYPE_NAMES.get(pr_type, f'Unknown type ({pr_type})')