forked from home-assistant/core
-
Notifications
You must be signed in to change notification settings - Fork 0
/
rtorrent.py
127 lines (103 loc) · 4.03 KB
/
rtorrent.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
"""Support for monitoring the rtorrent BitTorrent client API."""
import logging
import xmlrpc.client
import voluptuous as vol
from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.const import (
CONF_URL, CONF_NAME,
CONF_MONITORED_VARIABLES, STATE_IDLE)
from homeassistant.helpers.entity import Entity
import homeassistant.helpers.config_validation as cv
from homeassistant.exceptions import PlatformNotReady
_LOGGER = logging.getLogger(__name__)
SENSOR_TYPE_CURRENT_STATUS = 'current_status'
SENSOR_TYPE_DOWNLOAD_SPEED = 'download_speed'
SENSOR_TYPE_UPLOAD_SPEED = 'upload_speed'
DEFAULT_NAME = 'rtorrent'
SENSOR_TYPES = {
SENSOR_TYPE_CURRENT_STATUS: ['Status', None],
SENSOR_TYPE_DOWNLOAD_SPEED: ['Down Speed', 'kB/s'],
SENSOR_TYPE_UPLOAD_SPEED: ['Up Speed', 'kB/s'],
}
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_URL): cv.url,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_MONITORED_VARIABLES, default=SENSOR_TYPES): vol.All(
cv.ensure_list, [vol.In(SENSOR_TYPES)]),
})
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the rtorrent sensors."""
url = config[CONF_URL]
name = config[CONF_NAME]
try:
rtorrent = xmlrpc.client.ServerProxy(url)
except (xmlrpc.client.ProtocolError, ConnectionRefusedError):
_LOGGER.error("Connection to rtorrent daemon failed")
raise PlatformNotReady
dev = []
for variable in config[CONF_MONITORED_VARIABLES]:
dev.append(RTorrentSensor(variable, rtorrent, name))
add_entities(dev)
def format_speed(speed):
"""Return a bytes/s measurement as a human readable string."""
kb_spd = float(speed) / 1024
return round(kb_spd, 2 if kb_spd < 0.1 else 1)
class RTorrentSensor(Entity):
"""Representation of an rtorrent sensor."""
def __init__(self, sensor_type, rtorrent_client, client_name):
"""Initialize the sensor."""
self._name = SENSOR_TYPES[sensor_type][0]
self.client = rtorrent_client
self.type = sensor_type
self.client_name = client_name
self._state = None
self._unit_of_measurement = SENSOR_TYPES[sensor_type][1]
self.data = None
self._available = False
@property
def name(self):
"""Return the name of the sensor."""
return '{} {}'.format(self.client_name, self._name)
@property
def state(self):
"""Return the state of the sensor."""
return self._state
@property
def available(self):
"""Return true if device is available."""
return self._available
@property
def unit_of_measurement(self):
"""Return the unit of measurement of this entity, if any."""
return self._unit_of_measurement
def update(self):
"""Get the latest data from rtorrent and updates the state."""
multicall = xmlrpc.client.MultiCall(self.client)
multicall.throttle.global_up.rate()
multicall.throttle.global_down.rate()
try:
self.data = multicall()
self._available = True
except (xmlrpc.client.ProtocolError, ConnectionRefusedError):
_LOGGER.error("Connection to rtorrent lost")
self._available = False
return
upload = self.data[0]
download = self.data[1]
if self.type == SENSOR_TYPE_CURRENT_STATUS:
if self.data:
if upload > 0 and download > 0:
self._state = 'Up/Down'
elif upload > 0 and download == 0:
self._state = 'Seeding'
elif upload == 0 and download > 0:
self._state = 'Downloading'
else:
self._state = STATE_IDLE
else:
self._state = None
if self.data:
if self.type == SENSOR_TYPE_DOWNLOAD_SPEED:
self._state = format_speed(download)
elif self.type == SENSOR_TYPE_UPLOAD_SPEED:
self._state = format_speed(upload)