aliases + adaas stability + ystream refactor

This commit is contained in:
AF 2023-04-12 11:34:28 +00:00
parent 5615271518
commit e8c9f91a7c
4 changed files with 83 additions and 56 deletions

View File

@ -19,6 +19,7 @@ def get_of(mainservice: MainService) -> Callable[[str], command_type]:
at_of: AtOf[str, command_type] = AtOf() at_of: AtOf[str, command_type] = AtOf()
at, of = at_of() at, of = at_of()
@at("?")
@at("help") @at("help")
async def help_(ctx: Context, args: list[str]) -> None: async def help_(ctx: Context, args: list[str]) -> None:
match args: match args:
@ -28,6 +29,7 @@ def get_of(mainservice: MainService) -> Callable[[str], command_type]:
await ctx.reply(f"help for {name}: `{name} help`") await ctx.reply(f"help for {name}: `{name} help`")
@at("/") @at("/")
@at("p")
@at("play") @at("play")
async def play(ctx: Context, args: list[str]) -> None: async def play(ctx: Context, args: list[str]) -> None:
await catch( await catch(

View File

@ -1,7 +1,7 @@
import asyncio import asyncio
from collections import deque from collections import deque
from contextlib import AsyncExitStack from contextlib import AsyncExitStack
from typing import AsyncIterable, Iterable from typing import Any, AsyncIterable, Callable, Coroutine, Iterable
from v6d2ctx.context import * from v6d2ctx.context import *
from v6d2ctx.integration.responsetype import * from v6d2ctx.integration.responsetype import *
@ -83,20 +83,23 @@ class YState:
else: else:
return None return None
def cancel(self) -> None: def clear(self, /) -> None:
self.entries.clear() self.entries.clear()
self.playlists.clear() self.playlists.clear()
self.sources.clear() self.sources.clear()
def cancel(self, /) -> None:
self.clear()
self.results.put_nowait(_Stop()) self.results.put_nowait(_Stop())
class YStream(JobUnit): class YStream(JobUnit):
def __init__(self, state: YState) -> None: def __init__(self, state: YState, /) -> None:
self.state = state self.state = state
self.__running = False self.__running = False
self.__details: dict[str, ResponseType] = {"status": "stopped"} 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(): while self.state.playlists and self.state.playlists[0].done():
playlist = self.state.playlists.popleft() playlist = self.state.playlists.popleft()
if (e := playlist.exception()) is not None: if (e := playlist.exception()) is not None:
@ -107,61 +110,64 @@ class YStream(JobUnit):
for entry in playlist.result(): for entry in playlist.result():
self.state.entries.append(entry) self.state.entries.append(entry)
async def _run(self, context: JobContext, /) -> JobUnit | None: def _case_empty(self, context: JobContext, /) -> None:
if self.state.empty_processing(): self._set_details(context, {"status": "stopping"})
self.__details = {"status": "stopping"} if self.state.results.empty():
if self.state.results.empty(): self.state.results.put_nowait(_Stop())
self.state.results.put_nowait(_Stop())
return None def _case_has_entries(self, context: JobContext, /) -> Callable[[], Coroutine[Any, Any, None]]:
elif self.state.entries: entry = self.state.entries.popleft()
entry = self.state.entries.popleft() audiotask: asyncio.Future[YTAudio | None]
audiotask: asyncio.Future[YTAudio | None] if isinstance(entry, BaseException):
if isinstance(entry, BaseException): self._set_details(context, {"status": "breaking downstream audio creation"})
self._set_details(context, {"status": "breaking downstream audio creation"}) audiotask = asyncio.Future()
audiotask = asyncio.Future() audiotask.set_exception(entry)
audiotask.set_exception(entry) else:
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()
self._set_details( self._set_details(
context, context,
{ {
"status": "parsing playlist", "status": "creating audio",
"url": source.url, "info": cast_to_response(entry.info),
"effects": source.effects, "effects": entry.effects,
"already_read": source.already_read, "already_read": entry.already_read,
"tor": source.tor, "tor": entry.tor,
"ignore": source.ignore, "ignore": entry.ignore,
}, },
) )
playlisttask = asyncio.create_task(self.state.playlist(source)) audiotask = asyncio.create_task(self.state.result(entry))
self.state.playlists.append(playlisttask) 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: try:
await playlisttask await playlisttask
except: except:
self.state.sources.clear() self.state.sources.clear()
return self return
finally: finally:
self._unpack_playlists() self._unpack_playlists()
rescheduled = self.state.descheduled rescheduled = self.state.descheduled
@ -170,11 +176,30 @@ class YStream(JobUnit):
for _ in range(rescheduled): for _ in range(rescheduled):
await self.state.es.enter_async_context(YStream(self.state).at(self.state.pool)) await self.state.es.enter_async_context(YStream(self.state).at(self.state.pool))
self._set_details(context, {"status": "rescheduling self from sources"}) 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: else:
self._set_details(context, {"status": "descheduling"}) return self._case_idle(context)
self.state.descheduled += 1
return None 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: def _set_details(self, context: JobContext, details: dict[str, ResponseType], /) -> None:
self.__details = details self.__details = details

View File

@ -153,7 +153,7 @@ class YTAudio(discord.AudioSource):
def before_options(self) -> str: def before_options(self) -> str:
before_options = "" before_options = ""
if "https" in self.url and not self.unstable: if "http" in self.url and not self.unstable:
before_options += ( before_options += (
" -reconnect 1 -reconnect_at_eof 0 -reconnect_streamed 1 -reconnect_delay_max 60 -copy_unknown" " -reconnect 1 -reconnect_at_eof 0 -reconnect_streamed 1 -reconnect_delay_max 60 -copy_unknown"
) )

View File

@ -19,8 +19,8 @@
# 3.10 has no task groups ffs # 3.10 has no task groups ffs
# whaaeeee # whaaeeee
from typing import AsyncIterable, TypeVar, Generic
import asyncio import asyncio
from typing import AsyncIterable, Generic, TypeVar
T = TypeVar('T') T = TypeVar('T')