X-Git-Url: http://git.kaliko.me/?a=blobdiff_plain;f=musicpd.py;h=e779355ae24bf9fa00a49729e055e2ba0dee2f3f;hb=c353490d8fac7cb3f17193b1b2e3b9bb191d47ce;hp=e2d126504e9bfaa72908e399051013573d19979b;hpb=80fdad9f2cad5b3759351ab38bf32cad863cd3cb;p=python-musicpd.git diff --git a/musicpd.py b/musicpd.py index e2d1265..e779355 100644 --- a/musicpd.py +++ b/musicpd.py @@ -1,4 +1,4 @@ -# SPDX-FileCopyrightText: 2012-2022 kaliko +# SPDX-FileCopyrightText: 2012-2023 kaliko # SPDX-FileCopyrightText: 2021 Wonko der Verständige # SPDX-FileCopyrightText: 2019 Naglis Jonaitis # SPDX-FileCopyrightText: 2019 Bart Van Loon @@ -11,6 +11,9 @@ import socket import os from functools import wraps +# Type hint for python <= 3.8 +from typing import Any, Dict, List, Tuple +from typing import Optional, Union HELLO_PREFIX = "OK MPD " ERROR_PREFIX = "ACK " @@ -19,9 +22,9 @@ NEXT = "list_OK" 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): @@ -73,7 +76,7 @@ class IteratingError(MPDError): class Range: - def __init__(self, tpl): + def __init__(self, tpl: Tuple[int]): self.tpl = tpl self._check() @@ -145,13 +148,15 @@ class MPDClient: 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 @@ -286,19 +291,20 @@ class MPDClient: } 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] @@ -309,16 +315,16 @@ class MPDClient: 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 @@ -412,7 +418,7 @@ class MPDClient: 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: @@ -433,7 +439,7 @@ class MPDClient: 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 @@ -442,7 +448,7 @@ class MPDClient: raise ProtocolError(f"Could not parse pair: '{line}'") return pair - def _read_pairs(self, separator=": ", binary=False): + def _read_pairs(self, separator=": ", binary: bool =False): pair = self._read_pair(separator, binary=binary) while pair: yield pair @@ -461,8 +467,8 @@ class MPDClient: for _, value in self._read_pairs(":"): yield value - def _read_objects(self, delimiters=None): - obj = {} + def _read_objects(self, delimiters: Optional[List[str]] = None): + obj: Dict[str,Any] = {} if delimiters is None: delimiters = [] for key, value in self._read_pairs(): @@ -585,7 +591,7 @@ class MPDClient: 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 @@ -639,7 +645,7 @@ class MPDClient: 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) @@ -750,7 +756,7 @@ class MPDClient: 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: