/
context.py
140 lines (112 loc) · 4.16 KB
/
context.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
from __future__ import annotations
import logging
import re
from typing import (
TYPE_CHECKING,
Any,
Literal,
overload,
)
from mopidy_mpd import exceptions, types
if TYPE_CHECKING:
from collections.abc import Generator
import pykka
from mopidy.core import CoreProxy
from mopidy.models import Ref, Track
from mopidy.types import Uri
from mopidy_mpd.dispatcher import MpdDispatcher
from mopidy_mpd.session import MpdSession
from mopidy_mpd.uri_mapper import MpdUriMapper
logger = logging.getLogger(__name__)
class MpdContext:
"""
This object is passed as the first argument to all MPD command handlers to
give the command handlers access to important parts of Mopidy.
"""
#: The Mopidy config.
config: types.Config
#: The Mopidy core API.
core: CoreProxy
#: The current session instance.
session: MpdSession
#: The current dispatcher instance.
dispatcher: MpdDispatcher
#: Mapping of URIs to MPD names.
uri_map: MpdUriMapper
def __init__( # noqa: PLR0913
self,
config: types.Config,
core: CoreProxy,
uri_map: MpdUriMapper,
session: MpdSession,
dispatcher: MpdDispatcher,
) -> None:
self.config = config
self.core = core
self.uri_map = uri_map
self.session = session
self.dispatcher = dispatcher
@overload
def browse(
self, path: str | None, *, recursive: bool, lookup: Literal[True]
) -> Generator[tuple[str, pykka.Future[dict[Uri, list[Track]]] | None], Any, None]:
...
@overload
def browse(
self, path: str | None, *, recursive: bool, lookup: Literal[False]
) -> Generator[tuple[str, Ref | None], Any, None]:
...
def browse( # noqa: C901, PLR0912
self,
path: str | None,
*,
recursive: bool = True,
lookup: bool = True,
) -> Generator[Any, Any, None]:
"""
Browse the contents of a given directory path.
Returns a sequence of two-tuples ``(path, data)``.
If ``recursive`` is true, it returns results for all entries in the
given path.
If ``lookup`` is true and the ``path`` is to a track, the returned
``data`` is a future which will contain the results from looking up
the URI with :meth:`mopidy.core.LibraryController.lookup`. If
``lookup`` is false and the ``path`` is to a track, the returned
``data`` will be a :class:`mopidy.models.Ref` for the track.
For all entries that are not tracks, the returned ``data`` will be
:class:`None`.
"""
path_parts: list[str] = re.findall(r"[^/]+", path or "")
root_path: str = "/".join(["", *path_parts])
uri = self.uri_map.uri_from_name(root_path)
if uri is None:
for part in path_parts:
for ref in self.core.library.browse(uri).get():
if ref.type != ref.TRACK and ref.name == part:
uri = ref.uri
break
else:
raise exceptions.MpdNoExistError("Not found")
root_path = self.uri_map.insert(root_path, uri)
if recursive:
yield (root_path, None)
path_and_futures = [(root_path, self.core.library.browse(uri))]
while path_and_futures:
base_path, future = path_and_futures.pop()
for ref in future.get():
if ref.name is None or ref.uri is None:
continue
path = "/".join([base_path, ref.name.replace("/", "")])
path = self.uri_map.insert(path, ref.uri)
if ref.type == ref.TRACK:
if lookup:
# TODO: can we lookup all the refs at once now?
yield (path, self.core.library.lookup(uris=[ref.uri]))
else:
yield (path, ref)
else:
yield (path, None)
if recursive:
path_and_futures.append(
(path, self.core.library.browse(ref.uri))
)