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)

Validators:

• _derive_fields

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

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 __str__(self) -> str:
        return f'{self.__class__.__name__}({self.path or "unknown"})'

wait_udev(timeout=UDEV_SETTLE_TIMEOUT)

Wait for udev to finish processing device events.

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

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)

Validators:

• _derive_fields

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

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)

Validators:

• _derive_fields

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

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

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