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

flake8 compliance

Browse source
This commit is contained in:
tomaae 2019-12-02 17:59:49 +01:00
parent d8b9294483
commit 0a6c8d369d
5 changed files with 81 additions and 82 deletions
Download patch file Download diff file Expand all files Collapse all files

18
custom_components/mikrotik_router/__init__.py
View file

@ -21,9 +21,9 @@ from .const import (
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
#--------------------------- # ---------------------------
# async_setup # async_setup
#--------------------------- # ---------------------------
async def async_setup(hass, config): async def async_setup(hass, config):
"""Set up configured Mikrotik Controller.""" """Set up configured Mikrotik Controller."""
hass.data[DOMAIN] = {} hass.data[DOMAIN] = {}
@ -31,9 +31,9 @@ async def async_setup(hass, config):
return True return True
#--------------------------- # ---------------------------
# async_setup_entry # async_setup_entry
#--------------------------- # ---------------------------
async def async_setup_entry(hass, config_entry): async def async_setup_entry(hass, config_entry):
"""Set up Mikrotik Router as config entry.""" """Set up Mikrotik Router as config entry."""
name = config_entry.data[CONF_NAME] name = config_entry.data[CONF_NAME]
@ -52,9 +52,9 @@ async def async_setup_entry(hass, config_entry):
hass.data[DOMAIN][DATA_CLIENT][config_entry.entry_id] = mikrotik_controller hass.data[DOMAIN][DATA_CLIENT][config_entry.entry_id] = mikrotik_controller
#hass.async_create_task( # hass.async_create_task(
# hass.config_entries.async_forward_entry_setup(config_entry, "sensor") # hass.config_entries.async_forward_entry_setup(config_entry, "sensor")
#) # )
hass.async_create_task( hass.async_create_task(
hass.config_entries.async_forward_entry_setup(config_entry, "device_tracker") hass.config_entries.async_forward_entry_setup(config_entry, "device_tracker")
@ -72,9 +72,9 @@ async def async_setup_entry(hass, config_entry):
return True return True
#--------------------------- # ---------------------------
# async_unload_entry # async_unload_entry
#--------------------------- # ---------------------------
async def async_unload_entry(hass, config_entry): async def async_unload_entry(hass, config_entry):
"""Unload a config entry.""" """Unload a config entry."""
mikrotik_controller = hass.data[DOMAIN][DATA_CLIENT][config_entry.entry_id] mikrotik_controller = hass.data[DOMAIN][DATA_CLIENT][config_entry.entry_id]

26
custom_components/mikrotik_router/config_flow.py
View file

@ -1,8 +1,6 @@
"""Config flow to configure Mikrotik Router.""" """Config flow to configure Mikrotik Router."""
import logging import logging
_LOGGER = logging.getLogger(__name__)
import voluptuous as vol import voluptuous as vol
from homeassistant import config_entries from homeassistant import config_entries
from homeassistant.core import callback from homeassistant.core import callback
@ -25,9 +23,11 @@ from .const import (
from .mikrotikapi import MikrotikAPI from .mikrotikapi import MikrotikAPI
#--------------------------- _LOGGER = logging.getLogger(__name__)
# ---------------------------
# configured_instances # configured_instances
#--------------------------- # ---------------------------
@callback @callback
def configured_instances(hass): def configured_instances(hass):
"""Return a set of configured instances.""" """Return a set of configured instances."""
@ -36,9 +36,9 @@ def configured_instances(hass):
) )
#--------------------------- # ---------------------------
# MikrotikControllerConfigFlow # MikrotikControllerConfigFlow
#--------------------------- # ---------------------------
class MikrotikControllerConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): class MikrotikControllerConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
def __init__(self): def __init__(self):
"""Initialize.""" """Initialize."""
@ -58,16 +58,16 @@ class MikrotikControllerConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a flow initialized by the user.""" """Handle a flow initialized by the user."""
errors = {} errors = {}
if user_input is not None: if user_input is not None:
## Check if instance with this name already exists # Check if instance with this name already exists
if user_input[CONF_NAME] in configured_instances(self.hass): if user_input[CONF_NAME] in configured_instances(self.hass):
errors["base"] = "name_exists" errors["base"] = "name_exists"
## Test connection # Test connection
api = MikrotikAPI(host=user_input["host"], username=user_input["username"], password=user_input["password"], port=user_input["port"], use_ssl=user_input["ssl"]) api = MikrotikAPI(host=user_input["host"], username=user_input["username"], password=user_input["password"], port=user_input["port"], use_ssl=user_input["ssl"])
if not api.connect(): if not api.connect():
errors[CONF_HOST] = api.error errors[CONF_HOST] = api.error
## Save instance # Save instance
if not errors: if not errors:
return self.async_create_entry( return self.async_create_entry(
title=user_input[CONF_NAME], title=user_input[CONF_NAME],
@ -78,9 +78,9 @@ class MikrotikControllerConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return self._show_config_form(errors=errors) return self._show_config_form(errors=errors)
#--------------------------- # ---------------------------
# _show_config_form # _show_config_form
#--------------------------- # ---------------------------
def _show_config_form(self, host='10.0.0.1', username='admin', password='admin', port=0, name='Mikrotik', use_ssl=False, errors=None): def _show_config_form(self, host='10.0.0.1', username='admin', password='admin', port=0, name='Mikrotik', use_ssl=False, errors=None):
"""Show the configuration form to edit data.""" """Show the configuration form to edit data."""
return self.async_show_form( return self.async_show_form(
@ -97,9 +97,9 @@ class MikrotikControllerConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
) )
#--------------------------- # ---------------------------
# MikrotikControllerOptionsFlowHandler # MikrotikControllerOptionsFlowHandler
#--------------------------- # ---------------------------
class MikrotikControllerOptionsFlowHandler(config_entries.OptionsFlow): class MikrotikControllerOptionsFlowHandler(config_entries.OptionsFlow):
"""Handle options.""" """Handle options."""

15
custom_components/mikrotik_router/device_tracker.py
View file

@ -35,9 +35,9 @@ DEVICE_ATTRIBUTES = [
] ]
#--------------------------- # ---------------------------
# async_setup_entry # async_setup_entry
#--------------------------- # ---------------------------
async def async_setup_entry(hass, config_entry, async_add_entities): async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up device tracker for Mikrotik Router component.""" """Set up device tracker for Mikrotik Router component."""
name = config_entry.data[CONF_NAME] name = config_entry.data[CONF_NAME]
@ -49,7 +49,6 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
"""Update the values of the controller.""" """Update the values of the controller."""
update_items(name, mikrotik_controller, async_add_entities, tracked) update_items(name, mikrotik_controller, async_add_entities, tracked)
mikrotik_controller.listeners.append( mikrotik_controller.listeners.append(
async_dispatcher_connect(hass, mikrotik_controller.signal_update, update_controller) async_dispatcher_connect(hass, mikrotik_controller.signal_update, update_controller)
) )
@ -58,9 +57,9 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
return return
#--------------------------- # ---------------------------
# update_items # update_items
#--------------------------- # ---------------------------
@callback @callback
def update_items(name, mikrotik_controller, async_add_entities, tracked): def update_items(name, mikrotik_controller, async_add_entities, tracked):
"""Update tracked device state from the controller.""" """Update tracked device state from the controller."""
@ -83,9 +82,9 @@ def update_items(name, mikrotik_controller, async_add_entities, tracked):
return return
#--------------------------- # ---------------------------
# MikrotikControllerPortDeviceTracker # MikrotikControllerPortDeviceTracker
#--------------------------- # ---------------------------
class MikrotikControllerPortDeviceTracker(ScannerEntity): class MikrotikControllerPortDeviceTracker(ScannerEntity):
"""Representation of a network port.""" """Representation of a network port."""
@ -111,7 +110,7 @@ class MikrotikControllerPortDeviceTracker(ScannerEntity):
async def async_update(self): async def async_update(self):
"""Synchronize state with controller.""" """Synchronize state with controller."""
#await self.mikrotik_controller.async_update() # await self.mikrotik_controller.async_update()
return return
@property @property

82
custom_components/mikrotik_router/mikrotik_controller.py
View file

@ -4,7 +4,7 @@ from datetime import timedelta
import logging import logging
from homeassistant.helpers.dispatcher import async_dispatcher_send from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.event import async_track_time_interval from homeassistant.helpers.event import async_track_time_interval
#from homeassistant.util import Throttle # from homeassistant.util import Throttle
from .const import ( from .const import (
DOMAIN, DOMAIN,
@ -14,15 +14,15 @@ from .const import (
DEFAULT_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL,
) )
_LOGGER = logging.getLogger(__name__)
from .mikrotikapi import MikrotikAPI from .mikrotikapi import MikrotikAPI
#DEFAULT_SCAN_INTERVAL = timedelta(seconds=DEFAULT_SCAN_INTERVAL) _LOGGER = logging.getLogger(__name__)
# DEFAULT_SCAN_INTERVAL = timedelta(seconds=DEFAULT_SCAN_INTERVAL)
#--------------------------- # ---------------------------
# MikrotikControllerData # MikrotikControllerData
#--------------------------- # ---------------------------
class MikrotikControllerData(): class MikrotikControllerData():
def __init__(self, hass, config_entry, name, host, port, username, password, use_ssl): def __init__(self, hass, config_entry, name, host, port, username, password, use_ssl):
"""Initialize.""" """Initialize."""
@ -51,51 +51,51 @@ class MikrotikControllerData():
await self.async_update() await self.async_update()
return return
#--------------------------- # ---------------------------
# option_track_arp # option_track_arp
#--------------------------- # ---------------------------
@property @property
def option_track_arp(self): def option_track_arp(self):
"""Config entry option to not track ARP.""" """Config entry option to not track ARP."""
return self.config_entry.options.get(CONF_TRACK_ARP, DEFAULT_TRACK_ARP) return self.config_entry.options.get(CONF_TRACK_ARP, DEFAULT_TRACK_ARP)
#--------------------------- # ---------------------------
# option_scan_interval # option_scan_interval
#--------------------------- # ---------------------------
@property @property
def option_scan_interval(self): def option_scan_interval(self):
"""Config entry option scan interval.""" """Config entry option scan interval."""
scan_interval = self.config_entry.options.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL) scan_interval = self.config_entry.options.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)
return timedelta(seconds=scan_interval) return timedelta(seconds=scan_interval)
#--------------------------- # ---------------------------
# signal_update # signal_update
#--------------------------- # ---------------------------
@property @property
def signal_update(self): def signal_update(self):
"""Event specific per UniFi entry to signal new data.""" """Event specific per UniFi entry to signal new data."""
return f"{DOMAIN}-update-{self.name}" return f"{DOMAIN}-update-{self.name}"
#--------------------------- # ---------------------------
# connected # connected
#--------------------------- # ---------------------------
def connected(self): def connected(self):
"""Return connected boolean.""" """Return connected boolean."""
return self.api.connected() return self.api.connected()
#--------------------------- # ---------------------------
# hwinfo_update # hwinfo_update
#--------------------------- # ---------------------------
async def hwinfo_update(self): async def hwinfo_update(self):
"""Update Mikrotik hardware info.""" """Update Mikrotik hardware info."""
self.get_system_routerboard() self.get_system_routerboard()
self.get_system_resource() self.get_system_resource()
return return
#--------------------------- # ---------------------------
# async_update # async_update
#--------------------------- # ---------------------------
#@Throttle(DEFAULT_SCAN_INTERVAL) # @Throttle(DEFAULT_SCAN_INTERVAL)
async def async_update(self): async def async_update(self):
"""Update Mikrotik Controller data.""" """Update Mikrotik Controller data."""
@ -105,9 +105,9 @@ class MikrotikControllerData():
async_dispatcher_send(self.hass, self.signal_update) async_dispatcher_send(self.hass, self.signal_update)
return return
#--------------------------- # ---------------------------
# async_reset # async_reset
#--------------------------- # ---------------------------
async def async_reset(self): async def async_reset(self):
"""Reset this controller to default state.""" """Reset this controller to default state."""
for unsub_dispatcher in self.listeners: for unsub_dispatcher in self.listeners:
@ -116,9 +116,9 @@ class MikrotikControllerData():
self.listeners = [] self.listeners = []
return True return True
#--------------------------- # ---------------------------
# get_interfaces # get_interfaces
#--------------------------- # ---------------------------
def get_interfaces(self): def get_interfaces(self):
ifaces = self.api.path("/interface") ifaces = self.api.path("/interface")
for iface in ifaces: for iface in ifaces:
@ -152,9 +152,9 @@ class MikrotikControllerData():
return return
#--------------------------- # ---------------------------
# get_arp # get_arp
#--------------------------- # ---------------------------
def get_arp(self): def get_arp(self):
self.data['arp'] = {} self.data['arp'] = {}
if not self.option_track_arp: if not self.option_track_arp:
@ -167,20 +167,20 @@ class MikrotikControllerData():
bridge_used = False bridge_used = False
data = self.api.path("/ip/arp") data = self.api.path("/ip/arp")
for entry in data: for entry in data:
## Ignore invalid entries # Ignore invalid entries
if entry['invalid']: if entry['invalid']:
continue continue
## Do not add ARP detected on bridge # Do not add ARP detected on bridge
if entry['interface'] == "bridge": if entry['interface'] == "bridge":
bridge_used = True bridge_used = True
## Build address table on bridge # Build address table on bridge
if 'mac-address' in entry and 'address' in entry: if 'mac-address' in entry and 'address' in entry:
mac2ip[entry['mac-address']] = entry['address'] mac2ip[entry['mac-address']] = entry['address']
continue continue
## Get iface default-name from custom name # Get iface default-name from custom name
uid = None uid = None
for ifacename in self.data['interface']: for ifacename in self.data['interface']:
if self.data['interface'][ifacename]['name'] == entry['interface']: if self.data['interface'][ifacename]['name'] == entry['interface']:
@ -190,11 +190,11 @@ class MikrotikControllerData():
if not uid: if not uid:
continue continue
## Create uid arp dict # Create uid arp dict
if uid not in self.data['arp']: if uid not in self.data['arp']:
self.data['arp'][uid] = {} self.data['arp'][uid] = {}
## Add data # Add data
self.data['arp'][uid]['interface'] = uid self.data['arp'][uid]['interface'] = uid
if 'mac-address' in self.data['arp'][uid]: if 'mac-address' in self.data['arp'][uid]:
self.data['arp'][uid]['mac-address'] = "multiple" self.data['arp'][uid]['mac-address'] = "multiple"
@ -209,24 +209,24 @@ class MikrotikControllerData():
if bridge_used: if bridge_used:
self.update_bridge_hosts(mac2ip) self.update_bridge_hosts(mac2ip)
## Map ARP to ifaces # Map ARP to ifaces
for uid in self.data['interface']: for uid in self.data['interface']:
self.data['interface'][uid]['client-ip-address'] = self.data['arp'][uid]['address'] if uid in self.data['arp'] and 'address' in self.data['arp'][uid] else "" self.data['interface'][uid]['client-ip-address'] = self.data['arp'][uid]['address'] if uid in self.data['arp'] and 'address' in self.data['arp'][uid] else ""
self.data['interface'][uid]['client-mac-address'] = self.data['arp'][uid]['mac-address'] if uid in self.data['arp'] and 'mac-address' in self.data['arp'][uid] else "" self.data['interface'][uid]['client-mac-address'] = self.data['arp'][uid]['mac-address'] if uid in self.data['arp'] and 'mac-address' in self.data['arp'][uid] else ""
return True return True
#--------------------------- # ---------------------------
# update_bridge_hosts # update_bridge_hosts
#--------------------------- # ---------------------------
def update_bridge_hosts(self, mac2ip): def update_bridge_hosts(self, mac2ip):
data = self.api.path("/interface/bridge/host") data = self.api.path("/interface/bridge/host")
for entry in data: for entry in data:
## Ignore port MAC # Ignore port MAC
if entry['local']: if entry['local']:
continue continue
## Get iface default-name from custom name # Get iface default-name from custom name
uid = None uid = None
for ifacename in self.data['interface']: for ifacename in self.data['interface']:
if self.data['interface'][ifacename]['name'] == entry['interface']: if self.data['interface'][ifacename]['name'] == entry['interface']:
@ -236,11 +236,11 @@ class MikrotikControllerData():
if not uid: if not uid:
continue continue
## Create uid arp dict # Create uid arp dict
if uid not in self.data['arp']: if uid not in self.data['arp']:
self.data['arp'][uid] = {} self.data['arp'][uid] = {}
## Add data # Add data
self.data['arp'][uid]['interface'] = uid self.data['arp'][uid]['interface'] = uid
if 'mac-address' in self.data['arp'][uid]: if 'mac-address' in self.data['arp'][uid]:
self.data['arp'][uid]['mac-address'] = "multiple" self.data['arp'][uid]['mac-address'] = "multiple"
@ -254,9 +254,9 @@ class MikrotikControllerData():
return return
#--------------------------- # ---------------------------
# get_system_routerboard # get_system_routerboard
#--------------------------- # ---------------------------
def get_system_routerboard(self): def get_system_routerboard(self):
data = self.api.path("/system/routerboard") data = self.api.path("/system/routerboard")
for entry in data: for entry in data:
@ -267,9 +267,9 @@ class MikrotikControllerData():
return return
#--------------------------- # ---------------------------
# get_system_resource # get_system_resource
#--------------------------- # ---------------------------
def get_system_resource(self): def get_system_resource(self):
data = self.api.path("/system/resource") data = self.api.path("/system/resource")
for entry in data: for entry in data:

22
custom_components/mikrotik_router/mikrotikapi.py
View file

@ -6,9 +6,9 @@ import librouteros
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
#--------------------------- # ---------------------------
# MikrotikAPI # MikrotikAPI
#--------------------------- # ---------------------------
class MikrotikAPI: class MikrotikAPI:
"""Handle all communication with the Mikrotik API.""" """Handle all communication with the Mikrotik API."""
@ -27,13 +27,13 @@ class MikrotikAPI:
self._connected = False self._connected = False
self.error = "" self.error = ""
## Default ports # Default ports
if not self._port: if not self._port:
self._port = 8729 if self._use_ssl else 8728 self._port = 8729 if self._use_ssl else 8728
#--------------------------- # ---------------------------
# connect # connect
#--------------------------- # ---------------------------
def connect(self): def connect(self):
"""Connect to Mikrotik device.""" """Connect to Mikrotik device."""
self.error = "" self.error = ""
@ -72,9 +72,9 @@ class MikrotikAPI:
return self._connected return self._connected
#--------------------------- # ---------------------------
# errorToStrings # errorToStrings
#--------------------------- # ---------------------------
def errorToStrings(self, error): def errorToStrings(self, error):
self.error = "cannot_connect" self.error = "cannot_connect"
if error == "invalid user name or password (6)": if error == "invalid user name or password (6)":
@ -82,16 +82,16 @@ class MikrotikAPI:
return return
#--------------------------- # ---------------------------
# connected # connected
#--------------------------- # ---------------------------
def connected(self): def connected(self):
"""Return connected boolean.""" """Return connected boolean."""
return self._connected return self._connected
#--------------------------- # ---------------------------
# path # path
#--------------------------- # ---------------------------
def path(self, path): def path(self, path):
"""Retrieve data from Mikrotik API.""" """Retrieve data from Mikrotik API."""
if not self._connected or not self._connection: if not self._connected or not self._connection: