-# SPDX-FileCopyrightText: 2012-2022 kaliko <kaliko@azylum.org>
+# SPDX-FileCopyrightText: 2012-2023 kaliko <kaliko@azylum.org>
# SPDX-FileCopyrightText: 2021 Wonko der Verständige <wonko@hanstool.org>
# SPDX-FileCopyrightText: 2019 Naglis Jonaitis <naglis@mailbox.org>
# SPDX-FileCopyrightText: 2019 Bart Van Loon <bbb@bbbart.be>
# SPDX-FileCopyrightText: 2008-2010 J. Alexander Treuman <jat@spatialrift.net>
-# SPDX-License-Identifier: GPL-3.0-or-later
+# SPDX-License-Identifier: LGPL-3.0-or-later
"""python-musicpd: Python Music Player Daemon client library"""
import os
from functools import wraps
+# Type hint for python <= 3.8
+from typing import Any, Dict, List, Tuple
+from typing import Iterator, Optional, Union
HELLO_PREFIX = "OK MPD "
ERROR_PREFIX = "ACK "
SUCCESS = "OK"
NEXT = "list_OK"
-VERSION = '0.8.0b0'
+VERSION = '0.9.0b0'
#: Seconds before a connection attempt times out
#: (overriden by MPD_TIMEOUT env. var.)
-CONNECTION_TIMEOUT = 30
+CONNECTION_TIMEOUT: int = 30
#: Socket timeout in second (Default is None for no timeout)
-SOCKET_TIMEOUT = None
+SOCKET_TIMEOUT: Union[int, None] = None
def iterator_wrapper(func):
class Range:
- def __init__(self, tpl):
+ def __init__(self, tpl: Tuple[int]):
self.tpl = tpl
self._check()
will not be used as default value for the :py:meth:`password` method
"""
- def __init__(self):
- self.iterate = False
+ def __init__(self) -> None:
+ self.iterate: bool = False
#: Socket timeout value in seconds
self._socket_timeout = SOCKET_TIMEOUT
#: Current connection timeout value, defaults to
#: :py:obj:`CONNECTION_TIMEOUT` or env. var. ``MPD_TIMEOUT`` if provided
- self.mpd_timeout = None
+ self.mpd_timeout: Union[None,int] = None
+ #: Protocol version as exposed by the server
+ self.mpd_version: str = ''
self._reset()
self._commands = {
# Status Commands
}
self._get_envvars()
- def _get_envvars(self):
+ def _get_envvars(self) -> None:
"""
Retrieve MPD env. var. to overrides "localhost:6600"
Use MPD_HOST/MPD_PORT if set
else use MPD_HOST=${XDG_RUNTIME_DIR:-/run/}/mpd/socket if file exists
"""
- self.host = 'localhost'
- self.pwd = None
- self.port = os.getenv('MPD_PORT', '6600')
- if os.getenv('MPD_HOST'):
+ self.host: str = 'localhost'
+ self.pwd: Union[None, str] = None
+ self.port: Union[int,str] = os.getenv('MPD_PORT', '6600')
+ _host: str = os.getenv('MPD_HOST', '')
+ if _host:
# If password is set: MPD_HOST=pass@host
- if '@' in os.getenv('MPD_HOST'):
- mpd_host_env = os.getenv('MPD_HOST').split('@', 1)
+ if '@' in _host:
+ mpd_host_env = _host.split('@', 1)
if mpd_host_env[0]:
# A password is actually set
self.pwd = mpd_host_env[0]
self.host = '@'+mpd_host_env[1]
else:
# MPD_HOST is a plain host
- self.host = os.getenv('MPD_HOST')
+ self.host = _host
else:
# Is socket there
xdg_runtime_dir = os.getenv('XDG_RUNTIME_DIR', '/run')
rundir = os.path.join(xdg_runtime_dir, 'mpd/socket')
if os.path.exists(rundir):
self.host = rundir
- self.mpd_timeout = os.getenv('MPD_TIMEOUT')
- if self.mpd_timeout and self.mpd_timeout.isdigit():
- self.mpd_timeout = int(self.mpd_timeout)
+ _mpd_timeout = os.getenv('MPD_TIMEOUT', 'X')
+ if _mpd_timeout.isdigit():
+ self.mpd_timeout = int(_mpd_timeout)
else: # Use CONNECTION_TIMEOUT as default even if MPD_TIMEOUT carries gargage
self.mpd_timeout = CONNECTION_TIMEOUT
self._wfile.write(f"{line!s}\n")
self._wfile.flush()
- def _write_command(self, command, args=None):
+ def _write_command(self, command, args: Optional[list[str]] = None):
if args is None:
args = []
parts = [command]
raise CommandError('new line found in the command!')
self._write_line(" ".join(parts))
- def _read_binary(self, amount):
+ def _read_binary(self, amount: int):
chunk = bytearray()
while amount > 0:
result = self._rbfile.read(amount)
amount -= len(result)
return bytes(chunk)
- def _read_line(self, binary=False):
+ def _read_line(self, binary: bool = False):
if binary:
line = self._rbfile.readline().decode('utf-8')
else:
return None
return line
- def _read_pair(self, separator, binary=False):
+ def _read_pair(self, separator: str, binary: bool = False):
line = self._read_line(binary=binary)
if line is None:
return None
raise ProtocolError(f"Could not parse pair: '{line}'")
return pair
- def _read_pairs(self, separator=": ", binary=False):
+ def _read_pairs(self, separator: str =": ", binary: bool =False):
pair = self._read_pair(separator, binary=binary)
while pair:
yield pair
for _, value in self._read_pairs(":"):
yield value
- def _read_objects(self, delimiters=None):
- obj = {}
+ def _read_objects(self, delimiters: Optional[List[str]] = None) -> Iterator[Dict]:
+ obj: Dict[str,Any] = {}
if delimiters is None:
delimiters = []
for key, value in self._read_pairs():
def _fetch_playlist(self):
return self._read_playlist()
- def _fetch_object(self):
+ def _fetch_object(self) -> Dict:
objs = list(self._read_objects())
if not objs:
return {}
return objs[0]
@iterator_wrapper
- def _fetch_objects(self, delimiters):
+ def _fetch_objects(self, delimiters: List[str]):
return self._read_objects(delimiters)
def _fetch_changes(self):
self.mpd_version = line[len(HELLO_PREFIX):].strip()
def _reset(self):
- self.mpd_version = None
+ self.mpd_version = ''
self._iterating = False
self._pending = []
self._command_list = None
self._write_command("noidle")
return self._fetch_list()
- def connect(self, host=None, port=None):
+ def connect(self, host: Optional[str] = None, port: Optional[Union[int, str]] = None):
"""Connects the MPD server
:param str host: hostname, IP or FQDN (defaults to `localhost` or socket, see below for details)
return self._fetch_command_list()
-def escape(text):
+def escape(text: str) -> str:
return text.replace("\\", "\\\\").replace('"', '\\"')
# vim: set expandtab shiftwidth=4 softtabstop=4 textwidth=79:
home / dev / repositories / python-musicpd.git / blobdiff