mirror.MikroTik/tomaae.homeassistant-mikrotik_router
1
0
Fork 0
mirror of https://github.com/tomaae/homeassistant-mikrotik_router.git synced 2025-07-10 17:34:29 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

refactored switches

Browse source
This commit is contained in:
tomaae 2020-04-20 08:39:02 +02:00
parent 480ef5988c
commit 5d01524e0e
1 changed files with 75 additions and 141 deletions
Download patch file Download diff file Expand all files Collapse all files

216
custom_components/mikrotik_router/switch.py
View file

@ -104,8 +104,23 @@ def update_items(inst, mikrotik_controller, async_add_entities, switches):
new_switches = [] new_switches = []
# Add switches # Add switches
for sid, sid_func in zip( for sid, sid_uid, sid_name, sid_ref, sid_attr, sid_func in zip(
# Data point name
["interface", "nat", "script", "queue"], ["interface", "nat", "script", "queue"],
# Data point unique id
["name", "name", "name", "name"],
# Entry Name
["name", "name", "name", "name"],
# Entry Unique id
["port-mac-address", "name", "name", "name"],
# Attr
[
DEVICE_ATTRIBUTES_IFACE,
DEVICE_ATTRIBUTES_NAT,
DEVICE_ATTRIBUTES_SCRIPT,
DEVICE_ATTRIBUTES_QUEUE,
],
# Switch function
[ [
MikrotikControllerPortSwitch, MikrotikControllerPortSwitch,
MikrotikControllerNATSwitch, MikrotikControllerNATSwitch,
@ -114,14 +129,22 @@ def update_items(inst, mikrotik_controller, async_add_entities, switches):
], ],
): ):
for uid in mikrotik_controller.data[sid]: for uid in mikrotik_controller.data[sid]:
item_id = f"{inst}-{sid}-{mikrotik_controller.data[sid][uid]['name']}" item_id = f"{inst}-{sid}-{mikrotik_controller.data[sid][uid][sid_uid]}"
_LOGGER.debug("Updating switch %s", item_id) _LOGGER.debug("Updating switch %s", item_id)
if item_id in switches: if item_id in switches:
if switches[item_id].enabled: if switches[item_id].enabled:
switches[item_id].async_schedule_update_ha_state() switches[item_id].async_schedule_update_ha_state()
continue continue
switches[item_id] = sid_func(inst, uid, mikrotik_controller) # 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)
new_switches.append(switches[item_id]) new_switches.append(switches[item_id])
if new_switches: if new_switches:
@ -134,15 +157,25 @@ def update_items(inst, mikrotik_controller, async_add_entities, switches):
class MikrotikControllerSwitch(SwitchDevice, RestoreEntity): class MikrotikControllerSwitch(SwitchDevice, RestoreEntity):
"""Representation of a switch.""" """Representation of a switch."""
def __init__(self, inst, uid, mikrotik_controller): def __init__(self, inst, uid, mikrotik_controller, sid_data):
"""Set up switch.""" """Set up switch."""
self._sid_data = sid_data
self._inst = inst self._inst = inst
self._uid = uid
self._ctrl = mikrotik_controller self._ctrl = mikrotik_controller
self._data = mikrotik_controller.data[self._sid_data["sid"]][uid]
self._attrs = {
ATTR_ATTRIBUTION: ATTRIBUTION,
}
async def async_added_to_hass(self): async def async_added_to_hass(self):
"""Switch entity created.""" """Device tracker entity created."""
_LOGGER.debug("New switch %s (%s)", self._inst, self._uid) _LOGGER.debug(
"New switch %s (%s %s)",
self._inst,
self._sid_data["sid"],
self._data[self._sid_data["sid_uid"]],
)
async def async_update(self): async def async_update(self):
"""Synchronize state with controller.""" """Synchronize state with controller."""
@ -152,6 +185,32 @@ class MikrotikControllerSwitch(SwitchDevice, RestoreEntity):
"""Return if controller is available.""" """Return if controller is available."""
return self._ctrl.connected() return self._ctrl.connected()
@property
def name(self) -> str:
"""Return the name of the port."""
return f"{self._inst} {self._sid_data['sid']} {self._data[self._sid_data['sid_name']]}"
@property
def unique_id(self) -> str:
"""Return a unique identifier for this port."""
return f"{self._inst.lower()}-{self._sid_data['sid']}_switch-{self._data[self._sid_data['sid_ref']]}"
@property
def is_on(self):
"""Return true if the queue is on."""
return self._data["enabled"]
@property
def device_state_attributes(self):
"""Return the port state attributes."""
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
# --------------------------- # ---------------------------
# MikrotikControllerPortSwitch # MikrotikControllerPortSwitch
@ -159,28 +218,14 @@ class MikrotikControllerSwitch(SwitchDevice, RestoreEntity):
class MikrotikControllerPortSwitch(MikrotikControllerSwitch): class MikrotikControllerPortSwitch(MikrotikControllerSwitch):
"""Representation of a network port switch.""" """Representation of a network port switch."""
def __init__(self, inst, uid, mikrotik_controller): def __init__(self, inst, uid, mikrotik_controller, sid_data):
"""Set up tracked port.""" """Set up tracked port."""
super().__init__(inst, uid, mikrotik_controller) super().__init__(inst, uid, mikrotik_controller, sid_data)
self._data = mikrotik_controller.data["interface"][self._uid]
self._attrs = {
ATTR_ATTRIBUTION: ATTRIBUTION,
}
async def async_added_to_hass(self):
"""Port entity created."""
_LOGGER.debug(
"New port switch %s (%s %s)",
self._inst,
self._data["default-name"],
self._data["port-mac-address"],
)
@property @property
def name(self) -> str: def name(self) -> str:
"""Return the name of the port.""" """Return the name of the port."""
return f"{self._inst} port {self._data['default-name']}" return f"{self._inst} port {self._data[self._sid_data['sid_name']]}"
@property @property
def unique_id(self) -> str: def unique_id(self) -> str:
@ -211,17 +256,6 @@ class MikrotikControllerPortSwitch(MikrotikControllerSwitch):
} }
return info return info
@property
def device_state_attributes(self):
"""Return the port state attributes."""
attributes = self._attrs
for variable in DEVICE_ATTRIBUTES_IFACE:
if variable in self._data:
attributes[format_attribute(variable)] = self._data[variable]
return attributes
async def async_turn_on(self): async def async_turn_on(self):
"""Turn on the switch.""" """Turn on the switch."""
path = "/interface" path = "/interface"
@ -246,11 +280,6 @@ class MikrotikControllerPortSwitch(MikrotikControllerSwitch):
self._ctrl.set_value(path, param, value, mod_param, mod_value) self._ctrl.set_value(path, param, value, mod_param, mod_value)
await self._ctrl.async_update() await self._ctrl.async_update()
@property
def is_on(self):
"""Return true if device is on."""
return self._data["enabled"]
# --------------------------- # ---------------------------
# MikrotikControllerNATSwitch # MikrotikControllerNATSwitch
@ -258,29 +287,15 @@ class MikrotikControllerPortSwitch(MikrotikControllerSwitch):
class MikrotikControllerNATSwitch(MikrotikControllerSwitch): class MikrotikControllerNATSwitch(MikrotikControllerSwitch):
"""Representation of a NAT switch.""" """Representation of a NAT switch."""
def __init__(self, inst, uid, mikrotik_controller): def __init__(self, inst, uid, mikrotik_controller, sid_data):
"""Set up NAT switch.""" """Set up NAT switch."""
super().__init__(inst, uid, mikrotik_controller) super().__init__(inst, uid, mikrotik_controller, sid_data)
self._data = mikrotik_controller.data["nat"][self._uid]
self._attrs = {
ATTR_ATTRIBUTION: ATTRIBUTION,
}
async def async_added_to_hass(self):
"""NAT switch entity created."""
_LOGGER.debug("New port switch %s (%s)", self._inst, self._data["name"])
@property @property
def name(self) -> str: def name(self) -> str:
"""Return the name of the NAT switch.""" """Return the name of the NAT switch."""
return f"{self._inst} NAT {self._data['name']}" return f"{self._inst} NAT {self._data['name']}"
@property
def unique_id(self) -> str:
"""Return a unique identifier for this NAT switch."""
return f"{self._inst.lower()}-nat_switch-{self._data['name']}"
@property @property
def icon(self): def icon(self):
"""Return the icon.""" """Return the icon."""
@ -310,17 +325,6 @@ class MikrotikControllerNATSwitch(MikrotikControllerSwitch):
} }
return info return info
@property
def device_state_attributes(self):
"""Return the NAT switch state attributes."""
attributes = self._attrs
for variable in DEVICE_ATTRIBUTES_NAT:
if variable in self._data:
attributes[format_attribute(variable)] = self._data[variable]
return attributes
async def async_turn_on(self): async def async_turn_on(self):
"""Turn on the switch.""" """Turn on the switch."""
path = "/ip/firewall/nat" path = "/ip/firewall/nat"
@ -355,11 +359,6 @@ class MikrotikControllerNATSwitch(MikrotikControllerSwitch):
self._ctrl.set_value(path, param, value, mod_param, mod_value) self._ctrl.set_value(path, param, value, mod_param, mod_value)
await self._ctrl.async_update() await self._ctrl.async_update()
@property
def is_on(self):
"""Return true if device is on."""
return self._data["enabled"]
# --------------------------- # ---------------------------
# MikrotikControllerScriptSwitch # MikrotikControllerScriptSwitch
@ -367,28 +366,9 @@ class MikrotikControllerNATSwitch(MikrotikControllerSwitch):
class MikrotikControllerScriptSwitch(MikrotikControllerSwitch): class MikrotikControllerScriptSwitch(MikrotikControllerSwitch):
"""Representation of a script switch.""" """Representation of a script switch."""
def __init__(self, inst, uid, mikrotik_controller): def __init__(self, inst, uid, mikrotik_controller, sid_data):
"""Set up script switch.""" """Set up script switch."""
super().__init__(inst, uid, mikrotik_controller) super().__init__(inst, uid, mikrotik_controller, sid_data)
self._data = mikrotik_controller.data["script"][self._uid]
self._attrs = {
ATTR_ATTRIBUTION: ATTRIBUTION,
}
async def async_added_to_hass(self):
"""Script switch entity created."""
_LOGGER.debug("New script switch %s (%s)", self._inst, self._data["name"])
@property
def name(self) -> str:
"""Return the name of the script switch."""
return f"{self._inst} script {self._data['name']}"
@property
def unique_id(self) -> str:
"""Return a unique identifier for this script switch."""
return f"{self._inst.lower()}-script_switch-{self._data['name']}"
@property @property
def icon(self): def icon(self):
@ -414,17 +394,6 @@ class MikrotikControllerScriptSwitch(MikrotikControllerSwitch):
} }
return info return info
@property
def device_state_attributes(self):
"""Return the script switch state attributes."""
attributes = self._attrs
for variable in DEVICE_ATTRIBUTES_SCRIPT:
if variable in self._data:
attributes[format_attribute(variable)] = self._data[variable]
return attributes
async def async_turn_on(self): async def async_turn_on(self):
"""Turn on the switch.""" """Turn on the switch."""
self._ctrl.run_script(self._data["name"]) self._ctrl.run_script(self._data["name"])
@ -445,28 +414,9 @@ class MikrotikControllerScriptSwitch(MikrotikControllerSwitch):
class MikrotikControllerQueueSwitch(MikrotikControllerSwitch): class MikrotikControllerQueueSwitch(MikrotikControllerSwitch):
"""Representation of a queue switch.""" """Representation of a queue switch."""
def __init__(self, inst, uid, mikrotik_controller): def __init__(self, inst, uid, mikrotik_controller, sid_data):
"""Set up queue switch.""" """Set up queue switch."""
super().__init__(inst, uid, mikrotik_controller) super().__init__(inst, uid, mikrotik_controller, sid_data)
self._data = mikrotik_controller.data["queue"][self._uid]
self._attrs = {
ATTR_ATTRIBUTION: ATTRIBUTION,
}
async def async_added_to_hass(self):
"""Queue switch entity created."""
_LOGGER.debug("New queue switch %s (%s)", self._inst, self._data["name"])
@property
def name(self) -> str:
"""Return the name of the queue switch."""
return f"{self._inst} Queue {self._data['name']}"
@property
def unique_id(self) -> str:
"""Return a unique identifier for this queue switch."""
return f"{self._inst.lower()}-queue_switch-{self._data['name']}"
@property @property
def icon(self): def icon(self):
@ -497,17 +447,6 @@ class MikrotikControllerQueueSwitch(MikrotikControllerSwitch):
} }
return info return info
@property
def device_state_attributes(self):
"""Return the queue switch state attributes."""
attributes = self._attrs
for variable in DEVICE_ATTRIBUTES_QUEUE:
if variable in self._data:
attributes[format_attribute(variable)] = self._data[variable]
return attributes
async def async_turn_on(self): async def async_turn_on(self):
"""Turn on the queue switch.""" """Turn on the queue switch."""
path = "/queue/simple" path = "/queue/simple"
@ -535,8 +474,3 @@ class MikrotikControllerQueueSwitch(MikrotikControllerSwitch):
mod_value = True mod_value = True
self._ctrl.set_value(path, param, value, mod_param, mod_value) self._ctrl.set_value(path, param, value, mod_param, mod_value)
await self._ctrl.async_update() await self._ctrl.async_update()
@property
def is_on(self):
"""Return true if the queue is on."""
return self._data["enabled"]