#!/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

"""Manages the default speech server for cthulhu.  A script can use this
as its speech server, or it can feel free to create one of its own."""

from __future__ import annotations

__id__        = "$Id$"
__version__   = "$Revision$"
__date__      = "$Date$"
__copyright__ = "Copyright (c) 2005-2009 Sun Microsystems Inc."
__license__   = "LGPL"

import importlib
import time
from contextlib import contextmanager
from typing import TYPE_CHECKING, Optional, List, Dict, Any, Union, Callable

from . import debug
from . import logger
from . import settings
from . import speech_generator
from . import sound
from .sound_generator import Icon
from .speechserver import VoiceFamily

from .acss import ACSS
from . import speech_history

if TYPE_CHECKING:
    from .speechserver import SpeechServer
    from .logger import Logger

# Lazy initialization to avoid circular imports
_logger: Optional[Logger] = None
log: Optional[Any] = None # Logger.newLog returns a log object, keeping Any for now as logger isn't typed

def _ensureLogger() -> None:
    """Ensure logger is initialized."""
    global _logger, log
    if _logger is None:
        from . import cthulhu
        if cthulhu.cthulhuApp:
            _logger = cthulhu.cthulhuApp.logger
            log = _logger.newLog("speech")

# The speech server to use for all speech operations.
#
_speechserver: Optional[SpeechServer] = None
_echoSpeechserver: Optional[SpeechServer] = None

# The last time something was spoken.
_timestamp: float = 0.0

# Optional callback for live monitoring of spoken text.
_monitorWriteTextCallback: Optional[Callable[[str], None]] = None
_monitorWriteTextListeners: List[Callable[[str], None]] = []
_monitorSuppressionDepth = 0

def _isSpeechDispatcherFactory(moduleName: Optional[str]) -> bool:
    if not moduleName:
        return False
    return moduleName.split(".")[-1] == "speechdispatcherfactory"

def _matchesVoiceFamily(candidateFamily: Any, targetFamily: Any) -> bool:
    if not candidateFamily or not targetFamily:
        return False

    for key in (VoiceFamily.NAME, VoiceFamily.LANG, VoiceFamily.DIALECT, VoiceFamily.VARIANT):
        if candidateFamily.get(key) != targetFamily.get(key):
            return False

    return True

def _resolveSpeechDispatcherServerInfo(
    moduleName: Optional[str],
    speechServerInfo: Optional[Any],
    voice: Optional[Any] = None,
    fallbackServerInfo: Optional[Any] = None,
) -> Optional[Any]:

    if not _isSpeechDispatcherFactory(moduleName):
        return speechServerInfo

    if speechServerInfo and len(speechServerInfo) >= 2 and speechServerInfo[1] != "default":
        return speechServerInfo

    if fallbackServerInfo and len(fallbackServerInfo) >= 2 and fallbackServerInfo[1] != "default":
        tokens = [
            "SPEECH: Resolving Speech Dispatcher server via fallback server info:",
            fallbackServerInfo,
        ]
        debug.printTokens(debug.LEVEL_INFO, tokens, True)
        return fallbackServerInfo

    family = None
    if voice:
        try:
            family = ACSS(voice).get(ACSS.FAMILY)
        except Exception:
            debug.printException(debug.LEVEL_INFO)

    if not family:
        try:
            family = ACSS(settings.voices.get(settings.DEFAULT_VOICE, {})).get(ACSS.FAMILY)
        except Exception:
            debug.printException(debug.LEVEL_INFO)
            family = None

    if not family:
        return speechServerInfo

    factory = None
    try:
        factory = importlib.import_module(f"cthulhu.{moduleName}")
    except Exception:
        try:
            factory = importlib.import_module(moduleName)
        except Exception:
            debug.printException(debug.LEVEL_INFO)

    if factory is None:
        return speechServerInfo

    try:
        servers = factory.SpeechServer.getSpeechServers() # type: ignore[attr-defined]
    except Exception:
        debug.printException(debug.LEVEL_INFO)
        return speechServerInfo

    for server in servers:
        try:
            info = server.getInfo()
        except Exception:
            debug.printException(debug.LEVEL_INFO)
            continue

        if not info or len(info) < 2 or info[1] == "default":
            continue

        try:
            families = server.getVoiceFamilies() or []
        except Exception:
            debug.printException(debug.LEVEL_INFO)
            continue

        for candidate in families:
            if not _matchesVoiceFamily(candidate, family):
                continue

            tokens = [
                "SPEECH: Resolved Speech Dispatcher server info",
                info,
                "for voice family",
                family,
            ]
            debug.printTokens(debug.LEVEL_INFO, tokens, True)
            return info

    tokens = [
        "SPEECH: Could not resolve Speech Dispatcher server info for voice family",
        family,
    ]
    debug.printTokens(debug.LEVEL_INFO, tokens, True)
    return speechServerInfo

def _initSpeechServer(moduleName: Optional[str], speechServerInfo: Optional[Any]) -> SpeechServer:

    if not moduleName:
        raise Exception("ERROR: No speech server module name provided")

    factory = None
    try:
        factory = importlib.import_module(f'cthulhu.{moduleName}')
    except Exception:
        try:
            factory = importlib.import_module(moduleName)
        except Exception:
            debug.printException(debug.LEVEL_SEVERE)

    # Now, get the speech server we care about.
    #
    speech_server = None
    if factory:
        if speechServerInfo:
            speech_server = factory.SpeechServer.getSpeechServer(speechServerInfo) # type: ignore

        if not speech_server:
            speech_server = factory.SpeechServer.getSpeechServer() # type: ignore
            if speechServerInfo:
                tokens = ["SPEECH: Invalid speechServerInfo:", speechServerInfo]
                debug.printTokens(debug.LEVEL_INFO, tokens, True)

    if not speech_server:
        raise Exception(f"ERROR: No speech server for factory: {moduleName}")

    return speech_server

def _refreshEchoSpeechServer() -> None:
    global _echoSpeechserver
    _echoSpeechserver = None

    if not settings.enableSpeech or not settings.useCustomEchoSpeechServer:
        return

    if not _isSpeechDispatcherFactory(settings.speechServerFactory):
        return

    try:
        resolvedInfo = _resolveSpeechDispatcherServerInfo(
            settings.speechServerFactory,
            settings.echoSpeechServerInfo,
            settings.echoVoice,
            settings.speechServerInfo,
        )
        if resolvedInfo != settings.echoSpeechServerInfo:
            tokens = [
                "SPEECH: Updating echo speech server info from",
                settings.echoSpeechServerInfo,
                "to",
                resolvedInfo,
            ]
            debug.printTokens(debug.LEVEL_INFO, tokens, True)
            settings.echoSpeechServerInfo = resolvedInfo
        _echoSpeechserver = _initSpeechServer(
            settings.speechServerFactory,
            resolvedInfo
        )
    except Exception:
        debug.printException(debug.LEVEL_INFO)
        _echoSpeechserver = None

def init() -> None:
    global _speechserver
    debug.printMessage(debug.LEVEL_INFO, 'SPEECH: Initializing', True)
    if _speechserver:
        debug.printMessage(debug.LEVEL_INFO, 'SPEECH: Already initialized', True)
        _refreshEchoSpeechServer()
        return

    chosenModuleName = settings.speechServerFactory
    try:
        _speechserver = _initSpeechServer(chosenModuleName, settings.speechServerInfo)
    except Exception:
        moduleNames = settings.speechFactoryModules
        for moduleName in moduleNames:
            if moduleName != settings.speechServerFactory:
                try:
                    _speechserver = _initSpeechServer(moduleName, None)
                    if _speechserver:
                        chosenModuleName = moduleName
                        break
                except Exception:
                    debug.printException(debug.LEVEL_SEVERE)

    if _speechserver:
        if chosenModuleName != settings.speechServerFactory:
            settings.speechServerFactory = chosenModuleName
            settings.speechServerInfo = None
            tokens = [
                "SPEECH: Falling back to speech server factory:",
                chosenModuleName
            ]
            debug.printTokens(debug.LEVEL_INFO, tokens, True)
        tokens = ["SPEECH: Using speech server factory:", chosenModuleName]
        debug.printTokens(debug.LEVEL_INFO, tokens, True)
    else:
        msg = 'SPEECH: Not available'
        debug.printMessage(debug.LEVEL_INFO, msg, True)

    _refreshEchoSpeechServer()
    debug.printMessage(debug.LEVEL_INFO, 'SPEECH: Initialized', True)

def checkSpeechSetting() -> None:
    msg = "SPEECH: Checking speech setting."
    debug.printMessage(debug.LEVEL_INFO, msg, True)

    if not settings.enableSpeech:
        if _speechserver:
            shutdown()
        return

    if not _speechserver:
        init()
        return

    _refreshEchoSpeechServer()

def getSpeechServer() -> Optional[SpeechServer]:
    """Returns the speech server instance."""
    return _speechserver

def setSpeechServer(speechServer: SpeechServer) -> None:
    """Sets the speech server to be used.

    Arguments:
    - speechServer: the speech server to use.
    """
    global _speechserver
    _speechserver = speechServer
    _refreshEchoSpeechServer()

def set_monitor_callbacks(writeText: Optional[Callable[[str], None]] = None) -> None:
    """Sets runtime callbacks for live speech monitoring."""
    global _monitorWriteTextCallback
    _monitorWriteTextCallback = writeText

def add_monitor_callback(writeText: Callable[[str], None]) -> None:
    """Adds a runtime callback for live speech monitoring."""
    if writeText not in _monitorWriteTextListeners:
        _monitorWriteTextListeners.append(writeText)

def remove_monitor_callback(writeText: Callable[[str], None]) -> None:
    """Removes a runtime callback for live speech monitoring."""
    if writeText in _monitorWriteTextListeners:
        _monitorWriteTextListeners.remove(writeText)

@contextmanager
def suppress_monitor_callbacks():
    """Temporarily suppresses live speech monitoring callbacks."""
    global _monitorSuppressionDepth
    _monitorSuppressionDepth += 1
    try:
        yield
    finally:
        _monitorSuppressionDepth = max(0, _monitorSuppressionDepth - 1)

def _write_to_monitor(text: str) -> None:
    """Writes text to the active speech monitor callback if set."""
    if _monitorSuppressionDepth:
        return

    callbacks = []
    if _monitorWriteTextCallback is not None:
        callbacks.append(_monitorWriteTextCallback)
    callbacks.extend(
        callback for callback in _monitorWriteTextListeners
        if callback != _monitorWriteTextCallback
    )

    for callback in callbacks:
        try:
            callback(text)
        except Exception:
            debug.printException(debug.LEVEL_INFO)

def __resolveACSS(acss: Optional[Any] = None) -> ACSS:
    if isinstance(acss, ACSS):
        family = acss.get(acss.FAMILY)
        try:
            family = VoiceFamily(family)
        except Exception:
            family = VoiceFamily({})
        acss[acss.FAMILY] = family
        return acss
    elif isinstance(acss, list) and len(acss) == 1:
        return ACSS(acss[0])
    elif isinstance(acss, dict):
        return ACSS(acss)
    else:
        voices = settings.voices
        return ACSS(voices[settings.DEFAULT_VOICE])

def sayAll(utteranceIterator: Any, progressCallback: Any) -> None:
    _ensureLogger()
    if settings.silenceSpeech:
        return
    if _speechserver:
        def _speechHistorySayAllWrapper() -> Any:
            for [context, acss] in utteranceIterator:
                try:
                    utterance = getattr(context, "utterance", None)
                    if isinstance(utterance, str) and utterance.strip():
                        speech_history.add(utterance, source="sayAll") # type: ignore
                except Exception:
                    debug.printException(debug.LEVEL_INFO)
                yield [context, acss]

        _speechserver.sayAll(_speechHistorySayAllWrapper(), progressCallback) # type: ignore
    else:
        if log:
            for [context, acss] in utteranceIterator:
                logLine = f"SPEECH OUTPUT: '{context.utterance}'"
                debug.printMessage(debug.LEVEL_INFO, logLine, True)
                log.info(logLine)
                try:
                    utterance = getattr(context, "utterance", None)
                    if isinstance(utterance, str) and utterance.strip():
                        speech_history.add(utterance, source="sayAll-fallback") # type: ignore
                except Exception:
                    debug.printException(debug.LEVEL_INFO)

def _speak(text: str, acss: Optional[Any], interrupt: bool) -> None:
    """Speaks the individual string using the given ACSS."""

    _ensureLogger()

    # Block speech for applications in sleep mode, except sleep mode status messages
    from . import cthulhu_state
    from . import sleep_mode_manager
    if cthulhu_state.activeScript and hasattr(cthulhu_state.activeScript, 'app'):
        sleepModeManager = sleep_mode_manager.getManager()
        if sleepModeManager and sleepModeManager.isActiveForApp(cthulhu_state.activeScript.app):
            # Allow sleep mode status messages to get through
            if "Sleep mode enabled" in text or "Sleep mode disabled" in text:
                debug.printMessage(debug.LEVEL_INFO, f"SPEECH: Allowing sleep mode status: '{text}'", True)
            else:
                debug.printMessage(debug.LEVEL_INFO, f"SPEECH: Blocked by sleep mode: '{text}'", True)
                return

    try:
        speech_history.add(text, source="speak") # type: ignore
    except Exception:
        debug.printException(debug.LEVEL_INFO)

    _write_to_monitor(text)

    if not _speechserver:
        logLine = f"SPEECH OUTPUT: '{text}' {acss}"
        debug.printMessage(debug.LEVEL_INFO, logLine, True)
        if log:
            log.info(logLine)
        return

    voice = ACSS(settings.voices.get(settings.DEFAULT_VOICE))
    try:
        voice.update(__resolveACSS(acss))
    except Exception as error:
        msg = f"SPEECH: Exception updated voice with {acss}: {error}"
        debug.printMessage(debug.LEVEL_INFO, msg, True)

    resolvedVoice = __resolveACSS(voice)
    msg = f"SPEECH OUTPUT: '{text}' {resolvedVoice}"
    debug.printMessage(debug.LEVEL_INFO, msg, True)
    _speechserver.speak(text, resolvedVoice, interrupt) # type: ignore

