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
|
"""
`stream`: a video or a sound file of a group, played by a media player.
The file is served on this machine at an address printed on stdout, for VLC,
mpv or any player that reads HTTP (`client/relay.py`); `--player PROGRAM`
starts the player on it. Seeking is the player's: it asks for another byte
range. The stream runs until Ctrl+C, or until the player started with
`--player` is closed; the address works only while it runs.
"""
import asyncio
import contextlib
import json
import os
import shutil
import subprocess
import sys
from ..client.groups import my_groups, open_group
from ..client.paths import handle, parse, resolve_group
from ..client.relay import ChunkSource, Relay
from ..client.tree import Tree
from ..errors import NOT_FOUND, REFUSED, USAGE, CliError
from ..output import err, human_size, safe, sentence
from .context import Context
HELP = "play a video or a sound file in VLC, mpv or another player"
PLAYABLE = ("video", "audio")
def add_parser(sub, parent) -> None:
p = sub.add_parser("stream", parents=[parent], help=HELP, description=sentence(HELP)
+ ". The file is served on this machine at an address to open in "
"the player, until Ctrl+C. Nothing is written to disk.")
p.add_argument("path", metavar="PATH", help="group[@owner]/root/.../file")
p.add_argument("--player", metavar="PROGRAM",
help="start PROGRAM (vlc, mpv, ...) on the address; closing it stops "
"the stream")
def player_command(program: str) -> list[str]:
"""The player to start: a name on the PATH, or a path. VLC on Windows is
rarely on the PATH, so its usual places are looked in too."""
found = shutil.which(program)
if found is None and sys.platform == "win32" and program.lower() in ("vlc", "vlc.exe"):
for base in (os.environ.get("ProgramFiles"), os.environ.get("ProgramFiles(x86)")):
candidate = os.path.join(base or "", "VideoLAN", "VLC", "vlc.exe")
if base and os.path.isfile(candidate):
found = candidate
break
if found is None:
raise CliError(USAGE, f"{program} was not found; install it, or run without "
"--player and open the address yourself.")
return [found]
async def run(ctx: Context) -> int:
a = ctx.args
remote = parse(a.path)
if remote is None or not remote.parts:
raise CliError(USAGE, "Name the file to play: group/root/.../file.")
command = player_command(a.player) if a.player else None
await ctx.hub.check_version()
account = ctx.account()
session = source = relay = player = None
try:
group = resolve_group(await my_groups(account), remote.group)
session = await open_group(account, group)
entry = _the_file(Tree.from_index(await session.index()), remote, a.path)
name = entry["name"]
notes = (lambda _: None) if a.quiet else err
source = ChunkSource(session, entry, lambda: open_group(account, group), notes)
session = None # the source owns it now
relay = Relay(source, name, entry["type"])
url = await relay.start()
if a.json:
print(json.dumps({"url": url, "name": name, "size": int(entry["size"]),
"group": handle(group)}, ensure_ascii=False), flush=True)
else:
if not a.quiet:
err(f"Streaming {safe(name)} ({human_size(int(entry['size']))}) from "
f"{safe(handle(group))}."
+ ("" if command else " Open this address in VLC or mpv:"))
print(url, flush=True)
if not a.quiet:
err("Close the player or press Ctrl+C to stop." if command
else "Ctrl+C to stop.")
waits = [source.stopped]
if command:
player = subprocess.Popen(
command + [url], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL, start_new_session=sys.platform != "win32")
# A thread, not asyncio's subprocesses: Windows' selector loop has none.
waits.append(asyncio.ensure_future(asyncio.to_thread(player.wait)))
done, _ = await asyncio.wait(waits, return_when=asyncio.FIRST_COMPLETED)
if source.stopped in done:
# The player is told too, rather than left waiting on a closed socket.
await relay.close(grace=1)
raise source.stopped.exception()
if not a.quiet and not a.json:
err("The player was closed; the stream is stopped.")
return 0
finally:
if player is not None and player.poll() is None:
with contextlib.suppress(OSError):
player.terminate()
if relay is not None:
await relay.close()
if source is not None:
await source.close()
if session is not None:
await session.close()
await account.close()
def _the_file(tree: Tree, remote, text: str) -> dict:
found = tree.expand(remote.parts)
if not found and tree.file(remote.inner):
found = [remote.inner] # a literal name with brackets in it
files = [p for p in found if p and not tree.is_dir(p)]
if not files:
if found:
raise CliError(USAGE, f"{safe(text)} is a folder; name a file in it.")
raise CliError(NOT_FOUND, f"Nothing in this group matches {safe(text)}.")
if len(files) > 1:
raise CliError(USAGE, f"{safe(text)} names {len(files)} files; stream plays one.")
entry = tree.file(files[0])
if entry.get("type") not in PLAYABLE:
raise CliError(USAGE, f"{safe(entry['name'])} is not a video or a sound file; "
"fetch it with get.")
if not tree.root_available(files[0]):
raise CliError(REFUSED, f"{safe(text)}: its folder on the node is unavailable.")
return entry
|