Skip to content

Base Classes

Pydantic v2 base classes for all device types. Device provides path/name handling with lazy validation — construction is cheap and side-effect-free, runtime state is resolved later by discover().

Device Base Classes

sts.base

Base device classes: Device, NetworkDevice, StorageDevice.

Device pydantic-model

Bases: StsBaseModel

Base class for all devices.

Construction is intentionally cheap: Device(path='/dev/nonexistent') succeeds without checking whether the path exists on disk. Existence and other runtime properties are resolved later by discover() (or the subsystem-specific equivalent). This lazy-validation design keeps model construction fast and side-effect-free, which matters for test fixtures and for building device objects from report data before the device is ready.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Base class for all devices.\n\nConstruction is intentionally cheap: ``Device(path='/dev/nonexistent')``\nsucceeds without checking whether the path exists on disk.  Existence and\nother runtime properties are resolved later by ``discover()`` (or the\nsubsystem-specific equivalent).  This lazy-validation design keeps model\nconstruction fast and side-effect-free, which matters for test fixtures and\nfor building device objects from report data before the device is ready.",
  "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"
    }
  },
  "title": "Device",
  "type": "object"
}

Fields:

  • path (PathOrStr | None)
  • name (str | None)
Source code in sts_libs/src/sts/base.py 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
class Device(StsBaseModel):
    """Base class for all devices.

    Construction is intentionally cheap: ``Device(path='/dev/nonexistent')``
    succeeds without checking whether the path exists on disk.  Existence and
    other runtime properties are resolved later by ``discover()`` (or the
    subsystem-specific equivalent).  This lazy-validation design keeps model
    construction fast and side-effect-free, which matters for test fixtures and
    for building device objects from report data before the device is ready.
    """

    path: PathOrStr | None = None
    name: str | None = None

    # Standard Linux device paths
    DEV_PATH: ClassVar[Path] = Path('/dev')  # Device nodes (e.g. /dev/sda)
    SYS_PATH: ClassVar[Path] = Path('/sys')  # Sysfs device info (e.g. /sys/block/sda)

    @model_validator(mode='after')
    def _derive_fields(self) -> Self:
        """Derive name from path when only path is provided."""
        if self.path and not self.name:
            self.name = Path(self.path).name
        return self

    def wait_udev(self, timeout: int = UDEV_SETTLE_TIMEOUT) -> bool:
        """Wait for udev to finish processing device events."""
        # First try udevadm settle - this is faster but may fail if udev is busy
        result = run('udevadm settle')
        if result.succeeded:
            return True

        # If settle failed, poll for device existence
        # This is slower but more reliable, especially for slow devices
        if not self.path:
            return False

        start_time = time.time()
        while time.time() - start_time < timeout:
            if Path(self.path).exists():
                return True
            time.sleep(0.1)  # Small sleep to avoid busy waiting

        logger.warning(f'Timeout waiting for udev to settle on {self.path}')
        return False

    def wait_for_path(
        self,
        *,
        present: bool = True,
        path: PathOrStr | None = None,
        timeout: float = UDEV_SETTLE_TIMEOUT,
    ) -> bool:
        """Wait until a path reaches the requested existence state.

        Args:
            present: Whether the path is expected to exist.
            path: Explicit path to poll instead of this device's node path.
            timeout: Maximum time to wait in seconds.
        """
        target = Path(path) if path is not None else (Path(self.path) if self.path else None)
        if target is None:
            return False

        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            if target.exists() is present:
                return True
            time.sleep(0.1)

        if target.exists() is present:
            return True
        logger.warning(f'Timeout waiting for path to be {"present" if present else "absent"}: {target}')
        return False

    def __str__(self) -> str:
        return f'{self.__class__.__name__}({self.path or "unknown"})'

wait_for_path(*, present=True, path=None, timeout=UDEV_SETTLE_TIMEOUT)

Wait until a path reaches the requested existence state.

Parameters:

Name Type Description Default
present bool

Whether the path is expected to exist.

 True
path PathOrStr | None

Explicit path to poll instead of this device's node path.

 None
timeout float

Maximum time to wait in seconds.

 UDEV_SETTLE_TIMEOUT
Source code in sts_libs/src/sts/base.py 
 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
