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

Improve type hints in rtorrent (#145222)

Browse Source
This commit is contained in:
epenet 2025-05-19 15:53:41 +02:00 committed by GitHub
parent 8df447091d
commit a38e033e13
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 11 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
homeassistant/components/rtorrent/sensor.py
View File

@ -3,6 +3,7 @@
from __future__ import annotations from __future__ import annotations
import logging import logging
from typing import cast
import xmlrpc.client import xmlrpc.client
import voluptuous as vol import voluptuous as vol
@ -126,6 +127,9 @@ def format_speed(speed):
return round(kb_spd, 2 if kb_spd < 0.1 else 1) return round(kb_spd, 2 if kb_spd < 0.1 else 1)
type RTorrentData = tuple[float, float, list, list, list, list, list]
class RTorrentSensor(SensorEntity): class RTorrentSensor(SensorEntity):
"""Representation of an rtorrent sensor.""" """Representation of an rtorrent sensor."""
@ -135,12 +139,12 @@ class RTorrentSensor(SensorEntity):
"""Initialize the sensor.""" """Initialize the sensor."""
self.entity_description = description self.entity_description = description
self.client = rtorrent_client self.client = rtorrent_client
self.data = None self.data: RTorrentData | None = None
self._attr_name = f"{client_name} {description.name}" self._attr_name = f"{client_name} {description.name}"
self._attr_available = False self._attr_available = False
def update(self): def update(self) -> None:
"""Get the latest data from rtorrent and updates the state.""" """Get the latest data from rtorrent and updates the state."""
multicall = xmlrpc.client.MultiCall(self.client) multicall = xmlrpc.client.MultiCall(self.client)
multicall.throttle.global_up.rate() multicall.throttle.global_up.rate()
@ -152,7 +156,7 @@ class RTorrentSensor(SensorEntity):
multicall.d.multicall2("", "leeching", "d.down.rate=") multicall.d.multicall2("", "leeching", "d.down.rate=")
try: try:
self.data = multicall() self.data = cast(RTorrentData, multicall())
self._attr_available = True self._attr_available = True
except (xmlrpc.client.ProtocolError, OSError) as ex: except (xmlrpc.client.ProtocolError, OSError) as ex:
_LOGGER.error("Connection to rtorrent failed (%s)", ex) _LOGGER.error("Connection to rtorrent failed (%s)", ex)
@ -164,14 +168,16 @@ class RTorrentSensor(SensorEntity):
all_torrents = self.data[2] all_torrents = self.data[2]
stopped_torrents = self.data[3] stopped_torrents = self.data[3]
complete_torrents = self.data[4] complete_torrents = self.data[4]
up_torrents = self.data[5]
down_torrents = self.data[6]
uploading_torrents = 0 uploading_torrents = 0
for up_torrent in self.data[5]: for up_torrent in up_torrents:
if up_torrent[0]: if up_torrent[0]:
uploading_torrents += 1 uploading_torrents += 1
downloading_torrents = 0 downloading_torrents = 0
for down_torrent in self.data[6]: for down_torrent in down_torrents:
if down_torrent[0]: if down_torrent[0]:
downloading_torrents += 1 downloading_torrents += 1