-# python-musicpd: Python MPD client library
-# Copyright (C) 2012-2021 kaliko <kaliko@azylum.org>
-# Copyright (C) 2019 Naglis Jonaitis <naglis@mailbox.org>
-# Copyright (C) 2019 Bart Van Loon <bbb@bbbart.be>
-# Copyright (C) 2008-2010 J. Alexander Treuman <jat@spatialrift.net>
-#
-# python-musicpd is free software: you can redistribute it and/or modify
-# it under the terms of the GNU Lesser General Public License as published by
-# the Free Software Foundation, either version 3 of the License, or
-# (at your option) any later version.
-#
-# python-musicpd is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU Lesser General Public License for more details.
-#
-# You should have received a copy of the GNU Lesser General Public License
-# along with python-musicpd. If not, see <http://www.gnu.org/licenses/>.
+# 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: LGPL-3.0-or-later
+"""python-musicpd: Python Music Player Daemon client library"""
+
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 "
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
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=": ", 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):
+ obj: Dict[str,Any] = {}
if delimiters is None:
delimiters = []
for key, value in self._read_pairs():
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)
self._sock.close()
self._reset()
+ def __enter__(self):
+ self.connect()
+ return self
+
+ def __exit__(self, exception_type, exception_value, exception_traceback):
+ self.disconnect()
+
def fileno(self):
"""Return the socket’s file descriptor (a small integer).
This is useful with :py:obj:`select.select`.
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