/
tautulli.py
148 lines (119 loc) · 4.94 KB
/
tautulli.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
"""
A platform which allows you to get information from Tautulli.
For more details about this platform, please refer to the documentation at
https://www.home-assistant.io/components/sensor.tautulli/
"""
from datetime import timedelta
import logging
import voluptuous as vol
from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.const import (
CONF_API_KEY, CONF_HOST, CONF_MONITORED_CONDITIONS, CONF_NAME, CONF_PORT,
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 = ['pytautulli==0.4.0']
_LOGGER = logging.getLogger(__name__)
CONF_MONITORED_USERS = 'monitored_users'
DEFAULT_NAME = 'Tautulli'
DEFAULT_PORT = '8181'
DEFAULT_SSL = False
DEFAULT_VERIFY_SSL = True
TIME_BETWEEN_UPDATES = timedelta(seconds=10)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_API_KEY): cv.string,
vol.Required(CONF_HOST): cv.string,
vol.Optional(CONF_MONITORED_CONDITIONS):
vol.All(cv.ensure_list, [cv.string]),
vol.Optional(CONF_MONITORED_USERS): vol.All(cv.ensure_list, [cv.string]),
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.string,
vol.Optional(CONF_SSL, default=DEFAULT_SSL): cv.boolean,
vol.Optional(CONF_VERIFY_SSL, default=DEFAULT_VERIFY_SSL): cv.boolean,
})
async def async_setup_platform(
hass, config, async_add_entities, discovery_info=None):
"""Create the Tautulli sensor."""
from pytautulli import Tautulli
name = config.get(CONF_NAME)
host = config[CONF_HOST]
port = config.get(CONF_PORT)
api_key = config[CONF_API_KEY]
monitored_conditions = config.get(CONF_MONITORED_CONDITIONS)
user = config.get(CONF_MONITORED_USERS)
use_ssl = config.get(CONF_SSL)
verify_ssl = config.get(CONF_VERIFY_SSL)
session = async_get_clientsession(hass, verify_ssl)
tautulli = TautulliData(Tautulli(
host, port, api_key, hass.loop, session, use_ssl))
if not await tautulli.test_connection():
raise PlatformNotReady
sensor = [TautulliSensor(tautulli, name, monitored_conditions, user)]
async_add_entities(sensor, True)
class TautulliSensor(Entity):
"""Representation of a Tautulli sensor."""
def __init__(self, tautulli, name, monitored_conditions, users):
"""Initialize the Tautulli sensor."""
self.tautulli = tautulli
self.monitored_conditions = monitored_conditions
self.usernames = users
self.sessions = {}
self.home = {}
self._attributes = {}
self._name = name
self._state = None
async def async_update(self):
"""Get the latest data from the Tautulli API."""
await self.tautulli.async_update()
self.home = self.tautulli.api.home_data
self.sessions = self.tautulli.api.session_data
self._attributes['Top Movie'] = self.home[0]['rows'][0]['title']
self._attributes['Top TV Show'] = self.home[3]['rows'][0]['title']
self._attributes['Top User'] = self.home[7]['rows'][0]['user']
for key in self.sessions:
if 'sessions' not in key:
self._attributes[key] = self.sessions[key]
for user in self.tautulli.api.users:
if self.usernames is None or user in self.usernames:
userdata = self.tautulli.api.user_data
self._attributes[user] = {}
self._attributes[user]['Activity'] = userdata[user]['Activity']
if self.monitored_conditions:
for key in self.monitored_conditions:
try:
self._attributes[user][key] = userdata[user][key]
except (KeyError, TypeError):
self._attributes[user][key] = ''
@property
def name(self):
"""Return the name of the sensor."""
return self._name
@property
def state(self):
"""Return the state of the sensor."""
return self.sessions['stream_count']
@property
def icon(self):
"""Return the icon of the sensor."""
return 'mdi:plex'
@property
def device_state_attributes(self):
"""Return attributes for the sensor."""
return self._attributes
class TautulliData:
"""Get the latest data and update the states."""
def __init__(self, api):
"""Initialize the data object."""
self.api = api
@Throttle(TIME_BETWEEN_UPDATES)
async def async_update(self):
"""Get the latest data from Tautulli."""
await self.api.get_data()
async def test_connection(self):
"""Test connection to Tautulli."""
await self.api.test_connection()
connection_status = self.api.connection
return connection_status