diff options
Diffstat (limited to 'packages/meshbay-cli/src/meshbay_cli/commands')
| -rw-r--r-- | packages/meshbay-cli/src/meshbay_cli/commands/stream.py | 135 |
1 files changed, 135 insertions, 0 deletions
diff --git a/packages/meshbay-cli/src/meshbay_cli/commands/stream.py b/packages/meshbay-cli/src/meshbay_cli/commands/stream.py new file mode 100644 index 0000000..87f2b42 --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/commands/stream.py @@ -0,0 +1,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 |