PK!Tampdup/__init__.py# pylint: skip-file from .idle_client import IdleMPDClient, Subsystem from .mpd_client import MPDClient, Tag, Single, SearchType from .song import Song, SongId from .status import State, Status from .errors import * # noqa __all__ = [ 'IdleMPDClient', 'MPDClient', 'Single', 'Song', 'SongId', 'State', 'Status', 'Subsystem', 'Tag', *errors.__all__ # noqa ] PK!ӟampdup/base_client.pyfrom typing import AsyncGenerator, List from dataclasses import dataclass, field from curio.promise import Promise from curio.queue import Queue from curio.task import spawn, Task, TaskCancelled from .connection import Connection from .parsing import parse_error from .util import asynccontextmanager @dataclass class BaseMPDClient: pending_commands: Queue = field(default_factory=Queue) connection: Connection = None loop: Task = None @classmethod @asynccontextmanager async def make(cls, address: str, port: int) -> AsyncGenerator: c = cls() await c.connect(address, port) yield c await c.stop_loop() async def connect(self, address: str, port: int): self.connection = Connection() await self.connection.connect(address, port) await self._start_loop() async def run_command(self, command: str) -> List[str]: p = Promise() await self.pending_commands.put(p) await self.connection.write_line(command) result = await p.get() if isinstance(result, Exception): raise result return result async def stop_loop(self): if self.loop: await self.loop.cancel() async def _read_response(self) -> List[str]: lines: List[str] = [] while True: line = await self.connection.read_line() if line.startswith('OK'): break if line.startswith('ACK'): return parse_error(line, lines) lines.append(line) return lines async def _reply_loop(self): try: while True: reply = await self._read_response() pending = await self.pending_commands.get() await pending.set(reply) except TaskCancelled: return async def _start_loop(self): self.loop = await spawn(self._reply_loop()) PK!''Z##ampdup/connection.pyfrom curio.io import Socket from curio.network import open_connection from dataclasses import dataclass from .errors import ConnectionFailedError from .typing_local import AsyncBinaryIO @dataclass class Connection: connection: Socket = None stream: AsyncBinaryIO = None async def connect(self, address: str, port: int): self.connection = await open_connection(address, port) self.stream = self.connection.as_stream() result = await self.read_line() if not result.startswith('OK MPD'): raise ConnectionFailedError async def write_line(self, command: str): await self.stream.write(command.encode() + b'\n') async def read_line(self) -> str: line = await self.stream.readline() return line.decode().strip('\n') PK!XS?  ampdup/errors.py'''Classes for raising when errors happen.''' from enum import Enum from typing import Callable, Dict, List from dataclasses import dataclass __all__ = [ 'MPDError', 'ConnectionFailedError', 'ClientTypeError', 'NoCurrentSongError', 'CommandError', 'URINotFoundError', ] class ErrorCode(Enum): '''MPD Error codes included in ACK responses.''' NOT_LIST = 1 ARG = 2 PASSWORD = 3 PERMISSION = 4 UNKNOWN = 5 NO_EXIST = 50 PLAYLIST_MAX = 51 SYSTEM = 52 PLAYLIST_LOAD = 53 UPDATE_ALREADY = 54 PLAYER_SYNC = 55 EXIST = 56 class MPDError(Exception): '''Base class for errors raised by this library.''' class ConnectionFailedError(MPDError): '''Caused when a client fails to connect.''' class ClientTypeError(MPDError): '''Caused when trying to use the wrong type of client for an operation. Example: using the idle client for playback control and vice-versa. ''' class NoCurrentSongError(MPDError): '''Caused where there is no current song playing and one is expected.''' pass @dataclass class CommandError(MPDError): '''Wraps an error from MPD (an ACK response with error code and reason). Members: code: The MPD error code. line: Which line in a command list caused the error. command: The command that caused the error. message: The error message provided by MPD, unchanged. partial: The (maybe empty) list of lines representing a partial response. ''' code: ErrorCode line: int command: str message: str partial: List[str] def __post_init__(self): codetext = f'{self.code.name}/{self.code.value}' super().__init__( f'[{codetext}@{self.line}] {{{self.command}}} {self.message}' ) class URINotFoundError(CommandError): '''Wraps an error 50 from MPD.''' def __init__(self, *args): super().__init__(ErrorCode.NO_EXIST, *args) ErrorFactory = Callable[[int, str, str, List[str]], CommandError] ERRORS: Dict[ErrorCode, ErrorFactory] = { ErrorCode.NO_EXIST: URINotFoundError, } def get_error_constructor(error_code: ErrorCode) -> ErrorFactory: '''Get the error constructor for an error code, or a generic one. If the error code is not mapped to an exception type, a factory function for CommandError with the correct code is returned. Args: error_code: The error code from MPD. Returns: A function that constructs the correct exception with the remaining arguments. ''' return (ERRORS.get(error_code) or (lambda *args: CommandError(error_code, *args))) PK!wooampdup/idle_client.py'''Idle client module.''' from enum import Enum from typing import List from .base_client import BaseMPDClient from .errors import ClientTypeError from .parsing import split_item from .util import has_any_prefix class Subsystem(Enum): '''Enumeration of available subsystems for idle to listen to.''' DATABASE = 'database' UPDATE = 'update' STORED_PLAYLIST = 'stored_playlist' PLAYLIST = 'playlist' PLAYER = 'player' MIXER = 'mixer' OUTPUT = 'output' OPTIONS = 'options' PARTITION = 'partition' STICKER = 'sticker' SUBSCRIPTION = 'subscription' MESSAGE = 'message' class IdleMPDClient(BaseMPDClient): '''Client that is only capable of running the idle command. More information in the idle() method docstring. ''' async def run_command(self, command: str): if not has_any_prefix(command, ('idle', 'noidle')): raise ClientTypeError( 'Use an MPDClient to run commands other than idle and noidle.' ) return await super().run_command(command) async def idle(self, *subsystems: Subsystem) -> List[Subsystem]: '''Run the idle command, fetching events from the player. Args: *subsystems (Subsystem): Subsystems to listen to (variadic). If empty or omitted listens to all systems. Returns: List[Subsystem]: subsystems that changed since the command was called. ''' subsystem_names = [s.value for s in subsystems] command = ' '.join(['idle', *subsystem_names]) changed = (split_item(i) for i in await self.run_command(command)) return [Subsystem(s) for _, s in changed] async def noidle(self): '''Cancel the current idle command.''' return await self.connection.write_line('noidle') PK!L<<ampdup/mpd_client.py'''MPD Client module.''' from enum import Enum from typing import List, Tuple, Union, Optional from .base_client import BaseMPDClient from .errors import ClientTypeError, NoCurrentSongError from .song import Song, SongId, TimeRange from .parsing import from_lines, parse_playlist, parse_single from .stats import Stats from .status import Single, Status from .util import has_any_prefix Range = Tuple[int, int] PositionOrRange = Union[int, Range] class Tag(Enum): '''Tags supported by MPD.''' ARTIST = 'artist' ARTISTSORT = 'artistsort' ALBUM = 'album' ALBUMSORT = 'albumsort' ALBUMARTIST = 'albumartist' ALBUMARTISTSORT = 'albumartistsort' TITLE = 'title' TRACK = 'track' NAME = 'name' GENRE = 'genre' DATE = 'date' COMPOSER = 'composer' PERFORMER = 'performer' COMMENT = 'comment' DISC = 'disc' MUSICBRAINZ_ARTISTID = 'musicbrainz_artistid' MUSICBRAINZ_ALBUMID = 'musicbrainz_albumid' MUSICBRAINZ_ALBUMARTISTID = 'musicbrainz_albumartistid' MUSICBRAINZ_TRACKID = 'musicbrainz_trackid' MUSICBRAINZ_RELEASETRACKID = 'musicbrainz_releasetrackid' MUSICBRAINZ_WORKID = 'musicbrainz_workid' class SearchType(Enum): '''Special types for searching the database.''' ANY = 'any' FILE = 'file' BASE = 'base' MODIFIED_SINCE = 'modified-since' AnySearchType = Union[Tag, SearchType] def position_or_range_arg(arg: Optional[PositionOrRange]) -> str: '''Make argument string for commands that may take a position or a range. Args: arg: Either a position as an int or a range as a tuple. Returns: The correspondent string to the argument. ''' if arg is None: return '' if isinstance(arg, int): return f' {arg}' start, end = arg return f' {start}:{end}' def find_args( queries: List[Tuple[AnySearchType, str]] ) -> str: '''Make argument string for find and search. Args: queries: A list of queries (type and what). Returns: An argument string. ''' return ' '.join(f'{type.value} "{what}"' for type, what in queries) class MPDClient(BaseMPDClient): '''An async MPD Client object for any operations except idle/noidle.''' async def run_command(self, command: str) -> List[str]: if has_any_prefix(command, ('idle', 'noidle')): raise ClientTypeError('Use an IdleClient to use the idle command.') return await super().run_command(command) # MPD status async def current_song(self) -> Song: '''Displays the song info of the current song. Returns: The metadata of the song that is identified in status. ''' result = await self.run_command('currentsong') if not result: raise NoCurrentSongError('There is no current song playing.') return from_lines(Song, result) async def status(self) -> Status: '''Get the current player status. Returns: The current player status. ''' result = await self.run_command('status') return from_lines(Status, result) async def stats(self) -> Stats: '''Get player stats. Refer to the Stats type to see what is available. Returns: Statistics about the player. ''' result = await self.run_command('stats') return from_lines(Stats, result) # Playback options async def consume(self, state: bool): '''Enable or disable consume. When consume is enabled, played songs are removed from the playlist. Args: state: True for consume, otherwise False. ''' await self.run_command(f'consume {int(state)}') async def single(self, mode: Single): '''Enable, disable or set to oneshot single playback. Args: mode: True for random, False for sequential. ''' await self.run_command(f'single {mode.value}') async def random(self, state: bool): '''Enable or disable random playback. Args: state: True for random, False for sequential. ''' await self.run_command(f'random {int(state)}') async def repeat(self, state: bool): '''Enable or disable repeat. Args: state: True to repeat, False to play once. ''' await self.run_command(f'repeat {int(state)}') async def setvol(self, amount: int): '''Set volume. Args: amount: The new volume, from 0 to 100. ''' await self.run_command(f'setvol {amount}') # Playback control async def next(self): '''Play next song in the playlist.''' await self.run_command(f'next') async def pause(self, pause: bool): '''Pause or resume playback. Args: pause: Whether to pause (`True`) or resume (`False`). ''' await self.run_command(f'pause {int(pause)}') async def play(self, pos: Optional[int] = None): '''Begin playback. If supplied, start at `pos` in the playlist. Args: pos: The position in the playlist where to begin playback. If omitted and playback is paused, resume it. If playback was stopped, start from the beginning. ''' arg = '' if pos is None else f' {pos}' await self.run_command(f'play{arg}') async def play_id(self, song_id: Optional[SongId] = None): '''Begin playback. If supplied, start at the song with id `song_id`. Args: song_id: The id of the song in the playlist where to begin playback. If omitted and playback is paused, resume it. If playback was stopped, start from the beginning. ''' arg = '' if song_id is None else f' {song_id}' await self.run_command(f'playid{arg}') async def previous(self): '''Play previous song in the playlist.''' await self.run_command(f'previous') async def seek(self, pos: int, time: float): '''Seek to a certain time of the `pos` entry in the playlist. Args: pos: The position in the playlist of the song to seek. time: The timestamp to seek to in seconds (fractions allowed). ''' await self.run_command(f'seek {pos} {time}') async def seek_id(self, song_id: SongId, time: float): '''Seek to a certain time of the song with id `song_id` in the playlist. Args: song_id: The id of the song in the playlist to seek. time: The timestamp to seek to in seconds (fractions allowed). ''' await self.run_command(f'seekid {song_id} {time}') async def seek_cur(self, time: float): '''Seek to a certain time of the current song. Args: time: The timestamp to seek to in seconds (fractions allowed). ''' await self.run_command(f'seekcur {time}') async def stop(self): '''Stop playback.''' await self.run_command(f'stop') # Current playlist async def add(self, uri: str): '''Add a directory or a file to the current playlist. Args: uri: The URI of what to add. Directories are added recursively. ''' await self.run_command(f'add "{uri}"') async def add_id(self, song_uri: str, position: int = None) -> SongId: '''Add a directory or a file to the current playlist. Args: song_uri: The URI of the song to add. Must be a single file. position: Position on the playlist where to add. End of playlist if omitted. Returns: The id of the added song. ''' pos = '' if position is None else f' {position}' result = await self.run_command(f'addid "{song_uri}"{pos}') return parse_single(result, SongId) async def clear(self): '''Clears the current playlist.''' await self.run_command('clear') async def delete( self, position_or_range: PositionOrRange, ): '''Delete songs from the playlist. Args: position_or_range: either an integer pointing to a specific position in the playlist or an interval. ''' arg = position_or_range_arg(position_or_range) await self.run_command(f'delete{arg}') async def delete_id(self, song_id: SongId): '''Delete a song by its id. Args id: a song id. ''' await self.run_command(f'deleteid {song_id}') async def move(self, what: PositionOrRange, to: int): '''Move a song or a range to another position in the playlist. Args: what: A position or range of songs to move. to: The position on the current state of the playlist where the songs should be moved to. ''' what_arg = position_or_range_arg(what) await self.run_command(f'move {what_arg} {to}') async def move_id(self, song_id: SongId, to: int): '''Move a song by its id to a position in the playlist. If to is negative, it is relative to the current song in the playlist (if there is one). Args: song_id: An id of a song in the playlist. to: The position on the current state of the playlist where the songs should be moved to. ''' await self.run_command(f'moveid {song_id} {to}') async def playlist_find(self, tag: Tag, needle: str) -> List[Song]: '''Search strictly for `needle` among the `tag` values. Args: tag: Which tag to search for. needle: What to search for. Returns: ''' result = await self.run_command( f'playlistfind {tag.value} "{needle}"' ) return parse_playlist(result) async def playlist_id( self, song_id: Optional[SongId] = None ) -> List[Song]: '''Get information about a particular song in the playlist. Args: song_id: The id of the song to search for. If omitted, returns the whole playlist, like `playlist_info`. Returns: A list of Song objects representing the current playlist. ''' arg = f' {song_id}' if song_id else '' result = await self.run_command(f'playlistid{arg}') return parse_playlist(result) async def playlist_info( self, position_or_range: Optional[PositionOrRange] = None, ) -> List[Song]: '''Get information about every song in the current playlist. Args: position_or_range: either an integer pointing to a specific position in the playlist or an interval. If ommited, returns the whole playlist. Returns: A list of Song objects representing the current playlist. ''' arg = position_or_range_arg(position_or_range) result = await self.run_command(f'playlistinfo{arg}') return parse_playlist(result) async def playlist_search(self, tag: Tag, needle: str) -> List[Song]: '''Search case-insensitively for `needle` among the `tag` values. Args: tag: Which tag to search for. needle: What to search for. Returns: ''' result = await self.run_command( f'playlistsearch {tag.value} "{needle}"' ) return parse_playlist(result) async def prio(self, priority: int, song_range: Range): '''Set a song priority for random playback. Args: priority: A value between 0 and 255. song_range: The range of songs in the playlist to change. ''' start, end = song_range start_arg = '' if start is None else f'{start}' end_arg = '' if end is None else f'{end}' result = await self.run_command( f'prio {priority} {start_arg}:{end_arg}' ) return parse_playlist(result) async def prio_id(self, priority: int, song_id: SongId): '''Set a song priority for random playback. Args: priority: A value between 0 and 255. song_range: The range of songs in the playlist to change. ''' result = await self.run_command(f'prioid {priority} {song_id}') return parse_playlist(result) async def range_id( self, song_id: SongId, time_range: TimeRange = (None, None), ): '''Specify the portion of the song that shall be played. Args: song_id: The id of the target song. time_range: A pair of offsets in seconds (fractions allowed). If omitted, removes any range, ''' start, end = time_range start_arg = '' if start is None else f'{start}' end_arg = '' if end is None else f'{end}' await self.run_command(f'rangeid {song_id} {start_arg}:{end_arg}') async def shuffle(self, shuffle_range: Optional[Range] = None): '''Shuffle the current playlist. Args: shuffle_range: An optional range to shuffle. If ommitted, shuffles the whole playlist. ''' arg = position_or_range_arg(shuffle_range) await self.run_command(f'shuffle{arg}') async def swap(self, s1: int, s2: int): '''Swap two songs. Args: s1: the first song's position. s2: the second song's position. ''' await self.run_command(f'swap {s1} {s2}') async def swap_id(self, s1: SongId, s2: SongId): '''Swap two songs by id. Args: s1: the first song's id. s2: the second song's id. ''' await self.run_command(f'swapid {s1} {s2}') # Music database async def find( self, queries: List[Tuple[AnySearchType, str]] ) -> List[Song]: '''Search strictly in the music database. Args: Returns: ''' result = await self.run_command(f'find {find_args(queries)}') return parse_playlist(result) async def search( self, queries: List[Tuple[AnySearchType, str]] ) -> List[Song]: '''Search case-insensitively in the music database. Args: Returns: ''' result = await self.run_command(f'search {find_args(queries)}') return parse_playlist(result) async def update(self, uri: str = None) -> int: '''Update the database. Args: uri: An optional URI to specify which file or directory to update. If omitted, everything is updated. Returns: The id of the update job. ''' result = await self.run_command(f'update "{uri}"') return parse_single(result, int) async def rescan(self, uri: str = None) -> int: '''Update the database rescanning unmodified files as well. Args: uri: An optional URI to specify which file or directory to update. If omitted, everything is updated. Returns: The id of the update job. ''' result = await self.run_command(f'rescan "{uri}"') return parse_single(result, int) PK!  ampdup/parsing.py'''MPD output parsing utilities.''' import re from typing import overload, Callable, Iterable, List, Tuple, TypeVar, Union from .errors import ErrorCode, get_error_constructor from .song import Song from .util import from_json_like, split_on __all__ = [ 'normalize', 'split_item', 'from_lines', 'parse_single', 'parse_playlist', 'parse_error', ] def normalize(name: str) -> str: '''Normalize a value name to a valid Python (PEP8 compliant) identifier. Args: name: The name of a value returned by MPD. Returns: The normalized name, in all lowercase with - replaced by _. ''' return name.lower().replace('-', '_') def split_item(item: str) -> Tuple[str, str]: '''Split a key/value pair in a string into a tuple (key, value). This also strips space from both sides of either. Args: item: A key/value string in 'key: value' format. Returns: The (key, value) tuple, with both sides stripped. ''' lhs, rhs = item.split(':', maxsplit=1) return lhs.strip(), rhs.strip() def from_lines(cls: type, lines: Iterable[str]): '''Make a `cls` object from a list of lines in MPD output format.''' values = (split_item(l) for l in lines) normalized = {normalize(k): v for k, v in values} return from_json_like(cls, normalized) T = TypeVar('T') def parse_error(error_line: str, partial: List[str]): '''Parse an error from MPD. Errors are of format `ACK [CODE@LINE] {COMMAND} MESSAGE` Args: error_line: an ACK line from MPD. Returns: A CommandError (or subclass) object with the error data. ''' code, line, command, message = ERROR_RE.match(error_line).groups() error_code = ErrorCode(int(code)) return get_error_constructor(error_code)(int(line), command, message, partial) @overload def parse_single( lines: Iterable[str] # pylint: disable=unused-argument ) -> str: '''Overload.''' pass @overload # noqa: F811 def parse_single( # pylint: disable=function-redefined lines: Iterable[str], # pylint: disable=unused-argument cast: Callable[[str], T] # pylint: disable=unused-argument ) -> T: '''Overload.''' pass def parse_single( # noqa: F811, pylint: disable=function-redefined lines: Iterable[str], cast: Callable[[str], T] = None ) -> Union[str, T]: '''Parse a single return value and discard its name. Args: lines: The return from MPD as a list of a single line. cast: An optional function to read the string into another type. Returns: The value as a string or converted into the chosen type. ''' result, = lines _, value = split_item(result) if cast is None: return value return cast(value) def is_file(line: str) -> bool: '''Check if a return line is a song file.''' return line.startswith('file:') def parse_playlist(lines: Iterable[str]) -> List[Song]: '''Parse playlist information into a list of songs.''' split = split_on(is_file, lines) return [from_lines(Song, song_info) for song_info in split] ERROR_RE = re.compile(r'ACK\s+\[(\d+)@(\d+)\]\s+\{(.*)\}\s+(.*)') PK!<ampdup/song.py'''Song metadata representation.''' from typing import NamedTuple, Optional, Tuple class SongId(int): '''Strong alias for song ids.''' pass TimeRange = Tuple[Optional[float], Optional[float]] class Song(NamedTuple): '''Type representing the static data about a playable song in MPD.''' file: str last_modified: str time: int duration: float artist: str = '' artistsort: str = '' albumartist: str = '' albumartistsort: str = '' title: str = '' album: str = '' albumsort: str = '' genre: str = '' disc: int = None track: int = None date: int = None range: TimeRange = None pos: int = None id: SongId = None prio: int = None PK! ampdup/stats.py'''Player statistics.''' from typing import NamedTuple class Stats(NamedTuple): '''Statistics about the player.''' uptime: int playtime: int artists: int albums: int songs: int db_playtime: int db_update: int PK! kkampdup/status.py'''Player state representation.''' from typing import NamedTuple from enum import Enum from .song import SongId class Single(Enum): '''"single" setting state.''' DISABLED = '0' ENABLED = '1' ONESHOT = 'oneshot' class State(Enum): '''Player state.''' PLAY = 'play' STOP = 'stop' PAUSE = 'pause' class Status(NamedTuple): '''Type representing the static data about a playable song in MPD.''' volume: int repeat: bool random: bool single: Single consume: bool playlist: int playlistlength: int mixrampdb: float state: State song: int songid: SongId time: str elapsed: float bitrate: int duration: float audio: str nextsong: int = None nextsongid: SongId = None error: str = None mixrampdelay: int = None updating_db: int = None xfade: int = None PK!b$88ampdup/typing_local.py# pylint: skip-file from abc import abstractmethod from typing import AnyStr, Generic, List, Union class AsyncIO(Generic[AnyStr]): """Generic base class for async TextIO and BinaryIO.""" __slots__ = () @abstractmethod async def readline(self, limit: int = -1) -> AnyStr: pass @abstractmethod async def readlines(self, hint: int = -1) -> List[AnyStr]: pass @abstractmethod async def write(self, s: AnyStr) -> int: pass @abstractmethod async def writelines(self, lines: List[AnyStr]) -> None: pass @abstractmethod async def __aenter__(self) -> 'AsyncIO[AnyStr]': pass @abstractmethod async def __aexit__(self, type, value, traceback) -> None: pass class AsyncBinaryIO(AsyncIO[bytes]): """Typed version of an async IO object, counterparte to typing.BinaryIO.""" __slots__ = () @abstractmethod async def write(self, s: Union[bytes, bytearray]) -> int: pass @abstractmethod async def __aenter__(self) -> 'AsyncBinaryIO': pass PK!h&&ampdup/util.py'''Utility module.''' import sys from enum import Enum, EnumMeta from functools import lru_cache from itertools import groupby from operator import itemgetter from typing import Any, Callable, Iterable, List, Sequence, Tuple, TypeVar from .song import TimeRange if sys.version_info < (3, 7): from aiocontext import async_contextmanager as asynccontextmanager else: from contextlib import asynccontextmanager # noqa # pylint: disable=no-name-in-module __all__ = [ 'asynccontextmanager', ] class NoCommonTypeError(Exception): '''Happens when values in an Enum are not of the same type.''' pass class EmptyEnumError(Exception): '''Happens when an enum has no value.''' pass @lru_cache() def underlying_type(enum_type: EnumMeta) -> type: '''Get the underlying type of an enum. Returns: The type of every value in the enum if it is the same. Raises: NoCommonTypeError: If the values in the enum are not of the same type. EmptyEnumError: If the enum has no value. ''' try: first: Any = next(iter(enum_type)) except StopIteration: raise EmptyEnumError('No value in enum.') t = type(first.value) if any(not isinstance(v.value, t) for v in enum_type): # type: ignore raise NoCommonTypeError('No common type in enum.') return t def is_namedtuple(cls: type) -> bool: '''Checks if a type inherits typing.NamedTuple. Checking inspects the type and looks for the `_field_types` attribute. Args: cls: Type to inspect. Returns: Whether or not the type was generated by NamedTuple. ''' return hasattr(cls, '_field_types') def from_list(list_type, v): '''Make a list of typed objects from a JSON-like list, based on type hints. Args: list_type: A `List[T]`-like type to instantiate . v: A JSON-like list. Returns: list_type: An object of type `list_type`. ''' inner_type, = list_type.__args__ return [from_json_like(inner_type, value) for value in v] def from_dict(cls, d): '''Make an object from a dict, recursively, based on type hints. Args: cls: The type to instantiate. d: a dict of property names to values. Returns: cls: An object of type `cls`. ''' # pylint:disable=protected-access if hasattr(cls, '_renames'): for k in d: if k in cls._renames: d[cls._renames[k]] = d.pop(k) return cls(**{i: from_json_like(cls._field_types[i], v) for i, v in d.items()}) def time_range(s: str) -> TimeRange: '''Parse a time range from a song metadata. Args: s: The time range as a string. Returns: A (float, float) tuple with offsets in seconds. ''' start, end = s.split('-') return float(start), float(end) def from_json_like(cls, j): '''Make an object from a JSON-like value, recursively, based on type hints. Args: cls: The type to instantiate. j: the JSON-like object. Returns: cls: An object of type `cls`. ''' if cls is bool: return cls(int(j)) if any(issubclass(cls, t) for t in (int, float)): return cls(j) if cls is str: return j if cls is TimeRange: return time_range(j) if is_namedtuple(cls): return from_dict(cls, j) if issubclass(cls, Enum): return cls(underlying_type(cls)(j)) if issubclass(cls, List): return from_list(cls, j) raise TypeError(f'{j} cannot be converted into {cls}.') def has_any_prefix(s: str, prefixes: Sequence[str]) -> bool: '''Checks if a string has any of the provided prefixes. Args: s: The string to check. prefixes: A sequence of prefixes to check. Returns: Whether the string matches any of the prefixes. ''' return any(s.startswith(prefix) for prefix in prefixes) T = TypeVar('T') Predicate = Callable[[T], bool] def enumerate_on(pred: Predicate, iterable: Iterable[T], begin: int = 0) -> Iterable[Tuple[int, T]]: '''Generate enumerated tuples based on sentinel elements in an iterable. A sentinel element is one for which `pred` is `True`. From it onwards the enumeration is incremented. Args: pred: A predicate identifying a sentinel element. iterable: An iterable to enumerate. begin: where to begin the enumeration. Returns: The enumerated iterable of tuples. ''' i = begin it = iter(iterable) try: first = next(it) except StopIteration: return yield i, first for e in it: if pred(e): i += 1 yield i, e def split_on(pred: Predicate, iterable: Iterable[T]) -> Iterable[Iterable[T]]: '''Split an iterable based on sentnel elements. A sentinel element is one for which `pred` is `True`. From it onwards a new split is made. Args: pred: A predicate identifying a sentinel element. iterable: An iterable to enumerate. Returns: An iterable with an iterable for each split. ''' enumerated = enumerate_on(pred, iterable) return ((line for _, line in group) for _, group in groupby(enumerated, key=itemgetter(0))) PK!! !ampdup-0.2.2.dist-info/LICENSE.mdThe MIT License (MIT) ===================== Copyright © `2018` `Tarcísio Eduardo Moreira Crocomo ` Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. PK!HƣxSTampdup-0.2.2.dist-info/WHEEL A н#J;/"d&F]xzw>@Zpy3F ]n2H%_60{8&baPa>PK!H i`ampdup-0.2.2.dist-info/METADATAUn6|W,r6jNpP594HhK4n_"%b#<b( z ˙a|'<:UAg)w ޙ7쵲3wCAW#F*/jnP7Z.RJBPF4~L+7j"R#Dx zoQ>pux0DV;VtIL%dbx>+^al̮*L;e4eҊk!3Pi_:HG]Z F;aFb1 bt-^`d[5n!+QLhnAD9 UӃT 8!ZAV9/#G<ηZ7K8۫![q *˴ Y<"k^! ypcHje{q8 _=m!r̓o.SZ7F["U )xC6/xՓa3~X/R״nPK!Hҿ ampdup-0.2.2.dist-info/RECORDuI6->a20l0 o6iʒ9vzC~'Iٖc|ה?)p34V=/'u(%`IWtJqdMYV1G?AhJEZsf?'Q LPGk"ˮt%yZ/m$htWMkU*ːH$a+7tsG56rAKPW> Cм(2oEweֿKal92%(Pv+D>+yc:f&Z;fG.!ȟ1KCAUm`Iv-2Sd-mʫ ^Ő 8F0e9qP(yyYvDiu'4^[:6Ɯ2A'3}:I\W$GnR5(Vn0+ں!wqFuCSdc8WͶT:LnYiN$簨^Ȁ8;"X~2̮f*~e N46a?&R$/a_X