# Copyright (c) 2018-2020 Trim21 <i@trim21.me>
# Copyright (c) 2008-2014 Erik Svensson <erik.public@gmail.com>
# Licensed under the MIT license.
import datetime
from typing import Any, Tuple, NamedTuple
import transmission_rpc.constants as constants
from transmission_rpc.constants import LOGGER
UNITS = ['B', 'KiB', 'MiB', 'GiB', 'TiB', 'PiB', 'EiB']
[docs]def rpc_bool(arg: Any) -> int:
"""
Convert between Python boolean and Transmission RPC boolean.
"""
if isinstance(arg, str):
try:
arg = bool(int(arg))
except ValueError:
arg = arg.lower() in ['true', 'yes']
return 1 if bool(arg) else 0
TR_TYPE_MAP = {
'number': int,
'string': str,
'double': float,
'boolean': rpc_bool,
'array': list,
'object': dict,
}
[docs]def make_python_name(name):
"""
Convert Transmission RPC name to python compatible name.
"""
return name.replace('-', '_')
[docs]def make_rpc_name(name):
"""
Convert python compatible name to Transmission RPC name.
"""
return name.replace('_', '-')
[docs]def argument_value_convert(method, argument, value, rpc_version):
"""
Check and fix Transmission RPC issues with regards to methods, arguments and values.
"""
if method in ('torrent-add', 'torrent-get', 'torrent-set'):
args = constants.TORRENT_ARGS[method[-3:]]
elif method in ('session-get', 'session-set'):
args = constants.SESSION_ARGS[method[-3:]]
else:
return ValueError('Method "%s" not supported' % (method))
if argument in args:
info = args[argument]
invalid_version = True
while invalid_version:
invalid_version = False
replacement = None
if rpc_version < info[1]:
invalid_version = True
replacement = info[3]
if info[2] and info[2] <= rpc_version:
invalid_version = True
replacement = info[4]
if invalid_version:
if replacement:
LOGGER.warning(
f'Replacing requested argument "{argument}" with "{replacement}".'
)
argument = replacement
info = args[argument]
else:
raise ValueError(
f'Method "{method}" Argument "{argument}" does not exist in version {rpc_version:d}.'
)
return argument, TR_TYPE_MAP[info[0]](value)
else:
raise ValueError('Argument "%s" does not exists for method "%s".', (argument, method))
[docs]def get_arguments(method, rpc_version):
"""
Get arguments for method in specified Transmission RPC version.
"""
if method in ('torrent-add', 'torrent-get', 'torrent-set'):
args = constants.TORRENT_ARGS[method[-3:]]
elif method in ('session-get', 'session-set'):
args = constants.SESSION_ARGS[method[-3:]]
else:
return ValueError('Method "%s" not supported' % (method))
accessible = []
for argument, info in args.items():
valid_version = True
if rpc_version < info[1]:
valid_version = False
if info[2] and info[2] <= rpc_version:
valid_version = False
if valid_version:
accessible.append(argument)
return accessible
[docs]class Field(NamedTuple):
value: Any
dirty: bool