redo system sensors

This commit is contained in:
Tomaae 2022-02-01 10:12:38 +01:00
parent 123044fceb
commit 74b46ed562
2 changed files with 450 additions and 409 deletions

View file

@ -1,19 +1,17 @@
"""Support for the Mikrotik Router sensor service."""
"""Implementation of Mikrotik Router sensor entities."""
import logging
from typing import Any, Dict, Optional
from homeassistant.const import (
CONF_NAME,
CONF_HOST,
ATTR_ATTRIBUTION,
ATTR_DEVICE_CLASS,
TEMP_CELSIUS,
ELECTRIC_POTENTIAL_VOLT,
)
from homeassistant.helpers.entity import EntityCategory
from homeassistant.components.sensor import SensorDeviceClass
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.components.sensor import SensorEntity
from .const import (
CONF_SENSOR_PORT_TRAFFIC,
@ -21,11 +19,15 @@ from .const import (
)
from homeassistant.core import callback
from homeassistant.components.sensor import SensorEntity
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from .const import DOMAIN, DATA_CLIENT, ATTRIBUTION
from .sensor_types import (
MikrotikSensorEntityDescription,
SENSOR_TYPES,
DEVICE_ATTRIBUTES_CLIENT_TRAFFIC,
)
_LOGGER = logging.getLogger(__name__)
@ -46,191 +48,6 @@ def format_attribute(attr):
return res
ATTR_ICON = "icon"
ATTR_LABEL = "label"
ATTR_UNIT = "unit"
ATTR_UNIT_ATTR = "unit_attr"
ATTR_GROUP = "group"
ATTR_PATH = "data_path"
ATTR_ATTR = "data_attr"
ATTR_CTGR = "entity_category"
SENSOR_TYPES = {
"system_temperature": {
ATTR_DEVICE_CLASS: SensorDeviceClass.TEMPERATURE,
ATTR_ICON: "mdi:thermometer",
ATTR_LABEL: "Temperature",
ATTR_UNIT: TEMP_CELSIUS,
ATTR_GROUP: "System",
ATTR_PATH: "health",
ATTR_ATTR: "temperature",
ATTR_CTGR: None,
},
"system_voltage": {
ATTR_DEVICE_CLASS: SensorDeviceClass.VOLTAGE,
ATTR_ICON: "mdi:lightning-bolt",
ATTR_LABEL: "Voltage",
ATTR_UNIT: ELECTRIC_POTENTIAL_VOLT,
ATTR_GROUP: "System",
ATTR_PATH: "health",
ATTR_ATTR: "voltage",
ATTR_CTGR: EntityCategory.DIAGNOSTIC,
},
"system_cpu-temperature": {
ATTR_DEVICE_CLASS: SensorDeviceClass.TEMPERATURE,
ATTR_ICON: "mdi:thermometer",
ATTR_LABEL: "CPU temperature",
ATTR_UNIT: TEMP_CELSIUS,
ATTR_GROUP: "System",
ATTR_PATH: "health",
ATTR_ATTR: "cpu-temperature",
ATTR_CTGR: None,
},
"system_board-temperature1": {
ATTR_DEVICE_CLASS: SensorDeviceClass.TEMPERATURE,
ATTR_ICON: "mdi:thermometer",
ATTR_LABEL: "Board temperature",
ATTR_UNIT: TEMP_CELSIUS,
ATTR_GROUP: "System",
ATTR_PATH: "health",
ATTR_ATTR: "board-temperature1",
ATTR_CTGR: None,
},
"system_power-consumption": {
ATTR_DEVICE_CLASS: SensorDeviceClass.POWER,
ATTR_ICON: "mdi:transmission-tower",
ATTR_LABEL: "Power consumption",
ATTR_UNIT: "W",
ATTR_GROUP: "System",
ATTR_PATH: "health",
ATTR_ATTR: "power-consumption",
ATTR_CTGR: None,
},
"system_fan1-speed": {
ATTR_DEVICE_CLASS: None,
ATTR_ICON: "mdi:fan",
ATTR_LABEL: "Fan1 speed",
ATTR_UNIT: "RPM",
ATTR_GROUP: "System",
ATTR_PATH: "health",
ATTR_ATTR: "fan1-speed",
ATTR_CTGR: None,
},
"system_fan2-speed": {
ATTR_DEVICE_CLASS: None,
ATTR_ICON: "mdi:fan",
ATTR_LABEL: "Fan2 speed",
ATTR_UNIT: "RPM",
ATTR_GROUP: "System",
ATTR_PATH: "health",
ATTR_ATTR: "fan2-speed",
ATTR_CTGR: None,
},
"system_uptime": {
ATTR_DEVICE_CLASS: SensorDeviceClass.TIMESTAMP,
ATTR_ICON: None,
ATTR_LABEL: "Uptime",
ATTR_UNIT: None,
ATTR_GROUP: "System",
ATTR_PATH: "resource",
ATTR_ATTR: "uptime",
ATTR_CTGR: EntityCategory.DIAGNOSTIC,
},
"system_cpu-load": {
ATTR_DEVICE_CLASS: None,
ATTR_ICON: "mdi:speedometer",
ATTR_LABEL: "CPU load",
ATTR_UNIT: "%",
ATTR_GROUP: "System",
ATTR_PATH: "resource",
ATTR_ATTR: "cpu-load",
ATTR_CTGR: None,
},
"system_memory-usage": {
ATTR_DEVICE_CLASS: None,
ATTR_ICON: "mdi:memory",
ATTR_LABEL: "Memory usage",
ATTR_UNIT: "%",
ATTR_GROUP: "System",
ATTR_PATH: "resource",
ATTR_ATTR: "memory-usage",
ATTR_CTGR: None,
},
"system_hdd-usage": {
ATTR_DEVICE_CLASS: None,
ATTR_ICON: "mdi:harddisk",
ATTR_LABEL: "HDD usage",
ATTR_UNIT: "%",
ATTR_GROUP: "System",
ATTR_PATH: "resource",
ATTR_ATTR: "hdd-usage",
ATTR_CTGR: None,
},
"traffic_tx": {
ATTR_DEVICE_CLASS: None,
ATTR_ICON: "mdi:upload-network-outline",
ATTR_LABEL: "TX",
ATTR_UNIT: "ps",
ATTR_UNIT_ATTR: "tx-bits-per-second-attr",
ATTR_PATH: "interface",
ATTR_ATTR: "tx-bits-per-second",
ATTR_CTGR: None,
},
"traffic_rx": {
ATTR_DEVICE_CLASS: None,
ATTR_ICON: "mdi:download-network-outline",
ATTR_LABEL: "RX",
ATTR_UNIT: "ps",
ATTR_UNIT_ATTR: "rx-bits-per-second-attr",
ATTR_PATH: "interface",
ATTR_ATTR: "rx-bits-per-second",
ATTR_CTGR: None,
},
"client_traffic_lan_tx": {
ATTR_DEVICE_CLASS: None,
ATTR_ICON: "mdi:upload-network",
ATTR_LABEL: "LAN TX",
ATTR_UNIT: "ps",
ATTR_UNIT_ATTR: "tx-rx-attr",
ATTR_PATH: "client_traffic",
ATTR_ATTR: "lan-tx",
ATTR_CTGR: None,
},
"client_traffic_lan_rx": {
ATTR_DEVICE_CLASS: None,
ATTR_ICON: "mdi:download-network",
ATTR_LABEL: "LAN RX",
ATTR_UNIT: "ps",
ATTR_UNIT_ATTR: "tx-rx-attr",
ATTR_PATH: "client_traffic",
ATTR_ATTR: "lan-rx",
ATTR_CTGR: None,
},
"client_traffic_wan_tx": {
ATTR_DEVICE_CLASS: None,
ATTR_ICON: "mdi:upload-network",
ATTR_LABEL: "WAN TX",
ATTR_UNIT: "ps",
ATTR_UNIT_ATTR: "tx-rx-attr",
ATTR_PATH: "client_traffic",
ATTR_ATTR: "wan-tx",
ATTR_CTGR: None,
},
"client_traffic_wan_rx": {
ATTR_DEVICE_CLASS: None,
ATTR_ICON: "mdi:download-network",
ATTR_LABEL: "WAN RX",
ATTR_UNIT: "ps",
ATTR_UNIT_ATTR: "tx-rx-attr",
ATTR_PATH: "client_traffic",
ATTR_ATTR: "wan-rx",
ATTR_CTGR: None,
},
}
DEVICE_ATTRIBUTES_CLIENT_TRAFFIC = ["address", "mac-address", "host-name"]
# ---------------------------
# async_setup_entry
# ---------------------------
@ -264,53 +81,37 @@ def update_items(inst, config_entry, mikrotik_controller, async_add_entities, se
"""Update sensor state from the controller."""
new_sensors = []
for sid, sid_uid, sid_name, sid_val, sid_ref, sid_attr, sid_func in zip(
for sensor, sid_func in zip(
# Data point name
["environment"],
# Data point unique id
["name"],
# Entry Name
["name"],
# Entry Value
["value"],
# Entry Unique id
["name"],
# Attr
[
None,
],
# Switch function
[
MikrotikControllerEnvironmentSensor,
MikrotikControllerSensor,
],
):
for uid in mikrotik_controller.data[sid]:
item_id = f"{inst}-{sid}-{mikrotik_controller.data[sid][uid][sid_uid]}"
for uid in mikrotik_controller.data[sensor]:
item_id = f"{inst}-{sensor}-{mikrotik_controller.data[sensor][uid][SENSOR_TYPES[sensor].data_uid]}"
_LOGGER.debug("Updating sensor %s", item_id)
if item_id in sensors:
if sensors[item_id].enabled:
sensors[item_id].async_schedule_update_ha_state()
continue
# Create new entity
sid_data = {
"sid": sid,
"sid_uid": sid_uid,
"sid_name": sid_name,
"sid_ref": sid_ref,
"sid_attr": sid_attr,
"sid_val": sid_val,
}
sensors[item_id] = sid_func(inst, uid, mikrotik_controller, sid_data)
sensors[item_id] = sid_func(
inst=inst,
uid=uid,
mikrotik_controller=mikrotik_controller,
entity_description=SENSOR_TYPES[sensor],
)
new_sensors.append(sensors[item_id])
for sensor in SENSOR_TYPES:
if sensor.startswith("system_"):
if (
SENSOR_TYPES[sensor][ATTR_ATTR]
not in mikrotik_controller.data[SENSOR_TYPES[sensor][ATTR_PATH]]
or mikrotik_controller.data[SENSOR_TYPES[sensor][ATTR_PATH]][
SENSOR_TYPES[sensor][ATTR_ATTR]
SENSOR_TYPES[sensor].data_attribute
not in mikrotik_controller.data[SENSOR_TYPES[sensor].data_path]
or mikrotik_controller.data[SENSOR_TYPES[sensor].data_path][
SENSOR_TYPES[sensor].data_attribute
]
== "unknown"
):
@ -323,52 +124,55 @@ def update_items(inst, config_entry, mikrotik_controller, async_add_entities, se
continue
sensors[item_id] = MikrotikControllerSensor(
mikrotik_controller=mikrotik_controller, inst=inst, sid_data=sensor
inst=inst,
uid="",
mikrotik_controller=mikrotik_controller,
entity_description=SENSOR_TYPES[sensor],
)
new_sensors.append(sensors[item_id])
if sensor.startswith("traffic_"):
if not config_entry.options.get(
CONF_SENSOR_PORT_TRAFFIC, DEFAULT_SENSOR_PORT_TRAFFIC
):
continue
for uid in mikrotik_controller.data["interface"]:
if mikrotik_controller.data["interface"][uid]["type"] != "bridge":
item_id = f"{inst}-{sensor}-{mikrotik_controller.data['interface'][uid]['default-name']}"
_LOGGER.debug("Updating sensor %s", item_id)
if item_id in sensors:
if sensors[item_id].enabled:
sensors[item_id].async_schedule_update_ha_state()
continue
sensors[item_id] = MikrotikControllerTrafficSensor(
mikrotik_controller=mikrotik_controller,
inst=inst,
sensor=sensor,
uid=uid,
)
new_sensors.append(sensors[item_id])
if sensor.startswith("client_traffic_"):
for uid in mikrotik_controller.data["client_traffic"]:
item_id = f"{inst}-{sensor}-{mikrotik_controller.data['client_traffic'][uid]['mac-address']}"
if item_id in sensors:
if sensors[item_id].enabled:
sensors[item_id].async_schedule_update_ha_state()
continue
if (
SENSOR_TYPES[sensor][ATTR_ATTR]
in mikrotik_controller.data["client_traffic"][uid].keys()
):
sensors[item_id] = MikrotikClientTrafficSensor(
mikrotik_controller=mikrotik_controller,
inst=inst,
sensor=sensor,
uid=uid,
)
new_sensors.append(sensors[item_id])
# if sensor.startswith("traffic_"):
# if not config_entry.options.get(
# CONF_SENSOR_PORT_TRAFFIC, DEFAULT_SENSOR_PORT_TRAFFIC
# ):
# continue
#
# for uid in mikrotik_controller.data["interface"]:
# if mikrotik_controller.data["interface"][uid]["type"] != "bridge":
# item_id = f"{inst}-{sensor}-{mikrotik_controller.data['interface'][uid]['default-name']}"
# _LOGGER.debug("Updating sensor %s", item_id)
# if item_id in sensors:
# if sensors[item_id].enabled:
# sensors[item_id].async_schedule_update_ha_state()
# continue
#
# sensors[item_id] = MikrotikControllerTrafficSensor(
# mikrotik_controller=mikrotik_controller,
# inst=inst,
# sensor=SENSOR_TYPES[sensor],
# uid=uid,
# )
# new_sensors.append(sensors[item_id])
#
# if sensor.startswith("client_traffic_"):
# for uid in mikrotik_controller.data["client_traffic"]:
# item_id = f"{inst}-{sensor}-{mikrotik_controller.data['client_traffic'][uid]['mac-address']}"
# if item_id in sensors:
# if sensors[item_id].enabled:
# sensors[item_id].async_schedule_update_ha_state()
# continue
#
# if (
# SENSOR_TYPES[sensor].data_attribute
# in mikrotik_controller.data["client_traffic"][uid].keys()
# ):
# sensors[item_id] = MikrotikClientTrafficSensor(
# mikrotik_controller=mikrotik_controller,
# inst=inst,
# sensor=SENSOR_TYPES[sensor],
# uid=uid,
# )
# new_sensors.append(sensors[item_id])
if new_sensors:
async_add_entities(new_sensors, True)
@ -380,76 +184,49 @@ def update_items(inst, config_entry, mikrotik_controller, async_add_entities, se
class MikrotikControllerSensor(SensorEntity):
"""Define an Mikrotik Controller sensor."""
def __init__(self, mikrotik_controller, inst, sid_data):
def __init__(
self,
mikrotik_controller,
inst,
uid: "",
entity_description: MikrotikSensorEntityDescription,
):
"""Initialize."""
self._inst = inst
self._sensor = sid_data
self.entity_description = entity_description
self._ctrl = mikrotik_controller
if sid_data in SENSOR_TYPES:
self._data = mikrotik_controller.data[SENSOR_TYPES[sid_data][ATTR_PATH]]
self._type = SENSOR_TYPES[sid_data]
self._icon = self._type[ATTR_ICON]
self._attr = self._type[ATTR_ATTR]
self._dcls = self._type[ATTR_DEVICE_CLASS]
self._ctgr = self._type[ATTR_CTGR]
self._attr_extra_state_attributes = {ATTR_ATTRIBUTION: ATTRIBUTION}
self._uid = uid
if self._uid:
self._data = mikrotik_controller.data[self.entity_description.data_path][
self._uid
]
else:
self._type = {}
self._icon = None
self._attr = None
self._dcls = None
self._ctgr = None
self._state = None
self._attrs = {ATTR_ATTRIBUTION: ATTRIBUTION}
self._data = mikrotik_controller.data[self.entity_description.data_path]
@property
def name(self) -> str:
"""Return the name."""
return f"{self._inst} {self._type[ATTR_LABEL]}"
@property
def state(self) -> Optional[str]:
"""Return the state."""
val = "unknown"
if self._attr in self._data:
val = self._data[self._attr]
return val
@property
def extra_state_attributes(self) -> Dict[str, Any]:
"""Return the state attributes."""
return self._attrs
@property
def icon(self) -> str:
"""Return the icon."""
return self._icon
@property
def entity_category(self) -> str:
"""Return entity category"""
return self._ctgr
@property
def device_class(self) -> Optional[str]:
"""Return the device class."""
return self._dcls
if self._uid:
return f"{self._inst} {self._data[self.entity_description.data_reference]}"
else:
return f"{self._inst} {self.entity_description.name}"
@property
def unique_id(self) -> str:
"""Return a unique id for this entity."""
return f"{self._inst.lower()}-{self._sensor.lower()}"
if self._uid:
return f"{self._inst.lower()}-{self.entity_description.key}-{self._data[self.entity_description.data_reference].lower()}"
else:
return f"{self._inst.lower()}-{self.entity_description.key}"
@property
def unit_of_measurement(self):
"""Return the unit the value is expressed in."""
if ATTR_UNIT_ATTR in self._type:
return self._data[SENSOR_TYPES[self._sensor][ATTR_UNIT_ATTR]]
if ATTR_UNIT in self._type:
return self._type[ATTR_UNIT]
def state(self) -> Optional[str]:
"""Return the state."""
if self.entity_description.data_attribute:
return self._data[self.entity_description.data_attribute]
else:
return "unknown"
@property
def available(self) -> bool:
@ -457,40 +234,28 @@ class MikrotikControllerSensor(SensorEntity):
return self._ctrl.connected()
@property
def device_info(self) -> Dict[str, Any]:
def device_info(self) -> DeviceInfo:
"""Return a description for device registry."""
if self._type[ATTR_GROUP] == "System":
self._type[ATTR_GROUP] = self._ctrl.data["resource"]["board-name"]
di_domain = self.entity_description.data_reference
if self.entity_description.ha_group == "System":
self.entity_description.ha_group = self._ctrl.data["resource"]["board-name"]
di_domain = self._ctrl.data["routerboard"]["serial-number"]
info = {
"connections": {
(DOMAIN, f"{self._ctrl.data['routerboard']['serial-number']}")
},
"manufacturer": self._ctrl.data["resource"]["platform"],
"model": self._ctrl.data["resource"]["board-name"],
"name": f"{self._inst} {self._type[ATTR_GROUP]}",
"sw_version": self._ctrl.data["resource"]["version"],
"configuration_url": f"http://{self._ctrl.config_entry.data[CONF_HOST]}",
}
if ATTR_GROUP in self._type:
info["identifiers"] = {
(
DOMAIN,
"serial-number",
f"{self._ctrl.data['routerboard']['serial-number']}",
"sensor",
f"{self._inst} {self._type[ATTR_GROUP]}",
)
}
info = DeviceInfo(
connections={(DOMAIN, f"{di_domain}")},
name=f"{self._inst} {self.entity_description.ha_group}",
model=f"{self._ctrl.data['resource']['board-name']}",
manufacturer=f"{self._ctrl.data['resource']['platform']}",
sw_version=f"{self._ctrl.data['resource']['version']}",
configuration_url=f"http://{self._ctrl.config_entry.data[CONF_HOST]}",
via_device=(DOMAIN, f"{self._ctrl.data['routerboard']['serial-number']}"),
)
return info
async def async_update(self):
"""Synchronize state with controller."""
async def async_added_to_hass(self):
"""Run when entity about to be added to hass."""
_LOGGER.debug("New sensor %s (%s)", self._inst, self._sensor)
_LOGGER.debug("New sensor %s (%s)", self._inst, self.entity_description.name)
# ---------------------------
@ -502,18 +267,19 @@ class MikrotikControllerTrafficSensor(MikrotikControllerSensor):
def __init__(self, mikrotik_controller, inst, sensor, uid):
"""Initialize."""
super().__init__(mikrotik_controller, inst, sensor)
self._uid = uid
self._data = mikrotik_controller.data[SENSOR_TYPES[sensor][ATTR_PATH]][uid]
if uid:
self._uid = uid
self._data = mikrotik_controller.data[sensor.data_path][uid]
@property
def name(self) -> str:
"""Return the name."""
return f"{self._inst} {self._data['name']} {self._type[ATTR_LABEL]}"
return f"{self._inst} {self._data['name']} {self.entity_description.name}"
@property
def unique_id(self) -> str:
"""Return a unique id for this entity."""
return f"{self._inst.lower()}-{self._sensor.lower()}-{self._data['default-name'].lower()}"
return f"{self._inst.lower()}-{self.entity_description.lower()}-{self._data['default-name'].lower()}"
@property
def state_class(self) -> str:
@ -543,17 +309,17 @@ class MikrotikClientTrafficSensor(MikrotikControllerSensor):
"""Initialize."""
super().__init__(mikrotik_controller, inst, sensor)
self._uid = uid
self._data = mikrotik_controller.data[SENSOR_TYPES[sensor][ATTR_PATH]][uid]
self._data = mikrotik_controller.data[sensor.data_path][uid]
@property
def name(self) -> str:
"""Return the name."""
return f"{self._data['host-name']} {self._type[ATTR_LABEL]}"
return f"{self._data['host-name']} {self.entity_description.name}"
@property
def unique_id(self) -> str:
"""Return a unique id for this entity."""
return f"{self._inst.lower()}-{self._sensor.lower()}-{self._data['mac-address'].lower()}"
return f"{self._inst.lower()}-{self.entity_description.lower()}-{self._data['mac-address'].lower()}"
@property
def available(self) -> bool:
@ -590,57 +356,3 @@ class MikrotikClientTrafficSensor(MikrotikControllerSensor):
attributes[format_attribute(variable)] = self._data[variable]
return attributes
# ---------------------------
# MikrotikControllerEnvironmentSensor
# ---------------------------
class MikrotikControllerEnvironmentSensor(MikrotikControllerSensor):
"""Define an Enviroment variable sensor."""
def __init__(self, inst, uid, mikrotik_controller, sid_data):
"""Initialize."""
super().__init__(mikrotik_controller, inst, "")
self._uid = uid
self._sid_data = sid_data
self._data = mikrotik_controller.data[self._sid_data["sid"]][uid]
@property
def name(self) -> str:
"""Return the name."""
return f"{self._inst} {self._data[self._sid_data['sid_ref']]}"
@property
def unique_id(self) -> str:
"""Return a unique id for this entity."""
return f"{self._inst.lower()}-{self._sid_data['sid']}-{self._data[self._sid_data['sid_ref']]}"
@property
def state(self) -> Optional[str]:
"""Return the state."""
return self._data[self._sid_data["sid_val"]]
@property
def icon(self) -> str:
"""Return the icon."""
return "mdi:clipboard-list"
@property
def device_info(self) -> Dict[str, Any]:
"""Return a description for device registry."""
info = {
"identifiers": {
(
DOMAIN,
"serial-number",
f"{self._ctrl.data['routerboard']['serial-number']}",
"sensor",
"Environment",
)
},
"manufacturer": self._ctrl.data["resource"]["platform"],
"model": self._ctrl.data["resource"]["board-name"],
"name": f"{self._inst} Environment",
}
return info