jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-08 13:57:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add option to manually specify device detection method (#17852)

Browse Source 
* Add option to manually specify device detection method

* Fix style and lint issue
This commit is contained in:
Corey Edwards 2018-11-03 16:48:08 -06:00 committed by Fabian Affolter
parent 610b0b6494
commit 155df912e5
1 changed files with 41 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

96
homeassistant/components/device_tracker/mikrotik.py
View File

@ -12,19 +12,20 @@ import homeassistant.helpers.config_validation as cv
from homeassistant.components.device_tracker import ( from homeassistant.components.device_tracker import (
DOMAIN, PLATFORM_SCHEMA, DeviceScanner) DOMAIN, PLATFORM_SCHEMA, DeviceScanner)
from homeassistant.const import ( from homeassistant.const import (
CONF_HOST, CONF_PASSWORD, CONF_USERNAME, CONF_PORT) CONF_HOST, CONF_PASSWORD, CONF_USERNAME, CONF_PORT, CONF_METHOD)
REQUIREMENTS = ['librouteros==2.1.1'] REQUIREMENTS = ['librouteros==2.1.1']
MTK_DEFAULT_API_PORT = '8728'
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
MTK_DEFAULT_API_PORT = '8728'
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_HOST): cv.string, vol.Required(CONF_HOST): cv.string,
vol.Required(CONF_USERNAME): cv.string, vol.Required(CONF_USERNAME): cv.string,
vol.Required(CONF_PASSWORD): cv.string, vol.Required(CONF_PASSWORD): cv.string,
vol.Optional(CONF_PORT, default=MTK_DEFAULT_API_PORT): cv.port vol.Optional(CONF_METHOD): cv.string,
vol.Optional(CONF_PORT, default=MTK_DEFAULT_API_PORT): cv.port,
}) })
@ -45,6 +46,7 @@ class MikrotikScanner(DeviceScanner):
self.port = config[CONF_PORT] self.port = config[CONF_PORT]
self.username = config[CONF_USERNAME] self.username = config[CONF_USERNAME]
self.password = config[CONF_PASSWORD] self.password = config[CONF_PASSWORD]
self.method = config[CONF_METHOD]
self.connected = False self.connected = False
self.success_init = False self.success_init = False
@ -53,28 +55,18 @@ class MikrotikScanner(DeviceScanner):
self.success_init = self.connect_to_device() self.success_init = self.connect_to_device()
if self.success_init: if self.success_init:
_LOGGER.info( _LOGGER.info("Start polling Mikrotik (%s) router...", self.host)
"Start polling Mikrotik (%s) router...",
self.host
)
self._update_info() self._update_info()
else: else:
_LOGGER.error( _LOGGER.error("Connection to Mikrotik (%s) failed", self.host)
"Connection to Mikrotik (%s) failed",
self.host
)
def connect_to_device(self): def connect_to_device(self):
"""Connect to Mikrotik method.""" """Connect to Mikrotik method."""
import librouteros import librouteros
try: try:
self.client = librouteros.connect( self.client = librouteros.connect(
self.host, self.host, self.username, self.password, port=int(self.port),
self.username, encoding='utf-8')
self.password,
port=int(self.port),
encoding='utf-8'
)
try: try:
routerboard_info = self.client( routerboard_info = self.client(
@ -86,16 +78,15 @@ class MikrotikScanner(DeviceScanner):
raise raise
if routerboard_info: if routerboard_info:
_LOGGER.info("Connected to Mikrotik %s with IP %s", _LOGGER.info(
routerboard_info[0].get('model', 'Router'), "Connected to Mikrotik %s with IP %s",
self.host) routerboard_info[0].get('model', 'Router'), self.host)
self.connected = True self.connected = True
try: try:
self.capsman_exist = self.client( self.capsman_exist = self.client(
cmd='/caps-man/interface/getall' cmd='/caps-man/interface/getall')
)
except (librouteros.exceptions.TrapError, except (librouteros.exceptions.TrapError,
librouteros.exceptions.MultiTrapError, librouteros.exceptions.MultiTrapError,
librouteros.exceptions.ConnectionError): librouteros.exceptions.ConnectionError):
@ -103,27 +94,27 @@ class MikrotikScanner(DeviceScanner):
if not self.capsman_exist: if not self.capsman_exist:
_LOGGER.info( _LOGGER.info(
'Mikrotik %s: Not a CAPSman controller. Trying ' "Mikrotik %s: Not a CAPSman controller. Trying "
'local interfaces ', "local interfaces", self.host)
self.host
)
try: try:
self.wireless_exist = self.client( self.wireless_exist = self.client(
cmd='/interface/wireless/getall' cmd='/interface/wireless/getall')
)
except (librouteros.exceptions.TrapError, except (librouteros.exceptions.TrapError,
librouteros.exceptions.MultiTrapError, librouteros.exceptions.MultiTrapError,
librouteros.exceptions.ConnectionError): librouteros.exceptions.ConnectionError):
self.wireless_exist = False self.wireless_exist = False
if not self.wireless_exist: if not self.wireless_exist or self.method == 'ip':
_LOGGER.info( _LOGGER.info(
'Mikrotik %s: Wireless adapters not found. Try to ' "Mikrotik %s: Wireless adapters not found. Try to "
'use DHCP lease table as presence tracker source. ' "use DHCP lease table as presence tracker source. "
'Please decrease lease time as much as possible.', "Please decrease lease time as much as possible",
self.host self.host)
) if self.method:
_LOGGER.info(
"Mikrotik %s: Manually selected polling method %s",
self.host, self.method)
except (librouteros.exceptions.TrapError, except (librouteros.exceptions.TrapError,
librouteros.exceptions.MultiTrapError, librouteros.exceptions.MultiTrapError,
@ -143,28 +134,27 @@ class MikrotikScanner(DeviceScanner):
def _update_info(self): def _update_info(self):
"""Retrieve latest information from the Mikrotik box.""" """Retrieve latest information from the Mikrotik box."""
if self.capsman_exist: if self.method:
devices_tracker = 'capsman' devices_tracker = self.method
elif self.wireless_exist:
devices_tracker = 'wireless'
else: else:
devices_tracker = 'ip' if self.capsman_exist:
devices_tracker = 'capsman'
elif self.wireless_exist:
devices_tracker = 'wireless'
else:
devices_tracker = 'ip'
_LOGGER.info( _LOGGER.info(
"Loading %s devices from Mikrotik (%s) ...", "Loading %s devices from Mikrotik (%s) ...",
devices_tracker, devices_tracker, self.host)
self.host
)
device_names = self.client(cmd='/ip/dhcp-server/lease/getall') device_names = self.client(cmd='/ip/dhcp-server/lease/getall')
if devices_tracker == 'capsman': if devices_tracker == 'capsman':
devices = self.client( devices = self.client(
cmd='/caps-man/registration-table/getall' cmd='/caps-man/registration-table/getall')
)
elif devices_tracker == 'wireless': elif devices_tracker == 'wireless':
devices = self.client( devices = self.client(
cmd='/interface/wireless/registration-table/getall' cmd='/interface/wireless/registration-table/getall')
)
else: else:
devices = device_names devices = device_names
@ -172,21 +162,17 @@ class MikrotikScanner(DeviceScanner):
return False return False
mac_names = {device.get('mac-address'): device.get('host-name') mac_names = {device.get('mac-address'): device.get('host-name')
for device in device_names for device in device_names if device.get('mac-address')}
if device.get('mac-address')}
if self.wireless_exist or self.capsman_exist: if devices_tracker in ('wireless', 'capsman'):
self.last_results = { self.last_results = {
device.get('mac-address'): device.get('mac-address'):
mac_names.get(device.get('mac-address')) mac_names.get(device.get('mac-address'))
for device in devices for device in devices}
}
else: else:
self.last_results = { self.last_results = {
device.get('mac-address'): device.get('mac-address'):
mac_names.get(device.get('mac-address')) mac_names.get(device.get('mac-address'))
for device in device_names for device in device_names if device.get('active-address')}
if device.get('active-address')
}
return True return True