cthulhu/src/cthulhu/sound.py

#!/usr/bin/env python3
#
# Copyright (c) 2024 Stormux
# Copyright (c) 2010-2012 The Orca Team
# Copyright (c) 2012 Igalia, S.L.
# Copyright (c) 2005-2010 Sun Microsystems Inc.
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the
# Free Software Foundation, Inc., Franklin Street, Fifth Floor,
# Boston MA  02110-1301 USA.
#
# Forked from Orca screen reader.
# Cthulhu project: https://git.stormux.org/storm/cthulhu

"""Utilities for playing sounds."""

__id__        = "$Id:$"
__version__   = "$Revision:$"
__date__      = "$Date:$"
__copyright__ = "Copyright (c) 2016 Cthulhu Team"
__license__   = "LGPL"

import json
import os
import subprocess
import sys
import threading
from typing import Any, Optional, Tuple

import gi
from . import debug
from . import gstreamer_support
from . import settings
from . import sound_sink
from .sound_generator import Icon, Tone

try:
    gi.require_version('Gst', '1.0')
    from gi.repository import Gst
except Exception:
    _gstreamerAvailable: bool = False
else:
    _gstreamerAvailable = gstreamer_support.gst_init_check_available(Gst)

_soundSystemFailureReason: Optional[str] = None


class _PendingResponse:
    def __init__(self) -> None:
        self.event = threading.Event()
        self.response: Optional[dict[str, Any]] = None


