syng/syng/client.py
2022-12-05 23:57:42 +01:00

433 lines
13 KiB
Python

"""
Module for the playback client.
Excerp from the help::
usage: client.py [-h] [--room ROOM] [--secret SECRET] \
[--config-file CONFIG_FILE] server
positional arguments:
server
options:
-h, --help show this help message and exit
--room ROOM, -r ROOM
--secret SECRET, -s SECRET
--config-file CONFIG_FILE, -C CONFIG_FILE
The config file should be a json file in the following style::
{
"sources": {
"SOURCE1": { configuration for SOURCE },
"SOURCE2": { configuration for SOURCE },
...
},
},
"config": {
configuration for the client
}
}
"""
import asyncio
import datetime
import logging
import secrets
import string
import tempfile
from argparse import ArgumentParser
from dataclasses import dataclass
from dataclasses import field
from json import load
from traceback import print_exc
from typing import Any
from typing import Optional
import pyqrcode
import socketio
from PIL import Image
from . import json
from .entry import Entry
from .sources import configure_sources
from .sources import Source
sio: socketio.AsyncClient = socketio.AsyncClient(json=json)
logger: logging.Logger = logging.getLogger(__name__)
sources: dict[str, Source] = {}
currentLock: asyncio.Semaphore = asyncio.Semaphore(0)
@dataclass
class State:
"""This captures the current state of the playback client.
It doubles as a backup of the state of the :py:class:`syng.server` in case
the server needs to be restarted.
:param current_source: This holds a reference to the
:py:class:`syng.sources.source.Source` object, that is currently
playing. If no song is played, the value is `None`.
:type current_source: Optional[Source]
:param queue: A copy of the current playlist on the server.
:type queue: list[Entry]
:param recent: A copy of all played songs this session.
:type recent: list[Entry]
:param room: The room on the server this playback client is connected to.
:type room: str
:param secret: The passcode of the room. If a playback client reconnects to
a room, this must be identical. Also, if a webclient wants to have
admin privileges, this must be included.
:type secret: str
:param preview_duration: Amount of seconds the preview before a song be
displayed.
:type preview_duration: int
:param last_song: At what time should the server not accept any more songs.
`None` if no such limit should exist.
:type last_song: Optional[datetime.datetime]
"""
# pylint: disable=too-many-instance-attributes
current_source: Optional[Source] = None
queue: list[Entry] = field(default_factory=list)
recent: list[Entry] = field(default_factory=list)
room: str = ""
server: str = ""
secret: str = ""
preview_duration: int = 3
last_song: Optional[datetime.datetime] = None
def get_config(self) -> dict[str, Any]:
"""
Return a subset of values to be send to the server.
Currently this is:
- :py:attr:`State.preview_duration`
- :py:attr:`State.last_song` (As a timestamp)
:return: A dict resulting from the above values
:rtype: dict[str, Any]
"""
return {
"preview_duration": self.preview_duration,
"last_song": self.last_song.timestamp()
if self.last_song
else None,
}
state: State = State()
@sio.on("skip-current")
async def handle_skip_current(data: dict[str, Any]) -> None:
"""
Handle the "skip-current" message.
Skips the song, that is currently played. If playback currently waits for
buffering, the buffering is also aborted.
Since the ``queue`` could already be updated, when this evaluates, the
first entry in the queue is send explicitly.
:param data: An entry, that should be equivalent to the first entry of the
queue.
:rtype: None
"""
logger.info("Skipping current")
if state.current_source is not None:
await state.current_source.skip_current(Entry(**data))
@sio.on("state")
async def handle_state(data: dict[str, Any]) -> None:
"""
Handle the "state" message.
The "state" message forwards the current queue and recent list from the
server. This function saves a copy of both in the global
:py:class:`State`:.
After recieving the new state, a buffering task for the first elements of
the queue is started.
:param data: A dictionary with the `queue` and `recent` list.
:type data: dict[str, Any]
:rtype: None
"""
state.queue = [Entry(**entry) for entry in data["queue"]]
state.recent = [Entry(**entry) for entry in data["recent"]]
for entry in state.queue[:2]:
logger.info("Buffering: %s", entry.title)
await sources[entry.source].buffer(entry)
@sio.on("connect")
async def handle_connect() -> None:
"""
Handle the "connect" message.
Called when the client successfully connects or reconnects to the server.
Sends a `register-client` message to the server with the initial state and
configuration of the client, consiting of the currently saved
:py:attr:`State.queue` and :py:attr:`State.recent` field of the global
:py:class:`State`, as well a room code the client wants to connect to, a
secret to secure the access to the room and a config dictionary.
If the room code is `None`, the server will issue a room code.
This message will be handled by the
:py:func:`syng.server.handle_register_client` function of the server.
:rtype: None
"""
logging.info("Connected to server")
await sio.emit(
"register-client",
{
"queue": state.queue,
"recent": state.recent,
"room": state.room,
"secret": state.secret,
"config": state.get_config(),
},
)
@sio.on("get-meta-info")
async def handle_get_meta_info(data: dict[str, Any]) -> None:
"""
Handle a "get-meta-info" message.
Collects the metadata for a given :py:class:`Entry`, from its source, and
sends them back to the server in a "meta-info" message. On the server side
a :py:func:`syng.server.handle_meta_info` function is called.
:param data: A dictionary encoding the entry
:type data: dict[str, Any]
:rtype: None
"""
source: Source = sources[data["source"]]
meta_info: dict[str, Any] = await source.get_missing_metadata(
Entry(**data)
)
await sio.emit("meta-info", {"uuid": data["uuid"], "meta": meta_info})
async def preview(entry: Entry) -> None:
"""
Generate and play a preview for a given :py:class:`Entry`.
This function shows a black screen and prints the artist, title and
performer of the entry for a duration.
This is done by creating a black png file, and showing subtitles in the
middle of the screen.... don't ask, it works
:param entry: The entry to preview
:type entry: :py:class:`Entry`
:rtype: None
"""
background = Image.new("RGB", (1280, 720))
subtitle: str = f"""1
00:00:00,00 --> 00:05:00,00
{entry.artist} - {entry.title}
{entry.performer}"""
with tempfile.NamedTemporaryFile() as tmpfile:
background.save(tmpfile, "png")
process = await asyncio.create_subprocess_exec(
"mpv",
tmpfile.name,
f"--image-display-duration={state.preview_duration}",
"--sub-pos=50",
"--sub-file=-",
"--fullscreen",
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
await process.communicate(subtitle.encode())
@sio.on("play")
async def handle_play(data: dict[str, Any]) -> None:
"""
Handle the "play" message.
Plays the :py:class:`Entry`, that is encoded in the `data` parameter. If a
:py:attr:`State.preview_duration` is set, it shows a small preview before
that.
When the playback is done, the next song is requested from the server with
a "pop-then-get-next" message. This is handled by the
:py:func:`syng.server.handle_pop_then_get_next` function on the server.
If the entry is marked as skipped, emit a "get-first" message instead,
because the server already handled the removal of the first entry.
:param data: A dictionary encoding the entry
:type data: dict[str, Any]
:rtype: None
"""
entry: Entry = Entry(**data)
print(
f"Playing: {entry.artist} - {entry.title} [{entry.album}] "
f"({entry.source}) for {entry.performer}"
)
try:
state.current_source = sources[entry.source]
if state.preview_duration > 0:
await preview(entry)
await sources[entry.source].play(entry)
except Exception: # pylint: disable=broad-except
print_exc()
state.current_source = None
if entry.skip:
await sio.emit("get-first")
else:
await sio.emit("pop-then-get-next")
@sio.on("client-registered")
async def handle_client_registered(data: dict[str, Any]) -> None:
"""
Handle the "client-registered" massage.
If the registration was successfull (`data["success"]` == `True`), store
the room code in the global :py:class:`State` and print out a link to join
the webclient.
Start listing all configured :py:class:`syng.sources.source.Source` to the
server via a "sources" message. This message will be handled by the
:py:func:`syng.server.handle_sources` function and may request additional
configuration for each source.
If there is no song playing, start requesting the first song of the queue
with a "get-first" message. This will be handled on the server by the
:py:func:`syng.server.handle_get_first` function.
:param data: A dictionary containing a `success` and a `room` entry.
:type data: dict[str, Any]
:rtype: None
"""
if data["success"]:
logging.info("Registered")
print(f"Join here: {state.server}/{data['room']}")
print(
pyqrcode.create(f"{state.server}/{data['room']}").terminal(
quiet_zone=1
)
)
state.room = data["room"]
await sio.emit("sources", {"sources": list(sources.keys())})
if (
state.current_source is None
): # A possible race condition can occur here
await sio.emit("get-first")
else:
logging.warning("Registration failed")
await sio.disconnect()
@sio.on("request-config")
async def handle_request_config(data: dict[str, Any]) -> None:
"""
Handle the "request-config" message.
Sends the specific server side configuration for a given
:py:class:`syng.sources.source.Source`.
A Source can decide, that the config will be split up in multiple Parts.
If this is the case, multiple "config-chunk" messages will be send with a
running enumerator. Otherwise a singe "config" message will be send.
:param data: A dictionary with the entry `source` and a string, that
corresponds to the name of a source.
:type data: dict[str, Any]
:rtype: None
"""
if data["source"] in sources:
config: dict[str, Any] | list[dict[str, Any]] = await sources[
data["source"]
].get_config()
if isinstance(config, list):
num_chunks: int = len(config)
for current, chunk in enumerate(config):
await sio.emit(
"config-chunk",
{
"source": data["source"],
"config": chunk,
"number": current + 1,
"total": num_chunks,
},
)
else:
await sio.emit(
"config", {"source": data["source"], "config": config}
)
async def aiomain() -> None:
"""
Async main function.
Parses the arguments, reads a config file and sets default values. Then
connects to a specified server.
If no secret is given, a random secret will be generated and presented to
the user.
:rtype: None
"""
parser: ArgumentParser = ArgumentParser()
parser.add_argument("--room", "-r")
parser.add_argument("--secret", "-s")
parser.add_argument("--config-file", "-C", default="syng-client.json")
parser.add_argument("server")
args = parser.parse_args()
with open(args.config_file, encoding="utf8") as file:
config = load(file)
sources.update(configure_sources(config["sources"]))
if "config" in config:
if "last_song" in config["config"]:
state.last_song = datetime.datetime.fromisoformat(
config["config"]["last_song"]
)
if "preview_duration" in config["config"]:
state.preview_duration = config["config"]["preview_duration"]
if args.room:
state.room = args.room
if args.secret:
state.secret = args.secret
else:
state.secret = "".join(
secrets.choice(string.ascii_letters + string.digits)
for _ in range(8)
)
print(f"Generated secret: {state.secret}")
state.server = args.server
await sio.connect(args.server)
await sio.wait()
def main() -> None:
"""Entry point for the syng-client script."""
asyncio.run(aiomain())
if __name__ == "__main__":
main()