redo system binary sensors

This commit is contained in:
Tomaae 2022-02-04 22:06:45 +01:00
parent d74db8f57d
commit 8b2f75a0ff
2 changed files with 212 additions and 140 deletions

View file

@ -2,8 +2,8 @@
import logging import logging
from typing import Any, Dict, Optional from typing import Any, Dict, Optional
from collections.abc import Mapping
from homeassistant.helpers.entity import EntityCategory from homeassistant.helpers.entity import DeviceInfo, EntityCategory
from homeassistant.components.binary_sensor import ( from homeassistant.components.binary_sensor import (
BinarySensorEntity, BinarySensorEntity,
BinarySensorDeviceClass, BinarySensorDeviceClass,
@ -28,24 +28,13 @@ from .const import (
DEFAULT_SENSOR_PORT_TRACKER, DEFAULT_SENSOR_PORT_TRACKER,
) )
from .binary_sensor_types import (
MikrotikBinarySensorEntityDescription,
SENSOR_TYPES,
)
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
ATTR_LABEL = "label"
ATTR_GROUP = "group"
ATTR_PATH = "data_path"
ATTR_ATTR = "data_attr"
ATTR_CTGR = "entity_category"
SENSOR_TYPES = {
"system_fwupdate": {
ATTR_DEVICE_CLASS: BinarySensorDeviceClass.UPDATE,
ATTR_LABEL: "Firmware update",
ATTR_GROUP: "System",
ATTR_PATH: "fw-update",
ATTR_ATTR: "available",
ATTR_CTGR: EntityCategory.DIAGNOSTIC,
},
}
DEVICE_ATTRIBUTES_IFACE = [ DEVICE_ATTRIBUTES_IFACE = [
"running", "running",
@ -148,71 +137,129 @@ def update_items(inst, config_entry, mikrotik_controller, async_add_entities, se
"""Update sensor state from the controller.""" """Update sensor state from the controller."""
new_sensors = [] new_sensors = []
# Add switches # for sensor, sid_func in zip(
for sid, sid_uid, sid_name, sid_ref, sid_attr, sid_func in zip( # # Sensor type name
# Data point name # [
["ppp_secret", "interface"], # "environment",
# Data point unique id # ],
["name", "default-name"], # # Entity function
# Entry Name # [
["name", "name"], # MikrotikControllerSensor,
# Entry Unique id # ],
["name", "port-mac-address"], # ):
# Attr # if sensor.startswith("traffic_") and not config_entry.options.get(
[None, DEVICE_ATTRIBUTES_IFACE], # CONF_SENSOR_PORT_TRAFFIC, DEFAULT_SENSOR_PORT_TRAFFIC
# Tracker function # ):
[ # continue
MikrotikControllerPPPSecretBinarySensor, #
MikrotikControllerPortBinarySensor, # uid_sensor = SENSOR_TYPES[sensor]
], # for uid in mikrotik_controller.data[SENSOR_TYPES[sensor].data_path]:
): # uid_data = mikrotik_controller.data[SENSOR_TYPES[sensor].data_path]
if ( # if (
sid_func == MikrotikControllerPortBinarySensor # uid_sensor.data_path == "interface"
and not config_entry.options.get( # and uid_data[uid]["type"] == "bridge"
CONF_SENSOR_PORT_TRACKER, DEFAULT_SENSOR_PORT_TRACKER # ):
) # continue
): #
continue # item_id = f"{inst}-{sensor}-{uid_data[uid][uid_sensor.data_reference]}"
for uid in mikrotik_controller.data[sid]: # _LOGGER.debug("Updating binary sensor %s", item_id)
if ( # if item_id in sensors:
# Skip if interface is wlan # if sensors[item_id].enabled:
sid == "interface" # sensors[item_id].async_schedule_update_ha_state()
and mikrotik_controller.data[sid][uid]["type"] == "wlan" # continue
): #
continue # sensors[item_id] = sid_func(
# Update entity # inst=inst,
item_id = f"{inst}-{sid}-{mikrotik_controller.data[sid][uid][sid_uid]}" # uid=uid,
_LOGGER.debug("Updating binary_sensor %s", item_id) # mikrotik_controller=mikrotik_controller,
# entity_description=uid_sensor,
# )
# new_sensors.append(sensors[item_id])
for sensor in SENSOR_TYPES:
if sensor.startswith("system_"):
uid_sensor = SENSOR_TYPES[sensor]
item_id = f"{inst}-{sensor}"
_LOGGER.debug("Updating binary sensor %s", item_id)
if item_id in sensors: if item_id in sensors:
if sensors[item_id].enabled: if sensors[item_id].enabled:
sensors[item_id].async_schedule_update_ha_state() sensors[item_id].async_schedule_update_ha_state()
continue continue
# Create new entity sensors[item_id] = MikrotikControllerBinarySensor(
sid_data = { inst=inst,
"sid": sid, uid="",
"sid_uid": sid_uid, mikrotik_controller=mikrotik_controller,
"sid_name": sid_name, entity_description=uid_sensor,
"sid_ref": sid_ref,
"sid_attr": sid_attr,
}
sensors[item_id] = sid_func(
inst, uid, mikrotik_controller, config_entry, sid_data
) )
new_sensors.append(sensors[item_id]) new_sensors.append(sensors[item_id])
for sensor in SENSOR_TYPES: #
item_id = f"{inst}-{sensor}" # # Add switches
_LOGGER.debug("Updating binary_sensor %s", item_id) # for sid, sid_uid, sid_name, sid_ref, sid_attr, sid_func in zip(
if item_id in sensors: # # Data point name
if sensors[item_id].enabled: # ["ppp_secret", "interface"],
sensors[item_id].async_schedule_update_ha_state() # # Data point unique id
continue # ["name", "default-name"],
# # Entry Name
sensors[item_id] = MikrotikControllerBinarySensor( # ["name", "name"],
mikrotik_controller=mikrotik_controller, inst=inst, sid_data=sensor # # Entry Unique id
) # ["name", "port-mac-address"],
new_sensors.append(sensors[item_id]) # # Attr
# [None, DEVICE_ATTRIBUTES_IFACE],
# # Tracker function
# [
# MikrotikControllerPPPSecretBinarySensor,
# MikrotikControllerPortBinarySensor,
# ],
# ):
# if (
# sid_func == MikrotikControllerPortBinarySensor
# and not config_entry.options.get(
# CONF_SENSOR_PORT_TRACKER, DEFAULT_SENSOR_PORT_TRACKER
# )
# ):
# continue
# for uid in mikrotik_controller.data[sid]:
# if (
# # Skip if interface is wlan
# sid == "interface"
# and mikrotik_controller.data[sid][uid]["type"] == "wlan"
# ):
# continue
# # Update entity
# item_id = f"{inst}-{sid}-{mikrotik_controller.data[sid][uid][sid_uid]}"
# _LOGGER.debug("Updating binary_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,
# }
# sensors[item_id] = sid_func(
# inst, uid, mikrotik_controller, config_entry, sid_data
# )
# new_sensors.append(sensors[item_id])
#
# for sensor in SENSOR_TYPES:
# item_id = f"{inst}-{sensor}"
# _LOGGER.debug("Updating binary_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] = MikrotikControllerBinarySensor(
# mikrotik_controller=mikrotik_controller, inst=inst, sid_data=sensor
# )
# new_sensors.append(sensors[item_id])
if new_sensors: if new_sensors:
async_add_entities(new_sensors, True) async_add_entities(new_sensors, True)
@ -221,45 +268,44 @@ def update_items(inst, config_entry, mikrotik_controller, async_add_entities, se
class MikrotikControllerBinarySensor(BinarySensorEntity): class MikrotikControllerBinarySensor(BinarySensorEntity):
"""Define an Mikrotik Controller Binary Sensor.""" """Define an Mikrotik Controller Binary Sensor."""
def __init__(self, mikrotik_controller, inst, sid_data): def __init__(
self,
inst,
uid: "",
mikrotik_controller,
entity_description: MikrotikBinarySensorEntityDescription,
):
"""Initialize.""" """Initialize."""
self.entity_description = entity_description
self._inst = inst self._inst = inst
self._sensor = sid_data
self._ctrl = mikrotik_controller self._ctrl = mikrotik_controller
if sid_data in SENSOR_TYPES: self._attr_extra_state_attributes = {ATTR_ATTRIBUTION: ATTRIBUTION}
self._data = mikrotik_controller.data[SENSOR_TYPES[sid_data][ATTR_PATH]] self._uid = uid
self._type = SENSOR_TYPES[sid_data] if self._uid:
self._attr = SENSOR_TYPES[sid_data][ATTR_ATTR] self._data = mikrotik_controller.data[self.entity_description.data_path][
self._dcls = SENSOR_TYPES[sid_data][ATTR_DEVICE_CLASS] self._uid
self._ctgr = SENSOR_TYPES[sid_data][ATTR_CTGR] ]
else: else:
self._type = {} self._data = mikrotik_controller.data[self.entity_description.data_path]
self._attr = None
self._dcls = None
self._ctgr = None
self._state = None
self._attrs = {ATTR_ATTRIBUTION: ATTRIBUTION}
@property @property
def name(self) -> str: def name(self) -> str:
"""Return the name.""" """Return the name."""
return f"{self._inst} {self._type[ATTR_LABEL]}" if self._uid:
if self.entity_description.name:
return f"{self._inst} {self._data[self.entity_description.data_name]} {self.entity_description.name}"
@property return f"{self._inst} {self._data[self.entity_description.data_name]}"
def extra_state_attributes(self) -> Dict[str, Any]: else:
"""Return the state attributes.""" return f"{self._inst} {self.entity_description.name}"
return self._attrs
@property
def device_class(self) -> Optional[str]:
"""Return the device class."""
return self._dcls
@property @property
def unique_id(self) -> str: def unique_id(self) -> str:
"""Return a unique id for this entity.""" """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 @property
def available(self) -> bool: def available(self) -> bool:
@ -267,54 +313,80 @@ class MikrotikControllerBinarySensor(BinarySensorEntity):
return self._ctrl.connected() return self._ctrl.connected()
@property @property
def entity_category(self) -> str: def is_on(self) -> bool:
"""Return entity category""" """Return true if device is on."""
return self._ctgr return self._data[self.entity_description.data_is_on]
@property @property
def device_info(self) -> Dict[str, Any]: def device_info(self) -> DeviceInfo:
"""Return a description for device registry.""" """Return a description for device registry."""
if self._type[ATTR_GROUP] == "System": dev_connection = DOMAIN
self._type[ATTR_GROUP] = self._ctrl.data["resource"]["board-name"] dev_connection_value = self.entity_description.data_reference
dev_group = self.entity_description.ha_group
if self.entity_description.ha_group == "System":
dev_group = self._ctrl.data["resource"]["board-name"]
dev_connection_value = self._ctrl.data["routerboard"]["serial-number"]
info = { if self.entity_description.ha_group.startswith("data__"):
"connections": { dev_group = self.entity_description.ha_group[6:]
(DOMAIN, f"{self._ctrl.data['routerboard']['serial-number']}") if dev_group in self._data:
}, dev_group = self._data[dev_group]
"manufacturer": self._ctrl.data["resource"]["platform"], dev_connection_value = dev_group
"model": self._ctrl.data["resource"]["board-name"],
"name": f"{self._inst} {self._type[ATTR_GROUP]}", if self.entity_description.ha_connection:
"sw_version": self._ctrl.data["resource"]["version"], dev_connection = self.entity_description.ha_connection
"configuration_url": f"http://{self._ctrl.config_entry.data[CONF_HOST]}",
} if self.entity_description.ha_connection_value:
if ATTR_GROUP in self._type: dev_connection_value = self.entity_description.ha_connection_value
info["identifiers"] = { if dev_connection_value.startswith("data__"):
( dev_connection_value = dev_connection_value[6:]
dev_connection_value = self._data[dev_connection_value]
info = DeviceInfo(
connections={(dev_connection, f"{dev_connection_value}")},
identifiers={(dev_connection, f"{dev_connection_value}")},
default_name=f"{self._inst} {dev_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']}"),
)
if "mac-address" in self.entity_description.data_reference:
dev_group = self._data[self.entity_description.data_name]
dev_manufacturer = ""
if dev_connection_value in self._ctrl.data["host"]:
dev_group = self._ctrl.data["host"][dev_connection_value]["host-name"]
dev_manufacturer = self._ctrl.data["host"][dev_connection_value][
"manufacturer"
]
info = DeviceInfo(
connections={(dev_connection, f"{dev_connection_value}")},
default_name=f"{dev_group}",
manufacturer=f"{dev_manufacturer}",
via_device=(
DOMAIN, DOMAIN,
"serial-number",
f"{self._ctrl.data['routerboard']['serial-number']}", f"{self._ctrl.data['routerboard']['serial-number']}",
"sensor", ),
f"{self._inst} {self._type[ATTR_GROUP]}", )
)
}
return info return info
@property @property
def is_on(self) -> bool: def extra_state_attributes(self) -> Mapping[str, Any]:
"""Return true if device is on.""" """Return the state attributes."""
val = False attributes = super().extra_state_attributes
if self._attr in self._data: for variable in self.entity_description.data_attributes_list:
val = self._data[self._attr] if variable in self._data:
attributes[format_attribute(variable)] = self._data[variable]
return val return attributes
async def async_update(self):
"""Synchronize state with controller."""
async def async_added_to_hass(self): async def async_added_to_hass(self):
"""Run when entity about to be added to hass.""" """Run when entity about to be added to hass."""
_LOGGER.debug("New sensor %s (%s)", self._inst, self._sensor) _LOGGER.debug("New binary sensor %s (%s)", self._inst, self.unique_id)
# --------------------------- # ---------------------------

View file

@ -27,14 +27,14 @@ class MikrotikBinarySensorEntityDescription(BinarySensorEntityDescription):
SENSOR_TYPES = { SENSOR_TYPES = {
"system_temperature": MikrotikBinarySensorEntityDescription( "system_fwupdate": MikrotikBinarySensorEntityDescription(
key="system_temperature", key="system_fwupdate",
name="Temperature", name="Firmware update",
icon="mdi:thermometer", icon="mdi:thermometer",
device_class=BinarySensorDeviceClass.CONNECTIVITY, device_class=BinarySensorDeviceClass.UPDATE,
entity_category=EntityCategory.DIAGNOSTIC, entity_category=EntityCategory.DIAGNOSTIC,
ha_group="System", ha_group="System",
data_path="health", data_path="fw-update",
data_name="", data_name="",
data_uid="", data_uid="",
data_reference="", data_reference="",