Split installation into server and client. Added rough readme
This commit is contained in:
parent
ae5d82ce63
commit
e8453e0ffb
8 changed files with 130 additions and 284 deletions
26
README.md
26
README.md
|
|
@ -1 +1,27 @@
|
|||
# Syng
|
||||
|
||||
Syng is a all-in-one karaoke software, consisting of a *backend server*, a *web frontend* and a *playback client*.
|
||||
Karaoke performers can search a library using the web fronend, and add songs to the queue.
|
||||
The playback client retrieves songs from the backend server and plays them in order.
|
||||
|
||||
Currently songs can be accessed using the following sources:
|
||||
|
||||
- **Youtube.** The backend server queries YouTube for the song and forwards the URL to the playback client. The playback client then downloads the video from YouTube for playback.
|
||||
- **S3.** The backend server holds a list of all file paths accessible through the s3 storage, and forwards the chosen path to the playback client. The playback client then downloads the needed files from the s3 for playback.
|
||||
- **Files.** Same as S3, but all files reside locally on the playback client.
|
||||
|
||||
The playback client uses `mpv` for playback and can therefore play a variety of file formats, such as `mp3+cdg`, `webm`, `mp4`, ...
|
||||
|
||||
# Installation
|
||||
|
||||
## Server
|
||||
|
||||
pip install git+https://github.com/christofsteel/syng -E server
|
||||
|
||||
This installs the server part (`syng-server`), if you want to self-host a syng server. There is a publicly available syng instance at https://syng.rocks.
|
||||
|
||||
## Client
|
||||
|
||||
pip install git+https://github.com/christofsteel/syng -E client
|
||||
|
||||
This installs both the playback client (`syng-client`) and a configuration GUI (`syng-gui`).
|
||||
|
|
|
|||
|
|
@ -11,28 +11,37 @@ include = ["syng/static"]
|
|||
syng-client = "syng.client:main"
|
||||
syng-server = "syng.server:main"
|
||||
syng-gui = "syng.gui:main"
|
||||
syng-shell = "syng.webclientmockup:main"
|
||||
# syng-shell = "syng.webclientmockup:main"
|
||||
|
||||
[tool.poetry.dependencies]
|
||||
python = "^3.8"
|
||||
pytube = "*"
|
||||
aiohttp = "^3.8.3"
|
||||
python-socketio = "^5.7.2"
|
||||
minio = "^7.1.12"
|
||||
mutagen = "^1.46.0"
|
||||
aiocmd = "^0.1.5"
|
||||
pillow = "^9.3.0"
|
||||
yt-dlp = "*"
|
||||
customtkinter = "^5.2.1"
|
||||
qrcode = "^7.4.2"
|
||||
pymediainfo = "^6.1.0"
|
||||
pyyaml = "^6.0.1"
|
||||
async-tkinter-loop = "^0.9.2"
|
||||
tkcalendar = "^1.6.1"
|
||||
tktimepicker = "^2.0.2"
|
||||
python-socketio = "^5.10.0"
|
||||
aiohttp = "^3.9.1"
|
||||
pytube = { version = "*", optional = true }
|
||||
minio = { version = "^7.2.0", optional = true }
|
||||
mutagen = { version = "^1.47.0", optional = true }
|
||||
# aiocmd = "^0.1.5"
|
||||
pillow = { version = "^10.1.0", optional = true}
|
||||
yt-dlp = { version = "*", optional = true}
|
||||
customtkinter = { version = "^5.2.1", optional = true}
|
||||
qrcode = { version = "^7.4.2", optional = true }
|
||||
pymediainfo = { version = "^6.1.0", optional = true }
|
||||
pyyaml = { version = "^6.0.1", optional = true }
|
||||
# async-tkinter-loop = "^0.9.2"
|
||||
tkcalendar = { version = "^1.6.1", optional = true }
|
||||
tktimepicker = { version = "^2.0.2", optional = true }
|
||||
platformdirs = { version = "^4.0.0", optional = true }
|
||||
packaging = {version = "^23.2", optional = true}
|
||||
|
||||
[tool.poetry.group.dev.dependencies]
|
||||
types-pyyaml = "^6.0.12.12"
|
||||
types-pillow = "^10.1.0.2"
|
||||
platformdirs = "^4.0.0"
|
||||
|
||||
[tool.poetry.extras]
|
||||
client = ["minio", "mutagen", "pillow", "yt-dlp",
|
||||
"customtkinter", "qrcode", "pymediainfo", "pyyaml",
|
||||
"tkcalendar", "tktimepicker", "platformdirs", "packaging"]
|
||||
server = ["pytube"]
|
||||
|
||||
[build-system]
|
||||
requires = ["poetry-core"]
|
||||
|
|
|
|||
|
|
@ -1,4 +0,0 @@
|
|||
class QRCode:
|
||||
def terminal(self, quiet_zone: int) -> str: ...
|
||||
|
||||
def create(data: str) -> QRCode: ...
|
||||
|
|
@ -21,8 +21,14 @@ import platformdirs
|
|||
from .client import create_async_and_start_client, default_config
|
||||
|
||||
from .sources import available_sources
|
||||
|
||||
try:
|
||||
from .server import main as server_main
|
||||
|
||||
SERVER_AVAILABLE = True
|
||||
except ImportError:
|
||||
SERVER_AVAILABLE = False
|
||||
|
||||
|
||||
class DateAndTimePickerWindow(customtkinter.CTkToplevel): # type: ignore
|
||||
def __init__(
|
||||
|
|
@ -201,7 +207,7 @@ class OptionFrame(customtkinter.CTkScrollableFrame): # type:ignore
|
|||
self.date_time_options[name] = input_field
|
||||
try:
|
||||
datetime.fromisoformat(value)
|
||||
except TypeError:
|
||||
except (TypeError, ValueError):
|
||||
value = ""
|
||||
input_field.insert("0.0", value)
|
||||
|
||||
|
|
|
|||
|
|
@ -3,8 +3,13 @@ import asyncio
|
|||
import os
|
||||
from typing import Any, Optional
|
||||
|
||||
try:
|
||||
from pymediainfo import MediaInfo
|
||||
|
||||
PYMEDIAINFO_AVAILABLE = True
|
||||
except ImportError:
|
||||
PYMEDIAINFO_AVAILABLE = False
|
||||
|
||||
from .source import Source
|
||||
|
||||
|
||||
|
|
@ -27,9 +32,7 @@ class FileBasedSource(Source):
|
|||
"""Initialize the file module."""
|
||||
super().__init__(config)
|
||||
|
||||
self.extensions: list[str] = (
|
||||
config["extensions"] if "extensions" in config else ["mp3+cdg"]
|
||||
)
|
||||
self.extensions: list[str] = config["extensions"] if "extensions" in config else ["mp3+cdg"]
|
||||
self.extra_mpv_arguments = ["--scale=oversample"]
|
||||
|
||||
def has_correct_extension(self, path: str) -> bool:
|
||||
|
|
@ -40,29 +43,25 @@ class FileBasedSource(Source):
|
|||
:return: True iff path has correct extension.
|
||||
:rtype: bool
|
||||
"""
|
||||
return os.path.splitext(path)[1][1:] in [
|
||||
ext.split("+")[-1] for ext in self.extensions
|
||||
]
|
||||
return os.path.splitext(path)[1][1:] in [ext.split("+")[-1] for ext in self.extensions]
|
||||
|
||||
def get_video_audio_split(self, path: str) -> tuple[str, Optional[str]]:
|
||||
extension_of_path = os.path.splitext(path)[1][1:]
|
||||
splitted_extensions = [ext.split("+") for ext in self.extensions if "+" in ext]
|
||||
splitted_extensions_dict = {
|
||||
video: audio for [audio, video] in splitted_extensions
|
||||
}
|
||||
splitted_extensions_dict = {video: audio for [audio, video] in splitted_extensions}
|
||||
|
||||
if extension_of_path in splitted_extensions_dict:
|
||||
audio_path = (
|
||||
os.path.splitext(path)[0]
|
||||
+ "."
|
||||
+ splitted_extensions_dict[extension_of_path]
|
||||
os.path.splitext(path)[0] + "." + splitted_extensions_dict[extension_of_path]
|
||||
)
|
||||
return (path, audio_path)
|
||||
return (path, None)
|
||||
|
||||
async def get_duration(self, path: str) -> int:
|
||||
if not PYMEDIAINFO_AVAILABLE:
|
||||
return 180
|
||||
|
||||
def _get_duration(file: str) -> int:
|
||||
print(file)
|
||||
info: str | MediaInfo = MediaInfo.parse(file)
|
||||
if isinstance(info, str):
|
||||
return 180
|
||||
|
|
|
|||
|
|
@ -5,18 +5,18 @@ Adds it to the ``available_sources`` with the name ``s3``
|
|||
"""
|
||||
import asyncio
|
||||
import os
|
||||
from json import load
|
||||
from json import dump
|
||||
from typing import Any
|
||||
from typing import cast
|
||||
from typing import Optional
|
||||
from typing import Tuple
|
||||
from json import dump, load
|
||||
from typing import Any, Optional, Tuple, cast
|
||||
|
||||
try:
|
||||
from minio import Minio
|
||||
|
||||
from .filebased import FileBasedSource
|
||||
MINIO_AVAILABE = True
|
||||
except ImportError:
|
||||
MINIO_AVAILABE = False
|
||||
|
||||
from ..entry import Entry
|
||||
from .filebased import FileBasedSource
|
||||
from .source import available_sources
|
||||
|
||||
|
||||
|
|
@ -48,7 +48,12 @@ class S3Source(FileBasedSource):
|
|||
"""Create the source."""
|
||||
super().__init__(config)
|
||||
|
||||
if "endpoint" in config and "access_key" in config and "secret_key" in config:
|
||||
if (
|
||||
MINIO_AVAILABE
|
||||
and "endpoint" in config
|
||||
and "access_key" in config
|
||||
and "secret_key" in config
|
||||
):
|
||||
self.minio: Minio = Minio(
|
||||
config["endpoint"],
|
||||
access_key=config["access_key"],
|
||||
|
|
@ -56,13 +61,9 @@ class S3Source(FileBasedSource):
|
|||
secure=(config["secure"] if "secure" in config else True),
|
||||
)
|
||||
self.bucket: str = config["bucket"]
|
||||
self.tmp_dir: str = (
|
||||
config["tmp_dir"] if "tmp_dir" in config else "/tmp/syng"
|
||||
)
|
||||
self.tmp_dir: str = config["tmp_dir"] if "tmp_dir" in config else "/tmp/syng"
|
||||
|
||||
self.index_file: Optional[str] = (
|
||||
config["index_file"] if "index_file" in config else None
|
||||
)
|
||||
self.index_file: Optional[str] = config["index_file"] if "index_file" in config else None
|
||||
self.extra_mpv_arguments = ["--scale=oversample"]
|
||||
|
||||
async def get_file_list(self) -> list[str]:
|
||||
|
|
@ -131,18 +132,14 @@ class S3Source(FileBasedSource):
|
|||
video_dl_path: str = os.path.join(self.tmp_dir, video_path)
|
||||
os.makedirs(os.path.dirname(video_dl_path), exist_ok=True)
|
||||
video_dl_task: asyncio.Task[Any] = asyncio.create_task(
|
||||
asyncio.to_thread(
|
||||
self.minio.fget_object, self.bucket, entry.ident, video_dl_path
|
||||
)
|
||||
asyncio.to_thread(self.minio.fget_object, self.bucket, entry.ident, video_dl_path)
|
||||
)
|
||||
|
||||
if audio_path is not None:
|
||||
audio_dl_path: Optional[str] = os.path.join(self.tmp_dir, audio_path)
|
||||
|
||||
audio_dl_task: asyncio.Task[Any] = asyncio.create_task(
|
||||
asyncio.to_thread(
|
||||
self.minio.fget_object, self.bucket, audio_path, audio_dl_path
|
||||
)
|
||||
asyncio.to_thread(self.minio.fget_object, self.bucket, audio_path, audio_dl_path)
|
||||
)
|
||||
else:
|
||||
audio_dl_path = None
|
||||
|
|
|
|||
|
|
@ -6,32 +6,31 @@ used.
|
|||
|
||||
Adds it to the ``available_sources`` with the name ``youtube``.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import shlex
|
||||
from functools import partial
|
||||
from typing import Any
|
||||
from typing import Optional
|
||||
from typing import Tuple
|
||||
from typing import Any, Optional, Tuple
|
||||
|
||||
from pytube import Channel
|
||||
from pytube import innertube
|
||||
from pytube import Search
|
||||
from pytube import Stream
|
||||
from pytube import StreamQuery
|
||||
from pytube import YouTube
|
||||
from pytube import exceptions
|
||||
try:
|
||||
from pytube import Channel, Search, YouTube, exceptions, innertube
|
||||
|
||||
PYTUBE_AVAILABLE = True
|
||||
except ImportError:
|
||||
PYTUBE_AVAILABLE = False
|
||||
|
||||
try:
|
||||
from yt_dlp import YoutubeDL
|
||||
|
||||
USE_YT_DLP = True
|
||||
YT_DLP_AVAILABLE = True
|
||||
except ImportError:
|
||||
USE_YT_DLP = False
|
||||
print("No yt-dlp")
|
||||
YT_DLP_AVAILABLE = False
|
||||
|
||||
from ..entry import Entry
|
||||
from ..result import Result
|
||||
from .source import available_sources
|
||||
from .source import Source
|
||||
from .source import Source, available_sources
|
||||
|
||||
|
||||
class YoutubeSource(Source):
|
||||
|
|
@ -68,6 +67,7 @@ class YoutubeSource(Source):
|
|||
"""Create the source."""
|
||||
super().__init__(config)
|
||||
|
||||
if PYTUBE_AVAILABLE:
|
||||
self.innertube_client: innertube.InnerTube = innertube.InnerTube(client="WEB")
|
||||
self.channels: list[str] = config["channels"] if "channels" in config else []
|
||||
self.tmp_dir: str = config["tmp_dir"] if "tmp_dir" in config else "/tmp/syng"
|
||||
|
|
@ -76,10 +76,9 @@ class YoutubeSource(Source):
|
|||
config["start_streaming"] if "start_streaming" in config else False
|
||||
)
|
||||
self.formatstring = (
|
||||
f"bestvideo[height<={self.max_res}]+"
|
||||
f"bestaudio/best[height<={self.max_res}]"
|
||||
f"bestvideo[height<={self.max_res}]+" f"bestaudio/best[height<={self.max_res}]"
|
||||
)
|
||||
if USE_YT_DLP:
|
||||
if YT_DLP_AVAILABLE:
|
||||
self._yt_dlp = YoutubeDL(
|
||||
params={
|
||||
"paths": {"home": self.tmp_dir},
|
||||
|
|
@ -114,8 +113,7 @@ class YoutubeSource(Source):
|
|||
self.player = await self.play_mpv(
|
||||
entry.ident,
|
||||
None,
|
||||
"--script-opts=ytdl_hook-ytdl_path=yt-dlp,"
|
||||
"ytdl_hook-exclude='%.pls$'",
|
||||
"--script-opts=ytdl_hook-ytdl_path=yt-dlp," "ytdl_hook-exclude='%.pls$'",
|
||||
f"--ytdl-format={self.formatstring}",
|
||||
"--fullscreen",
|
||||
)
|
||||
|
|
@ -139,6 +137,9 @@ class YoutubeSource(Source):
|
|||
"""
|
||||
|
||||
def _get_entry(performer: str, url: str) -> Optional[Entry]:
|
||||
if not PYTUBE_AVAILABLE:
|
||||
return None
|
||||
|
||||
try:
|
||||
yt_song = YouTube(url)
|
||||
try:
|
||||
|
|
@ -190,15 +191,10 @@ class YoutubeSource(Source):
|
|||
|
||||
results: list[YouTube] = []
|
||||
results_lists: list[list[YouTube]] = await asyncio.gather(
|
||||
*[
|
||||
asyncio.to_thread(self._channel_search, query, channel)
|
||||
for channel in self.channels
|
||||
],
|
||||
*[asyncio.to_thread(self._channel_search, query, channel) for channel in self.channels],
|
||||
asyncio.to_thread(self._yt_search, query),
|
||||
)
|
||||
results = [
|
||||
search_result for yt_result in results_lists for search_result in yt_result
|
||||
]
|
||||
results = [search_result for yt_result in results_lists for search_result in yt_result]
|
||||
|
||||
results.sort(key=partial(_contains_index, query))
|
||||
|
||||
|
|
@ -242,11 +238,9 @@ class YoutubeSource(Source):
|
|||
results: dict[str, Any] = self.innertube_client._call_api(
|
||||
endpoint, self.innertube_client.base_params, data
|
||||
)
|
||||
items: list[dict[str, Any]] = results["contents"][
|
||||
"twoColumnBrowseResultsRenderer"
|
||||
]["tabs"][-1]["expandableTabRenderer"]["content"]["sectionListRenderer"][
|
||||
"contents"
|
||||
]
|
||||
items: list[dict[str, Any]] = results["contents"]["twoColumnBrowseResultsRenderer"]["tabs"][
|
||||
-1
|
||||
]["expandableTabRenderer"]["content"]["sectionListRenderer"]["contents"]
|
||||
|
||||
list_of_videos: list[YouTube] = []
|
||||
for item in items:
|
||||
|
|
@ -257,16 +251,14 @@ class YoutubeSource(Source):
|
|||
):
|
||||
yt_url: str = (
|
||||
"https://youtube.com/watch?v="
|
||||
+ item["itemSectionRenderer"]["contents"][0]["videoRenderer"][
|
||||
"videoId"
|
||||
]
|
||||
+ item["itemSectionRenderer"]["contents"][0]["videoRenderer"]["videoId"]
|
||||
)
|
||||
author: str = item["itemSectionRenderer"]["contents"][0][
|
||||
"videoRenderer"
|
||||
]["ownerText"]["runs"][0]["text"]
|
||||
title: str = item["itemSectionRenderer"]["contents"][0][
|
||||
"videoRenderer"
|
||||
]["title"]["runs"][0]["text"]
|
||||
author: str = item["itemSectionRenderer"]["contents"][0]["videoRenderer"][
|
||||
"ownerText"
|
||||
]["runs"][0]["text"]
|
||||
title: str = item["itemSectionRenderer"]["contents"][0]["videoRenderer"][
|
||||
"title"
|
||||
]["runs"][0]["text"]
|
||||
yt_song: YouTube = YouTube(yt_url)
|
||||
yt_song.author = author
|
||||
yt_song.title = title
|
||||
|
|
@ -276,9 +268,12 @@ class YoutubeSource(Source):
|
|||
pass
|
||||
return list_of_videos
|
||||
|
||||
async def _buffer_with_yt_dlp(self, entry: Entry) -> Tuple[str, Optional[str]]:
|
||||
async def do_buffer(self, entry: Entry) -> Tuple[str, Optional[str]]:
|
||||
"""
|
||||
Download the video using yt-dlp.
|
||||
Download the video.
|
||||
|
||||
Downloads the highest quality stream respecting the ``max_res``.
|
||||
For higher resolution videos (1080p and above).
|
||||
|
||||
Yt-dlp automatically merges the audio and video, so only the video
|
||||
location exists, the return value for the audio part will always be
|
||||
|
|
@ -286,68 +281,12 @@ class YoutubeSource(Source):
|
|||
|
||||
:param entry: The entry to download.
|
||||
:type entry: Entry
|
||||
:return: The location of the video file and ```None```
|
||||
:return: The location of the video file and ``None``.
|
||||
:rtype: Tuple[str, Optional[str]]
|
||||
"""
|
||||
info = await asyncio.to_thread(self._yt_dlp.extract_info, entry.ident)
|
||||
combined_path = info["requested_downloads"][0]["filepath"]
|
||||
return combined_path, None
|
||||
|
||||
async def do_buffer(self, entry: Entry) -> Tuple[str, Optional[str]]:
|
||||
"""
|
||||
Download the video.
|
||||
|
||||
Downloads the highest quality stream respecting the ``max_res``.
|
||||
For higher resolution videos (1080p and above), YouTube will give you
|
||||
the video and audio seperatly. If that is the case, both will be
|
||||
downloaded.
|
||||
|
||||
If yt-dlp is installed it will be used, otherwise pytube will be used.
|
||||
|
||||
|
||||
:param entry: The entry to download.
|
||||
:type entry: Entry
|
||||
:return: The location of the video file and (if applicable) the
|
||||
location of the audio file.
|
||||
:rtype: Tuple[str, Optional[str]]
|
||||
"""
|
||||
if USE_YT_DLP:
|
||||
return await self._buffer_with_yt_dlp(entry)
|
||||
|
||||
yt_song: YouTube = YouTube(entry.ident)
|
||||
|
||||
streams: StreamQuery = await asyncio.to_thread(lambda: yt_song.streams)
|
||||
|
||||
video_streams: StreamQuery = streams.filter(
|
||||
type="video",
|
||||
custom_filter_functions=[lambda s: int(s.resolution[:-1]) <= self.max_res],
|
||||
)
|
||||
audio_streams: StreamQuery = streams.filter(only_audio=True)
|
||||
|
||||
best_video_stream: Stream = sorted(
|
||||
video_streams,
|
||||
key=lambda s: int(s.resolution[:-1]) + (1 if s.is_progressive else 0),
|
||||
)[-1]
|
||||
best_audio_stream: Stream = sorted(
|
||||
audio_streams, key=lambda s: int(s.abr[:-4])
|
||||
)[-1]
|
||||
|
||||
audio: Optional[str] = (
|
||||
await asyncio.to_thread(
|
||||
best_audio_stream.download,
|
||||
output_path=self.tmp_dir,
|
||||
filename_prefix="audio-",
|
||||
)
|
||||
if best_video_stream.is_adaptive
|
||||
else None
|
||||
)
|
||||
|
||||
video: str = await asyncio.to_thread(
|
||||
best_video_stream.download,
|
||||
output_path=self.tmp_dir,
|
||||
)
|
||||
|
||||
return video, audio
|
||||
|
||||
|
||||
available_sources["youtube"] = YoutubeSource
|
||||
|
|
|
|||
|
|
@ -1,126 +0,0 @@
|
|||
# pylint: disable=missing-function-docstring
|
||||
# pylint: disable=missing-module-docstring
|
||||
# pylint: disable=missing-class-docstring
|
||||
import asyncio
|
||||
from typing import Any, Optional
|
||||
|
||||
from aiocmd import aiocmd
|
||||
import socketio
|
||||
|
||||
from .result import Result
|
||||
from .entry import Entry
|
||||
|
||||
sio: socketio.AsyncClient = socketio.AsyncClient()
|
||||
state: dict[str, Any] = {}
|
||||
|
||||
|
||||
@sio.on("search-results")
|
||||
async def handle_search_results(data: dict[str, Any]) -> None:
|
||||
for raw_item in data["results"]:
|
||||
item = Result(**raw_item)
|
||||
print(f"{item.artist} - {item.title} [{item.album}]")
|
||||
print(f"{item.source}: {item.ident}")
|
||||
|
||||
|
||||
@sio.on("state")
|
||||
async def handle_state(data: dict[str, Any]) -> None:
|
||||
print("New Queue")
|
||||
for raw_item in data["queue"]:
|
||||
item = Entry(**raw_item)
|
||||
print(f"\t{item.performer}: {item.artist} - {item.title} ({item.duration})")
|
||||
print("Waiting Room")
|
||||
for raw_item in data["shadow_queue"]:
|
||||
item = Entry(**raw_item)
|
||||
print(f"\t{item.performer}: {item.artist} - {item.title} ({item.duration})")
|
||||
print("Recent")
|
||||
for raw_item in data["recent"]:
|
||||
item = Entry(**raw_item)
|
||||
print(f"\t{item.performer}: {item.artist} - {item.title} ({item.duration})")
|
||||
|
||||
|
||||
@sio.on("msg")
|
||||
async def handle_msg(data: dict[str, Any]) -> None:
|
||||
print(data["msg"])
|
||||
|
||||
|
||||
@sio.on("connect")
|
||||
async def handle_connect() -> None:
|
||||
print("Connected")
|
||||
await sio.emit("register-web", {"room": state["room"]})
|
||||
|
||||
|
||||
@sio.on("register-admin")
|
||||
async def handle_register_admin(data: dict[str, Any]) -> None:
|
||||
if data["success"]:
|
||||
print("Logged in")
|
||||
else:
|
||||
print("Log in failed")
|
||||
|
||||
|
||||
class SyngShell(aiocmd.PromptToolkitCmd):
|
||||
prompt = "syng> "
|
||||
|
||||
def do_exit(self) -> bool:
|
||||
return True
|
||||
|
||||
async def do_stuff(self) -> None:
|
||||
await sio.emit(
|
||||
"append",
|
||||
{
|
||||
"performer": "Hammy",
|
||||
"source": "youtube",
|
||||
"uid": "mockup",
|
||||
# https://youtube.com/watch?v=x5bM5Bdizi4",
|
||||
"ident": "https://www.youtube.com/watch?v=rqZqHXJm-UA",
|
||||
},
|
||||
)
|
||||
|
||||
async def do_search(self, query: str) -> None:
|
||||
await sio.emit("search", {"query": query})
|
||||
|
||||
async def do_append(
|
||||
self, source: str, ident: str, uid: Optional[str] = None
|
||||
) -> None:
|
||||
await sio.emit(
|
||||
"append",
|
||||
{
|
||||
"performer": "Mockup",
|
||||
"source": source,
|
||||
"ident": ident,
|
||||
"uid": uid if uid is not None else "mockup",
|
||||
},
|
||||
)
|
||||
|
||||
async def do_waiting_room(
|
||||
self, source: str, ident: str, uid: Optional[str] = None
|
||||
) -> None:
|
||||
await sio.emit(
|
||||
"shadow-append",
|
||||
{
|
||||
"performer": "Mockup",
|
||||
"source": source,
|
||||
"ident": ident,
|
||||
"uid": uid if uid is not None else "mockup",
|
||||
},
|
||||
)
|
||||
|
||||
async def do_admin(self, data: str) -> None:
|
||||
await sio.emit("register-admin", {"secret": data})
|
||||
|
||||
async def do_connect(self, server: str, room: str) -> None:
|
||||
state["room"] = room
|
||||
await sio.connect(server)
|
||||
|
||||
async def do_skip(self) -> None:
|
||||
await sio.emit("skip")
|
||||
|
||||
async def do_queue(self) -> None:
|
||||
await sio.emit("get-state")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
asyncio.run(SyngShell().run())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Loading…
Add table
Reference in a new issue