Mechanical: ruff format, and ruff check --fix with the rule set every NAPALM driver uses (E, F, I, UP, B at line length 100, Python 3.10). Mapping and Iterable come from collections.abc, imports are sorted. What ruff cannot fix comes in the next commit.
136 lines
4.9 KiB
Python
136 lines
4.9 KiB
Python
"""Hardware of the managed server: Processors, Memory, Storage, Drives, SimpleStorage."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
from typing import Any
|
|
|
|
from .common import boolean, ident, integer, link, link_list, location_label, present, status, text
|
|
|
|
|
|
def parse_processor(cpu: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""One member of ``Systems/{id}/Processors``."""
|
|
state, health = status(cpu)
|
|
return {
|
|
"id": text(cpu.get("Id")),
|
|
"name": text(cpu.get("Name")),
|
|
"socket": text(cpu.get("Socket")) or location_label(cpu),
|
|
"processor_type": text(cpu.get("ProcessorType")),
|
|
"manufacturer": ident(cpu.get("Manufacturer")),
|
|
"model": ident(cpu.get("Model")),
|
|
"architecture": text(cpu.get("ProcessorArchitecture")),
|
|
"instruction_set": text(cpu.get("InstructionSet")),
|
|
"max_speed_mhz": integer(cpu.get("MaxSpeedMHz")),
|
|
"cores": integer(cpu.get("TotalCores")),
|
|
"threads": integer(cpu.get("TotalThreads")),
|
|
"serial_number": ident(cpu.get("SerialNumber")),
|
|
"present": present(cpu),
|
|
"state": state,
|
|
"health": health,
|
|
}
|
|
|
|
|
|
def parse_memory(dimm: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""One member of ``Systems/{id}/Memory``.
|
|
|
|
An enabled slot with a capacity of 0 counts as empty: some BMCs list empty slots so.
|
|
"""
|
|
state, health = status(dimm)
|
|
capacity = integer(dimm.get("CapacityMiB"))
|
|
return {
|
|
"id": text(dimm.get("Id")),
|
|
"name": text(dimm.get("Name")),
|
|
"location": (
|
|
location_label(dimm) or text(dimm.get("DeviceLocator")) or text(dimm.get("Name"))
|
|
),
|
|
"capacity_mib": capacity,
|
|
"memory_device_type": text(dimm.get("MemoryDeviceType")),
|
|
"base_module_type": text(dimm.get("BaseModuleType")),
|
|
"speed_mhz": integer(dimm.get("OperatingSpeedMhz")),
|
|
"manufacturer": ident(dimm.get("Manufacturer")),
|
|
"part_number": ident(dimm.get("PartNumber")),
|
|
"serial_number": ident(dimm.get("SerialNumber")),
|
|
"present": present(dimm) and capacity != 0,
|
|
"state": state,
|
|
"health": health,
|
|
}
|
|
|
|
|
|
def _controller(ctrl: Mapping[str, Any]) -> dict[str, Any]:
|
|
state, health = status(ctrl)
|
|
return {
|
|
"name": text(ctrl.get("Name")),
|
|
"manufacturer": ident(ctrl.get("Manufacturer")),
|
|
"model": ident(ctrl.get("Model")),
|
|
"serial_number": ident(ctrl.get("SerialNumber")),
|
|
"firmware_version": text(ctrl.get("FirmwareVersion")),
|
|
"speed_gbps": ctrl.get("SpeedGbps")
|
|
if isinstance(ctrl.get("SpeedGbps"), (int, float))
|
|
else None,
|
|
"state": state,
|
|
"health": health,
|
|
}
|
|
|
|
|
|
def parse_storage(storage: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""One member of ``Systems/{id}/Storage``.
|
|
|
|
Drives are separate resources: ``drive_paths`` lists them for ``parse_drive``. Storage
|
|
1.15 and later move controllers into a collection at ``controllers_path``.
|
|
"""
|
|
state, health = status(storage)
|
|
controllers = storage.get("StorageControllers")
|
|
return {
|
|
"id": text(storage.get("Id")),
|
|
"name": text(storage.get("Name")),
|
|
"state": state,
|
|
"health": health,
|
|
"controllers": [_controller(c) for c in controllers if isinstance(c, Mapping)]
|
|
if isinstance(controllers, list)
|
|
else [],
|
|
"controllers_path": link(storage, "Controllers"),
|
|
"drive_paths": link_list(storage, "Drives"),
|
|
}
|
|
|
|
|
|
def parse_drive(drive: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""A Drive resource, as linked from Storage."""
|
|
state, health = status(drive)
|
|
return {
|
|
"id": text(drive.get("Id")),
|
|
"name": text(drive.get("Name")),
|
|
"manufacturer": ident(drive.get("Manufacturer")),
|
|
"model": ident(drive.get("Model")),
|
|
"serial_number": ident(drive.get("SerialNumber")),
|
|
"part_number": ident(drive.get("PartNumber")),
|
|
"revision": text(drive.get("Revision")),
|
|
"capacity_bytes": integer(drive.get("CapacityBytes")),
|
|
"media_type": text(drive.get("MediaType")),
|
|
"protocol": text(drive.get("Protocol")),
|
|
"failure_predicted": boolean(drive.get("FailurePredicted")),
|
|
"present": present(drive),
|
|
"state": state,
|
|
"health": health,
|
|
}
|
|
|
|
|
|
def parse_simple_storage(storage: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""One member of ``Systems/{id}/SimpleStorage`` (older services, no Storage).
|
|
|
|
Its devices come out with the same keys as ``parse_drive``.
|
|
"""
|
|
state, health = status(storage)
|
|
devices = storage.get("Devices")
|
|
drives = (
|
|
[parse_drive(dev) for dev in devices if isinstance(dev, Mapping)]
|
|
if isinstance(devices, list)
|
|
else []
|
|
)
|
|
return {
|
|
"id": text(storage.get("Id")),
|
|
"name": text(storage.get("Name")),
|
|
"state": state,
|
|
"health": health,
|
|
"drives": drives,
|
|
}
|