Dateien nach "custom_components/proxmox_pve" hochladen
This commit is contained in:
647
custom_components/proxmox_pve/sensor.py
Normal file
647
custom_components/proxmox_pve/sensor.py
Normal file
@@ -0,0 +1,647 @@
|
||||
import asyncio
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from homeassistant.components.sensor import SensorEntity
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import PERCENTAGE, UnitOfInformation
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import device_registry as dr, entity_registry as er
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from .const import DOMAIN
|
||||
from .coordinator import ProxmoxGuestCoordinator, ProxmoxNodeCoordinator
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _bytes_to_mb(value: int | float) -> float:
|
||||
return round(float(value) / (1024.0 * 1024.0), 2)
|
||||
|
||||
|
||||
def _bytes_to_gb_3(value: int | float) -> float:
|
||||
return round(float(value) / (1024.0 * 1024.0 * 1024.0), 3)
|
||||
|
||||
|
||||
def _format_uptime(seconds: int) -> str:
|
||||
if seconds < 0:
|
||||
seconds = 0
|
||||
days = seconds // 86400
|
||||
rem = seconds % 86400
|
||||
hours = rem // 3600
|
||||
minutes = (rem % 3600) // 60
|
||||
return f"{days}d {hours}h {minutes:02d}m"
|
||||
|
||||
|
||||
# -----------------------
|
||||
# Node helpers
|
||||
# -----------------------
|
||||
def _node_id(node: str) -> str:
|
||||
return f"node:{node}"
|
||||
|
||||
|
||||
def _node_name(node: str) -> str:
|
||||
return f"Proxmox Node {node}"
|
||||
|
||||
|
||||
async def _get_node_coordinator(hass: HomeAssistant, entry: ConfigEntry, node: str) -> ProxmoxNodeCoordinator:
|
||||
data = hass.data[DOMAIN][entry.entry_id]
|
||||
if node in data["node_coordinators"]:
|
||||
return data["node_coordinators"][node]
|
||||
|
||||
coord = ProxmoxNodeCoordinator(
|
||||
hass=hass,
|
||||
client=data["client"],
|
||||
node=node,
|
||||
scan_interval=int(data["scan_interval"]),
|
||||
)
|
||||
data["node_coordinators"][node] = coord
|
||||
hass.async_create_task(coord.async_config_entry_first_refresh())
|
||||
return coord
|
||||
|
||||
|
||||
# -----------------------
|
||||
# Guest helpers
|
||||
# -----------------------
|
||||
def _guest_display_name(resource: dict) -> str:
|
||||
name = resource.get("name") or f"{resource.get('type')} {resource.get('vmid')}"
|
||||
return f"{name} (VMID {resource.get('vmid')})"
|
||||
|
||||
|
||||
def _guest_id(resource: dict) -> str:
|
||||
return f"{resource.get('node')}:{resource.get('type')}:{resource.get('vmid')}"
|
||||
|
||||
|
||||
def _guest_key(resource: dict) -> tuple[str, str, int]:
|
||||
return (resource["node"], resource["type"], int(resource["vmid"]))
|
||||
|
||||
|
||||
def _model_for(resource: dict) -> str:
|
||||
return "Virtual Machine" if resource.get("type") == "qemu" else "Container"
|
||||
|
||||
|
||||
async def _get_guest_coordinator(hass: HomeAssistant, entry: ConfigEntry, resource: dict) -> ProxmoxGuestCoordinator:
|
||||
data = hass.data[DOMAIN][entry.entry_id]
|
||||
key = _guest_key(resource)
|
||||
|
||||
if key in data["guest_coordinators"]:
|
||||
return data["guest_coordinators"][key]
|
||||
|
||||
coord = ProxmoxGuestCoordinator(
|
||||
hass=hass,
|
||||
client=data["client"],
|
||||
node=key[0],
|
||||
vmtype=key[1],
|
||||
vmid=key[2],
|
||||
scan_interval=int(data["scan_interval"]),
|
||||
ip_mode=str(data["ip_mode"]),
|
||||
ip_prefix=str(data["ip_prefix"]),
|
||||
)
|
||||
data["guest_coordinators"][key] = coord
|
||||
hass.async_create_task(coord.async_config_entry_first_refresh())
|
||||
return coord
|
||||
|
||||
|
||||
def _update_device_name(hass: HomeAssistant, guest_id: str, new_name: str, model: str) -> None:
|
||||
dev_reg = dr.async_get(hass)
|
||||
device = dev_reg.async_get_device(identifiers={(DOMAIN, guest_id)})
|
||||
if device and (device.name != new_name or device.model != model):
|
||||
dev_reg.async_update_device(device.id, name=new_name, model=model)
|
||||
|
||||
|
||||
async def _purge_guest_entity_registry(hass: HomeAssistant, entry: ConfigEntry, guest_id: str) -> None:
|
||||
ent_reg = er.async_get(hass)
|
||||
prefix = f"{entry.entry_id}_{guest_id}_"
|
||||
|
||||
to_remove: list[str] = []
|
||||
for entity_id, ent in ent_reg.entities.items():
|
||||
if ent.config_entry_id != entry.entry_id:
|
||||
continue
|
||||
if ent.unique_id and ent.unique_id.startswith(prefix):
|
||||
to_remove.append(entity_id)
|
||||
|
||||
for entity_id in to_remove:
|
||||
ent_reg.async_remove(entity_id)
|
||||
|
||||
await asyncio.sleep(0)
|
||||
|
||||
|
||||
async def _remove_device(hass: HomeAssistant, device_ident: str) -> None:
|
||||
dev_reg = dr.async_get(hass)
|
||||
device = dev_reg.async_get_device(identifiers={(DOMAIN, device_ident)})
|
||||
if device:
|
||||
dev_reg.async_remove_device(device.id)
|
||||
|
||||
|
||||
# -----------------------
|
||||
# Node Entities
|
||||
# -----------------------
|
||||
class ProxmoxNodeBase(CoordinatorEntity):
|
||||
def __init__(self, coordinator: ProxmoxNodeCoordinator, entry: ConfigEntry, node: str) -> None:
|
||||
super().__init__(coordinator)
|
||||
self._entry = entry
|
||||
self._node = node
|
||||
|
||||
@property
|
||||
def device_info(self):
|
||||
return {
|
||||
"identifiers": {(DOMAIN, _node_id(self._node))},
|
||||
"name": _node_name(self._node),
|
||||
"manufacturer": "Proxmox VE",
|
||||
"model": "Node",
|
||||
}
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self) -> dict[str, Any]:
|
||||
return {"node": self._node}
|
||||
|
||||
|
||||
class ProxmoxNodeCpuSensor(ProxmoxNodeBase, SensorEntity):
|
||||
_attr_icon = "mdi:cpu-64-bit"
|
||||
_attr_native_unit_of_measurement = PERCENTAGE
|
||||
|
||||
def __init__(self, coordinator, entry, node: str) -> None:
|
||||
super().__init__(coordinator, entry, node)
|
||||
self._attr_unique_id = f"{entry.entry_id}_{_node_id(node)}_cpu"
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return f"{_node_name(self._node)} CPU"
|
||||
|
||||
@property
|
||||
def native_value(self) -> float | None:
|
||||
cpu = (self.coordinator.data or {}).get("cpu")
|
||||
return None if cpu is None else round(float(cpu) * 100.0, 2)
|
||||
|
||||
|
||||
class ProxmoxNodeUptimeSensor(ProxmoxNodeBase, SensorEntity):
|
||||
_attr_icon = "mdi:timer-outline"
|
||||
|
||||
def __init__(self, coordinator, entry, node: str) -> None:
|
||||
super().__init__(coordinator, entry, node)
|
||||
self._attr_unique_id = f"{entry.entry_id}_{_node_id(node)}_uptime"
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return f"{_node_name(self._node)} Uptime"
|
||||
|
||||
@property
|
||||
def native_value(self) -> str | None:
|
||||
up = (self.coordinator.data or {}).get("uptime")
|
||||
return None if up is None else _format_uptime(int(up))
|
||||
|
||||
|
||||
class ProxmoxNodeLoad1Sensor(ProxmoxNodeBase, SensorEntity):
|
||||
_attr_icon = "mdi:gauge"
|
||||
|
||||
def __init__(self, coordinator, entry, node: str) -> None:
|
||||
super().__init__(coordinator, entry, node)
|
||||
self._attr_unique_id = f"{entry.entry_id}_{_node_id(node)}_load1"
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return f"{_node_name(self._node)} Load (1m)"
|
||||
|
||||
@property
|
||||
def native_value(self) -> float | None:
|
||||
la = (self.coordinator.data or {}).get("loadavg")
|
||||
if not la:
|
||||
return None
|
||||
try:
|
||||
if isinstance(la, list) and la:
|
||||
return float(la[0])
|
||||
if isinstance(la, str):
|
||||
return float(la.split()[0])
|
||||
except Exception:
|
||||
return None
|
||||
return None
|
||||
|
||||
|
||||
# ---- RAM (MB)
|
||||
class ProxmoxNodeRamUsedSensor(ProxmoxNodeBase, SensorEntity):
|
||||
_attr_icon = "mdi:memory"
|
||||
_attr_native_unit_of_measurement = UnitOfInformation.MEGABYTES
|
||||
|
||||
def __init__(self, coordinator, entry, node: str) -> None:
|
||||
super().__init__(coordinator, entry, node)
|
||||
self._attr_unique_id = f"{entry.entry_id}_{_node_id(node)}_ram_used_mb"
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return f"{_node_name(self._node)} RAM Used"
|
||||
|
||||
@property
|
||||
def native_value(self) -> float | None:
|
||||
mem = (self.coordinator.data or {}).get("memory", {}).get("used")
|
||||
return None if mem is None else _bytes_to_mb(int(mem))
|
||||
|
||||
|
||||
class ProxmoxNodeRamTotalSensor(ProxmoxNodeBase, SensorEntity):
|
||||
_attr_icon = "mdi:memory"
|
||||
_attr_native_unit_of_measurement = UnitOfInformation.MEGABYTES
|
||||
|
||||
def __init__(self, coordinator, entry, node: str) -> None:
|
||||
super().__init__(coordinator, entry, node)
|
||||
self._attr_unique_id = f"{entry.entry_id}_{_node_id(node)}_ram_total_mb"
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return f"{_node_name(self._node)} RAM Total"
|
||||
|
||||
@property
|
||||
def native_value(self) -> float | None:
|
||||
total = (self.coordinator.data or {}).get("memory", {}).get("total")
|
||||
return None if total is None else _bytes_to_mb(int(total))
|
||||
|
||||
|
||||
class ProxmoxNodeRamFreeSensor(ProxmoxNodeBase, SensorEntity):
|
||||
_attr_icon = "mdi:memory"
|
||||
_attr_native_unit_of_measurement = UnitOfInformation.MEGABYTES
|
||||
|
||||
def __init__(self, coordinator, entry, node: str) -> None:
|
||||
super().__init__(coordinator, entry, node)
|
||||
self._attr_unique_id = f"{entry.entry_id}_{_node_id(node)}_ram_free_mb"
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return f"{_node_name(self._node)} RAM Free"
|
||||
|
||||
@property
|
||||
def native_value(self) -> float | None:
|
||||
free = (self.coordinator.data or {}).get("memory", {}).get("free")
|
||||
return None if free is None else _bytes_to_mb(int(free))
|
||||
|
||||
|
||||
# ---- Swap (MB)
|
||||
class ProxmoxNodeSwapUsedSensor(ProxmoxNodeBase, SensorEntity):
|
||||
_attr_icon = "mdi:swap-horizontal"
|
||||
_attr_native_unit_of_measurement = UnitOfInformation.MEGABYTES
|
||||
|
||||
def __init__(self, coordinator, entry, node: str) -> None:
|
||||
super().__init__(coordinator, entry, node)
|
||||
self._attr_unique_id = f"{entry.entry_id}_{_node_id(node)}_swap_used_mb"
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return f"{_node_name(self._node)} Swap Used"
|
||||
|
||||
@property
|
||||
def native_value(self) -> float | None:
|
||||
used = (self.coordinator.data or {}).get("swap", {}).get("used")
|
||||
return None if used is None else _bytes_to_mb(int(used))
|
||||
|
||||
|
||||
class ProxmoxNodeSwapTotalSensor(ProxmoxNodeBase, SensorEntity):
|
||||
_attr_icon = "mdi:swap-horizontal"
|
||||
_attr_native_unit_of_measurement = UnitOfInformation.MEGABYTES
|
||||
|
||||
def __init__(self, coordinator, entry, node: str) -> None:
|
||||
super().__init__(coordinator, entry, node)
|
||||
self._attr_unique_id = f"{entry.entry_id}_{_node_id(node)}_swap_total_mb"
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return f"{_node_name(self._node)} Swap Total"
|
||||
|
||||
@property
|
||||
def native_value(self) -> float | None:
|
||||
total = (self.coordinator.data or {}).get("swap", {}).get("total")
|
||||
return None if total is None else _bytes_to_mb(int(total))
|
||||
|
||||
|
||||
class ProxmoxNodeSwapFreeSensor(ProxmoxNodeBase, SensorEntity):
|
||||
_attr_icon = "mdi:swap-horizontal"
|
||||
_attr_native_unit_of_measurement = UnitOfInformation.MEGABYTES
|
||||
|
||||
def __init__(self, coordinator, entry, node: str) -> None:
|
||||
super().__init__(coordinator, entry, node)
|
||||
self._attr_unique_id = f"{entry.entry_id}_{_node_id(node)}_swap_free_mb"
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return f"{_node_name(self._node)} Swap Free"
|
||||
|
||||
@property
|
||||
def native_value(self) -> float | None:
|
||||
free = (self.coordinator.data or {}).get("swap", {}).get("free")
|
||||
return None if free is None else _bytes_to_mb(int(free))
|
||||
|
||||
|
||||
# ---- RootFS / Node Storage (GB, 3 decimals)
|
||||
class ProxmoxNodeStorageUsedSensor(ProxmoxNodeBase, SensorEntity):
|
||||
_attr_icon = "mdi:harddisk"
|
||||
_attr_native_unit_of_measurement = "GB"
|
||||
|
||||
def __init__(self, coordinator, entry, node: str) -> None:
|
||||
super().__init__(coordinator, entry, node)
|
||||
self._attr_unique_id = f"{entry.entry_id}_{_node_id(node)}_storage_used_gb"
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return f"{_node_name(self._node)} Storage Used"
|
||||
|
||||
@property
|
||||
def native_value(self) -> float | None:
|
||||
used = (self.coordinator.data or {}).get("rootfs", {}).get("used")
|
||||
return None if used is None else _bytes_to_gb_3(int(used))
|
||||
|
||||
|
||||
class ProxmoxNodeStorageTotalSensor(ProxmoxNodeBase, SensorEntity):
|
||||
_attr_icon = "mdi:harddisk"
|
||||
_attr_native_unit_of_measurement = "GB"
|
||||
|
||||
def __init__(self, coordinator, entry, node: str) -> None:
|
||||
super().__init__(coordinator, entry, node)
|
||||
self._attr_unique_id = f"{entry.entry_id}_{_node_id(node)}_storage_total_gb"
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return f"{_node_name(self._node)} Storage Total"
|
||||
|
||||
@property
|
||||
def native_value(self) -> float | None:
|
||||
total = (self.coordinator.data or {}).get("rootfs", {}).get("total")
|
||||
return None if total is None else _bytes_to_gb_3(int(total))
|
||||
|
||||
|
||||
class ProxmoxNodeStorageFreeSensor(ProxmoxNodeBase, SensorEntity):
|
||||
_attr_icon = "mdi:harddisk"
|
||||
_attr_native_unit_of_measurement = "GB"
|
||||
|
||||
def __init__(self, coordinator, entry, node: str) -> None:
|
||||
super().__init__(coordinator, entry, node)
|
||||
self._attr_unique_id = f"{entry.entry_id}_{_node_id(node)}_storage_free_gb"
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return f"{_node_name(self._node)} Storage Free"
|
||||
|
||||
@property
|
||||
def native_value(self) -> float | None:
|
||||
free = (self.coordinator.data or {}).get("rootfs", {}).get("free")
|
||||
return None if free is None else _bytes_to_gb_3(int(free))
|
||||
|
||||
|
||||
# -----------------------
|
||||
# Guest Entities
|
||||
# -----------------------
|
||||
class ProxmoxBaseGuestEntity(CoordinatorEntity):
|
||||
def __init__(self, coordinator, entry: ConfigEntry, resource: dict) -> None:
|
||||
super().__init__(coordinator)
|
||||
self._entry = entry
|
||||
self._resource = dict(resource)
|
||||
|
||||
def update_resource(self, resource: dict) -> None:
|
||||
self._resource = dict(resource)
|
||||
|
||||
@property
|
||||
def device_info(self):
|
||||
node = self._resource.get("node")
|
||||
via = (DOMAIN, _node_id(node)) if node else None
|
||||
|
||||
info = {
|
||||
"identifiers": {(DOMAIN, _guest_id(self._resource))},
|
||||
"name": _guest_display_name(self._resource),
|
||||
"manufacturer": "Proxmox VE",
|
||||
"model": _model_for(self._resource),
|
||||
}
|
||||
if via:
|
||||
info["via_device"] = via
|
||||
return info
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self) -> dict[str, Any]:
|
||||
return {"vmid": self._resource.get("vmid"), "node": self._resource.get("node"), "type": self._resource.get("type")}
|
||||
|
||||
|
||||
class ProxmoxGuestStatusSensor(ProxmoxBaseGuestEntity, SensorEntity):
|
||||
_attr_icon = "mdi:power"
|
||||
|
||||
def __init__(self, coordinator, entry: ConfigEntry, resource: dict) -> None:
|
||||
super().__init__(coordinator, entry, resource)
|
||||
self._attr_unique_id = f"{entry.entry_id}_{_guest_id(resource)}_status"
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return f"{_guest_display_name(self._resource)} Status"
|
||||
|
||||
@property
|
||||
def native_value(self) -> str | None:
|
||||
return (self.coordinator.data or {}).get("status")
|
||||
|
||||
|
||||
class ProxmoxGuestCpuSensor(ProxmoxBaseGuestEntity, SensorEntity):
|
||||
_attr_icon = "mdi:cpu-64-bit"
|
||||
_attr_native_unit_of_measurement = PERCENTAGE
|
||||
|
||||
def __init__(self, coordinator, entry: ConfigEntry, resource: dict) -> None:
|
||||
super().__init__(coordinator, entry, resource)
|
||||
self._attr_unique_id = f"{entry.entry_id}_{_guest_id(resource)}_cpu"
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return f"{_guest_display_name(self._resource)} CPU"
|
||||
|
||||
@property
|
||||
def native_value(self) -> float | None:
|
||||
cpu = (self.coordinator.data or {}).get("cpu")
|
||||
return None if cpu is None else round(float(cpu) * 100.0, 2)
|
||||
|
||||
|
||||
class ProxmoxGuestRamUsedMB(ProxmoxBaseGuestEntity, SensorEntity):
|
||||
_attr_icon = "mdi:memory"
|
||||
_attr_native_unit_of_measurement = UnitOfInformation.MEGABYTES
|
||||
|
||||
def __init__(self, coordinator, entry: ConfigEntry, resource: dict) -> None:
|
||||
super().__init__(coordinator, entry, resource)
|
||||
self._attr_unique_id = f"{entry.entry_id}_{_guest_id(resource)}_ram_used_mb"
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return f"{_guest_display_name(self._resource)} RAM Used"
|
||||
|
||||
@property
|
||||
def native_value(self) -> float | None:
|
||||
mem = (self.coordinator.data or {}).get("mem")
|
||||
return None if mem is None else _bytes_to_mb(int(mem))
|
||||
|
||||
|
||||
class ProxmoxGuestUptimePretty(ProxmoxBaseGuestEntity, SensorEntity):
|
||||
_attr_icon = "mdi:timer-outline"
|
||||
|
||||
def __init__(self, coordinator, entry: ConfigEntry, resource: dict) -> None:
|
||||
super().__init__(coordinator, entry, resource)
|
||||
self._attr_unique_id = f"{entry.entry_id}_{_guest_id(resource)}_uptime_pretty"
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return f"{_guest_display_name(self._resource)} Uptime"
|
||||
|
||||
@property
|
||||
def native_value(self) -> str | None:
|
||||
uptime = (self.coordinator.data or {}).get("uptime")
|
||||
return None if uptime is None else _format_uptime(int(uptime))
|
||||
|
||||
|
||||
class ProxmoxGuestNetInMB(ProxmoxBaseGuestEntity, SensorEntity):
|
||||
_attr_icon = "mdi:download-network"
|
||||
_attr_native_unit_of_measurement = UnitOfInformation.MEGABYTES
|
||||
|
||||
def __init__(self, coordinator, entry: ConfigEntry, resource: dict) -> None:
|
||||
super().__init__(coordinator, entry, resource)
|
||||
self._attr_unique_id = f"{entry.entry_id}_{_guest_id(resource)}_netin_mb"
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return f"{_guest_display_name(self._resource)} Network In"
|
||||
|
||||
@property
|
||||
def native_value(self) -> float | None:
|
||||
v = (self.coordinator.data or {}).get("netin")
|
||||
return None if v is None else _bytes_to_mb(int(v))
|
||||
|
||||
|
||||
class ProxmoxGuestNetOutMB(ProxmoxBaseGuestEntity, SensorEntity):
|
||||
_attr_icon = "mdi:upload-network"
|
||||
_attr_native_unit_of_measurement = UnitOfInformation.MEGABYTES
|
||||
|
||||
def __init__(self, coordinator, entry: ConfigEntry, resource: dict) -> None:
|
||||
super().__init__(coordinator, entry, resource)
|
||||
self._attr_unique_id = f"{entry.entry_id}_{_guest_id(resource)}_netout_mb"
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return f"{_guest_display_name(self._resource)} Network Out"
|
||||
|
||||
@property
|
||||
def native_value(self) -> float | None:
|
||||
v = (self.coordinator.data or {}).get("netout")
|
||||
return None if v is None else _bytes_to_mb(int(v))
|
||||
|
||||
|
||||
class ProxmoxGuestPreferredIP(ProxmoxBaseGuestEntity, SensorEntity):
|
||||
_attr_icon = "mdi:ip-network"
|
||||
|
||||
def __init__(self, coordinator, entry: ConfigEntry, resource: dict) -> None:
|
||||
super().__init__(coordinator, entry, resource)
|
||||
self._attr_unique_id = f"{entry.entry_id}_{_guest_id(resource)}_ip_preferred"
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return f"{_guest_display_name(self._resource)} IP"
|
||||
|
||||
@property
|
||||
def native_value(self) -> str | None:
|
||||
return (self.coordinator.data or {}).get("_preferred_ip")
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self) -> dict[str, Any]:
|
||||
attrs = super().extra_state_attributes
|
||||
attrs["ip_addresses"] = (self.coordinator.data or {}).get("_ip_addresses", [])
|
||||
return attrs
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback) -> None:
|
||||
data = hass.data[DOMAIN][entry.entry_id]
|
||||
resources_coord = data["resources"]
|
||||
nodes_coord = data["nodes"]
|
||||
|
||||
platform_cache = data.setdefault("platform_cache", {})
|
||||
guest_cache: dict[tuple[str, str, int], list[SensorEntity]] = platform_cache.setdefault("sensor_guest", {})
|
||||
node_cache: dict[str, list[SensorEntity]] = platform_cache.setdefault("sensor_node", {})
|
||||
|
||||
async def _sync_nodes() -> None:
|
||||
nodes = nodes_coord.data or []
|
||||
current_nodes = {n.get("node") for n in nodes if n.get("node")}
|
||||
|
||||
new_entities: list[SensorEntity] = []
|
||||
for node in sorted(current_nodes):
|
||||
if node in node_cache:
|
||||
continue
|
||||
|
||||
node_c = await _get_node_coordinator(hass, entry, node)
|
||||
ents = [
|
||||
ProxmoxNodeCpuSensor(node_c, entry, node),
|
||||
ProxmoxNodeLoad1Sensor(node_c, entry, node),
|
||||
ProxmoxNodeRamUsedSensor(node_c, entry, node),
|
||||
ProxmoxNodeRamTotalSensor(node_c, entry, node),
|
||||
ProxmoxNodeRamFreeSensor(node_c, entry, node),
|
||||
ProxmoxNodeSwapUsedSensor(node_c, entry, node),
|
||||
ProxmoxNodeSwapTotalSensor(node_c, entry, node),
|
||||
ProxmoxNodeSwapFreeSensor(node_c, entry, node),
|
||||
ProxmoxNodeStorageUsedSensor(node_c, entry, node),
|
||||
ProxmoxNodeStorageTotalSensor(node_c, entry, node),
|
||||
ProxmoxNodeStorageFreeSensor(node_c, entry, node),
|
||||
ProxmoxNodeUptimeSensor(node_c, entry, node),
|
||||
]
|
||||
node_cache[node] = ents
|
||||
new_entities.extend(ents)
|
||||
|
||||
if new_entities:
|
||||
async_add_entities(new_entities, update_before_add=False)
|
||||
|
||||
removed = [n for n in list(node_cache.keys()) if n not in current_nodes]
|
||||
for n in removed:
|
||||
for ent in node_cache[n]:
|
||||
await ent.async_remove()
|
||||
del node_cache[n]
|
||||
await _remove_device(hass, _node_id(n))
|
||||
|
||||
async def _sync_guests() -> None:
|
||||
resources = resources_coord.data or []
|
||||
current: dict[tuple[str, str, int], dict] = {}
|
||||
|
||||
for r in resources:
|
||||
if r.get("type") not in ("qemu", "lxc"):
|
||||
continue
|
||||
if r.get("node") is None or r.get("vmid") is None:
|
||||
continue
|
||||
current[_guest_key(r)] = r
|
||||
|
||||
for key, r in current.items():
|
||||
if key not in guest_cache:
|
||||
continue
|
||||
gid = _guest_id(r)
|
||||
_update_device_name(hass, gid, _guest_display_name(r), _model_for(r))
|
||||
for ent in guest_cache[key]:
|
||||
ent.update_resource(r)
|
||||
ent.async_write_ha_state()
|
||||
|
||||
new_entities: list[SensorEntity] = []
|
||||
for key, r in current.items():
|
||||
if key in guest_cache:
|
||||
continue
|
||||
guest_coord = await _get_guest_coordinator(hass, entry, r)
|
||||
ents = [
|
||||
ProxmoxGuestStatusSensor(guest_coord, entry, r),
|
||||
ProxmoxGuestCpuSensor(guest_coord, entry, r),
|
||||
ProxmoxGuestRamUsedMB(guest_coord, entry, r),
|
||||
ProxmoxGuestUptimePretty(guest_coord, entry, r),
|
||||
ProxmoxGuestNetInMB(guest_coord, entry, r),
|
||||
ProxmoxGuestNetOutMB(guest_coord, entry, r),
|
||||
ProxmoxGuestPreferredIP(guest_coord, entry, r),
|
||||
]
|
||||
guest_cache[key] = ents
|
||||
new_entities.extend(ents)
|
||||
|
||||
if new_entities:
|
||||
async_add_entities(new_entities, update_before_add=False)
|
||||
|
||||
removed = [k for k in list(guest_cache.keys()) if k not in current]
|
||||
for k in removed:
|
||||
gid = f"{k[0]}:{k[1]}:{k[2]}"
|
||||
for ent in guest_cache[k]:
|
||||
await ent.async_remove()
|
||||
del guest_cache[k]
|
||||
|
||||
data["guest_coordinators"].pop(k, None)
|
||||
await _purge_guest_entity_registry(hass, entry, gid)
|
||||
await _remove_device(hass, gid)
|
||||
|
||||
await _sync_nodes()
|
||||
await _sync_guests()
|
||||
|
||||
unsub_nodes = nodes_coord.async_add_listener(lambda: hass.async_create_task(_sync_nodes()))
|
||||
unsub_guests = resources_coord.async_add_listener(lambda: hass.async_create_task(_sync_guests()))
|
||||
platform_cache.setdefault("sensor_unsub", []).extend([unsub_nodes, unsub_guests])
|
||||
Reference in New Issue
Block a user