class Player:
    """Plays Icons and Tones through a persistent worker process."""

    def __init__(self) -> None:
        self._initialized = False
        self._workerProcess: Optional[subprocess.Popen[str]] = None
        self._workerSink: Optional[str] = None
        self._workerRestartRequired = False
        self._workerRestartReason: Optional[str] = None
        self._workerLock = threading.RLock()
        self._responseLock = threading.Lock()
        self._pendingResponses: dict[int, _PendingResponse] = {}
        self._nextRequestId = 1

        if not _gstreamerAvailable:
            debug.printMessage(debug.LEVEL_INFO, 'SOUND ERROR: Gstreamer is not available', True)
            return

    @staticmethod
    def _get_configured_volume() -> float:
        """Returns the configured sound volume with a safe fallback."""

        try:
            from . import cthulhu
            if cthulhu.cthulhuApp is not None:
                volume = cthulhu.cthulhuApp.settingsManager.getSetting('soundVolume')
                return max(0.0, float(volume))
        except Exception:
            pass

        return max(0.0, float(settings.soundVolume))

    def init(self) -> None:
        """(Re)Initializes the persistent worker."""

        if not _gstreamerAvailable:
            return

        with self._workerLock:
            if self._ensureWorkerLocked():
                self._initialized = True
                clearSoundSystemFailure()

    def play(self, item: Any, interrupt: bool = True) -> None:
        """Plays a sound, interrupting the current play first unless specified."""

        if isinstance(item, Icon):
            self._playIcon(item, interrupt)
        elif isinstance(item, Tone):
            self._playTone(item, interrupt)
        else:
            debug.printTokens(debug.LEVEL_INFO, ["SOUND ERROR:", item, "is not an Icon or Tone"], True)

    def playAndWait(self, item: Any, interrupt: bool = True, timeout_seconds: int = 10) -> bool:
        """Plays a sound and blocks until completion or timeout."""

        if isinstance(item, Icon):
            return self._playIconAndWait(item, interrupt=interrupt, timeout_seconds=timeout_seconds)
        if isinstance(item, Tone):
            return self._playToneAndWait(item, interrupt=interrupt, timeout_seconds=timeout_seconds)
        self.play(item, interrupt)
        return False

    def stop(self, _element: Any = None) -> None:
        """Stops current sound playback."""

        with self._workerLock:
            if self._workerProcess is None or self._workerProcess.poll() is not None:
                return

        self._sendWorkerCommand({"action": "stop"}, waitForResponse=False)

    def shutdown(self) -> None:
        """Shuts down the sound utilities."""

        if not _gstreamerAvailable:
            return

        with self._workerLock:
            self._stopWorkerLocked("Sound system shutdown")
            self._initialized = False

    def _playIcon(self, icon: Icon, interrupt: bool = True) -> None:
        if not icon.isValid():
            return

        success, reason = self._sendWorkerCommand(
            {
                "action": "play_file",
                "path": icon.path,
                "volume": self._get_configured_volume(),
                "interrupt": interrupt,
            },
            waitForResponse=False,
        )
        if not success and reason:
            debug.printMessage(debug.LEVEL_INFO, f"SOUND ERROR: {reason}", True)

    def _playIconAndWait(
        self,
        icon: Icon,
        interrupt: bool = True,
        timeout_seconds: Optional[int] = 10,
    ) -> bool:
        if not icon.isValid():
            return False

        timeout = float((timeout_seconds or 10) + 2)
        success, reason = self._sendWorkerCommand(
            {
                "action": "play_file",
                "path": icon.path,
                "volume": self._get_configured_volume(),
                "interrupt": interrupt,
            },
            waitForResponse=True,
            timeout=timeout,
        )
        if not success and reason:
            debug.printMessage(debug.LEVEL_INFO, f"SOUND ERROR: {reason}", True)
        return success

    def _playTone(self, tone: Tone, interrupt: bool = True) -> None:
        success, reason = self._sendWorkerCommand(
            self._buildToneCommand(tone, interrupt),
            waitForResponse=False,
        )
        if not success and reason:
            debug.printMessage(debug.LEVEL_INFO, f"SOUND ERROR: {reason}", True)

    def _playToneAndWait(
        self,
        tone: Tone,
        interrupt: bool = True,
        timeout_seconds: Optional[int] = 10,
    ) -> bool:
        timeout = max(float(timeout_seconds or 10), float(tone.duration) + 2.0)
        success, reason = self._sendWorkerCommand(
            self._buildToneCommand(tone, interrupt),
            waitForResponse=True,
            timeout=timeout,
        )
        if not success and reason:
            debug.printMessage(debug.LEVEL_INFO, f"SOUND ERROR: {reason}", True)
        return success

    def _buildToneCommand(self, tone: Tone, interrupt: bool) -> dict[str, Any]:
        return {
            "action": "play_tone",
            "duration": tone.duration,
            "frequency": tone.frequency,
            "volume": tone.volume,
            "wave": tone.wave,
            "interrupt": interrupt,
        }

    def _ensureWorkerLocked(self) -> bool:
        configuredSink = sound_sink.get_configured_sound_sink()

        if self._workerProcess is not None and self._workerProcess.poll() is None:
            if self._workerRestartRequired:
                reason = self._consumeWorkerRestartReason()
                debug.printMessage(
                    debug.LEVEL_INFO,
                    f"SOUND: Restarting persistent worker after recovery request: {reason}",
                    True,
                )
                self._stopWorkerLocked(reason)
            elif self._workerSink == configuredSink:
                return True
            else:
                debug.printMessage(
                    debug.LEVEL_INFO,
                    f"SOUND: Restarting persistent worker for soundSink={configuredSink}",
                    True,
                )
                self._stopWorkerLocked("Sound sink changed")

        return self._startWorkerLocked(configuredSink)

    def _startWorkerLocked(self, configuredSink: str) -> bool:
        environment = _buildSoundHelperEnvironment()
        command = [
            sys.executable,
            "-m",
            "cthulhu.sound_helper",
            "--worker",
            "--sound-sink",
            configuredSink,
        ]

        try:
            process = subprocess.Popen(
                command,
                stdin=subprocess.PIPE,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                bufsize=1,
                env=environment,
            )
        except Exception as error:
            reason = f"Failed to start persistent sound worker: {error}"
            disableSoundSystem(reason)
            debug.printMessage(debug.LEVEL_INFO, f"SOUND ERROR: {reason}", True)
            return False

        if process.stdin is None or process.stdout is None or process.stderr is None:
            try:
                process.terminate()
            except Exception:
                pass
            reason = "Persistent sound worker is missing stdio pipes"
            disableSoundSystem(reason)
            debug.printMessage(debug.LEVEL_INFO, f"SOUND ERROR: {reason}", True)
            return False

        self._workerProcess = process
        self._workerSink = configuredSink
        self._workerRestartRequired = False
        self._workerRestartReason = None
        self._startWorkerThreads(process)

        success, reason = self._sendWorkerCommandLocked(
            {"action": "ping"},
            waitForResponse=True,
            timeout=2.0,
            allowRestart=False,
        )
        if not success:
            self._stopWorkerLocked(reason or "Persistent sound worker failed to initialize")
            disableSoundSystem(reason or "Persistent sound worker failed to initialize")
            if reason:
                debug.printMessage(debug.LEVEL_INFO, f"SOUND ERROR: {reason}", True)
            return False

        debug.printMessage(
            debug.LEVEL_INFO,
            f"SOUND: Using persistent worker for icon playback (soundSink={configuredSink})",
            True,
        )
        debug.printMessage(
            debug.LEVEL_INFO,
            f"SOUND: Using persistent worker for tone playback (soundSink={configuredSink})",
            True,
        )
        return True

    def _startWorkerThreads(self, process: subprocess.Popen[str]) -> None:
        threading.Thread(
            target=self._readWorkerStdout,
            args=(process,),
            daemon=True,
        ).start()
        threading.Thread(
            target=self._readWorkerStderr,
            args=(process,),
            daemon=True,
        ).start()
        threading.Thread(
            target=self._watchWorkerExit,
            args=(process,),
            daemon=True,
        ).start()

    def _readWorkerStdout(self, process: subprocess.Popen[str]) -> None:
        assert process.stdout is not None
        for line in process.stdout:
            line = line.strip()
            if not line:
                continue
            try:
                response = json.loads(line)
            except Exception:
                debug.printMessage(debug.LEVEL_INFO, f"SOUND ERROR: Invalid worker response: {line}", True)
                continue

            requestId = response.get("id")
            if requestId is None:
                continue

            with self._responseLock:
                pending = self._pendingResponses.pop(int(requestId), None)

            if pending is None:
                continue

            pending.response = response
            pending.event.set()

    def _readWorkerStderr(self, process: subprocess.Popen[str]) -> None:
        assert process.stderr is not None
        for line in process.stderr:
            message = line.strip()
            if message:
                self._handleWorkerDiagnostic(message)
                debug.printMessage(debug.LEVEL_INFO, f"SOUND WORKER: {message}", True)

    def _watchWorkerExit(self, process: subprocess.Popen[str]) -> None:
        returnCode = process.wait()
        reason = _formatWorkerFailure(returnCode)

        with self._workerLock:
            if self._workerProcess is process:
                self._workerProcess = None
                self._workerSink = None

        self._failPendingResponses(reason)

        if returnCode != 0:
            debug.printMessage(debug.LEVEL_INFO, f"SOUND ERROR: {reason}", True)

    def _failPendingResponses(self, reason: str) -> None:
        with self._responseLock:
            pendingResponses = list(self._pendingResponses.values())
            self._pendingResponses.clear()

        for pending in pendingResponses:
            pending.response = {"ok": False, "error": reason}
            pending.event.set()

    def _stopWorkerLocked(self, reason: str) -> None:
        process = self._workerProcess
        self._workerProcess = None
        self._workerSink = None
        self._workerRestartRequired = False
        self._workerRestartReason = None

        if process is None:
            return

        if process.poll() is None:
            try:
                assert process.stdin is not None
                process.stdin.write(json.dumps({"action": "shutdown"}) + "\n")
                process.stdin.flush()
                process.wait(timeout=1.0)
            except Exception:
                try:
                    process.terminate()
                    process.wait(timeout=1.0)
                except Exception:
                    try:
                        process.kill()
                        process.wait(timeout=1.0)
                    except Exception:
                        pass

        self._failPendingResponses(reason)

    def _sendWorkerCommand(
        self,
        command: dict[str, Any],
        waitForResponse: bool,
        timeout: float = 2.0,
    ) -> Tuple[bool, Optional[str]]:
        with self._workerLock:
            return self._sendWorkerCommandLocked(command, waitForResponse, timeout)

    def _sendWorkerCommandLocked(
        self,
        command: dict[str, Any],
        waitForResponse: bool,
        timeout: float = 2.0,
        allowRestart: bool = True,
    ) -> Tuple[bool, Optional[str]]:
        if not self._ensureWorkerLocked():
            return False, getSoundSystemFailureReason() or "Persistent sound worker is unavailable"

        process = self._workerProcess
        if process is None or process.stdin is None:
            if allowRestart:
                self._stopWorkerLocked("Worker pipes disappeared")
                if self._ensureWorkerLocked():
                    return self._sendWorkerCommandLocked(command, waitForResponse, timeout, allowRestart=False)
            return False, "Persistent sound worker stdin is unavailable"

        requestId: Optional[int] = None
        pending: Optional[_PendingResponse] = None
        payload = dict(command)

        if waitForResponse:
            requestId = self._allocateRequestId()
            pending = _PendingResponse()
            payload["id"] = requestId
            with self._responseLock:
                self._pendingResponses[requestId] = pending

        try:
            process.stdin.write(json.dumps(payload) + "\n")
            process.stdin.flush()
        except Exception as error:
            if requestId is not None:
                with self._responseLock:
                    self._pendingResponses.pop(requestId, None)
            if allowRestart:
                self._stopWorkerLocked(f"Worker write failed: {error}")
                if self._ensureWorkerLocked():
                    return self._sendWorkerCommandLocked(command, waitForResponse, timeout, allowRestart=False)
            return False, f"Worker write failed: {error}"

        if not waitForResponse or pending is None:
            return True, None

        if pending.event.wait(timeout):
            response = pending.response or {"ok": False, "error": "No worker response"}
            if not bool(response.get("ok")):
                self._maybeMarkWorkerRestartRequired(response.get("error"))
            return bool(response.get("ok")), response.get("error")

        with self._responseLock:
            self._pendingResponses.pop(requestId, None)

        if allowRestart and (self._workerProcess is None or self._workerProcess.poll() is not None):
            self._stopWorkerLocked("Worker exited while waiting for response")
            if self._ensureWorkerLocked():
                return self._sendWorkerCommandLocked(command, waitForResponse, timeout, allowRestart=False)

        reason = f"Persistent sound worker timed out after {timeout:.1f} seconds"
        self._markWorkerRestartRequired(reason)
        return False, reason

    def _allocateRequestId(self) -> int:
        with self._responseLock:
            requestId = self._nextRequestId
            self._nextRequestId += 1
        return requestId

    def _handleWorkerDiagnostic(self, message: str) -> None:
        normalizedMessage = str(message).strip()
        if not normalizedMessage.startswith("RECOVERY REQUIRED:"):
            return

        reason = normalizedMessage.split(":", 1)[1].strip() or "Worker requested recovery"
        self._markWorkerRestartRequired(reason)

    def _markWorkerRestartRequired(self, reason: Optional[str]) -> None:
        normalizedReason = str(reason or "").strip() or "Worker requested recovery"
        with self._workerLock:
            self._workerRestartRequired = True
            self._workerRestartReason = normalizedReason

    def _consumeWorkerRestartReason(self) -> str:
        reason = self._workerRestartReason or "Worker requested recovery"
        self._workerRestartRequired = False
        self._workerRestartReason = None
        return reason

    def _maybeMarkWorkerRestartRequired(self, reason: Optional[str]) -> None:
        normalizedReason = str(reason or "").strip().lower()
        if normalizedReason in {"", "interrupted", "stopped", "shutdown", "stdin closed"}:
            return

        self._markWorkerRestartRequired(reason)