def speak(content: Union[str, List[Any]], acss: Optional[Any] = None, interrupt: bool = False) -> None:
    """Speaks the given content.  The content can be either a simple
    string or an array of arrays of objects returned by a speech
    generator. Speech queues by default; callers that need to cancel
    current output should pass interrupt=True or call presentationInterrupt()."""

    if settings.silenceSpeech:
        return

    # Block speech for applications in sleep mode, except sleep mode status messages
    from . import cthulhu_state
    from . import sleep_mode_manager
    if cthulhu_state.activeScript and hasattr(cthulhu_state.activeScript, 'app'):
        sleepModeManager = sleep_mode_manager.getManager()
        if sleepModeManager and sleepModeManager.isActiveForApp(cthulhu_state.activeScript.app):
            # Allow sleep mode status messages to get through
            if isinstance(content, str):
                if "Sleep mode enabled" in content or "Sleep mode disabled" in content:
                    debug.printMessage(debug.LEVEL_INFO, f"SPEECH: Allowing sleep mode status: '{content}'", True)
                else:
                    debug.printMessage(debug.LEVEL_INFO, f"SPEECH: Blocked by sleep mode: '{content}'", True)
                    return
            elif isinstance(content, list):
                # Check if any element in the content list contains sleep mode messages
                content_str = str(content)
                if "Sleep mode enabled" in content_str or "Sleep mode disabled" in content_str:
                    debug.printMessage(debug.LEVEL_INFO, f"SPEECH: Allowing sleep mode status list: '{content_str}'", True)
                else:
                    debug.printMessage(debug.LEVEL_INFO, f"SPEECH: Blocked by sleep mode list: '{content_str}'", True)
                    return
            else:
                debug.printMessage(debug.LEVEL_INFO, f"SPEECH: Blocked by sleep mode: '{content}'", True)
                return

    validTypes = (str, list, speech_generator.Pause,
                  speech_generator.LineBreak, ACSS, Icon)
    error = "SPEECH: bad content sent to speak(): '%s'"
    if not isinstance(content, validTypes): # type: ignore
        debug.printMessage(debug.LEVEL_INFO, error % content, True)
        return

    global _timestamp
    if _timestamp:
        msg = f"SPEECH: Last spoke {time.time() - _timestamp:.4f} seconds ago"
        debug.printMessage(debug.LEVEL_INFO, msg, True)
    _timestamp = time.time()

    if isinstance(content, str):
        msg = f"SPEECH: Speak '{content}' acss: {acss}"
        debug.printMessage(debug.LEVEL_INFO, msg, True)
    else:
        tokens = ["SPEECH: Speak", content, ", acss:", acss]
        debug.printTokens(debug.LEVEL_INFO, tokens, True)

    if isinstance(content, str):
        _speak(content, acss, interrupt)
    if not isinstance(content, list):
        return

    shouldInterrupt = interrupt
    toSpeak: List[str] = []
    activeVoice = acss
    if acss is not None:
        activeVoice = ACSS(acss)

    for element in content:
        if not isinstance(element, validTypes): # type: ignore
            debug.printMessage(debug.LEVEL_INFO, error % element, True)
        elif isinstance(element, list):
            speak(element, acss, shouldInterrupt)
            shouldInterrupt = False
        elif isinstance(element, str):
            if len(element):
                toSpeak.append(element)
        elif isinstance(element, Icon):
            if toSpeak:
                string = " ".join(toSpeak)
                _speak(string, activeVoice, shouldInterrupt)
                shouldInterrupt = False
                toSpeak = []
            if element.isValid():
                player = sound.getPlayer()
                if player:
                    player.play(element, interrupt=interrupt)
        elif toSpeak:
            newVoice = ACSS(acss)
            newItemsToSpeak = []
            if isinstance(element, speech_generator.Pause):
                if toSpeak[-1] and toSpeak[-1][-1].isalnum():
                    toSpeak[-1] += '.'
            elif isinstance(element, ACSS):
                newVoice.update(element)
                if activeVoice is None:
                    activeVoice = newVoice
                if newVoice == activeVoice:
                    continue
                tokens = ["SPEECH: New voice", newVoice, " != active voice", activeVoice]
                debug.printTokens(debug.LEVEL_INFO, tokens, True)
                newItemsToSpeak.append(toSpeak.pop())

            if toSpeak:
                string = " ".join(toSpeak)
                _speak(string, activeVoice, shouldInterrupt)
                shouldInterrupt = False
            activeVoice = newVoice
            toSpeak = newItemsToSpeak

    if toSpeak:
        string = " ".join(toSpeak)
        _speak(string, activeVoice, shouldInterrupt)

def speakKeyEvent(event: Any, acss: Optional[Any] = None) -> None:
    """Speaks a key event immediately.

    Arguments:
    - event: input_event.KeyboardEvent to speak.
    """

    _ensureLogger()

    if settings.silenceSpeech:
        return

    global _timestamp
    if _timestamp:
        msg = f"SPEECH: Last spoke {time.time() - _timestamp:.4f} seconds ago"
        debug.printMessage(debug.LEVEL_INFO, msg, True)
    _timestamp = time.time()

    keyname = event.getKeyName()
    lockingStateString = event.getLockingStateString()
    acss = __resolveACSS(acss)
    msg = f"{keyname} {lockingStateString}"
    logLine = f"SPEECH OUTPUT: '{msg.strip()}' {acss}"
    debug.printMessage(debug.LEVEL_INFO, logLine, True)
    if log:
        log.info(logLine)

    _write_to_monitor(msg.strip())

    if _speechserver:
        _speechserver.speakKeyEvent(event, acss) # type: ignore

def speakEchoKeyEvent(event: Any, acss: Optional[Any] = None) -> None:
    """Speaks a key event for key/typing echo using echo speech settings."""

    _ensureLogger()

    if settings.silenceSpeech:
        return

    global _timestamp
    if _timestamp:
        msg = f"SPEECH: Last spoke {time.time() - _timestamp:.4f} seconds ago"
        debug.printMessage(debug.LEVEL_INFO, msg, True)
    _timestamp = time.time()

    keyname = event.getKeyName()
    lockingStateString = event.getLockingStateString()
    acss = __resolveACSS(acss)
    msg = f"{keyname} {lockingStateString}"
    logLine = f"SPEECH ECHO OUTPUT: '{msg.strip()}' {acss}"
    debug.printMessage(debug.LEVEL_INFO, logLine, True)
    if log:
        log.info(logLine)

    _write_to_monitor(msg.strip())

    server = _echoSpeechserver or _speechserver
    if server:
        server.speakKeyEvent(event, acss) # type: ignore

