tomaae.homeassistant-mikrot.../custom_components/mikrotik_router/switch.py

1010 lines
31 KiB
Python
Raw Normal View History

"""Support for the Mikrotik Router switches."""
2020-12-25 20:28:36 +01:00
import logging
2020-12-25 23:31:51 +01:00
from typing import Any, Dict, Optional
from homeassistant.components.switch import SwitchEntity
2020-04-11 05:45:36 +02:00
from homeassistant.const import CONF_NAME, ATTR_ATTRIBUTION
from homeassistant.core import callback
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.restore_state import RestoreEntity
2020-04-11 05:45:36 +02:00
from .const import DOMAIN, DATA_CLIENT, ATTRIBUTION
_LOGGER = logging.getLogger(__name__)
2019-12-03 18:30:45 +01:00
DEVICE_ATTRIBUTES_IFACE = [
"running",
"enabled",
"comment",
"client-ip-address",
"client-mac-address",
"port-mac-address",
"last-link-down-time",
"last-link-up-time",
"link-downs",
"actual-mtu",
"type",
"name",
]
DEVICE_ATTRIBUTES_IFACE_ETHER = [
"running",
"enabled",
"comment",
"client-ip-address",
"client-mac-address",
"port-mac-address",
"last-link-down-time",
"last-link-up-time",
"link-downs",
"actual-mtu",
"type",
"name",
"status",
"auto-negotiation",
"rate",
"full-duplex",
"default-name",
"poe-out",
]
2021-04-12 14:28:39 +02:00
DEVICE_ATTRIBUTES_IFACE_SFP = [
"status",
"auto-negotiation",
"advertising",
"link-partner-advertising",
"sfp-temperature",
"sfp-supply-voltage",
"sfp-module-present",
"sfp-tx-bias-current",
"sfp-tx-power",
"sfp-rx-power",
"sfp-rx-loss",
"sfp-tx-fault",
"sfp-type",
"sfp-connector-type",
"sfp-vendor-name",
"sfp-vendor-part-number",
"sfp-vendor-revision",
"sfp-vendor-serial",
"sfp-manufacturing-date",
"eeprom-checksum",
]
2019-12-03 18:30:45 +01:00
DEVICE_ATTRIBUTES_NAT = [
"protocol",
"dst-port",
"in-interface",
"to-addresses",
"to-ports",
"comment",
]
2020-12-18 19:58:54 +01:00
DEVICE_ATTRIBUTES_MANGLE = [
"chain",
"action",
"passthrough",
"protocol",
"src-address",
2020-12-18 19:58:54 +01:00
"src-port",
"dst-address",
2020-12-18 19:58:54 +01:00
"dst-port",
"comment",
]
2021-04-12 12:40:45 +02:00
DEVICE_ATTRIBUTES_FILTER = [
"chain",
"action",
"address-list",
"protocol",
"layer7-protocol",
"tcp-flags",
"connection-state",
"in-interface",
"src-address",
"src-port",
"out-interface",
"dst-address",
"dst-port",
"comment",
]
2020-12-25 11:48:24 +01:00
DEVICE_ATTRIBUTES_PPP_SECRET = [
"connected",
"service",
"profile",
"comment",
"caller-id",
"encoding",
]
2020-12-25 18:42:57 +01:00
DEVICE_ATTRIBUTES_KIDCONTROL = [
"rate-limit",
"mon",
"tue",
"wed",
"thu",
"fri",
"sat",
"sun",
]
DEVICE_ATTRIBUTES_QUEUE = [
"target",
"download-rate",
"upload-rate",
"download-max-limit",
"upload-max-limit",
"upload-limit-at",
"download-limit-at",
"upload-burst-limit",
"download-burst-limit",
"upload-burst-threshold",
"download-burst-threshold",
"upload-burst-time",
"download-burst-time",
"packet-marks",
"parent",
"comment",
]
# ---------------------------
# format_attribute
# ---------------------------
2019-12-07 20:50:00 +01:00
def format_attribute(attr):
res = attr.replace("-", " ")
res = res.capitalize()
res = res.replace(" ip ", " IP ")
res = res.replace(" mac ", " MAC ")
res = res.replace(" mtu", " MTU")
2021-04-12 14:29:15 +02:00
res = res.replace("Sfp", "SFP")
res = res.replace("Poe", "POE")
res = res.replace(" tx", " TX")
res = res.replace(" rx", " RX")
2019-12-07 20:50:00 +01:00
return res
# ---------------------------
# async_setup_entry
# ---------------------------
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up switches for Mikrotik Router component."""
2019-12-06 01:22:34 +01:00
inst = config_entry.data[CONF_NAME]
mikrotik_controller = hass.data[DOMAIN][DATA_CLIENT][config_entry.entry_id]
switches = {}
@callback
def update_controller():
"""Update the values of the controller."""
2019-12-06 01:22:34 +01:00
update_items(inst, mikrotik_controller, async_add_entities, switches)
mikrotik_controller.listeners.append(
2020-03-16 04:51:41 +01:00
async_dispatcher_connect(
hass, mikrotik_controller.signal_update, update_controller
)
)
update_controller()
# ---------------------------
# update_items
# ---------------------------
@callback
2019-12-06 01:22:34 +01:00
def update_items(inst, mikrotik_controller, async_add_entities, switches):
"""Update device switch state from the controller."""
new_switches = []
2019-12-05 22:10:42 +01:00
# Add switches
2020-04-20 08:39:02 +02:00
for sid, sid_uid, sid_name, sid_ref, sid_attr, sid_func in zip(
# Data point name
2021-04-12 12:40:45 +02:00
[
"interface",
"nat",
"mangle",
"filter",
"ppp_secret",
"queue",
"kid-control",
"kid-control",
2021-04-12 12:40:45 +02:00
],
2020-04-20 08:39:02 +02:00
# Data point unique id
[
"name",
"uniq-id",
"uniq-id",
"uniq-id",
"name",
"name",
"name",
"name",
],
2020-04-20 08:39:02 +02:00
# Entry Name
[
"name",
"name",
"name",
"name",
"name",
"name",
"name",
"name",
],
2020-04-20 08:39:02 +02:00
# Entry Unique id
2021-04-12 12:40:45 +02:00
[
"port-mac-address",
"uniq-id",
"uniq-id",
"uniq-id",
"name",
"name",
"name",
"name",
],
2020-04-20 08:39:02 +02:00
# Attr
[
DEVICE_ATTRIBUTES_IFACE,
DEVICE_ATTRIBUTES_NAT,
2020-12-18 19:58:54 +01:00
DEVICE_ATTRIBUTES_MANGLE,
2021-04-12 12:40:45 +02:00
DEVICE_ATTRIBUTES_FILTER,
2020-12-25 11:48:24 +01:00
DEVICE_ATTRIBUTES_PPP_SECRET,
2020-04-20 08:39:02 +02:00
DEVICE_ATTRIBUTES_QUEUE,
2020-12-25 18:42:57 +01:00
DEVICE_ATTRIBUTES_KIDCONTROL,
DEVICE_ATTRIBUTES_KIDCONTROL,
2020-04-20 08:39:02 +02:00
],
# Switch function
2020-03-16 04:51:41 +01:00
[
MikrotikControllerPortSwitch,
MikrotikControllerNATSwitch,
2020-12-18 19:58:54 +01:00
MikrotikControllerMangleSwitch,
2021-04-12 12:40:45 +02:00
MikrotikControllerFilterSwitch,
2020-12-25 11:48:24 +01:00
MikrotikControllerPPPSecretSwitch,
MikrotikControllerQueueSwitch,
2020-12-25 18:42:57 +01:00
MikrotikControllerKidcontrolSwitch,
MikrotikControllerKidcontrolPauseSwitch,
2020-03-16 04:51:41 +01:00
],
2019-12-09 09:52:45 +01:00
):
2019-12-05 22:10:42 +01:00
for uid in mikrotik_controller.data[sid]:
2020-04-20 08:39:02 +02:00
item_id = f"{inst}-{sid}-{mikrotik_controller.data[sid][uid][sid_uid]}"
if sid_func.__name__ == "MikrotikControllerKidcontrolPauseSwitch":
item_id = f"{inst}-kid-control-pause-{mikrotik_controller.data[sid][uid][sid_uid]}"
2020-04-04 19:46:40 +02:00
_LOGGER.debug("Updating switch %s", item_id)
if item_id in switches:
if switches[item_id].enabled:
switches[item_id].async_schedule_update_ha_state()
continue
2020-04-20 08:39:02 +02:00
# Create new entity
sid_data = {
"sid": sid,
"sid_uid": sid_uid,
"sid_name": sid_name,
"sid_ref": sid_ref,
"sid_attr": sid_attr,
}
switches[item_id] = sid_func(inst, uid, mikrotik_controller, sid_data)
2019-12-05 22:10:42 +01:00
new_switches.append(switches[item_id])
if new_switches:
async_add_entities(new_switches)
# ---------------------------
2019-12-03 18:30:45 +01:00
# MikrotikControllerSwitch
# ---------------------------
class MikrotikControllerSwitch(SwitchEntity, RestoreEntity):
2019-12-06 01:22:34 +01:00
"""Representation of a switch."""
2020-04-20 08:39:02 +02:00
def __init__(self, inst, uid, mikrotik_controller, sid_data):
2020-12-25 20:28:36 +01:00
"""Initialize."""
2020-04-20 08:39:02 +02:00
self._sid_data = sid_data
2019-12-06 01:22:34 +01:00
self._inst = inst
self._ctrl = mikrotik_controller
2020-04-20 08:39:02 +02:00
self._data = mikrotik_controller.data[self._sid_data["sid"]][uid]
self._attrs = {
ATTR_ATTRIBUTION: ATTRIBUTION,
}
2019-12-03 18:30:45 +01:00
async def async_added_to_hass(self):
2020-12-25 20:28:36 +01:00
"""Run when entity about to be added to hass."""
2020-04-20 08:39:02 +02:00
_LOGGER.debug(
"New switch %s (%s %s)",
self._inst,
self._sid_data["sid"],
self._data[self._sid_data["sid_uid"]],
)
2019-12-03 18:30:45 +01:00
async def async_update(self):
"""Synchronize state with controller."""
2019-12-03 18:30:45 +01:00
@property
def available(self) -> bool:
"""Return if controller is available."""
2019-12-06 01:22:34 +01:00
return self._ctrl.connected()
2019-12-03 18:30:45 +01:00
2020-04-20 08:39:02 +02:00
@property
def name(self) -> str:
2020-12-25 20:28:36 +01:00
"""Return the name."""
2020-04-20 08:39:02 +02:00
return f"{self._inst} {self._sid_data['sid']} {self._data[self._sid_data['sid_name']]}"
@property
def unique_id(self) -> str:
2020-12-25 20:28:36 +01:00
"""Return a unique id for this entity."""
2020-04-20 08:39:02 +02:00
return f"{self._inst.lower()}-{self._sid_data['sid']}_switch-{self._data[self._sid_data['sid_ref']]}"
@property
2020-12-25 20:28:36 +01:00
def is_on(self) -> bool:
"""Return true if device is on."""
2020-04-20 08:39:02 +02:00
return self._data["enabled"]
@property
def extra_state_attributes(self) -> Dict[str, Any]:
2020-12-25 20:28:36 +01:00
"""Return the state attributes."""
2020-04-20 08:39:02 +02:00
attributes = self._attrs
for variable in self._sid_data["sid_attr"]:
if variable in self._data:
attributes[format_attribute(variable)] = self._data[variable]
return attributes
2020-12-25 23:31:51 +01:00
def turn_off(self, **kwargs: Any) -> None:
"""Required abstract method."""
pass
def turn_on(self, **kwargs: Any) -> None:
"""Required abstract method."""
pass
2019-12-03 18:30:45 +01:00
# ---------------------------
# MikrotikControllerPortSwitch
# ---------------------------
class MikrotikControllerPortSwitch(MikrotikControllerSwitch):
"""Representation of a network port switch."""
2020-04-20 08:39:02 +02:00
def __init__(self, inst, uid, mikrotik_controller, sid_data):
2020-12-25 20:28:36 +01:00
"""Initialize."""
2020-04-20 08:39:02 +02:00
super().__init__(inst, uid, mikrotik_controller, sid_data)
@property
def name(self) -> str:
2020-12-25 20:28:36 +01:00
"""Return the name."""
2020-04-20 08:39:02 +02:00
return f"{self._inst} port {self._data[self._sid_data['sid_name']]}"
@property
def unique_id(self) -> str:
2020-12-25 20:28:36 +01:00
"""Return a unique id for this entity."""
2020-12-02 11:43:10 +01:00
return f"{self._inst.lower()}-enable_switch-{self._data['port-mac-address']}_{self._data['default-name']}"
2021-04-12 14:28:39 +02:00
@property
def extra_state_attributes(self) -> Dict[str, Any]:
2021-04-12 14:28:39 +02:00
"""Return the state attributes."""
attributes = self._attrs
if self._data["type"] == "ether":
for variable in DEVICE_ATTRIBUTES_IFACE_ETHER:
if variable in self._data:
attributes[format_attribute(variable)] = self._data[variable]
2021-04-12 14:28:39 +02:00
if "sfp-shutdown-temperature" in self._data:
for variable in DEVICE_ATTRIBUTES_IFACE_SFP:
if variable in self._data:
attributes[format_attribute(variable)] = self._data[variable]
else:
for variable in self._sid_data["sid_attr"]:
2021-04-12 14:28:39 +02:00
if variable in self._data:
attributes[format_attribute(variable)] = self._data[variable]
return attributes
@property
2020-12-25 20:28:36 +01:00
def icon(self) -> str:
"""Return the icon."""
2020-03-16 04:51:41 +01:00
if self._data["running"]:
icon = "mdi:lan-connect"
else:
2020-03-16 04:51:41 +01:00
icon = "mdi:lan-pending"
2020-03-16 04:51:41 +01:00
if not self._data["enabled"]:
icon = "mdi:lan-disconnect"
return icon
@property
2020-12-25 20:28:36 +01:00
def device_info(self) -> Dict[str, Any]:
"""Return a description for device registry."""
info = {
2020-04-11 05:45:36 +02:00
"connections": {(CONNECTION_NETWORK_MAC, self._data["port-mac-address"])},
2020-03-16 04:51:41 +01:00
"manufacturer": self._ctrl.data["resource"]["platform"],
"model": self._ctrl.data["resource"]["board-name"],
"name": f"{self._inst} {self._data['default-name']}",
}
return info
2020-12-25 23:31:51 +01:00
async def async_turn_on(self) -> Optional[str]:
"""Turn on the switch."""
2020-03-16 04:51:41 +01:00
path = "/interface"
param = "default-name"
if self._data["about"] == "managed by CAPsMAN":
_LOGGER.error("Unable to enable %s, managed by CAPsMAN", self._data[param])
return "managed by CAPsMAN"
if "-" in self._data["port-mac-address"]:
param = "name"
2019-12-06 01:22:34 +01:00
value = self._data[param]
2020-03-16 04:51:41 +01:00
mod_param = "disabled"
mod_value = False
2019-12-06 01:22:34 +01:00
self._ctrl.set_value(path, param, value, mod_param, mod_value)
if "poe-out" in self._data and self._data["poe-out"] == "off":
path = "/interface/ethernet"
self._ctrl.set_value(path, param, value, "poe-out", "auto-on")
2019-12-06 01:22:34 +01:00
await self._ctrl.force_update()
2020-12-25 23:31:51 +01:00
async def async_turn_off(self) -> Optional[str]:
2020-12-25 20:28:36 +01:00
"""Turn off the switch."""
2020-03-16 04:51:41 +01:00
path = "/interface"
param = "default-name"
if self._data["about"] == "managed by CAPsMAN":
_LOGGER.error("Unable to disable %s, managed by CAPsMAN", self._data[param])
return "managed by CAPsMAN"
if "-" in self._data["port-mac-address"]:
param = "name"
2019-12-06 01:22:34 +01:00
value = self._data[param]
2020-03-16 04:51:41 +01:00
mod_param = "disabled"
mod_value = True
2019-12-06 01:22:34 +01:00
self._ctrl.set_value(path, param, value, mod_param, mod_value)
if "poe-out" in self._data and self._data["poe-out"] == "auto-on":
path = "/interface/ethernet"
self._ctrl.set_value(path, param, value, "poe-out", "off")
2019-12-06 01:22:34 +01:00
await self._ctrl.async_update()
2020-12-02 15:38:17 +01:00
2019-12-03 18:30:45 +01:00
# ---------------------------
# MikrotikControllerNATSwitch
# ---------------------------
class MikrotikControllerNATSwitch(MikrotikControllerSwitch):
"""Representation of a NAT switch."""
2020-04-20 08:39:02 +02:00
def __init__(self, inst, uid, mikrotik_controller, sid_data):
2020-12-25 20:28:36 +01:00
"""Initialize."""
2020-04-20 08:39:02 +02:00
super().__init__(inst, uid, mikrotik_controller, sid_data)
2019-12-03 18:30:45 +01:00
@property
def name(self) -> str:
2020-12-25 20:28:36 +01:00
"""Return the name."""
if self._data["comment"]:
return f"{self._inst} NAT {self._data['comment']}"
return f"{self._inst} NAT {self._data['name']}"
@property
def unique_id(self) -> str:
2020-12-25 20:28:36 +01:00
"""Return a unique id for this entity."""
return f"{self._inst.lower()}-enable_nat-{self._data['uniq-id']}"
2019-12-03 18:30:45 +01:00
@property
2020-12-25 20:28:36 +01:00
def icon(self) -> str:
2019-12-03 18:30:45 +01:00
"""Return the icon."""
2020-03-16 04:51:41 +01:00
if not self._data["enabled"]:
icon = "mdi:network-off-outline"
2019-12-03 18:30:45 +01:00
else:
2020-03-16 04:51:41 +01:00
icon = "mdi:network-outline"
2019-12-03 18:30:45 +01:00
return icon
2019-12-03 18:30:45 +01:00
@property
2020-12-25 20:28:36 +01:00
def device_info(self) -> Dict[str, Any]:
"""Return a description for device registry."""
2019-12-03 18:30:45 +01:00
info = {
2020-03-16 04:51:41 +01:00
"identifiers": {
(
DOMAIN,
"serial-number",
2022-01-21 15:50:25 +01:00
f"{self._ctrl.data['routerboard']['serial-number']}",
2020-03-16 04:51:41 +01:00
"switch",
"NAT",
)
},
"manufacturer": self._ctrl.data["resource"]["platform"],
"model": self._ctrl.data["resource"]["board-name"],
"name": f"{self._inst} NAT",
2019-12-03 18:30:45 +01:00
}
return info
2020-12-25 20:28:36 +01:00
async def async_turn_on(self) -> None:
2019-12-03 18:30:45 +01:00
"""Turn on the switch."""
2020-03-16 04:51:41 +01:00
path = "/ip/firewall/nat"
param = ".id"
value = None
2020-03-16 04:51:41 +01:00
for uid in self._ctrl.data["nat"]:
2020-12-25 23:31:51 +01:00
if self._ctrl.data["nat"][uid]["uniq-id"] == (
f"{self._data['chain']},{self._data['action']},{self._data['protocol']},"
f"{self._data['in-interface']}:{self._data['dst-port']}-"
f"{self._data['out-interface']}:{self._data['to-addresses']}:{self._data['to-ports']}"
2020-03-16 04:51:41 +01:00
):
value = self._ctrl.data["nat"][uid][".id"]
2020-03-16 04:51:41 +01:00
mod_param = "disabled"
2019-12-03 18:30:45 +01:00
mod_value = False
2019-12-06 01:22:34 +01:00
self._ctrl.set_value(path, param, value, mod_param, mod_value)
await self._ctrl.force_update()
2019-12-03 18:30:45 +01:00
2020-12-25 20:28:36 +01:00
async def async_turn_off(self) -> None:
"""Turn off the switch."""
2020-03-16 04:51:41 +01:00
path = "/ip/firewall/nat"
param = ".id"
value = None
2020-03-16 04:51:41 +01:00
for uid in self._ctrl.data["nat"]:
2020-12-25 23:31:51 +01:00
if self._ctrl.data["nat"][uid]["uniq-id"] == (
f"{self._data['chain']},{self._data['action']},{self._data['protocol']},"
f"{self._data['in-interface']}:{self._data['dst-port']}-"
f"{self._data['out-interface']}:{self._data['to-addresses']}:{self._data['to-ports']}"
2020-03-16 04:51:41 +01:00
):
value = self._ctrl.data["nat"][uid][".id"]
2020-03-16 04:51:41 +01:00
mod_param = "disabled"
2019-12-03 18:30:45 +01:00
mod_value = True
2019-12-06 01:22:34 +01:00
self._ctrl.set_value(path, param, value, mod_param, mod_value)
await self._ctrl.async_update()
2019-12-03 18:30:45 +01:00
2019-12-04 20:13:11 +01:00
2020-12-18 19:58:54 +01:00
# ---------------------------
# MikrotikControllerMangleSwitch
# ---------------------------
class MikrotikControllerMangleSwitch(MikrotikControllerSwitch):
"""Representation of a Mangle switch."""
def __init__(self, inst, uid, mikrotik_controller, sid_data):
2020-12-25 20:28:36 +01:00
"""Initialize."""
2020-12-18 19:58:54 +01:00
super().__init__(inst, uid, mikrotik_controller, sid_data)
@property
def name(self) -> str:
2020-12-25 20:28:36 +01:00
"""Return the name."""
if self._data["comment"]:
return f"{self._inst} Mangle {self._data['comment']}"
2020-12-18 19:58:54 +01:00
return f"{self._inst} Mangle {self._data['name']}"
@property
def unique_id(self) -> str:
2020-12-25 20:28:36 +01:00
"""Return a unique id for this entity."""
return f"{self._inst.lower()}-enable_mangle-{self._data['uniq-id']}"
2020-12-18 19:58:54 +01:00
@property
2020-12-25 20:28:36 +01:00
def icon(self) -> str:
2020-12-18 19:58:54 +01:00
"""Return the icon."""
if not self._data["enabled"]:
icon = "mdi:bookmark-off-outline"
else:
icon = "mdi:bookmark-outline"
return icon
@property
2020-12-25 20:28:36 +01:00
def device_info(self) -> Dict[str, Any]:
"""Return a description for device registry."""
2020-12-18 19:58:54 +01:00
info = {
"identifiers": {
(
DOMAIN,
"serial-number",
2022-01-21 15:50:25 +01:00
f"{self._ctrl.data['routerboard']['serial-number']}",
2020-12-18 19:58:54 +01:00
"switch",
"Mangle",
)
},
"manufacturer": self._ctrl.data["resource"]["platform"],
"model": self._ctrl.data["resource"]["board-name"],
"name": f"{self._inst} Mangle",
}
return info
2020-12-25 20:28:36 +01:00
async def async_turn_on(self) -> None:
2020-12-18 19:58:54 +01:00
"""Turn on the switch."""
path = "/ip/firewall/mangle"
param = ".id"
value = None
for uid in self._ctrl.data["mangle"]:
2020-12-25 23:31:51 +01:00
if self._ctrl.data["mangle"][uid]["uniq-id"] == (
f"{self._data['chain']},{self._data['action']},{self._data['protocol']},"
f"{self._data['src-address']}:{self._data['src-port']}-"
2021-08-24 18:24:41 +07:00
f"{self._data['dst-address']}:{self._data['dst-port']},"
2021-08-24 19:24:50 +07:00
f"{self._data['src-address-list']}-{self._data['dst-address-list']}"
2020-12-18 19:58:54 +01:00
):
value = self._ctrl.data["mangle"][uid][".id"]
mod_param = "disabled"
mod_value = False
self._ctrl.set_value(path, param, value, mod_param, mod_value)
await self._ctrl.force_update()
2020-12-25 20:28:36 +01:00
async def async_turn_off(self) -> None:
"""Turn off the switch."""
2020-12-18 19:58:54 +01:00
path = "/ip/firewall/mangle"
param = ".id"
value = None
for uid in self._ctrl.data["mangle"]:
2020-12-25 23:31:51 +01:00
if self._ctrl.data["mangle"][uid]["uniq-id"] == (
f"{self._data['chain']},{self._data['action']},{self._data['protocol']},"
f"{self._data['src-address']}:{self._data['src-port']}-"
2021-08-24 18:24:41 +07:00
f"{self._data['dst-address']}:{self._data['dst-port']},"
2021-08-24 19:24:50 +07:00
f"{self._data['src-address-list']}-{self._data['dst-address-list']}"
2020-12-18 19:58:54 +01:00
):
value = self._ctrl.data["mangle"][uid][".id"]
mod_param = "disabled"
mod_value = True
self._ctrl.set_value(path, param, value, mod_param, mod_value)
await self._ctrl.async_update()
2021-04-12 12:40:45 +02:00
# ---------------------------
# MikrotikControllerFilterSwitch
# ---------------------------
class MikrotikControllerFilterSwitch(MikrotikControllerSwitch):
"""Representation of a Filter switch."""
def __init__(self, inst, uid, mikrotik_controller, sid_data):
"""Initialize."""
super().__init__(inst, uid, mikrotik_controller, sid_data)
@property
def name(self) -> str:
"""Return the name."""
if self._data["comment"]:
return f"{self._inst} Filter {self._data['comment']}"
return f"{self._inst} Filter {self._data['name']}"
@property
def unique_id(self) -> str:
"""Return a unique id for this entity."""
return f"{self._inst.lower()}-enable_filter-{self._data['uniq-id']}"
@property
def icon(self) -> str:
"""Return the icon."""
if not self._data["enabled"]:
icon = "mdi:filter-variant-remove"
else:
icon = "mdi:filter-variant"
return icon
@property
def device_info(self) -> Dict[str, Any]:
"""Return a description for device registry."""
info = {
"identifiers": {
(
DOMAIN,
"serial-number",
2022-01-21 15:50:25 +01:00
f"{self._ctrl.data['routerboard']['serial-number']}",
2021-04-12 12:40:45 +02:00
"switch",
"Filter",
)
},
"manufacturer": self._ctrl.data["resource"]["platform"],
"model": self._ctrl.data["resource"]["board-name"],
"name": f"{self._inst} Filter",
}
return info
async def async_turn_on(self) -> None:
"""Turn on the switch."""
path = "/ip/firewall/filter"
param = ".id"
value = None
for uid in self._ctrl.data["filter"]:
if self._ctrl.data["filter"][uid]["uniq-id"] == (
f"{self._data['chain']},{self._data['action']},{self._data['protocol']},{self._data['layer7-protocol']},"
f"{self._data['in-interface']}:{self._data['src-address']}:{self._data['src-port']}-"
f"{self._data['out-interface']}:{self._data['dst-address']}:{self._data['dst-port']}"
):
value = self._ctrl.data["filter"][uid][".id"]
mod_param = "disabled"
mod_value = False
self._ctrl.set_value(path, param, value, mod_param, mod_value)
await self._ctrl.force_update()
async def async_turn_off(self) -> None:
"""Turn off the switch."""
path = "/ip/firewall/filter"
param = ".id"
value = None
for uid in self._ctrl.data["filter"]:
if self._ctrl.data["filter"][uid]["uniq-id"] == (
f"{self._data['chain']},{self._data['action']},{self._data['protocol']},{self._data['layer7-protocol']},"
f"{self._data['in-interface']}:{self._data['src-address']}:{self._data['src-port']}-"
f"{self._data['out-interface']}:{self._data['dst-address']}:{self._data['dst-port']}"
):
value = self._ctrl.data["filter"][uid][".id"]
mod_param = "disabled"
mod_value = True
self._ctrl.set_value(path, param, value, mod_param, mod_value)
await self._ctrl.async_update()
2020-12-25 11:48:24 +01:00
# ---------------------------
# MikrotikControllerPPPSecretSwitch
# ---------------------------
class MikrotikControllerPPPSecretSwitch(MikrotikControllerSwitch):
"""Representation of a PPP Secret switch."""
def __init__(self, inst, uid, mikrotik_controller, sid_data):
2020-12-25 20:28:36 +01:00
"""Initialize."""
2020-12-25 11:48:24 +01:00
super().__init__(inst, uid, mikrotik_controller, sid_data)
@property
def name(self) -> str:
2020-12-25 20:28:36 +01:00
"""Return the name."""
2020-12-25 11:48:24 +01:00
return f"{self._inst} PPP Secret {self._data['name']}"
@property
def unique_id(self) -> str:
2020-12-25 20:28:36 +01:00
"""Return a unique id for this entity."""
2020-12-25 11:48:24 +01:00
return f"{self._inst.lower()}-enable_ppp_secret-{self._data['name']}"
@property
2020-12-25 20:28:36 +01:00
def icon(self) -> str:
2020-12-25 11:48:24 +01:00
"""Return the icon."""
if not self._data["enabled"]:
icon = "mdi:account-off-outline"
else:
icon = "mdi:account-outline"
return icon
@property
2020-12-25 20:28:36 +01:00
def device_info(self) -> Dict[str, Any]:
"""Return a description for device registry."""
2020-12-25 11:48:24 +01:00
info = {
"identifiers": {
(
DOMAIN,
"serial-number",
2022-01-21 15:50:25 +01:00
f"{self._ctrl.data['routerboard']['serial-number']}",
2020-12-25 11:48:24 +01:00
"switch",
2020-12-25 15:30:54 +01:00
"PPP",
2020-12-25 11:48:24 +01:00
)
},
"manufacturer": self._ctrl.data["resource"]["platform"],
"model": self._ctrl.data["resource"]["board-name"],
2020-12-25 15:30:54 +01:00
"name": f"{self._inst} PPP",
2020-12-25 11:48:24 +01:00
}
return info
2020-12-25 20:28:36 +01:00
async def async_turn_on(self) -> None:
2020-12-25 11:48:24 +01:00
"""Turn on the switch."""
path = "/ppp/secret"
param = "name"
value = self._data["name"]
mod_param = "disabled"
mod_value = False
self._ctrl.set_value(path, param, value, mod_param, mod_value)
await self._ctrl.force_update()
2020-12-25 20:28:36 +01:00
async def async_turn_off(self) -> None:
"""Turn off the switch."""
2020-12-25 11:48:24 +01:00
path = "/ppp/secret"
param = "name"
value = self._data["name"]
mod_param = "disabled"
mod_value = True
self._ctrl.set_value(path, param, value, mod_param, mod_value)
await self._ctrl.async_update()
# ---------------------------
# MikrotikControllerQueueSwitch
# ---------------------------
class MikrotikControllerQueueSwitch(MikrotikControllerSwitch):
"""Representation of a queue switch."""
2020-04-20 08:39:02 +02:00
def __init__(self, inst, uid, mikrotik_controller, sid_data):
2020-12-25 20:28:36 +01:00
"""Initialize."""
2020-04-20 08:39:02 +02:00
super().__init__(inst, uid, mikrotik_controller, sid_data)
@property
2020-12-25 20:28:36 +01:00
def icon(self) -> str:
"""Return the icon."""
if not self._data["enabled"]:
icon = "mdi:leaf-off"
else:
icon = "mdi:leaf"
return icon
@property
2020-12-25 20:28:36 +01:00
def device_info(self) -> Dict[str, Any]:
"""Return a description for device registry."""
info = {
"identifiers": {
(
DOMAIN,
"serial-number",
2022-01-21 15:50:25 +01:00
f"{self._ctrl.data['routerboard']['serial-number']}",
"switch",
"Queue",
)
},
"manufacturer": self._ctrl.data["resource"]["platform"],
"model": self._ctrl.data["resource"]["board-name"],
"name": f"{self._inst} Queue",
}
return info
2020-12-25 20:28:36 +01:00
async def async_turn_on(self) -> None:
"""Turn on the switch."""
path = "/queue/simple"
param = ".id"
value = None
for uid in self._ctrl.data["queue"]:
2020-04-11 05:45:36 +02:00
if self._ctrl.data["queue"][uid]["name"] == f"{self._data['name']}":
value = self._ctrl.data["queue"][uid][".id"]
mod_param = "disabled"
mod_value = False
self._ctrl.set_value(path, param, value, mod_param, mod_value)
await self._ctrl.force_update()
2020-12-25 20:28:36 +01:00
async def async_turn_off(self) -> None:
"""Turn off the switch."""
path = "/queue/simple"
param = ".id"
value = None
for uid in self._ctrl.data["queue"]:
2020-04-11 05:45:36 +02:00
if self._ctrl.data["queue"][uid]["name"] == f"{self._data['name']}":
value = self._ctrl.data["queue"][uid][".id"]
mod_param = "disabled"
mod_value = True
self._ctrl.set_value(path, param, value, mod_param, mod_value)
await self._ctrl.async_update()
2020-12-25 18:42:57 +01:00
# ---------------------------
# MikrotikControllerKidcontrolSwitch
# ---------------------------
class MikrotikControllerKidcontrolSwitch(MikrotikControllerSwitch):
"""Representation of a queue switch."""
def __init__(self, inst, uid, mikrotik_controller, sid_data):
2020-12-25 20:28:36 +01:00
"""Initialize."""
2020-12-25 18:42:57 +01:00
super().__init__(inst, uid, mikrotik_controller, sid_data)
@property
def icon(self) -> str:
"""Return the icon."""
if not self._data["enabled"]:
icon = "mdi:account-off"
else:
icon = "mdi:account"
return icon
@property
def device_info(self) -> Dict[str, Any]:
"""Return a description for device registry."""
info = {
"identifiers": {
(
DOMAIN,
"serial-number",
2022-01-21 15:50:25 +01:00
f"{self._ctrl.data['routerboard']['serial-number']}",
"switch",
"Kidcontrol",
)
},
"manufacturer": self._ctrl.data["resource"]["platform"],
"model": self._ctrl.data["resource"]["board-name"],
"name": f"{self._inst} Kidcontrol",
}
return info
async def async_turn_on(self) -> None:
"""Turn on the switch."""
path = "/ip/kid-control"
param = "name"
value = self._data["name"]
mod_param = "disabled"
mod_value = False
self._ctrl.set_value(path, param, value, mod_param, mod_value)
await self._ctrl.force_update()
async def async_turn_off(self) -> None:
"""Turn off the switch."""
path = "/ip/kid-control"
param = "name"
value = self._data["name"]
mod_param = "disabled"
mod_value = True
self._ctrl.set_value(path, param, value, mod_param, mod_value)
await self._ctrl.async_update()
# ---------------------------
# MikrotikControllerKidcontrolPauseSwitch
# ---------------------------
class MikrotikControllerKidcontrolPauseSwitch(MikrotikControllerSwitch):
"""Representation of a queue switch."""
def __init__(self, inst, uid, mikrotik_controller, sid_data):
"""Initialize."""
super().__init__(inst, uid, mikrotik_controller, sid_data)
@property
def name(self) -> str:
"""Return the name."""
return f"{self._inst} {self._sid_data['sid']} Pause {self._data[self._sid_data['sid_name']]}"
@property
def unique_id(self) -> str:
"""Return a unique id for this entity."""
return f"{self._inst.lower()}-{self._sid_data['sid']}-pause_switch-{self._data[self._sid_data['sid_ref']]}"
@property
def is_on(self) -> bool:
"""Return true if device is on."""
return self._data["paused"]
2020-12-25 18:42:57 +01:00
@property
2020-12-25 20:28:36 +01:00
def icon(self) -> str:
2020-12-25 18:42:57 +01:00
"""Return the icon."""
if not self._data["enabled"]:
icon = "mdi:account-off-outline"
else:
icon = "mdi:account-outline"
return icon
@property
2020-12-25 20:28:36 +01:00
def device_info(self) -> Dict[str, Any]:
"""Return a description for device registry."""
2020-12-25 18:42:57 +01:00
info = {
"identifiers": {
(
DOMAIN,
"serial-number",
2022-01-21 15:50:25 +01:00
f"{self._ctrl.data['routerboard']['serial-number']}",
2020-12-25 18:42:57 +01:00
"switch",
"Kidcontrol",
)
},
"manufacturer": self._ctrl.data["resource"]["platform"],
"model": self._ctrl.data["resource"]["board-name"],
"name": f"{self._inst} Kidcontrol",
}
return info
2020-12-25 20:28:36 +01:00
async def async_turn_on(self) -> None:
"""Turn on the switch."""
2020-12-25 18:42:57 +01:00
path = "/ip/kid-control"
param = "name"
value = self._data["name"]
command = "resume"
self._ctrl.execute(path, command, param, value)
2020-12-25 18:42:57 +01:00
await self._ctrl.force_update()
2020-12-25 20:28:36 +01:00
async def async_turn_off(self) -> None:
"""Turn off the switch."""
2020-12-25 18:42:57 +01:00
path = "/ip/kid-control"
param = "name"
value = self._data["name"]
command = "pause"
self._ctrl.execute(path, command, param, value)
2020-12-25 18:42:57 +01:00
await self._ctrl.async_update()