diff --git a/v6d3music/commands.py b/v6d3music/commands.py index 9dcdeb2..bc22313 100644 --- a/v6d3music/commands.py +++ b/v6d3music/commands.py @@ -19,6 +19,7 @@ def get_of(mainservice: MainService) -> Callable[[str], command_type]: at_of: AtOf[str, command_type] = AtOf() at, of = at_of() + @at("?") @at("help") async def help_(ctx: Context, args: list[str]) -> None: match args: @@ -28,6 +29,7 @@ def get_of(mainservice: MainService) -> Callable[[str], command_type]: await ctx.reply(f"help for {name}: `{name} help`") @at("/") + @at("p") @at("play") async def play(ctx: Context, args: list[str]) -> None: await catch( diff --git a/v6d3music/core/ystate.py b/v6d3music/core/ystate.py index 5c0e1b2..636d6ce 100644 --- a/v6d3music/core/ystate.py +++ b/v6d3music/core/ystate.py @@ -1,7 +1,7 @@ import asyncio from collections import deque from contextlib import AsyncExitStack -from typing import AsyncIterable, Iterable +from typing import Any, AsyncIterable, Callable, Coroutine, Iterable from v6d2ctx.context import * from v6d2ctx.integration.responsetype import * @@ -83,20 +83,23 @@ class YState: else: return None - def cancel(self) -> None: + def clear(self, /) -> None: self.entries.clear() self.playlists.clear() self.sources.clear() + + def cancel(self, /) -> None: + self.clear() self.results.put_nowait(_Stop()) class YStream(JobUnit): - def __init__(self, state: YState) -> None: + def __init__(self, state: YState, /) -> None: self.state = state self.__running = False self.__details: dict[str, ResponseType] = {"status": "stopped"} - def _unpack_playlists(self) -> None: + def _unpack_playlists(self, /) -> None: while self.state.playlists and self.state.playlists[0].done(): playlist = self.state.playlists.popleft() if (e := playlist.exception()) is not None: @@ -107,61 +110,64 @@ class YStream(JobUnit): for entry in playlist.result(): self.state.entries.append(entry) - async def _run(self, context: JobContext, /) -> JobUnit | None: - if self.state.empty_processing(): - self.__details = {"status": "stopping"} - if self.state.results.empty(): - self.state.results.put_nowait(_Stop()) - return None - elif self.state.entries: - entry = self.state.entries.popleft() - audiotask: asyncio.Future[YTAudio | None] - if isinstance(entry, BaseException): - self._set_details(context, {"status": "breaking downstream audio creation"}) - audiotask = asyncio.Future() - audiotask.set_exception(entry) - else: - self._set_details( - context, - { - "status": "creating audio", - "info": cast_to_response(entry.info), - "effects": entry.effects, - "already_read": entry.already_read, - "tor": entry.tor, - "ignore": entry.ignore, - }, - ) - audiotask = asyncio.create_task(self.state.result(entry)) - self.state.results.put_nowait(audiotask) - try: - await audiotask - except: - self.state.entries.clear() - self.state.playlists.clear() - self.state.sources.clear() - self._set_details(context, {"status": "rescheduling self from entries"}) - return self - elif self.state.sources: - source = self.state.sources.popleft() + def _case_empty(self, context: JobContext, /) -> None: + self._set_details(context, {"status": "stopping"}) + if self.state.results.empty(): + self.state.results.put_nowait(_Stop()) + + def _case_has_entries(self, context: JobContext, /) -> Callable[[], Coroutine[Any, Any, None]]: + entry = self.state.entries.popleft() + audiotask: asyncio.Future[YTAudio | None] + if isinstance(entry, BaseException): + self._set_details(context, {"status": "breaking downstream audio creation"}) + audiotask = asyncio.Future() + audiotask.set_exception(entry) + else: self._set_details( context, { - "status": "parsing playlist", - "url": source.url, - "effects": source.effects, - "already_read": source.already_read, - "tor": source.tor, - "ignore": source.ignore, + "status": "creating audio", + "info": cast_to_response(entry.info), + "effects": entry.effects, + "already_read": entry.already_read, + "tor": entry.tor, + "ignore": entry.ignore, }, ) - playlisttask = asyncio.create_task(self.state.playlist(source)) - self.state.playlists.append(playlisttask) + audiotask = asyncio.create_task(self.state.result(entry)) + self.state.results.put_nowait(audiotask) + + async def task() -> None: + try: + await audiotask + except: + self.state.clear() + self._set_details(context, {"status": "rescheduling self from entries"}) + + return task + + def _case_has_sources(self, context: JobContext, /) -> Callable[[], Coroutine[Any, Any, None]]: + source = self.state.sources.popleft() + self._set_details( + context, + { + "status": "parsing playlist", + "url": source.url, + "effects": source.effects, + "already_read": source.already_read, + "tor": source.tor, + "ignore": source.ignore, + }, + ) + playlisttask = asyncio.create_task(self.state.playlist(source)) + self.state.playlists.append(playlisttask) + + async def task() -> None: try: await playlisttask except: self.state.sources.clear() - return self + return finally: self._unpack_playlists() rescheduled = self.state.descheduled @@ -170,11 +176,30 @@ class YStream(JobUnit): for _ in range(rescheduled): await self.state.es.enter_async_context(YStream(self.state).at(self.state.pool)) self._set_details(context, {"status": "rescheduling self from sources"}) - return self + + return task + + def _case_idle(self, context: JobContext, /) -> None: + self._set_details(context, {"status": "descheduling"}) + self.state.descheduled += 1 + + def _next_task(self, context: JobContext, /) -> None | Callable[[], Coroutine[Any, Any, None]]: + if self.state.empty_processing(): + return self._case_empty(context) + elif self.state.entries: + return self._case_has_entries(context) + elif self.state.sources: + return self._case_has_sources(context) else: - self._set_details(context, {"status": "descheduling"}) - self.state.descheduled += 1 - return None + return self._case_idle(context) + + async def _run(self, context: JobContext, /) -> JobUnit | None: + match self._next_task(context): + case None: + return None + case task: + await task() + return self def _set_details(self, context: JobContext, details: dict[str, ResponseType], /) -> None: self.__details = details diff --git a/v6d3music/core/ytaudio.py b/v6d3music/core/ytaudio.py index ece829f..a98df31 100644 --- a/v6d3music/core/ytaudio.py +++ b/v6d3music/core/ytaudio.py @@ -153,7 +153,7 @@ class YTAudio(discord.AudioSource): def before_options(self) -> str: before_options = "" - if "https" in self.url and not self.unstable: + if "http" in self.url and not self.unstable: before_options += ( " -reconnect 1 -reconnect_at_eof 0 -reconnect_streamed 1 -reconnect_delay_max 60 -copy_unknown" ) diff --git a/v6d3music/processing/yprocess.py b/v6d3music/processing/yprocess.py index e326e11..7257857 100644 --- a/v6d3music/processing/yprocess.py +++ b/v6d3music/processing/yprocess.py @@ -19,8 +19,8 @@ # 3.10 has no task groups ffs # whaaeeee -from typing import AsyncIterable, TypeVar, Generic import asyncio +from typing import AsyncIterable, Generic, TypeVar T = TypeVar('T')