def speakCharacter(character: str, acss: Optional[Any] = None) -> None:
    """Speaks a single character immediately.

    Arguments:
    - character: text to be spoken
    - acss:      acss.ACSS instance; if None,
                 the default voice settings will be used.
                 Otherwise, the acss settings will be
                 used to augment/override the default
                 voice settings.
    """

    _ensureLogger()

    if settings.silenceSpeech:
        return

    global _timestamp
    if _timestamp:
        msg = f"SPEECH: Last spoke {time.time() - _timestamp:.4f} seconds ago"
        debug.printMessage(debug.LEVEL_INFO, msg, True)
    _timestamp = time.time()

    acss = __resolveACSS(acss)
    tokens = [f"SPEECH OUTPUT: '{character}'", acss]
    debug.printTokens(debug.LEVEL_INFO, tokens, True)
    if log:
        log.info(f"SPEECH OUTPUT: '{character}'")

    _write_to_monitor(character)

    if _speechserver:
        _speechserver.speakCharacter(character, acss=acss) # type: ignore

def speakEchoCharacter(character: str, acss: Optional[Any] = None) -> None:
    """Speaks a single character for key/typing echo using echo speech settings."""

    _ensureLogger()

    if settings.silenceSpeech:
        return

    global _timestamp
    if _timestamp:
        msg = f"SPEECH: Last spoke {time.time() - _timestamp:.4f} seconds ago"
        debug.printMessage(debug.LEVEL_INFO, msg, True)
    _timestamp = time.time()

    acss = __resolveACSS(acss)
    tokens = [f"SPEECH ECHO OUTPUT: '{character}'", acss]
    debug.printTokens(debug.LEVEL_INFO, tokens, True)
    if log:
        log.info(f"SPEECH ECHO OUTPUT: '{character}'")

    _write_to_monitor(character)

    server = _echoSpeechserver or _speechserver
    if server:
        server.speakCharacter(character, acss=acss) # type: ignore

def speakEchoText(text: str, acss: Optional[Any] = None, interrupt: bool = True) -> None:
    """Speaks text for typing echo using echo speech settings."""

    _ensureLogger()

    if settings.silenceSpeech:
        return

    global _timestamp
    if _timestamp:
        msg = f"SPEECH: Last spoke {time.time() - _timestamp:.4f} seconds ago"
        debug.printMessage(debug.LEVEL_INFO, msg, True)
    _timestamp = time.time()

    acss = __resolveACSS(acss)
    logLine = f"SPEECH ECHO OUTPUT: '{text}' {acss}"
    debug.printMessage(debug.LEVEL_INFO, logLine, True)
    if log:
        log.info(logLine)

    _write_to_monitor(text)

    server = _echoSpeechserver or _speechserver
    if server:
        server.speak(text, acss, interrupt) # type: ignore

def getInfo() -> Optional[Any]:
    info = None
    if _speechserver:
        info = _speechserver.getInfo() # type: ignore

    return info

def stop() -> None:
    if _speechserver:
        _speechserver.stop() # type: ignore
    if _echoSpeechserver and _echoSpeechserver != _speechserver:
        _echoSpeechserver.stop() # type: ignore
    player = sound.getPlayer()
    player.stop()

def shutdown() -> None:
    debug.printMessage(debug.LEVEL_INFO, 'SPEECH: Shutting down', True)
    global _speechserver, _echoSpeechserver
    if _speechserver:
        _speechserver.shutdownActiveServers() # type: ignore
        _speechserver = None
    elif _echoSpeechserver:
        _echoSpeechserver.shutdownActiveServers() # type: ignore
    _echoSpeechserver = None

def reset(text: Optional[str] = None, acss: Optional[Any] = None) -> None:
    if _speechserver:
        _speechserver.reset(text, acss) # type: ignore