def wait_for_path(
    self,
    *,
    present: bool = True,
    path: PathOrStr | None = None,
    timeout: float = UDEV_SETTLE_TIMEOUT,
) -> bool:
    """Wait until a path reaches the requested existence state.

    Args:
        present: Whether the path is expected to exist.
        path: Explicit path to poll instead of this device's node path.
        timeout: Maximum time to wait in seconds.
    """
    target = Path(path) if path is not None else (Path(self.path) if self.path else None)
    if target is None:
        return False

    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if target.exists() is present:
            return True
        time.sleep(0.1)

    if target.exists() is present:
        return True
    logger.warning(f'Timeout waiting for path to be {"present" if present else "absent"}: {target}')
    return False

wait_udev(timeout=UDEV_SETTLE_TIMEOUT)

Wait for udev to finish processing device events.

Source code in sts_libs/src/sts/base.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def wait_udev(self, timeout: int = UDEV_SETTLE_TIMEOUT) -> bool:
    """Wait for udev to finish processing device events."""
    # First try udevadm settle - this is faster but may fail if udev is busy
    result = run('udevadm settle')
    if result.succeeded:
        return True

    # If settle failed, poll for device existence
    # This is slower but more reliable, especially for slow devices
    if not self.path:
        return False

    start_time = time.time()
    while time.time() - start_time < timeout:
        if Path(self.path).exists():
            return True
        time.sleep(0.1)  # Small sleep to avoid busy waiting

    logger.warning(f'Timeout waiting for udev to settle on {self.path}')
    return False

NetworkDevice pydantic-model

Bases: Device

Network-capable device.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Network-capable device.",
  "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"
    },
    "ip": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Ip"
    },
    "port": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Port"
    }
  },
  "title": "NetworkDevice",
  "type": "object"
}

Fields:

  • path (PathOrStr | None)
  • name (str | None)
  • ip (str | None)
  • port (int | None)
Source code in sts_libs/src/sts/base.py 
108
109
110
111
112
113
114
115
class NetworkDevice(Device):
    """Network-capable device."""

    ip: str | None = None
    port: int | None = None

    # Network devices are managed through sysfs
    NET_PATH: ClassVar[Path] = Path('/sys/class/net')

StorageDevice pydantic-model

Bases: Device

Storage device with size and model information.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Storage device with size and model information.",
  "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": [
        {
          "minimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "model": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Model"
    }
  },
  "title": "StorageDevice",
  "type": "object"
}

Fields:

  • path (PathOrStr | None)
  • name (str | None)
  • size (int | None)
  • model (str | None)
Source code in sts_libs/src/sts/base.py 
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
class StorageDevice(Device):
    """Storage device with size and model information."""

    size: int | None = Field(default=None, ge=0)
    model: str | None = None

    # Block devices are managed through sysfs block subsystem
    BLOCK_PATH: ClassVar[Path] = Path('/sys/block')

    @property
    def size_human(self) -> str:
        """Human-readable size (e.g. '1.0 TB'), or 'Unknown' if unset."""
        if self.size is None:
            return 'Unknown'

        # Convert bytes to human readable format using binary prefixes
        # (1024-based: KiB, MiB, etc. but displayed as KB, MB for simplicity)
        size = float(self.size)
        for unit in ('B', 'KB', 'MB', 'GB', 'TB', 'PB'):
            if size < BYTES_PER_UNIT:
                return f'{size:.1f} {unit}'
            size /= BYTES_PER_UNIT
        return f'{size:.1f} EB'

    def check_sector_zero(self) -> bool:
        """Check that the first 512 bytes of the device can be read."""
        if not self.path:
            logger.warning('Cannot check sector zero: path is not set')
            return False
        try:
            with Path(self.path).open('rb') as f:
                # Read the first 512 bytes, which represent sector zero
                sector_zero = f.read(512)
            # Ensure the read data is exactly 512 bytes in size
            if len(sector_zero) < 512:
                return False
        except OSError:
            logger.warning(f'No such device or address: {self.path}')
            return False
        else:
            return True

size_human property

Human-readable size (e.g. '1.0 TB'), or 'Unknown' if unset.

check_sector_zero()

Check that the first 512 bytes of the device can be read.

Source code in sts_libs/src/sts/base.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def check_sector_zero(self) -> bool:
    """Check that the first 512 bytes of the device can be read."""
    if not self.path:
        logger.warning('Cannot check sector zero: path is not set')
        return False
    try:
        with Path(self.path).open('rb') as f:
            # Read the first 512 bytes, which represent sector zero
            sector_zero = f.read(512)
        # Ensure the read data is exactly 512 bytes in size
        if len(sector_zero) < 512:
            return False
    except OSError:
        logger.warning(f'No such device or address: {self.path}')
        return False
    else:
        return True