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

Use pihole module to get data (#14809)

Browse Source
This commit is contained in:
Fabian Affolter 2018-06-04 18:49:26 +02:00 committed by Martin Hjelmare
parent bd1b1a9ff9
commit 816efa02d1
2 changed files with 82 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

148
homeassistant/components/sensor/pi_hole.py
View File

@ -1,23 +1,26 @@
""" """
Support for getting statistical data from a Pi-Hole system. Support for getting statistical data from a Pi-hole system.
For more details about this platform, please refer to the documentation at For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/sensor.pi_hole/ https://home-assistant.io/components/sensor.pi_hole/
""" """
import logging
import json
from datetime import timedelta from datetime import timedelta
import logging
import voluptuous as vol import voluptuous as vol
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity
from homeassistant.components.sensor import PLATFORM_SCHEMA from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.const import ( from homeassistant.const import (
CONF_NAME, CONF_HOST, CONF_SSL, CONF_VERIFY_SSL, CONF_MONITORED_CONDITIONS) CONF_HOST, CONF_MONITORED_CONDITIONS, CONF_NAME, CONF_SSL, CONF_VERIFY_SSL)
from homeassistant.exceptions import PlatformNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity
from homeassistant.util import Throttle
REQUIREMENTS = ['pihole==0.1.2']
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
_ENDPOINT = '/api.php'
ATTR_BLOCKED_DOMAINS = 'domains_blocked' ATTR_BLOCKED_DOMAINS = 'domains_blocked'
ATTR_PERCENTAGE_TODAY = 'percentage_today' ATTR_PERCENTAGE_TODAY = 'percentage_today'
@ -32,25 +35,27 @@ DEFAULT_NAME = 'Pi-Hole'
DEFAULT_SSL = False DEFAULT_SSL = False
DEFAULT_VERIFY_SSL = True DEFAULT_VERIFY_SSL = True
SCAN_INTERVAL = timedelta(minutes=5) MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=5)
MONITORED_CONDITIONS = { MONITORED_CONDITIONS = {
'dns_queries_today': ['DNS Queries Today', 'ads_blocked_today':
'queries', 'mdi:comment-question-outline'], ['Ads Blocked Today', 'ads', 'mdi:close-octagon-outline'],
'ads_blocked_today': ['Ads Blocked Today', 'ads_percentage_today':
'ads', 'mdi:close-octagon-outline'], ['Ads Percentage Blocked Today', '%', 'mdi:close-octagon-outline'],
'ads_percentage_today': ['Ads Percentage Blocked Today', 'clients_ever_seen':
'%', 'mdi:close-octagon-outline'], ['Seen Clients', 'clients', 'mdi:account-outline'],
'domains_being_blocked': ['Domains Blocked', 'dns_queries_today':
'domains', 'mdi:block-helper'], ['DNS Queries Today', 'queries', 'mdi:comment-question-outline'],
'queries_cached': ['DNS Queries Cached', 'domains_being_blocked':
'queries', 'mdi:comment-question-outline'], ['Domains Blocked', 'domains', 'mdi:block-helper'],
'queries_forwarded': ['DNS Queries Forwarded', 'queries_cached':
'queries', 'mdi:comment-question-outline'], ['DNS Queries Cached', 'queries', 'mdi:comment-question-outline'],
'unique_clients': ['DNS Unique Clients', 'queries_forwarded':
'clients', 'mdi:account-outline'], ['DNS Queries Forwarded', 'queries', 'mdi:comment-question-outline'],
'unique_domains': ['DNS Unique Domains', 'unique_clients':
'domains', 'mdi:domain'], ['DNS Unique Clients', 'clients', 'mdi:account-outline'],
'unique_domains':
['DNS Unique Domains', 'domains', 'mdi:domain'],
} }
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
@ -65,100 +70,105 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
}) })
def setup_platform(hass, config, add_devices, discovery_info=None): async def async_setup_platform(
"""Set up the Pi-Hole sensor.""" hass, config, async_add_devices, discovery_info=None):
"""Set up the Pi-hole sensor."""
from pihole import PiHole
name = config.get(CONF_NAME) name = config.get(CONF_NAME)
host = config.get(CONF_HOST) host = config.get(CONF_HOST)
use_ssl = config.get(CONF_SSL) use_tls = config.get(CONF_SSL)
location = config.get(CONF_LOCATION) location = config.get(CONF_LOCATION)
verify_ssl = config.get(CONF_VERIFY_SSL) verify_tls = config.get(CONF_VERIFY_SSL)
api = PiHoleAPI('{}/{}'.format(host, location), use_ssl, verify_ssl) session = async_get_clientsession(hass)
pi_hole = PiHoleData(PiHole(
host, hass.loop, session, location=location, tls=use_tls,
verify_tls=verify_tls))
sensors = [PiHoleSensor(hass, api, name, condition) await pi_hole.async_update()
if pi_hole.api.data is None:
raise PlatformNotReady
sensors = [PiHoleSensor(pi_hole, name, condition)
for condition in config[CONF_MONITORED_CONDITIONS]] for condition in config[CONF_MONITORED_CONDITIONS]]
add_devices(sensors, True) async_add_devices(sensors, True)
class PiHoleSensor(Entity): class PiHoleSensor(Entity):
"""Representation of a Pi-Hole sensor.""" """Representation of a Pi-hole sensor."""
def __init__(self, hass, api, name, variable): def __init__(self, pi_hole, name, condition):
"""Initialize a Pi-Hole sensor.""" """Initialize a Pi-hole sensor."""
self._hass = hass self.pi_hole = pi_hole
self._api = api
self._name = name self._name = name
self._var_id = variable self._condition = condition
variable_info = MONITORED_CONDITIONS[variable] variable_info = MONITORED_CONDITIONS[condition]
self._var_name = variable_info[0] self._condition_name = variable_info[0]
self._var_units = variable_info[1] self._unit_of_measurement = variable_info[1]
self._var_icon = variable_info[2] self._icon = variable_info[2]
self.data = {}
@property @property
def name(self): def name(self):
"""Return the name of the sensor.""" """Return the name of the sensor."""
return "{} {}".format(self._name, self._var_name) return "{} {}".format(self._name, self._condition_name)
@property @property
def icon(self): def icon(self):
"""Icon to use in the frontend, if any.""" """Icon to use in the frontend, if any."""
return self._var_icon return self._icon
@property @property
def unit_of_measurement(self): def unit_of_measurement(self):
"""Return the unit the value is expressed in.""" """Return the unit the value is expressed in."""
return self._var_units return self._unit_of_measurement
# pylint: disable=no-member
@property @property
def state(self): def state(self):
"""Return the state of the device.""" """Return the state of the device."""
try: try:
return round(self._api.data[self._var_id], 2) return round(self.data[self._condition], 2)
except TypeError: except TypeError:
return self._api.data[self._var_id] return self.data[self._condition]
# pylint: disable=no-member
@property @property
def device_state_attributes(self): def device_state_attributes(self):
"""Return the state attributes of the Pi-Hole.""" """Return the state attributes of the Pi-Hole."""
return { return {
ATTR_BLOCKED_DOMAINS: self._api.data['domains_being_blocked'], ATTR_BLOCKED_DOMAINS: self.data['domains_being_blocked'],
} }
@property @property
def available(self): def available(self):
"""Could the device be accessed during the last update call.""" """Could the device be accessed during the last update call."""
return self._api.available return self.pi_hole.available
def update(self): async def async_update(self):
"""Get the latest data from the Pi-Hole API.""" """Get the latest data from the Pi-hole API."""
self._api.update() await self.pi_hole.async_update()
self.data = self.pi_hole.api.data
class PiHoleAPI(object): class PiHoleData(object):
"""Get the latest data and update the states.""" """Get the latest data and update the states."""
def __init__(self, host, use_ssl, verify_ssl): def __init__(self, api):
"""Initialize the data object.""" """Initialize the data object."""
from homeassistant.components.sensor.rest import RestData self.api = api
uri_scheme = 'https://' if use_ssl else 'http://'
resource = "{}{}{}".format(uri_scheme, host, _ENDPOINT)
self._rest = RestData('GET', resource, None, None, None, verify_ssl)
self.data = None
self.available = True self.available = True
self.update()
def update(self): @Throttle(MIN_TIME_BETWEEN_UPDATES)
"""Get the latest data from the Pi-Hole.""" async def async_update(self):
"""Get the latest data from the Pi-hole."""
from pihole.exceptions import PiHoleError
try: try:
self._rest.update() await self.api.get_data()
self.data = json.loads(self._rest.data)
self.available = True self.available = True
except TypeError: except PiHoleError:
_LOGGER.error("Unable to fetch data from Pi-Hole") _LOGGER.error("Unable to fetch data from Pi-hole")
self.available = False self.available = False

3
requirements_all.txt
View File

@ -633,6 +633,9 @@ pifacedigitalio==3.0.5
# homeassistant.components.light.piglow # homeassistant.components.light.piglow
piglow==1.2.4 piglow==1.2.4
# homeassistant.components.sensor.pi_hole
pihole==0.1.2
# homeassistant.components.pilight # homeassistant.components.pilight
pilight==0.1.1 pilight==0.1.1