def disableSoundSystem(reason: str) -> None:
    global _soundSystemFailureReason

    if _soundSystemFailureReason == reason:
        return

    _soundSystemFailureReason = reason
    debug.printMessage(debug.LEVEL_INFO, f"SOUND: Disabling sound system. Reason: {reason}", True)


def getSoundSystemFailureReason() -> Optional[str]:
    return _soundSystemFailureReason


def isSoundSystemAvailable() -> bool:
    return _soundSystemFailureReason is None


def clearSoundSystemFailure() -> None:
    global _soundSystemFailureReason
    _soundSystemFailureReason = None


def _buildSoundHelperEnvironment() -> dict[str, str]:
    pythonPathEntries = []
    for path in sys.path:
        if not path:
            path = os.getcwd()
        if os.path.isdir(path) and path not in pythonPathEntries:
            pythonPathEntries.append(path)

    environment = os.environ.copy()
    if pythonPathEntries:
        environment["PYTHONPATH"] = os.pathsep.join(pythonPathEntries)
    return environment


def _formatWorkerFailure(returnCode: int) -> str:
    if returnCode < 0:
        return f"Sound worker exited via signal {-returnCode}"
    return f"Sound worker exited with status {returnCode}"


_player = Player()


def getPlayer() -> Player:
    return _player


def play(item: Any, interrupt: bool = True) -> None:
    _player.play(item, interrupt)


def playIconSafely(icon: Icon, timeoutSeconds: int = 10) -> Tuple[bool, Optional[str]]:
    if not icon.isValid():
        return False, f"Invalid sound icon: {icon.path}"

    return _player._sendWorkerCommand(
        {
            "action": "play_file",
            "path": icon.path,
            "volume": Player._get_configured_volume(),
            "interrupt": True,
        },
        waitForResponse=True,
        timeout=max(0.1, float(timeoutSeconds)),
    )