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
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
|
#!/usr/bin/env python3
import io
import json
import os
import socket
import sys
import click
MPV_SOCKET_PATH = "/tmp/mpvd"
NEWLINE = b"\n"[0]
def terminate_on_failure(response):
if response["error"] != "success":
sys.stderr.write(f"{response['error']}")
sys.stderr.flush()
sys.exit(1)
def open_socket():
the_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM, 0)
try:
the_socket.connect(MPV_SOCKET_PATH)
return the_socket
except OSError:
the_socket.close()
raise
def snarf_available(the_socket):
buffer = io.BytesIO()
received = []
while True:
chunk = the_socket.recv(65536)
buffer.write(chunk)
if chunk[-1] != NEWLINE:
continue
buffer.seek(0)
received = [json.loads(x) for x in buffer]
buffer.seek(0)
buffer.truncate()
received = [x for x in received if "request_id" in x]
if not received:
continue
return received[0]
return {"error": "no data"}
def send_command(command, *args):
command_json = (
json.dumps({"command": [command] + list(args)}).encode("UTF-8") + b"\n"
)
with open_socket() as the_socket:
the_socket.send(command_json)
return snarf_available(the_socket)
def set_property(property_name, value):
return send_command("set_property", property_name, value)
def update_playlist_helper(opt, args):
if not args:
args = [x.rstrip("\n") for x in sys.stdin]
for entry in args:
terminate_on_failure(send_command("loadfile", entry, opt))
@click.group()
def cli():
pass
def get_property(property_name):
return send_command("get_property", property_name)
@cli.command(help="Show an mpv property.")
@click.argument("property_name")
def prop(property_name):
resp = get_property(property_name)
terminate_on_failure(resp)
print(resp["data"])
@cli.command(help="Set or show the volume.")
@click.argument("new_volume", type=float, required=False, default=None)
def volume(new_volume):
if new_volume is not None:
terminate_on_failure(set_property("volume", new_volume))
else:
current_volume = get_property("volume")
terminate_on_failure(current_volume)
print(f"Current volume: {current_volume['data']}")
@cli.command(help="Stop playback.")
def stop():
terminate_on_failure(send_command("stop"))
@cli.command(name="next", help="Jump to the next item in the playlist.")
def playlist_next():
terminate_on_failure(send_command("playlist-next"))
@cli.command(name="prev", help="Jump to the previous item in the playlist.")
def playlist_prev():
terminate_on_failure(send_command("playlist-prev"))
@cli.command(name="shuffle", help="Shuffle the playlist.")
def playlist_shuffle():
terminate_on_failure(send_command("playlist-shuffle"))
@cli.command(name="clear", help="Clear the playlist.")
def playlist_clear():
terminate_on_failure(send_command("playlist-clear"))
@cli.command(help="Jump to a specific entry in playlist, or resume playback.")
@click.argument("playlist_position", required=False, type=int, default=None)
def play(playlist_position):
if playlist_position is not None:
terminate_on_failure(set_property("playlist-pos-1", playlist_position))
terminate_on_failure(set_property("pause", False))
@cli.command(help="Pause playback.")
def pause():
terminate_on_failure(set_property("pause", True))
@cli.command(name="list", help="Print the playlist.")
def cmd_list():
resp = send_command("get_property", "playlist")
terminate_on_failure(resp)
for entry in resp["data"]:
char_current = "*" if entry.get("current", None) else " "
char_playing = "!" if entry.get("playing", None) else " "
print(f"{char_playing}{char_current}{entry['filename']}")
@cli.command(help="Start a standalone mpv.")
def launch_mpv():
MPV_NO_VIDEO = ("--no-video",)
MPV_COMMAND = (
"mpv",
"--idle=yes",
"--no-terminal",
f"--input-ipc-server={MPV_SOCKET_PATH}",
) + MPV_NO_VIDEO
os.execvp("mpv", MPV_COMMAND)
@cli.command(name="add", help="Append one or more items to the playlist.")
@click.argument("items", nargs=-1)
def append_to_playlist(items):
update_playlist_helper("append-play", items)
@cli.command(name="replace", help="Replace the playlist with the given arguments.")
@click.argument("items", nargs=-1)
def replace_playlist(items):
update_playlist_helper("replace", items)
@cli.command(context_settings=dict(ignore_unknown_options=True))
@click.argument("amount", type=click.FLOAT)
@click.option("--absolute/--relative", default=False)
@click.option("--percent", is_flag=True, default=False)
def seek(amount, absolute, percent):
flag_words = {
(False, False): "relative",
(False, True): "relative-percent",
(True, False): "absolute",
(True, True): "absolute-percent",
}
terminate_on_failure(send_command("seek", amount, flag_words[(absolute, percent)]))
if __name__ == "__main__":
cli()
|
