# Copyright (c) 2008-2014 Erik Svensson <erik.public@gmail.com>
# Licensed under the MIT license.
import datetime
from typing import TYPE_CHECKING, Any, Dict, List, Union, Optional
from transmission_rpc.utils import format_timedelta
from transmission_rpc.constants import PRIORITY, IDLE_LIMIT, RATIO_LIMIT
from transmission_rpc.lib_types import File, Field, _Timeout
if TYPE_CHECKING:
from transmission_rpc.client import Client
def get_status_old(code: int) -> str:
"""Get the torrent status using old status codes"""
mapping = {
(1 << 0): 'check pending',
(1 << 1): 'checking',
(1 << 2): 'downloading',
(1 << 3): 'seeding',
(1 << 4): 'stopped',
}
return mapping[code]
def get_status_new(code: int) -> str:
"""Get the torrent status using new status codes"""
mapping = {
0: 'stopped',
1: 'check pending',
2: 'checking',
3: 'download pending',
4: 'downloading',
5: 'seed pending',
6: 'seeding',
}
return mapping[code]
[docs]class Torrent:
"""
Torrent is a class holding the data received from Transmission regarding a bittorrent transfer.
All fetched torrent fields are accessible through this class using attributes.
This class has a few convenience properties using the torrent data.
"""
def __init__(self, client: 'Client', fields: Dict[str, Any]):
if 'id' not in fields:
raise ValueError('Torrent requires an id')
self._fields: Dict[str, Field] = {}
self._update_fields(fields)
self._incoming_pending = False
self._outgoing_pending = False
self._client = client
@property
def id(self) -> int:
return self._fields['id'].value
def _get_name_string(self) -> Optional[str]:
"""Get the name"""
name = None
# try to find name
if 'name' in self._fields:
name = self._fields['name'].value
return name
def __repr__(self) -> str:
tid = self._fields['id'].value
name = self._get_name_string()
if name is not None:
return f'<Torrent {tid} "{name}">'
else:
return f'<Torrent {tid}>'
def __str__(self) -> str:
name = self._get_name_string()
if name is not None:
return f'Torrent "{name}"'
else:
return 'Torrent'
def __copy__(self) -> 'Torrent':
return Torrent(self._client, self._fields)
def __getattr__(self, name: str) -> Any:
try:
return self._fields[name].value
except KeyError:
raise AttributeError('No attribute %s' % name)
def _rpc_version(self) -> int:
"""Get the Transmission RPC API version."""
if self._client:
return self._client.rpc_version
return 2
def _dirty_fields(self) -> List[str]:
"""Enumerate changed fields"""
outgoing_keys = [
'bandwidthPriority', 'downloadLimit', 'downloadLimited', 'peer_limit', 'queuePosition',
'seedIdleLimit', 'seedIdleMode', 'seedRatioLimit', 'seedRatioMode', 'uploadLimit',
'uploadLimited'
]
fields = []
for key in outgoing_keys:
if key in self._fields and self._fields[key].dirty:
fields.append(key)
return fields
def _push(self) -> None:
"""Push changed fields to the server"""
dirty = self._dirty_fields()
args = {}
for key in dirty:
args[key] = self._fields[key].value
self._fields[key] = self._fields[key]._replace(dirty=False)
if len(args) > 0:
self._client.change_torrent(self.id, **args)
def _update_fields(self, other: Union['Torrent', Dict[str, Any]]) -> None:
"""
Update the torrent data from a Transmission JSON-RPC arguments dictionary
"""
if isinstance(other, dict):
for key, value in other.items():
self._fields[key.replace('-', '_')] = Field(value, False)
elif isinstance(other, Torrent):
for key in list(other._fields.keys()):
self._fields[key] = Field(other._fields[key].value, False)
else:
raise ValueError('Cannot update with supplied data')
self._incoming_pending = False
def _status(self) -> str:
"""Get the torrent status"""
code = self._fields['status'].value
if self._rpc_version() >= 14:
return get_status_new(code)
else:
return get_status_old(code)
[docs] def files(self) -> List[File]:
"""
Get list of files for this torrent.
This function returns a dictionary with file information for each file.
The file information is has following fields:
"""
result: List[File] = []
if 'files' in self._fields:
files = self._fields['files'].value
indices = range(len(files))
priorities = self._fields['priorities'].value
wanted = self._fields['wanted'].value
for item in zip(indices, files, priorities, wanted):
selected = True if item[3] else False
priority = PRIORITY[item[2]]
result.append(
File(
selected=selected,
priority=priority,
size=item[1]['length'],
name=item[1]['name'],
completed=item[1]['bytesCompleted'],
)
)
return result
@property
def status(self) -> str:
"""
Returns the torrent status. Is either one of 'check pending', 'checking',
'downloading', 'seeding' or 'stopped'. The first two is related to
verification.
"""
return self._status()
@property
def progress(self) -> float:
"""download progress in percent."""
try:
size = self._fields['sizeWhenDone'].value
left = self._fields['leftUntilDone'].value
return 100.0 * (size - left) / float(size)
except ZeroDivisionError:
return 0.0
@property
def ratio(self) -> float:
"""upload/download ratio."""
return float(self._fields['uploadRatio'].value)
@property
def eta(self) -> datetime.timedelta:
"""the "eta" as datetime.timedelta."""
eta = self._fields['eta'].value
if eta >= 0:
return datetime.timedelta(seconds=eta)
else:
raise ValueError('eta not valid')
@property
def date_active(self) -> datetime.datetime:
"""the attribute "activityDate" as datetime.datetime."""
return datetime.datetime.fromtimestamp(self._fields['activityDate'].value)
@property
def date_added(self) -> datetime.datetime:
"""the attribute "addedDate" as datetime.datetime."""
return datetime.datetime.fromtimestamp(self._fields['addedDate'].value)
@property
def date_started(self) -> datetime.datetime:
"""the attribute "startDate" as datetime.datetime."""
return datetime.datetime.fromtimestamp(self._fields['startDate'].value)
@property
def date_done(self) -> Optional[datetime.datetime]:
"""the attribute "doneDate" as datetime.datetime. returns None if "doneDate" is invalid."""
done_date = self._fields['doneDate'].value
# Transmission might forget to set doneDate which is initialized to zero, so if doneDate is zero return None
if done_date == 0:
return None
else:
return datetime.datetime.fromtimestamp(done_date)
@property
def download_limit(self) -> Optional[int]:
"""The download limit.
Can be a number or None.
"""
if self._fields['downloadLimited'].value:
return self._fields['downloadLimit'].value
else:
return None
@download_limit.setter
def download_limit(self, limit: int) -> None:
"""Download limit in Kbps or None."""
if isinstance(limit, int):
self._fields['downloadLimited'] = Field(True, True)
self._fields['downloadLimit'] = Field(limit, True)
self._push()
elif limit is None:
self._fields['downloadLimited'] = Field(False, True)
self._push()
else:
raise ValueError('Not a valid limit')
@property
def peer_limit(self) -> int:
"""the peer limit."""
return self._fields['peer_limit'].value
@peer_limit.setter
def peer_limit(self, limit: int) -> None:
"""
Set the peer limit.
"""
if isinstance(limit, int):
self._fields['peer_limit'] = Field(limit, True)
self._push()
else:
raise ValueError('Not a valid limit')
@property
def priority(self) -> str:
"""Bandwidth priority as string. Can be one of 'low', 'normal', 'high'. This is a mutator."""
return PRIORITY[self._fields['bandwidthPriority'].value]
@priority.setter
def priority(self, priority: str) -> None:
if isinstance(priority, str):
self._fields['bandwidthPriority'] = Field(PRIORITY[priority], True)
self._push()
@property
def seed_idle_limit(self) -> int:
"""
seed idle limit in minutes.
"""
return self._fields['seedIdleLimit'].value
@seed_idle_limit.setter
def seed_idle_limit(self, limit: int) -> None:
"""
Set the seed idle limit in minutes.
"""
if isinstance(limit, int):
self._fields['seedIdleLimit'] = Field(limit, True)
self._push()
else:
raise ValueError('Not a valid limit')
@property
def seed_idle_mode(self) -> str:
"""
Seed idle mode as string. Can be one of 'global', 'single' or 'unlimited'.
* global, use session seed idle limit.
* single, use torrent seed idle limit. See seed_idle_limit.
* unlimited, no seed idle limit.
"""
return IDLE_LIMIT[self._fields['seedIdleMode'].value]
@seed_idle_mode.setter
def seed_idle_mode(self, mode: Union[str, int]) -> None:
"""
Set the seed ratio mode. Can be one of ``global``, ``single`` or ``unlimited``, or ``0``, ``1``, ``2``.
"""
if isinstance(mode, str):
self._fields['seedIdleMode'] = Field(IDLE_LIMIT[mode], True)
self._push()
elif isinstance(mode, int):
self._fields['seedIdleMode'] = Field(IDLE_LIMIT[mode], True)
self._push()
else:
raise ValueError('Not a valid limit')
@property
def seed_ratio_limit(self) -> float:
"""Torrent seed ratio limit as float. Also see seed_ratio_mode. This is a mutator."""
return float(self._fields['seedRatioLimit'].value)
@seed_ratio_limit.setter
def seed_ratio_limit(self, limit: Union[int, float]) -> None:
"""
Set the seed ratio limit as float.
"""
if isinstance(limit, (int, float)) and limit >= 0.0:
self._fields['seedRatioLimit'] = Field(float(limit), True)
self._push()
else:
raise ValueError('Not a valid limit')
@property
def seed_ratio_mode(self) -> str:
"""
Seed ratio mode as string. Can be one of 'global', 'single' or 'unlimited'.
* global, use session seed ratio limit.
* single, use torrent seed ratio limit. See seed_ratio_limit.
* unlimited, no seed ratio limit.
This is a mutator.
"""
return RATIO_LIMIT[self._fields['seedRatioMode'].value]
@seed_ratio_mode.setter
def seed_ratio_mode(self, mode: Union[str, int]) -> None:
"""
Set the seed ratio mode.
Can be one of ``'global'``, ``'single'`` or ``'unlimited'``, or ``0``, ``1``, ``2``.
"""
if isinstance(mode, str):
self._fields['seedRatioMode'] = Field(RATIO_LIMIT[mode], True)
self._push()
elif isinstance(mode, int):
self._fields['seedRatioMode'] = Field(mode, True)
self._push()
else:
raise ValueError('Not a valid limit')
@property
def upload_limit(self) -> Optional[int]:
"""
upload limit.
Can be a number or None.
"""
if self._fields['uploadLimited'].value:
return self._fields['uploadLimit'].value
else:
return None
@upload_limit.setter
def upload_limit(self, limit: Optional[int]) -> None:
"""Upload limit in Kbps or None."""
if isinstance(limit, int):
self._fields['uploadLimited'] = Field(True, True)
self._fields['uploadLimit'] = Field(limit, True)
self._push()
elif limit is None:
self._fields['uploadLimited'] = Field(False, True)
self._push()
else:
raise ValueError('Not a valid limit')
@property
def queue_position(self) -> int:
"""queue position for this torrent."""
if self._rpc_version() >= 14:
return self._fields['queuePosition'].value
else:
return 0
@queue_position.setter
def queue_position(self, position: str) -> None:
"""Queue position"""
if self._rpc_version() >= 14:
if isinstance(position, int):
self._fields['queuePosition'] = Field(position, True)
self._push()
else:
raise ValueError('Not a valid position')
else:
pass
[docs] def update(self, timeout: _Timeout = None) -> None:
"""Update the torrent information."""
self._push()
torrent = self._client.get_torrent(self.id, timeout=timeout)
self._update_fields(torrent)
[docs] def start(self, bypass_queue: bool = False, timeout: _Timeout = None) -> None:
"""
Start the torrent.
"""
self._incoming_pending = True
self._client.start_torrent(self.id, bypass_queue=bypass_queue, timeout=timeout)
[docs] def stop(self, timeout: _Timeout = None) -> None:
"""Stop the torrent."""
self._incoming_pending = True
self._client.stop_torrent(self.id, timeout=timeout)
[docs] def move_data(self, location: str, timeout: _Timeout = None) -> None:
"""Move torrent data to location."""
self._incoming_pending = True
self._client.move_torrent_data(self.id, location, timeout=timeout)
[docs] def locate_data(self, location: str, timeout: _Timeout = None) -> None:
"""Locate torrent data at location."""
self._incoming_pending = True
self._client.locate_torrent_data(self.id, location, timeout=timeout)