Add coordinator to QNAP (#94413)
* Create coordinator.py * Update sensor.py * Update sensor.py * Update sensor.py * Update homeassistant/components/qnap/coordinator.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update homeassistant/components/qnap/coordinator.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update homeassistant/components/qnap/coordinator.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update homeassistant/components/qnap/coordinator.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update homeassistant/components/qnap/coordinator.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update homeassistant/components/qnap/coordinator.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update homeassistant/components/qnap/sensor.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update homeassistant/components/qnap/sensor.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update homeassistant/components/qnap/sensor.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update homeassistant/components/qnap/sensor.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update homeassistant/components/qnap/sensor.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Add import * Update coordinator.py * Update coordinator.py * Add platformnotready * Walrus * Update sensor.py * Undo walres CI didnt like it * Update coordinator.py black fix * Update sensor.py fix black * Update sensor.py fix ruff * Update coordinator.py fix lint * Update sensor.py fix ruff * Update homeassistant/components/qnap/coordinator.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update homeassistant/components/qnap/sensor.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update sensor.py * Update coordinator.py * Update coordinator.py * Update sensor.py * Update homeassistant/components/qnap/sensor.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update sensor.py --------- Co-authored-by: epenet <6771947+epenet@users.noreply.github.com>pull/94615/head
disforw
GitHub
parent
b149fffa08
commit
a79e37c240
|
|
@ -0,0 +1,59 @@
|
|||
"""Data coordinator for the qnap integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from qnapstats import QNAPStats
|
||||
|
||||
from homeassistant.const import (
|
||||
CONF_HOST,
|
||||
CONF_PASSWORD,
|
||||
CONF_PORT,
|
||||
CONF_SSL,
|
||||
CONF_TIMEOUT,
|
||||
CONF_USERNAME,
|
||||
CONF_VERIFY_SSL,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
|
||||
|
||||
from .const import DOMAIN
|
||||
|
||||
UPDATE_INTERVAL = timedelta(minutes=1)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class QnapCoordinator(DataUpdateCoordinator[dict[str, dict[str, Any]]]):
|
||||
"""Custom coordinator for the qnap integration."""
|
||||
|
||||
def __init__(self, hass: HomeAssistant, config: ConfigType) -> None:
|
||||
"""Initialize the qnap coordinator."""
|
||||
super().__init__(hass, _LOGGER, name=DOMAIN, update_interval=UPDATE_INTERVAL)
|
||||
|
||||
protocol = "https" if config[CONF_SSL] else "http"
|
||||
self._api = QNAPStats(
|
||||
f"{protocol}://{config.get(CONF_HOST)}",
|
||||
config.get(CONF_PORT),
|
||||
config.get(CONF_USERNAME),
|
||||
config.get(CONF_PASSWORD),
|
||||
verify_ssl=config.get(CONF_VERIFY_SSL),
|
||||
timeout=config.get(CONF_TIMEOUT),
|
||||
)
|
||||
|
||||
def _sync_update(self) -> dict[str, dict[str, Any]]:
|
||||
"""Get the latest data from the Qnap API."""
|
||||
return {
|
||||
"system_stats": self._api.get_system_stats(),
|
||||
"system_health": self._api.get_system_health(),
|
||||
"smart_drive_health": self._api.get_smart_disk_health(),
|
||||
"volumes": self._api.get_volumes(),
|
||||
"bandwidth": self._api.get_bandwidth(),
|
||||
}
|
||||
|
||||
async def _async_update_data(self) -> dict[str, dict[str, Any]]:
|
||||
"""Get the latest data from the Qnap API."""
|
||||
return await self.hass.async_add_executor_job(self._sync_update)
|
||||
|
|
@ -1,10 +1,6 @@
|
|||
"""Support for QNAP NAS Sensors."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
|
||||
from qnapstats import QNAPStats
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.sensor import (
|
||||
|
|
@ -33,9 +29,10 @@ from homeassistant.exceptions import PlatformNotReady
|
|||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||
from homeassistant.util import Throttle
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from .const import DEFAULT_PORT, DEFAULT_TIMEOUT
|
||||
from .coordinator import QnapCoordinator
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -59,8 +56,6 @@ CONF_DRIVES = "drives"
|
|||
CONF_NICS = "nics"
|
||||
CONF_VOLUMES = "volumes"
|
||||
|
||||
MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=1)
|
||||
|
||||
NOTIFICATION_ID = "qnap_notification"
|
||||
NOTIFICATION_TITLE = "QNAP Sensor Setup"
|
||||
|
||||
|
|
@ -201,18 +196,16 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
|||
)
|
||||
|
||||
|
||||
def setup_platform(
|
||||
async def async_setup_platform(
|
||||
hass: HomeAssistant,
|
||||
config: ConfigType,
|
||||
add_entities: AddEntitiesCallback,
|
||||
discovery_info: DiscoveryInfoType | None = None,
|
||||
) -> None:
|
||||
"""Set up the QNAP NAS sensor."""
|
||||
api = QNAPStatsAPI(config)
|
||||
api.update()
|
||||
|
||||
# QNAP is not available
|
||||
if not api.data:
|
||||
coordinator = QnapCoordinator(hass, config)
|
||||
await coordinator.async_refresh()
|
||||
if not coordinator.last_update_success:
|
||||
raise PlatformNotReady
|
||||
|
||||
monitored_conditions = config[CONF_MONITORED_CONDITIONS]
|
||||
|
|
@ -221,21 +214,21 @@ def setup_platform(
|
|||
# Basic sensors
|
||||
sensors.extend(
|
||||
[
|
||||
QNAPSystemSensor(api, description)
|
||||
QNAPSystemSensor(coordinator, description)
|
||||
for description in _SYSTEM_MON_COND
|
||||
if description.key in monitored_conditions
|
||||
]
|
||||
)
|
||||
sensors.extend(
|
||||
[
|
||||
QNAPCPUSensor(api, description)
|
||||
QNAPCPUSensor(coordinator, description)
|
||||
for description in _CPU_MON_COND
|
||||
if description.key in monitored_conditions
|
||||
]
|
||||
)
|
||||
sensors.extend(
|
||||
[
|
||||
QNAPMemorySensor(api, description)
|
||||
QNAPMemorySensor(coordinator, description)
|
||||
for description in _MEMORY_MON_COND
|
||||
if description.key in monitored_conditions
|
||||
]
|
||||
|
|
@ -244,8 +237,8 @@ def setup_platform(
|
|||
# Network sensors
|
||||
sensors.extend(
|
||||
[
|
||||
QNAPNetworkSensor(api, description, nic)
|
||||
for nic in config.get(CONF_NICS, api.data["system_stats"]["nics"])
|
||||
QNAPNetworkSensor(coordinator, description, nic)
|
||||
for nic in config.get(CONF_NICS, coordinator.data["system_stats"]["nics"])
|
||||
for description in _NETWORK_MON_COND
|
||||
if description.key in monitored_conditions
|
||||
]
|
||||
|
|
@ -254,8 +247,8 @@ def setup_platform(
|
|||
# Drive sensors
|
||||
sensors.extend(
|
||||
[
|
||||
QNAPDriveSensor(api, description, drive)
|
||||
for drive in config.get(CONF_DRIVES, api.data["smart_drive_health"])
|
||||
QNAPDriveSensor(coordinator, description, drive)
|
||||
for drive in config.get(CONF_DRIVES, coordinator.data["smart_drive_health"])
|
||||
for description in _DRIVE_MON_COND
|
||||
if description.key in monitored_conditions
|
||||
]
|
||||
|
|
@ -264,8 +257,8 @@ def setup_platform(
|
|||
# Volume sensors
|
||||
sensors.extend(
|
||||
[
|
||||
QNAPVolumeSensor(api, description, volume)
|
||||
for volume in config.get(CONF_VOLUMES, api.data["volumes"])
|
||||
QNAPVolumeSensor(coordinator, description, volume)
|
||||
for volume in config.get(CONF_VOLUMES, coordinator.data["volumes"])
|
||||
for description in _VOLUME_MON_COND
|
||||
if description.key in monitored_conditions
|
||||
]
|
||||
|
|
@ -284,62 +277,27 @@ def round_nicely(number):
|
|||
return round(number)
|
||||
|
||||
|
||||
class QNAPStatsAPI:
|
||||
"""Class to interface with the API."""
|
||||
|
||||
def __init__(self, config):
|
||||
"""Initialize the API wrapper."""
|
||||
|
||||
protocol = "https" if config[CONF_SSL] else "http"
|
||||
self._api = QNAPStats(
|
||||
f"{protocol}://{config.get(CONF_HOST)}",
|
||||
config.get(CONF_PORT),
|
||||
config.get(CONF_USERNAME),
|
||||
config.get(CONF_PASSWORD),
|
||||
verify_ssl=config.get(CONF_VERIFY_SSL),
|
||||
timeout=config.get(CONF_TIMEOUT),
|
||||
)
|
||||
|
||||
self.data = {}
|
||||
|
||||
@Throttle(MIN_TIME_BETWEEN_UPDATES)
|
||||
def update(self):
|
||||
"""Update API information and store locally."""
|
||||
try:
|
||||
self.data["system_stats"] = self._api.get_system_stats()
|
||||
self.data["system_health"] = self._api.get_system_health()
|
||||
self.data["smart_drive_health"] = self._api.get_smart_disk_health()
|
||||
self.data["volumes"] = self._api.get_volumes()
|
||||
self.data["bandwidth"] = self._api.get_bandwidth()
|
||||
except Exception: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Failed to fetch QNAP stats from the NAS")
|
||||
|
||||
|
||||
class QNAPSensor(SensorEntity):
|
||||
class QNAPSensor(CoordinatorEntity[QnapCoordinator], SensorEntity):
|
||||
"""Base class for a QNAP sensor."""
|
||||
|
||||
def __init__(
|
||||
self, api, description: SensorEntityDescription, monitor_device=None
|
||||
self,
|
||||
coordinator: QnapCoordinator,
|
||||
description: SensorEntityDescription,
|
||||
monitor_device: str | None = None,
|
||||
) -> None:
|
||||
"""Initialize the sensor."""
|
||||
super().__init__(coordinator)
|
||||
self.entity_description = description
|
||||
self.monitor_device = monitor_device
|
||||
self._api = api
|
||||
self.device_name = self.coordinator.data["system_stats"]["system"]["name"]
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the sensor, if any."""
|
||||
server_name = self._api.data["system_stats"]["system"]["name"]
|
||||
|
||||
if self.monitor_device is not None:
|
||||
return (
|
||||
f"{server_name} {self.entity_description.name} ({self.monitor_device})"
|
||||
)
|
||||
return f"{server_name} {self.entity_description.name}"
|
||||
|
||||
def update(self) -> None:
|
||||
"""Get the latest data for the states."""
|
||||
self._api.update()
|
||||
return f"{self.device_name} {self.entity_description.name} ({self.monitor_device})"
|
||||
return f"{self.device_name} {self.entity_description.name}"
|
||||
|
||||
|
||||
class QNAPCPUSensor(QNAPSensor):
|
||||
|
|
@ -349,9 +307,9 @@ class QNAPCPUSensor(QNAPSensor):
|
|||
def native_value(self):
|
||||
"""Return the state of the sensor."""
|
||||
if self.entity_description.key == "cpu_temp":
|
||||
return self._api.data["system_stats"]["cpu"]["temp_c"]
|
||||
return self.coordinator.data["system_stats"]["cpu"]["temp_c"]
|
||||
if self.entity_description.key == "cpu_usage":
|
||||
return self._api.data["system_stats"]["cpu"]["usage_percent"]
|
||||
return self.coordinator.data["system_stats"]["cpu"]["usage_percent"]
|
||||
|
||||
|
||||
class QNAPMemorySensor(QNAPSensor):
|
||||
|
|
@ -360,11 +318,11 @@ class QNAPMemorySensor(QNAPSensor):
|
|||
@property
|
||||
def native_value(self):
|
||||
"""Return the state of the sensor."""
|
||||
free = float(self._api.data["system_stats"]["memory"]["free"]) / 1024
|
||||
free = float(self.coordinator.data["system_stats"]["memory"]["free"]) / 1024
|
||||
if self.entity_description.key == "memory_free":
|
||||
return round_nicely(free)
|
||||
|
||||
total = float(self._api.data["system_stats"]["memory"]["total"]) / 1024
|
||||
total = float(self.coordinator.data["system_stats"]["memory"]["total"]) / 1024
|
||||
|
||||
used = total - free
|
||||
if self.entity_description.key == "memory_used":
|
||||
|
|
@ -376,8 +334,8 @@ class QNAPMemorySensor(QNAPSensor):
|
|||
@property
|
||||
def extra_state_attributes(self):
|
||||
"""Return the state attributes."""
|
||||
if self._api.data:
|
||||
data = self._api.data["system_stats"]["memory"]
|
||||
if self.coordinator.data:
|
||||
data = self.coordinator.data["system_stats"]["memory"]
|
||||
size = round_nicely(float(data["total"]) / 1024)
|
||||
return {ATTR_MEMORY_SIZE: f"{size} {UnitOfInformation.GIBIBYTES}"}
|
||||
|
||||
|
|
@ -389,10 +347,10 @@ class QNAPNetworkSensor(QNAPSensor):
|
|||
def native_value(self):
|
||||
"""Return the state of the sensor."""
|
||||
if self.entity_description.key == "network_link_status":
|
||||
nic = self._api.data["system_stats"]["nics"][self.monitor_device]
|
||||
nic = self.coordinator.data["system_stats"]["nics"][self.monitor_device]
|
||||
return nic["link_status"]
|
||||
|
||||
data = self._api.data["bandwidth"][self.monitor_device]
|
||||
data = self.coordinator.data["bandwidth"][self.monitor_device]
|
||||
if self.entity_description.key == "network_tx":
|
||||
return round_nicely(data["tx"] / 1024 / 1024)
|
||||
|
||||
|
|
@ -402,8 +360,8 @@ class QNAPNetworkSensor(QNAPSensor):
|
|||
@property
|
||||
def extra_state_attributes(self):
|
||||
"""Return the state attributes."""
|
||||
if self._api.data:
|
||||
data = self._api.data["system_stats"]["nics"][self.monitor_device]
|
||||
if self.coordinator.data:
|
||||
data = self.coordinator.data["system_stats"]["nics"][self.monitor_device]
|
||||
return {
|
||||
ATTR_IP: data["ip"],
|
||||
ATTR_MASK: data["mask"],
|
||||
|
|
@ -422,16 +380,16 @@ class QNAPSystemSensor(QNAPSensor):
|
|||
def native_value(self):
|
||||
"""Return the state of the sensor."""
|
||||
if self.entity_description.key == "status":
|
||||
return self._api.data["system_health"]
|
||||
return self.coordinator.data["system_health"]
|
||||
|
||||
if self.entity_description.key == "system_temp":
|
||||
return int(self._api.data["system_stats"]["system"]["temp_c"])
|
||||
return int(self.coordinator.data["system_stats"]["system"]["temp_c"])
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self):
|
||||
"""Return the state attributes."""
|
||||
if self._api.data:
|
||||
data = self._api.data["system_stats"]
|
||||
if self.coordinator.data:
|
||||
data = self.coordinator.data["system_stats"]
|
||||
days = int(data["uptime"]["days"])
|
||||
hours = int(data["uptime"]["hours"])
|
||||
minutes = int(data["uptime"]["minutes"])
|
||||
|
|
@ -450,7 +408,7 @@ class QNAPDriveSensor(QNAPSensor):
|
|||
@property
|
||||
def native_value(self):
|
||||
"""Return the state of the sensor."""
|
||||
data = self._api.data["smart_drive_health"][self.monitor_device]
|
||||
data = self.coordinator.data["smart_drive_health"][self.monitor_device]
|
||||
|
||||
if self.entity_description.key == "drive_smart_status":
|
||||
return data["health"]
|
||||
|
|
@ -461,7 +419,7 @@ class QNAPDriveSensor(QNAPSensor):
|
|||
@property
|
||||
def name(self):
|
||||
"""Return the name of the sensor, if any."""
|
||||
server_name = self._api.data["system_stats"]["system"]["name"]
|
||||
server_name = self.coordinator.data["system_stats"]["system"]["name"]
|
||||
|
||||
return (
|
||||
f"{server_name} {self.entity_description.name} (Drive"
|
||||
|
|
@ -471,8 +429,8 @@ class QNAPDriveSensor(QNAPSensor):
|
|||
@property
|
||||
def extra_state_attributes(self):
|
||||
"""Return the state attributes."""
|
||||
if self._api.data:
|
||||
data = self._api.data["smart_drive_health"][self.monitor_device]
|
||||
if self.coordinator.data:
|
||||
data = self.coordinator.data["smart_drive_health"][self.monitor_device]
|
||||
return {
|
||||
ATTR_DRIVE: data["drive_number"],
|
||||
ATTR_MODEL: data["model"],
|
||||
|
|
@ -487,7 +445,7 @@ class QNAPVolumeSensor(QNAPSensor):
|
|||
@property
|
||||
def native_value(self):
|
||||
"""Return the state of the sensor."""
|
||||
data = self._api.data["volumes"][self.monitor_device]
|
||||
data = self.coordinator.data["volumes"][self.monitor_device]
|
||||
|
||||
free_gb = int(data["free_size"]) / 1024 / 1024 / 1024
|
||||
if self.entity_description.key == "volume_size_free":
|
||||
|
|
@ -505,8 +463,8 @@ class QNAPVolumeSensor(QNAPSensor):
|
|||
@property
|
||||
def extra_state_attributes(self):
|
||||
"""Return the state attributes."""
|
||||
if self._api.data:
|
||||
data = self._api.data["volumes"][self.monitor_device]
|
||||
if self.coordinator.data:
|
||||
data = self.coordinator.data["volumes"][self.monitor_device]
|
||||
total_gb = int(data["total_size"]) / 1024 / 1024 / 1024
|
||||
|
||||
return {
|
||||
|
|
|
|||
Loading…
Reference in New Issue