音声の再生¶

import bostie
from bostie import Context, voice

bot = bostie.Bot()
tracks: dict[int, voice.Track] = {}


@bot.command
async def play(ctx: Context, url: str) -> None:
    """Play a media URL."""
    vc = ctx.voice.current or await ctx.voice.connect()
    track = await vc.play(voice.Url(url), volume=0.8, meta=ctx.user.id)
    tracks[vc.guild_id] = track
    await ctx.respond(f"Playing {track.source.tags.title or url}.")
    ended = await track.wait()
    if ended.reason is voice.EndReason.ERROR:
        await ctx.followup(f"The track failed: {ended.error}")


@bot.command
async def pause(ctx: Context) -> None:
    """Pause or resume the track."""
    track = tracks.get(ctx.guild_id or 0)
    if track is None or track.ended is not None:
        await ctx.respond("Nothing is playing.", ephemeral=True)
        return
    if track.paused:
        track.resume()
    else:
        track.pause()
    await ctx.respond(f"At {track.position:.0f} s.")


@bot.command
async def seek(ctx: Context, seconds: float) -> None:
    """Jump to a point in the track."""
    track = tracks.get(ctx.guild_id or 0)
    if track is not None and track.source.seekable:
        await track.seek(seconds)
    await ctx.respond(f"Now at {seconds:.0f} s.")


@bot.command
async def bass(ctx: Context, gain: float = 6.0) -> None:
    """Boost the bass and slow down a little."""
    track = tracks.get(ctx.guild_id or 0)
    if track is not None:
        track.filters = voice.Filters(bass=gain, speed=0.9)
    await ctx.respond("Changed the filters.")

import asyncio
from collections.abc import AsyncIterator
from pathlib import Path

import bostie
from bostie import Context, voice

bot = bostie.Bot()
CHIME = Path("chime.ogg")


@bot.command
async def chime(ctx: Context) -> None:
    """Play a short sound from memory."""
    vc = ctx.voice.current or await ctx.voice.connect()
    data = await asyncio.to_thread(CHIME.read_bytes)
    await vc.play(voice.Bytes(data, hint="ogg"))
    await ctx.respond("Ding.")


@bot.command
async def radio(ctx: Context) -> None:
    """Play a station through ffmpeg."""
    vc = ctx.voice.current or await ctx.voice.connect()
    station = voice.FFmpeg(["-i", "https://example.com/radio.m3u8"])
    await vc.play(station)
    await ctx.respond("On air.")


async def silence(seconds: float) -> AsyncIterator[bytes]:
    frame = bytes(3840)
    for _ in range(int(seconds * 50)):
        yield frame


@bot.command
async def hush(ctx: Context) -> None:
    """Play three seconds of generated silence."""
    vc = ctx.voice.current or await ctx.voice.connect()
    track = await vc.play(voice.Stream(silence(3), voice.PCM()))
    await ctx.respond("Shh.")
    await track.wait()

import bostie
from bostie import Context, voice

bot = bostie.Bot(voice=voice.Defaults(crossfade=3.0, history=20))
queues: dict[int, voice.Queue] = {}


async def queue_of(ctx: Context) -> voice.Queue:
    vc = ctx.voice.current or await ctx.voice.connect()
    queue = queues.get(vc.guild_id)
    if queue is None or queue.closed:
        queue = queues[vc.guild_id] = voice.Queue(vc)
    return queue


@bot.command
async def add(ctx: Context, url: str, next_up: bool = False) -> None:
    """Add a URL to the queue, or play it next."""
    queue = await queue_of(ctx)
    queue.add(voice.Url(url), meta=ctx.user.id, index=0 if next_up else None)
    await ctx.respond(f"{len(queue.upcoming)} tracks up next.")


@bot.command
async def queue(ctx: Context) -> None:
    """Show what plays now and next."""
    player = await queue_of(ctx)
    lines = [f"Now: {player.now.source!r}" if player.now else "Nothing playing"]
    lines += [f"{i}. {track.source!r}" for i, track in enumerate(player.upcoming, 1)]
    await ctx.respond("\n".join(lines))


@bot.command
async def back(ctx: Context) -> None:
    """Play the previous track again."""
    player = await queue_of(ctx)
    previous = player.back()
    await ctx.respond("Going back." if previous else "Nothing played before.")


@bot.command
async def shuffle(ctx: Context) -> None:
    """Shuffle the upcoming tracks."""
    (await queue_of(ctx)).shuffle()
    await ctx.respond("Shuffled.")