esl-reader/lib/sounds.py

from typing import Optional, Self, Type, cast

from PyQt6.QtCore import (
    QBuffer,
    QByteArray,
    QCryptographicHash,
    QDir,
    QObject,
    QStandardPaths,
    QUrl,
    pyqtSlot,
)
from PyQt6.QtMultimedia import (
    QAudioOutput,
    QMediaDevices,
    QMediaPlayer,
    QSoundEffect,
)
from PyQt6.QtNetwork import (
    QNetworkAccessManager,
    QNetworkDiskCache,
    QNetworkReply,
    QNetworkRequest,
)
from trycast import trycast

# from PyQt6.QtWidgets import QWidget


class SoundOff(QObject):
    _instance = None

    def __new__(cls: Type[Self]) -> Self:
        if cls._instance:
            return cls._instance
        cls._instance = super(SoundOff, cls).__new__(cls)
        #
        # Setup devices
        #
        cls.virtualDevice = None

        for output in QMediaDevices.audioOutputs():
            if output.id().data().decode("utf-8") == "virt-input":
                cls.virtualDevice = output
            if output.isDefault():
                cls.localDevice = output
        cls.alertEffect = QSoundEffect()
        cls.alertEffect.setSource(QUrl("qrc:/beep.wav"))
        cls.alertEffect.setAudioDevice(cls.localDevice)
        cls.alertEffect.setVolume(0.25)
        cls.alertEffect.setLoopCount(1)

        cls.localPlayer = QMediaPlayer()
        cls.localPlayer.setObjectName("localPlayer")
        cls.localOutput = QAudioOutput()
        cls.localOutput.setDevice(cls.localDevice)
        cls.localPlayer.setAudioOutput(cls.localOutput)
        if cls.virtualDevice:
            cls.virtualPlayer = QMediaPlayer()
            cls.virtualPlayer.setObjectName("virtualPlayer")
            cls.virtualOutput = QAudioOutput()
            cls.virtualOutput.setVolume(1.0)
            cls.virtualOutput.setDevice(cls.virtualDevice)
            cls.virtualPlayer.setAudioOutput(cls.virtualOutput)

        cacheDir = QDir(
            QStandardPaths.writableLocation(
                QStandardPaths.StandardLocation.GenericCacheLocation
            )
        )
        cacheDir.mkdir("Troglodite")
        cacheDir = QDir(cacheDir.path() + QDir.separator() + "Troglodite")
        netCache = QNetworkDiskCache()
        netCache.setCacheDirectory(cacheDir.path())
        cls.nam = QNetworkAccessManager()
        cls.nam.setCache(netCache)
        return cls._instance

    def __init__(self) -> None:
        super().__init__()
        if self.localPlayer.receivers(self.localPlayer.errorOccurred) > 0:
            print("SoundOff, __init__() after __init__()")

        #
        # Connections
        #
        self.localPlayer.errorOccurred.connect(self.mediaError)
        self.localPlayer.mediaStatusChanged.connect(self.mediaStatus)
        self.localPlayer.playbackStateChanged.connect(self.playbackState)
        if self.virtualDevice:
            self.virtualPlayer.errorOccurred.connect(self.mediaError)
            self.virtualPlayer.mediaStatusChanged.connect(self.mediaStatus)
            self.virtualPlayer.playbackStateChanged.connect(self.playbackState)
        return

    @pyqtSlot(QMediaPlayer.Error, str)
    def mediaError(self, error: QMediaPlayer.Error, msg: str) -> None:
        print(error)
        print(msg)
        return

    @pyqtSlot(QMediaPlayer.MediaStatus)
    def mediaStatus(self, status: QMediaPlayer.MediaStatus) -> None:
        if status == QMediaPlayer.MediaStatus.LoadedMedia:
            player = trycast(QMediaPlayer, self.sender())
            if player is None:
                player = trycast(QMediaPlayer, self.lastSender)
            assert player is not None
            player.play()
        self.lastSender = self.sender()
        return

    @pyqtSlot(QMediaPlayer.PlaybackState)
    def playbackState(self, state: QMediaPlayer.PlaybackState) -> None:
        # print(f"playbackState: {state}")
        return

    #
    # Communications slots
    #
    @pyqtSlot()
    def soundAlert(self) -> None:
        self.alertEffect.play()
        return

    @pyqtSlot(str)
    @pyqtSlot(QUrl)
    def playSound(self, url: str | QUrl) -> None:
        if isinstance(url, str):
            url = QUrl(url)
        if not self.localPlayer.audioOutput():
            self.localPlayer.setAudioOutput(self.localOutput)
        if self.virtualPlayer and not self.virtualPlayer.audioOutput():
            self.virtualPlayer.setAudioOutput(self.virtualOutput)
        if url != self._lastUrl:
            request = QNetworkRequest(url)
            reply = self.nam.get(request)
            assert reply is not None
            self._lastUrl = url
            reply.finished.connect(self.finished)
            return
        for player in [self.localPlayer, self.virtualPlayer]:
            if not player:
                continue
            player.setPosition(0)
            if player.mediaStatus() == QMediaPlayer.MediaStatus.LoadedMedia:
                player.play()
        return

    @pyqtSlot()
    def alert(self) -> None:
        self.alertEffect.play()
        return

    #
    # Network slots
    #
    _storage: dict[QMediaPlayer, QByteArray] = {}
    _buffer: dict[QMediaPlayer, QBuffer] = {}
    _lastUrl = QUrl()

    @pyqtSlot()
    def finished(self) -> None:
        reply = trycast(QNetworkReply, self.sender())
        assert reply is not None
        code = reply.attribute(QNetworkRequest.Attribute.HttpStatusCodeAttribute)
        print(f"HttpStatusCodeAttribute: {code}, error: {reply.error()}")
        self._reply = reply.readAll()
        url = reply.request().url()
        reply.close()
        if self._reply.isEmpty() or self._reply.isNull():
            return
        for player in [self.localPlayer, self.virtualPlayer]:
            if not player:
                continue
            self._storage[player] = QByteArray(self._reply)
            self._buffer[player] = QBuffer(self._storage[player])
            player.setSourceDevice(self._buffer[player], url)
            player.setPosition(0)
            if player.mediaStatus() == QMediaPlayer.MediaStatus.LoadedMedia:
                player.play()
                print("play")
        return