S.T.A.R.K. - Speech And Text Algorithmic Recognition Kit
import asyncer

from stark.interfaces.protocols import (
    SpeechRecognizer,
    SpeechRecognizerDelegate,
    SpeechSynthesizer,
    SpeechSynthesizerResult
)   
from stark.core import (
    Command,
    Pattern,
    Response,
    ResponseStatus,
    CommandsContext,
    CommandsManager
)
from stark.voice_assistant import (
    VoiceAssistant,
    Mode
)
from stark.general.blockage_detector import BlockageDetector


async def run(
    manager: CommandsManager,
    speech_recognizer: SpeechRecognizer,
    speech_synthesizer: SpeechSynthesizer
):
    async with asyncer.create_task_group() as main_task_group:
        context = CommandsContext(
            task_group = main_task_group, 
            commands_manager = manager
        )
        voice_assistant = VoiceAssistant(
            speech_recognizer = speech_recognizer,
            speech_synthesizer = speech_synthesizer,
            commands_context = context
        )
        speech_recognizer.delegate = voice_assistant
        context.delegate = voice_assistant
        
        main_task_group.soonify(speech_recognizer.start_listening)()
        main_task_group.soonify(context.handle_responses)()
        
        detector = BlockageDetector()
        main_task_group.soonify(detector.monitor)()

stark/__init__.py
MarkParker5/STARK
from .patterns import Pattern, expressions
from . import types
from .command import (
    Command,
    Response,
    ResponseStatus,
    ResponseHandler,
    AsyncResponseHandler
)
from .commands_manager import (
    CommandsManager
)
from .commands_context import (
    CommandsContext,
    CommandsContextLayer,
    CommandsContextDelegate
)


Pattern.add_parameter_type(types.String)
Pattern.add_parameter_type(types.Word)

stark/core/__init__.py
MarkParker5/STARK
from __future__ import annotations
from typing import cast, Callable, Awaitable, Any, Protocol, AsyncGenerator, Generator, Optional, TypeVar, Generic
from functools import wraps, update_wrapper
import warnings
from uuid import UUID, uuid4
from enum import auto, Enum
from datetime import datetime
import json
import inspect

from pydantic import BaseModel, Field
import asyncer

from ..general.classproperty import classproperty
from .patterns import Pattern


ResponseOptions = Optional['Response'] | Generator[Optional['Response'], None, None] | AsyncGenerator[Optional['Response'], None]
AwaitResponse = Awaitable[ResponseOptions]
AsyncCommandRunner = Callable[..., AwaitResponse]
SyncCommandRunner = Callable[..., Optional['Response']]
CommandRunner = TypeVar('CommandRunner', bound = SyncCommandRunner | AsyncCommandRunner)

class Command(Generic[CommandRunner]):
    name: str
    pattern: Pattern
    _runner: CommandRunner

    def __init__(self, name: str, pattern: Pattern, runner: CommandRunner):
        assert isinstance(pattern, Pattern)
        self.name = name
        self.pattern = pattern
        self._runner = runner
        update_wrapper(self, runner)

    def run(self, parameters_dict: dict[str, Any] | None = None, / , **kwparameters: dict[str, Any]) -> AwaitResponse:
        # allow call both with and without dict unpacking 
        # e.g. command.run(foo = bar, lorem = ipsum), command.run(**parameters) and command.run(parameters)
        # where parameters is dict {'foo': bar, 'lorem': ipsum}
        
        parameters = parameters_dict or {}
        parameters.update(kwparameters)
        
        runner: AsyncCommandRunner
        
        if inspect.iscoroutinefunction(self._runner) or inspect.isasyncgen(self._runner):
             # async functions (coroutines) and async generators are remain as is
            runner = cast(AsyncCommandRunner, self._runner)
        else:
            # sync functions are wrapped with asyncer.asyncify to make them async (coroutines)
            # async generators are not supported yet by asyncer.asyncify (https://github.com/tiangolo/asyncer/discussions/86)
            runner = asyncer.asyncify(cast(SyncCommandRunner, self._runner))
            
        if any(p.kind == p.VAR_KEYWORD for p in inspect.signature(self._runner).parameters.values()):
            # if command runner accepts **kwargs, pass all parameters
            coroutine = runner(**parameters)
        else:
            # otherwise pass only parameters that are in command runner signature to prevent TypeError: got an unexpected keyword argument
            coroutine = runner(**{k: v for k, v in parameters.items() if k in self._runner.__code__.co_varnames})
            
        @wraps(runner)
        async def coroutine_wrapper() -> ResponseOptions:
            try:
                response = await coroutine
            except Exception as e:
                response = Response(
                    text = f'Command {self} raised an exception: {e.__class__.__name__}',
                    voice = f'Command {self} raised an exception: {e.__class__.__name__}',
                    status = ResponseStatus.error
                )
            if inspect.isgenerator(response):
                message = f'[WARNING] Command {self} is a sync GeneratorType that is not fully supported and may block the main thread. ' + \
                            'Consider using the ResponseHandler.respond() or async approach instead.'
                warnings.warn(message, UserWarning)
            return response
            
        return coroutine_wrapper()
    
    def __call__(self, *args, **kwargs) -> AwaitResponse:
        # just syntactic sugar for command(...) instead of command.run(...)
        return self.run(*args, **kwargs)
    
    def __repr__(self):
        return f'<Command {self.name}>'

class ResponseStatus(Enum):
    none = auto()
    not_found = auto()
    failed = auto()
    success = auto()
    info = auto()
    error = auto()

class Response(BaseModel):
    voice: str = ''
    text: str = ''
    status: ResponseStatus = ResponseStatus.success
    needs_user_input: bool = False
    commands: list[Command] = []
    parameters: dict[str, Any] = {}
    
    id: UUID = Field(default_factory=uuid4)
    time: datetime = Field(default_factory=datetime.now)
    
    _repeat_last: Response | None = None # static instance
    
    @classproperty
    def repeat_last(cls) -> Response:
        if not Response._repeat_last:
            Response._repeat_last = Response()
            return Response._repeat_last
        return Response._repeat_last
    
    class Config:
        arbitrary_types_allowed = True
    
    def __eq__(self, other):
        return self.id == other.id
        
class ResponseHandler(Protocol):
    def respond(self, response: Response): pass
    def unrespond(self, response: Response): pass
    def pop_context(self): pass

class AsyncResponseHandler(Protocol):
    async def respond(self, response: Response): pass
    async def unrespond(self, response: Response): pass
    async def pop_context(self): pass

stark/core/command.py
MarkParker5/STARK
from __future__ import annotations
from types import GeneratorType, AsyncGeneratorType
from typing import Any, Protocol, runtime_checkable
from dataclasses import dataclass
import warnings

import anyio
from asyncer import syncify
from asyncer._main import TaskGroup

from ..general.dependencies import DependencyManager, default_dependency_manager
from .commands_manager import CommandsManager, SearchResult
from .command import Command, Response, ResponseHandler, AsyncResponseHandler, CommandRunner, ResponseOptions


@dataclass
class CommandsContextLayer:
    commands: list[Command]
    parameters: dict[str, Any]

@runtime_checkable
class CommandsContextDelegate(Protocol):
    async def commands_context_did_receive_response(self, response: Response): pass
    def remove_response(self, response: Response): pass

class CommandsContext:
    
    is_stopped = False
    commands_manager: CommandsManager
    dependency_manager: DependencyManager
    last_response: Response | None = None
    fallback_command: Command | None = None
    
    _delegate: CommandsContextDelegate | None = None
    _response_queue: list[Response]
    _context_queue: list[CommandsContextLayer]
    _task_group: TaskGroup
    
    def __init__(self, task_group: TaskGroup, commands_manager: CommandsManager, dependency_manager: DependencyManager = default_dependency_manager):
        assert isinstance(task_group, TaskGroup), task_group
        assert isinstance(commands_manager, CommandsManager)
        assert isinstance(dependency_manager, DependencyManager)
        self.commands_manager = commands_manager
        self._context_queue = [self.root_context]
        self._response_queue = []
        self._task_group = task_group
        self.dependency_manager = dependency_manager
        self.dependency_manager.add_dependency(None, AsyncResponseHandler, self)
        self.dependency_manager.add_dependency(None, ResponseHandler, SyncResponseHandler(self))
        self.dependency_manager.add_dependency('inject_dependencies', None, self.inject_dependencies)
        
    @property
    def delegate(self):
        return self._delegate
    
    @delegate.setter
    def delegate(self, delegate: CommandsContextDelegate | None):
        assert isinstance(delegate, CommandsContextDelegate) or delegate is None
        self._delegate = delegate

    @property
    def root_context(self):
        return CommandsContextLayer(self.commands_manager.commands, {})

    async def process_string(self, string: str):
        
        if not self._context_queue:
            self._context_queue.append(self.root_context)

        # search commands
        while self._context_queue:
            
            current_context = self._context_queue[0]
            search_results = await self.commands_manager.search(string = string, commands = current_context.commands)
            
            if search_results:
                break
            else:
                self._context_queue.pop(0)
                
        if not search_results and self.fallback_command and (matches := await self.fallback_command.pattern.match(string)):
            for match in matches:
                search_results = [SearchResult(
                    command = self.fallback_command,
                    match_result = match,
                    index = 0
                )]

        for search_result in search_results or []:

            parameters = current_context.parameters
            parameters.update(search_result.match_result.parameters)
            parameters.update(self.dependency_manager.resolve(search_result.command._runner))
            
            self.run_command(search_result.command, parameters)
            
    def inject_dependencies(self, runner: Command[CommandRunner] | CommandRunner) -> CommandRunner:
        def injected_func(**kwargs) -> ResponseOptions:
            kwargs.update(self.dependency_manager.resolve(runner._runner if isinstance(runner, Command) else runner))
            return runner(**kwargs) # type: ignore
        return injected_func # type: ignore
                
    def run_command(self, command: Command, parameters: dict[str, Any] = {}):
        async def command_runner():
            command_return = await command(parameters)
            
            if isinstance(command_return, Response):
                await self.respond(command_return)
            
            elif isinstance(command_return, AsyncGeneratorType):
                async for response in command_return:
                    if response:
                        await self.respond(response)
                        
            elif isinstance(command_return, GeneratorType):
                message = f'[WARNING] Command {command} is a sync GeneratorType that is not fully supported and may block the main thread. ' + \
                            'Consider using the ResponseHandler.respond() or async approach instead.'
                warnings.warn(message, UserWarning)
                for response in command_return:
                    if response:
                        await self.respond(response)
            
            elif command_return is None:            
                pass
            
            else:
                raise TypeError(f'Command {command} returned {command_return} of type {type(command_return)} instead of Response or AsyncGeneratorType[Response]')
                
        self._task_group.soonify(command_runner)()

    # ResponseHandler
    
    async def respond(self, response: Response): # async forces to run in main thread
        assert isinstance(response, Response)
        self._response_queue.append(response)
    
    async def unrespond(self, response: Response):
        if response in self._response_queue:
            self._response_queue.remove(response)
        self.delegate.remove_response(response)
    
    async def pop_context(self):
        self._context_queue.pop(0)
    
    # Context
    
    def pop_to_root_context(self):
        self._context_queue = [self.root_context]
    
    def add_context(self, context: CommandsContextLayer):
        self._context_queue.insert(0, context)
        
    # ResponseQueue
    
    async def handle_responses(self):
        self.is_stopped = False
        while not self.is_stopped:
            while self._response_queue:
                await self._process_response(self._response_queue.pop(0))
            await anyio.sleep(0.01)
            
    def stop(self):
        self.is_stopped = True
    
    async def _process_response(self, response: Response):
        if response is Response.repeat_last and self.last_response:
            await self._process_response(self.last_response)
            return
        
        if not response is Response.repeat_last:    
            self.last_response = response
        
        if response.commands:
            newContext = CommandsContextLayer(response.commands, response.parameters)
            self._context_queue.insert(0, newContext)
        
        await self.delegate.commands_context_did_receive_response(response)

class SyncResponseHandler: # needs for changing thread from worker to main in commands ran with asyncify
    
    async_response_handler: ResponseHandler
    
    def __init__(self, async_response_handler: ResponseHandler):
        self.async_response_handler = async_response_handler
    
    # ResponseHandler
    
    def respond(self, response: Response):
        syncify(self.async_response_handler.respond)(response)
        
    def unrespond(self, response: Response):
        syncify(self.async_response_handler.unrespond)(response)
        
    def pop_context(self):
        syncify(self.async_response_handler.pop_context)()

stark/core/commands_context.py
MarkParker5/STARK
from __future__ import annotations
from dataclasses import dataclass
import inspect
from asyncer import create_task_group, SoonValue
import json

from .patterns import Pattern, MatchResult
from .types import Object
from .command import Command, CommandRunner, ResponseHandler, AsyncResponseHandler


@dataclass
class SearchResult:
    command: Command
    match_result: MatchResult
    index: int = 0

class CommandsManager:
    
    name: str
    commands: list[Command]
    
    def __init__(self, name: str = ''):
        self.name = name or 'CommandsManager'
        self.commands = []
        
    def get_by_name(self, name: str) -> Command | None:
        for command in self.commands:
            if command.name in {f'{self.name}.{name}', name}:
                return command
        return None

    async def search(self, string: str, commands: list[Command] | None = None) -> list[SearchResult]:
        
        if not commands:
            commands = self.commands
        
        objects_cache: dict[str, Object] = {}
        results: list[SearchResult] = []
        futures: list[tuple[Command, SoonValue[list[MatchResult]]]] = []
        
        # run concurent commands match
        async with create_task_group() as group:
            for command in commands:
                futures.append((command, group.soonify(command.pattern.match)(string, objects_cache)))
        
        # read all finished matches
        i = 0
        for command, future in futures:
            for match in future.value: # may be empty for most of commands
                results.append(SearchResult(
                    command = command,
                    match_result = match,
                    index = i
                ))
                i += 1
                
        # resolve overlapped results
        
        results = sorted(results, key = lambda result: result.match_result.start)
        
        # copy to prevent affecting iteration by removing items; slice makes copy automatically
        for prev, current in zip(results.copy(), results[1:]): # TODO: concurrent; optimize using cache
            if prev.match_result.start == current.match_result.start or prev.match_result.end > current.match_result.start:
                
                # constrain prev end to current start
                prev_cut = await prev.command.pattern.match(string[prev.match_result.start:current.match_result.start], objects_cache) 
                # constrain current start to prev end
                current_cut = await current.command.pattern.match(string[prev.match_result.end:current.match_result.end], objects_cache)

                # less index = more priority to save full match
                priority1, priority2 = (prev, current) if prev.index < current.index else (current, prev)
                priority1_cut, priority2_cut = (prev_cut, current_cut) if prev.index < current.index else (current_cut, prev_cut)
                
                if new_matches := priority2_cut: # if can cut less priority
                    priority2.match_result = new_matches[0]
                elif new_matches := priority1_cut: # else if can cut more priority
                    priority1.match_result = new_matches[0]
                else: # else remove less priority
                    results.remove(priority2)
        
        return results 
    
    def new(self, pattern_str: str, hidden: bool = False):
        def creator(runner: CommandRunner) -> Command:
            pattern = Pattern(pattern_str)
            
            # check that runner has all parameters from pattern
            
            error_msg = f'Command {self.name}.{runner.__name__} must have all parameters from pattern: {pattern.parameters=} {runner.__annotations__=}'
            assert pattern.parameters.items() <= runner.__annotations__.items(), error_msg
            
            # additional checks for DI
            
            if ResponseHandler in runner.__annotations__.values() and inspect.iscoroutinefunction(runner):
                raise TypeError(f'`ResponseHandler` is not compatible with command {self.name}.{runner.__name__} because it is async, use `AsyncResponseHandler` instead')
            
            if AsyncResponseHandler in runner.__annotations__.values() and not inspect.iscoroutinefunction(runner):
                raise TypeError(f'`AsyncResponseHandler` is not compatible with command {self.name}.{runner.__name__} because it is sync, use `ResponseHandler` instead')
            
            # create command
            
            cmd = Command(f'{self.name}.{runner.__name__}', pattern, runner)
            
            if not hidden:
                self.commands.append(cmd)
            
            return cmd
        return creator
    
    def extend(self, other_manager: CommandsManager):
        self.commands.extend(other_manager.commands)

stark/core/commands_manager.py
MarkParker5/STARK
from . import expressions
from .pattern import Pattern, MatchResult

stark/core/patterns/__init__.py
MarkParker5/STARK
alphanumerics = r'A-zА-яЁё0-9'
specials = r'\(\)\[\]\{\}'
any = alphanumerics + specials

dictionary = {
    #   one of the list (a|b|c)
    r'\(((?:.*\|)*.*)\)': r'(?:\1)',
    
    #   one or more of the list, space-splitted {a|b|c}
    r'\{((?:.*\|?)*?.*?)\}': r'(?:(?:\1)\\s?)+',
    
    #   stars *
    r'\*\*': fr'[{alphanumerics}\\s]*', # ** for any few words
    fr'([^{specials}]|^)\*': fr'\1[{alphanumerics}]*', # * for any word    
}

stark/core/patterns/expressions.py
MarkParker5/STARK
from __future__ import annotations
from typing import Type, Generator, TypeAlias, TYPE_CHECKING
from dataclasses import dataclass
import re
from asyncer import create_task_group, SoonValue
import json

from .expressions import dictionary
if TYPE_CHECKING:
    from ..types import Object, ParseResult
    ObjectType: TypeAlias = Type[Object]


@dataclass
class MatchResult:
    substring: str
    start: int
    end: int
    parameters: dict[str, Object]

class Pattern:
    
    parameters: dict[str, ObjectType]
    compiled: str
    
    _origin: str
    _parameter_regex: re.Pattern
    _parameter_types: dict[str, ObjectType] = {} # static

    def __init__(self, origin: str):
        self._origin = origin
        self._parameter_regex = self._get_parameter_regex()
        self.parameters = dict(self._get_parameters())
        self.compiled = self._compile()
        
    async def match(self, string: str, objects_cache: dict[str, Object] | None = None) -> list[MatchResult]:
        
        if objects_cache is None:
            objects_cache = {}
            
        matches: list[MatchResult] = []
        
        for match in sorted(re.finditer(self.compiled, string), key = lambda match: match.start()):
            
            if match.start() == match.end():
                continue # skip empty
            
            # start and end in string, not in match.group(0) 
            match_start = match.start()
            match_end = match.end()
            match_str_groups = match.groupdict()
            
            # parse parameters
            
            parameters: dict[str, Object] = {}
            substrings: dict[str, str] = {}
            futures: list[tuple[str, SoonValue[ParseResult]]] = []
            
            # run concurrent objects parsing
            async with create_task_group() as group:
                for name, object_type in self.parameters.items():
                    if not match_str_groups.get(name):
                        continue
                    
                    parameter_str = match_str_groups[name].strip()
                    
                    for parsed_substr, parsed_obj in objects_cache.items():
                        if parsed_substr in parameter_str:
                            parameters[name] = parsed_obj.copy()
                            parameter_str = parsed_substr
                            substrings[name] = parsed_substr
                            break
                    else:
                        async def parse(from_string: str, parameters: dict[str, str]):
                            parse_result = await object_type.parse(from_string = from_string, parameters = parameters)
                            objects_cache[parse_result.substring] = parse_result.obj
                            return parse_result
                        futures.append((name, group.soonify(parse)(parameter_str, match_str_groups)))
            
            # read futures
            for name, future in futures:
                parse_result = future.value
                parameters[name] = parse_result.obj
                substrings[name] = parse_result.substring
                
            # save parameters
            for name in parameters.keys():
                parameter_str = substrings[name]
                parameter_start = match_str_groups[name].find(parameter_str)
                parameter_end = parameter_start + len(parameter_str)
                
                # adjust start, end and substring after parsing parameters
                if match.start(name) == match.start() and parameter_start != 0:
                    match_start = match.start(name) + parameter_start
                if match.end(name) == match.end() and parameter_end != len(parameter_str):
                    match_end = match.start(name) + parameter_start + parameter_end 
                    
            # strip original string
            substring = string[match_start:match_end].strip()
            start = match_start + string[match_start:match_end].find(substring)
            end = start + len(substring)
            
            matches.append(MatchResult(
                substring = substring,
                start = start,
                end = end,
                parameters = parameters
            ))
            
        # filter overlapping matches
        
        for prev, current in zip(matches.copy(), matches[1:]): # copy to prevent affecting iteration by removing items; slice makes copy automatically
            if prev.start == current.start or prev.end > current.start: # if overlap 
                matches.remove(min(prev, current, key = lambda m: len(m.substring))) # remove shorter
            
        return sorted(matches, key = lambda m: len(m.substring), reverse = True)
    
    @classmethod
    def add_parameter_type(cls, object_type: ObjectType):
        error_msg = f'Can`t add parameter type "{object_type.__name__}": pattern parameters do not match properties annotated in class'
        assert object_type.pattern.parameters.items() <= object_type.__annotations__.items(), error_msg
        exists_type = cls._parameter_types.get(object_type.__name__)
        assert exists_type in {object_type, None}, f'Can`t add parameter type: {object_type.__name__} already exists'
        cls._parameter_types[object_type.__name__] = object_type
        
    # private
    
    def _get_parameter_regex(self) -> re.Pattern:
        # types = '|'.join(Pattern._parameter_types.keys())
        # return re.compile(r'\$(?P<name>[A-z][A-z0-9]*)\:(?P<type>(?:' + types + r'))')
        # do not use types list because it prevents validation of unknown types
        return re.compile(r'\$(?P<name>[A-z][A-z0-9]*)\:(?P<type>[A-z][A-z0-9]*)')
    
    def _get_parameters(self) -> Generator[tuple[str, ObjectType], None, None]:
        for match in re.finditer(self._parameter_regex, self._origin):
            arg_name = match.group('name')
            arg_type_name = match.group('type')
            arg_type: ObjectType | None = Pattern._parameter_types.get(arg_type_name)
            
            if not arg_type: 
                raise NameError(f'Unknown type: "{arg_type_name}" for parameter: "{arg_name}" in pattern: "{self._origin}"')
            
            yield arg_name, arg_type
            
    def _compile(self) -> str: # transform Pattern to classic regex with named groups
        
        pattern: str = self._origin

        #   transform core expressions to regex
        
        for pattern_re, regex in dictionary.items():
            pattern = re.sub(pattern_re, regex, pattern)

        #   find and transform parameters like $name:Type
        
        for name, object_type in self.parameters.items():
            
            arg_declaration = f'\\${name}\\:{object_type.__name__}'
            arg_pattern = object_type.pattern.compiled.replace('\\', r'\\')
            pattern = re.sub(arg_declaration, f'(?P<{name}>{arg_pattern})', pattern)
        
        return pattern
    
    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Pattern):
            raise NotImplementedError(f'Can`t compare Pattern with {type(other)}')
        return self._origin == other._origin
    
    def __repr__(self) -> str:
        return f'<Pattern \'{self._origin}\'>'

stark/core/patterns/pattern.py
MarkParker5/STARK
from .object import Object, ParseResult, ParseError
from .string import String
from .word import Word
# from .Number import Number
# from .TimeInterval import TimeInterval
# from .Time import Time

stark/core/types/__init__.py
MarkParker5/STARK
# from .object import Object, Pattern, classproperty

# class Number(Object):
#     value: float
#     isPercentage: bool = False

#     def __init__(self, value: float):
#         self.value = value

#     @classmethod
#     def parse(cls, fromString: str):
#         string = string.replace(' ', '').replace(',', '.').replace('%', '')
#         value = float(string)

#         if '%' in stringValue:
#             value /= 100
#             isPercentage = True

#         acNumber = Number(value = value)
#         acNumber.isPercentage = isPercentage
#         acNumber.stringValue = stringValue

#         return acNumber

#     @classproperty
#     def pattern() -> Pattern:
#         return Pattern('[:digits:,\.%\s]')

stark/core/types/number.py
MarkParker5/STARK
from __future__ import annotations
from typing import Any
from collections import namedtuple
from abc import ABC
import copy

from stark.general.classproperty import classproperty
from .. import Pattern


ParseResult = namedtuple('ParseResult', ['obj', 'substring'])

class ParseError(Exception):
    pass

class Object(ABC):

    value: Any
    
    @classproperty
    def pattern(cls) -> Pattern:
        return Pattern('**')

    def __init__(self, value: Any):
        '''Just init with wrapped value.'''
        self.value = value
        
    async def did_parse(self, from_string: str) -> str:
        '''
        This method is called after parsing from string and setting parameters found in pattern. 
        You will very rarely, if ever, need to call this method directly.
        
        Override this method for more complex parsing from string. 
        
        Returns:
            Minimal substring that is required to parse value.
        
        Raises:
            ParseError: if parsing failed.
        '''
        self.value = from_string
        return from_string

    @classmethod
    async def parse(cls, from_string: str, parameters: dict[str, str] | None = None) -> ParseResult:
        '''
        For internal use only.
        You will very rarely, if ever, need to override or even call this method.
        Override `def did_parse(self, from_string: str) -> str` if you need complex parsing.
        
        Raises:
            ParseError: if parsing failed.
        '''
        
        obj = cls(None)
        parameters = parameters or {}
        
        for name, object_type in cls.pattern.parameters.items():
            if not parameters.get(name):
                continue
            value = parameters.pop(name)
            setattr(obj, name, (await object_type.parse(from_string = value, parameters = parameters)).obj)
        
        substring = await obj.did_parse(from_string)
        
        return ParseResult(obj, substring)
    
    def copy(self) -> Object:
        return copy.copy(self)
    
    def __format__(self, spec) -> str:
        return f'{self.value:{spec}}'

    def __repr__(self):
        strValue = f'"{str(self.value)}"' if type(self.value) == str else str(self.value)
        return f'<{type(self).__name__} value: {strValue}>'

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, type(self)):
            raise NotImplementedError(f'Cannot compare {type(self)} with {type(other)}')
        return self.value == other.value
    
stark/core/types/object.py
MarkParker5/STARK
from .object import Object


class String(Object):
    value: str
stark/core/types/string.py
MarkParker5/STARK
# from __future__ import annotations
# from typing import Optional
# from datetime import datetime, timedelta

# from .object import Object, Pattern, classproperty
# from .time_interval import TimeInterval

# class Time(Object):
#     value: datetime

#     def __init__(self, value: Optional[datetime] = None):
#         self.value = value or datetime.now()

#     def addingInterval(self, timeinterval: Union[Object, Number]) -> Time:
#         seconds = timeinterval.value if isinstance(timeinterval, TimeInterval) else timeinterval
#         return Time(self.value + timedelta(seconds = seconds))

#     def addInterval(self, timeinterval: Union[Object, Number]):
#         seconds = timeinterval.value if isinstance(timeinterval, TimeInterval) else timeinterval
#         self.value += timedelta(seconds = seconds)

#     @classmethod
#     def parse(cls, fromString: str) -> Object:
#         raise NotImplementedError

#     @classproperty
#     def pattern() -> Pattern:
#         raise NotImplementedError

#     def __sub__(self, other: Time) -> TimeInterval:
#         return TimeInterval((self.value - other.value).total_seconds())

stark/core/types/time.py
MarkParker5/STARK
# from __future__ import annotations
# from typing import Union
# from numbers import Number

# from .object import Object, Pattern, classproperty

# class TimeInterval(Object):
#     value: float # seconds

#     @classmethod
#     def initWith(
#         cls,
#         seconds: float = 0,
#         minutes: float = 0,
#         hours: float = 0,
#         days: float = 0,
#         weeks: float = 0
#     ) -> TimeInterval:

#         days += weeks * 7
#         hours += days * 24
#         minutes += hours * 60
#         seconds += minutes * 60
#         return cls(value = seconds)

#     @classmethod
#     def parse(cls, fromString: str) -> Object:
#         raise NotImplementedError

#     @classproperty
#     def pattern() -> Pattern:
#         raise NotImplementedError

#     def __lt__(self, other: Object) -> bool:
#         value = other.value if isinstance(other, TimeInterval) else other
#         return self.value < value

#     def __le__(self, other: Object) -> bool:
#         value = other.value if isinstance(other, TimeInterval) else other
#         return self.value <= value

#     def __gt__(self, other: Object) -> bool:
#         value = other.value if isinstance(other, TimeInterval) else other
#         return self.value > value

#     def __ge__(self, other: Object) -> bool:
#         value = other.value if isinstance(other, TimeInterval) else other
#         return self.value >= value

#     def __eq__(self, other: Object) -> bool:
#         value = other.value if isinstance(other, TimeInterval) else other
#         return self.value == value

#     def __ne__(self, other: Object) -> bool:
#         value = other.value if isinstance(other, TimeInterval) else other
#         return self.value != value

#     def __add__(self, other: Union[Object, Number]) -> TimeInterval:
#         value = other.value if isinstance(other, TimeInterval) else other
#         return TimeInterval(self.value + value)

#     def __sub__(self, other: Union[Object, Number]) -> TimeInterval:
#         value = other.value if isinstance(other, TimeInterval) else other
#         return TimeInterval(self.value - value)

#     def __mul__(self, other: Union[Object, Number]) -> TimeInterval:
#         value = other.value if isinstance(other, TimeInterval) else other
#         return TimeInterval(self.value * value)

#     def __mod__(self, other: Union[Object, Number]) -> TimeInterval:
#         value = other.value if isinstance(other, TimeInterval) else other
#         return TimeInterval(self.value % value)

#     def __floordiv__(self, other: Union[Object, Number]) -> Union[Object, Number]:
#         return self.value // other.value if isinstance(other, TimeInterval) else TimeInterval(self.value // other)

#     def __truediv__(self, other: Union[Object, Number]) -> Union[Object, Number]:
#         return self.value / other.value if isinstance(other, TimeInterval) else TimeInterval(self.value / other)

stark/core/types/time_interval.py
MarkParker5/STARK
from .object import Object, Pattern, classproperty


class Word(Object):
    value: str

    @classproperty
    def pattern(cls) -> Pattern:
        return Pattern('*')

stark/core/types/word.py
MarkParker5/STARK
import time
import anyio


class BlockageDetector:
    def __init__(self, threshold = 1.0):
        self._threshold = threshold
        self._last_update = time.time()
        self._running = True

    async def monitor(self):
        self._running = True
        while self._running:
            current_time = time.time()
            if current_time - self._last_update > self._threshold:
                self.handle_blockage()
            self._last_update = current_time
            await anyio.sleep(self._threshold / 4)

    def handle_blockage(self):
        print(f'[RuntimeWarning] Main thread was blocked for more than {self._threshold}s!')

    def stop(self):
        self._running = False

stark/general/blockage_detector.py
MarkParker5/STARK
from typing import Callable, Any
from dataclasses import dataclass


@dataclass
class Dependency:
    name: str | None
    annotation: type | None
    value: Any
    
    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Dependency):
            raise TypeError(f"Cannot compare Dependency with {type(other)}")
        return self.name == other.name and self.annotation == other.annotation
    
    def __hash__(self):
        return hash((self.name, self.annotation))

class DependencyManager:
    
    dependencies: set[Dependency]
    
    def __init__(self):
        self.dependencies = set()
        
    def find(self, name: str | None, annotation: type | None) -> Dependency | None:
        for dependency in self.dependencies:
            name_match = name == dependency.name
            annotation_match = annotation == dependency.annotation
            if (not dependency.name or name_match) and annotation_match:
                return dependency
        return None
    
    def resolve(self, func: Callable) -> dict[str, Any]:
        parameters = {}
        
        annotations = {name: None for name in func.__code__.co_varnames}
        annotations.update(func.__annotations__)
        
        for name, annotation in annotations.items():
            if dependency := self.find(name, annotation):
                parameters[name] = dependency.value

        return parameters
        
    def add_dependency(self, name: str | None, annotation: type | None, value: Any):
        assert (name or annotation) and value
        self.dependencies.add(Dependency(name, annotation, value))
                
default_dependency_manager = DependencyManager()
default_dependency_manager.add_dependency(None, DependencyManager, default_dependency_manager)

stark/general/dependencies.py
MarkParker5/STARK
# from typing import _SpecialGenericAlias
import json
import inspect
from stark import Pattern, Command, CommandsManager


class StarkJsonEncoder(json.JSONEncoder):
    def default(self, obj):
        if hasattr(obj, '__json__'):
            return obj.__json__()
        
        elif isinstance(obj, Pattern):
            return {
                'origin': obj._origin
            }
            
        elif isinstance(obj, Command):
            return {
                'name': obj.name,
                'pattern': obj.pattern,
                'declaration': self.get_function_declaration(obj._runner),
                'docstring': inspect.getdoc(obj._runner) or ''
            }
            
        elif isinstance(obj, CommandsManager):
            return {
                'name': obj.name,
                'commands': obj.commands,
            }
            
        else:
            return super().default(obj)
        
    def get_function_declaration(self, func):
        get_name = lambda x: x.__name__ if hasattr(x, '__name__') else x
        
        # return inspect.getsourcelines(func)[0][:2]
        
        signature = inspect.signature(func)
        parameters = []
        
        for name, parameter in signature.parameters.items():
            
            # if issubclass(parameter.annotation, _SpecialGenericAlias): for typing.Generator and etc
            #     parameters.append(str(parameter))
            #     continue
            
            param_str = name
            
            if parameter.annotation != inspect.Parameter.empty:
                annotation = get_name(parameter.annotation)
                param_str += f': {annotation}'
            
            if parameter.default != inspect.Parameter.empty:
                default_value = parameter.default
                
                if default_value is None:
                    default_str = 'None'
                elif isinstance(default_value, str):
                    default_str = f"'{default_value}'"
                else:
                    default_str = str(default_value)
                
                param_str += f' = {default_str}'
            
            parameters.append(param_str)
            
        func_type = 'async def' if inspect.iscoroutinefunction(func) else 'def'
        name = func.__name__
        parameters_str = ', '.join(parameters)
        return_annotation = get_name(signature.return_annotation)
        
        if signature.return_annotation != inspect.Parameter.empty:
            return f'{func_type} {name}({parameters_str}) -> {return_annotation}'
        else:
            return f'{func_type} {name}({parameters_str})'


stark/general/json_encoder.py
MarkParker5/STARK
import os
from google.cloud import texttospeech
import sounddevice
import soundfile
import asyncer
from .protocols import SpeechSynthesizer, SpeechSynthesizerResult

class Speech(SpeechSynthesizerResult):

    def __init__(self, text, voice, path):
        self.text = text
        self.voice = voice
        self.path = path

    def play(self):
        try:
            sounddevice.play(*soundfile.read(self.path, dtype='float32'))
            sounddevice.wait()
        except Exception as e:
            print('\n[Error] Can`t play audio file\n', e)

    def stop(self):
        pass

class GCloudSpeechSynthesizer(SpeechSynthesizer):
    
    def __init__(self, voice_name: str, language_code: str, json_key_path: str):
        os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = json_key_path
        self._client       = texttospeech.TextToSpeechClient()
        self._audio_config = texttospeech.AudioConfig(audio_encoding = texttospeech.AudioEncoding.LINEAR16)
        self._language_code= language_code
        self._name         = voice_name
        self._voice        = texttospeech.VoiceSelectionParams(
            language_code  = self._language_code,
            name           = self._name,
            ssml_gender    = texttospeech.SsmlVoiceGender.FEMALE
        )

    async def synthesize(self, text) -> Speech:
        folder = f'audio/{self._name}'
        path = f'{folder}/{self._transliterate(text)[:100]}.wav'

        if os.path.exists(path):
            return Speech(text, self._name, path)

        synthesis_input = texttospeech.SynthesisInput(text = text)

        try:
            synthesize_speech_async = asyncer.asyncify(self._client.synthesize_speech)
            response = await synthesize_speech_async(input = synthesis_input, voice = self._voice, audio_config = self._audio_config)
            if not os.path.exists(folder): 
                os.makedirs(folder)
            with open(path, 'wb') as out:
                out.write(response.audio_content)
        except Exception as e:
            print("\n[ERROR] TTS Error: google cloud tts response error. Check Cloud Platform Console\n", e)

        return Speech(text, self._name, path)
    
    @staticmethod
    def _transliterate(name):
        dict = {'а':'a','б':'b','в':'v','г':'g','д':'d','е':'e','ё':'e',
          'ж':'zh','з':'z','и':'i','й':'i','к':'k','л':'l','м':'m','н':'n',
          'о':'o','п':'p','р':'r','с':'s','т':'t','у':'u','ф':'f','х':'h',
          'ц':'c','ч':'ch','ш':'sh','щ':'sch','ы':'y','э':'e',
          'ю':'u','я':'ja', ' ':'_'}
        allowed = 'abcdefghijklmnopqrstuvxyz'
        name = name.lower()
        for i, letter in enumerate(name):
            if letter in allowed: 
                continue;
            elif letter in dict.keys():
                name = name.replace(letter, dict[letter])
            else:
                name = name.replace(letter, '')
        return name


stark/interfaces/gcloud.py
MarkParker5/STARK
from typing import Protocol, runtime_checkable


@runtime_checkable
class SpeechRecognizerDelegate(Protocol):
    async def speech_recognizer_did_receive_final_result(self, result: str): pass
    async def speech_recognizer_did_receive_partial_result(self, result: str): pass
    async def speech_recognizer_did_receive_empty_result(self): pass
   
@runtime_checkable
class SpeechRecognizer(Protocol):
    is_recognizing: bool
    delegate: SpeechRecognizerDelegate | None
    
    async def start_listening(self): pass
    def stop_listening(self): pass
 
@runtime_checkable
class SpeechSynthesizerResult(Protocol):
    async def play(self): pass

@runtime_checkable   
class SpeechSynthesizer(Protocol):
    async def synthesize(self, text: str) -> SpeechSynthesizerResult: pass

stark/interfaces/protocols.py
MarkParker5/STARK
import os
import numpy
import torch
import sounddevice
import asyncer
from .protocols import SpeechSynthesizer, SpeechSynthesizerResult


class Speech(SpeechSynthesizerResult):

    def __init__(self, audio: numpy.ndarray, sample_rate: int):
        self.audio = audio
        self.sample_rate = sample_rate

    async def play(self):
        play_async = asyncer.asyncify(sounddevice.play)
        await play_async(self.audio, self.sample_rate, blocking = True)

    def stop(self):
        sounddevice.stop()

class SileroSpeechSynthesizer(SpeechSynthesizer):
    
    def __init__(self, model_url: str, speaker: str = 'baya', threads: int = 4, device ='cpu', torch_backends_quantized_engine: str | None = 'qnnpack'):
        if torch_backends_quantized_engine:
            torch.backends.quantized.engine = torch_backends_quantized_engine
        device = torch.device(device)
        torch.set_num_threads(threads)
        local_file = 'downloads/' + model_url.split('/')[-1]
        
        if not os.path.isdir('downloads'):
            os.mkdir('downloads')

        if not os.path.isfile(local_file):
            torch.hub.download_url_to_file(model_url, local_file)
            
        self.model = torch.package.PackageImporter(local_file).load_pickle('tts_models', 'model')
        self.model.to(device)
        self.sample_rate = 24000
        self.speaker = speaker

    async def synthesize(self, text) -> Speech:
        synthesize_async = asyncer.asyncify(self.model.apply_tts)
        audio = await synthesize_async(text = text, speaker = self.speaker, sample_rate = self.sample_rate)
        return Speech(audio, self.sample_rate)

stark/interfaces/silero.py
MarkParker5/STARK
from typing import Optional, cast
from datetime import datetime, timedelta
import os
import urllib.request
import zipfile
import anyio
import json
from queue import Queue, Empty

import sounddevice
import vosk
from pydantic import BaseModel, Field, ValidationError

from .protocols import SpeechRecognizer, SpeechRecognizerDelegate


vosk.SetLogLevel(-1)

class KaldiTranscriptionWord(BaseModel):
    word: str
    start: float
    end: float
    conf: float | None = None # only for KaldiMBR
    
class KaldiMBR(BaseModel):
    text: str
    result: list[KaldiTranscriptionWord] = Field(default_factory = list)
    spk: list[float]
    spk_frames: int
    
    @property
    def confidence(self):
        return sum(word.conf for word in self.result) / len(self.result)

class KaldiTranscription(BaseModel):
    text: str
    result: list[KaldiTranscriptionWord] = Field(default_factory = list)
    confidence: float
    
class KaldiResult(BaseModel):
    alternatives: list[KaldiTranscription]

class VoskSpeechRecognizer(SpeechRecognizer):

    _delegate: SpeechRecognizerDelegate | None = None

    audio_queue: Queue

    samplerate: int
    blocksize = 8000
    dtype = 'int16'
    channels = 1
    kaldiRecognizer: vosk.KaldiRecognizer

    last_result: Optional[str] = ''
    last_partial_result: str = ''
    last_partial_update_time: Optional[datetime] = None

    is_recognizing = True
    _is_listening = False
    
    _stored_speakers: dict[int, list[int]] = {}
    _speaker_trashold = 0.75

    def __init__(self, model_url: str, speaker_model_url: str | None = None):
        downloads = 'downloads'
        model_path = downloads + '/' + model_url.split('/')[-1].replace('.zip', '')
        zip_path = model_path + '.zip'
        speaker_model_path = downloads + '/' + speaker_model_url.split('/')[-1].replace('.zip', '') if speaker_model_url else None
        
        if not os.path.isdir('downloads'):
            os.mkdir('downloads')

        if not os.path.isdir(model_path):
            print('VOSK: Downloading model...')
            urllib.request.urlretrieve(model_url, zip_path)
            zip_file = zipfile.ZipFile(zip_path)
            zip_file.extractall(downloads)
            os.remove(zip_path)
            print('VOSK: Model downloaded!')
            
        if speaker_model_url and not os.path.isdir(cast(str, speaker_model_path)):
            print('VOSK: Downloading speaker model...')
            urllib.request.urlretrieve(speaker_model_url, zip_path)
            zip_file = zipfile.ZipFile(zip_path)
            zip_file.extractall(downloads)
            os.remove(zip_path)
            print('VOSK: Speaker model downloaded!')
        
        self.audio_queue = Queue()
        self.samplerate = int(sounddevice.query_devices(kind = 'input')['default_samplerate'])
        vosk_model = vosk.Model(model_path)
        speaker_model = vosk.SpkModel(speaker_model_path) if speaker_model_path else None
        self.kaldiRecognizer = vosk.KaldiRecognizer(vosk_model, self.samplerate)
        self.kaldiRecognizer.SetMaxAlternatives(0) # 0 (default) returns KaldiMBR; 1+ returns KaldiResult (with bad confidence implementation)
        self.kaldiRecognizer.SetWords(True) # needs to calculate MBR average confidence; (default: False)
        if speaker_model_url:
            self.kaldiRecognizer.SetSpkModel(speaker_model)
        
    @property
    def delegate(self):
        return self._delegate
    
    @delegate.setter
    def delegate(self, delegate: SpeechRecognizerDelegate | None):
        assert delegate is None or isinstance(delegate, SpeechRecognizerDelegate)
        self._delegate = delegate
        
    @property
    def sounddevice_parameters(self):
        return {
            'samplerate': self.samplerate,
            'blocksize': self.blocksize,
            'dtype': self.dtype,
            'channels': self.channels,
            'callback': self._audio_input_callback
        }

    def stop_listening(self):
        self._is_listening = False
        self.audio_queue = Queue()

    async def start_listening(self):
        if self._is_listening: return

        self.last_partial_result = ''
        self.last_partial_update_time = None
        self._is_listening = True

        with sounddevice.RawInputStream(**self.sounddevice_parameters):
            while self._is_listening:
                try:
                    if data := self.audio_queue.get(block = False):
                        await self._transcribe(data)
                except Empty:
                    pass
                await anyio.sleep(0.01)
                
    async def _transcribe(self, data):
        delegate = self.delegate
        if not delegate: return
        
        if self.kaldiRecognizer.AcceptWaveform(data):
            self.last_partial_update_time = None
            raw_json = self.kaldiRecognizer.Result()
            text: str | None = None
            
            try:
                result = KaldiMBR.parse_raw(raw_json)
                text = result.text
                # print('\nConfidence:', result.confidence)
            except ValidationError:
                try:
                    result = KaldiResult.parse_raw(raw_json)
                    transcription = result.alternatives[0]
                    text = transcription.text
                except ValidationError:
                    text = json.loads(raw_json).get('text')
            
            if text:
                # if result.spk:
                #     speaker, similarity = self._get_speaker(result.spk)
                #     print(f'\nSpeaker: {speaker} ({similarity * 100:.2f}%)\n')
                
                self.last_result = text
                await delegate.speech_recognizer_did_receive_final_result(text)
            else:
                self.last_result = None
                await delegate.speech_recognizer_did_receive_empty_result()
                
        else:
            result = json.loads(self.kaldiRecognizer.PartialResult())
            # partial always returns {"partial": "..."}
            if (string := result.get('partial')) and string != self.last_partial_result:
                self.last_partial_result = string
                self.last_partial_update_time = datetime.now()
                await delegate.speech_recognizer_did_receive_partial_result(string)

    def _audio_input_callback(self, indata, frames, time, status):
        if not self.is_recognizing: return
        self.audio_queue.put(bytes(indata))
        
    def _get_speaker(self, vector: list[int]) -> tuple[int, float]:
        # TODO: search Is not working good, need to improve the algorithm
        
        matched_speaker_id = None
        best_similarity = -1.0
        
        for speaker_id, vec in self._stored_speakers.items():
            similarity = self._cosine_similarity(vector, vec)
            if similarity > best_similarity:
                best_similarity = similarity
                matched_speaker_id = speaker_id
                
        if not best_similarity or best_similarity < self._speaker_trashold:
            matched_speaker_id = len(self._stored_speakers)
            self._stored_speakers[matched_speaker_id] = vector
            # print(f'New speaker: {matched_speaker_id}, similarity: {best_similarity * 100:.2f}%')
            best_similarity = 1
            
        return cast(int, matched_speaker_id), best_similarity
        
    def _cosine_similarity(self, vector_a, vector_b) -> float:
        dot_product = sum(a*b for a, b in zip(vector_a, vector_b))
        norm_a = sum(a*a for a in vector_a) ** 0.5
        norm_b = sum(b*b for b in vector_b) ** 0.5
        return dot_product / (norm_a * norm_b)

stark/interfaces/vosk.py
MarkParker5/STARK
from .mode import Mode
from .voice_assistant import VoiceAssistant
stark/voice_assistant/__init__.py
MarkParker5/STARK
from __future__ import annotations
from typing import Optional, Callable
from pydantic import BaseModel
from ..general.classproperty import classproperty


class Mode(BaseModel):
    
    play_responses: bool = True
    collect_responses: bool = False
    explicit_interaction_pattern: Optional[str] = None
    timeout_after_interaction: int = 20 # seconds
    timeout_before_repeat: int = 5 # seconds
    mode_on_timeout: Callable[[], Mode] | None = None
    mode_on_interaction: Callable[[], Mode] | None = None
    stop_after_interaction: bool = False
    
    @classproperty
    def active(cls) -> Mode:
        return Mode(
            mode_on_timeout = lambda: Mode.waiting,
        )
    
    @classproperty
    def waiting(cls) -> Mode:
        return Mode(
            collect_responses = True,
            mode_on_interaction = lambda: Mode.active,
        )
    
    @classproperty
    def inactive(cls) -> Mode:
        return Mode(
            play_responses = False,
            collect_responses = True,
            timeout_after_interaction = 0, # start collecting responses immediately
            timeout_before_repeat = 0, # repeat all
            mode_on_interaction = lambda: Mode.active,
        )
    
    @classmethod
    def sleeping(cls, pattern: str) -> Mode:
        return Mode(
            play_responses = False,
            collect_responses = True,
            timeout_after_interaction = 0, # start collecting responses immediately
            timeout_before_repeat = 0, # repeat all
            explicit_interaction_pattern = pattern,
            mode_on_interaction = lambda: Mode.active,
        )
    
    @classmethod
    def explicit(cls, pattern: str) -> Mode:
        return Mode(
            explicit_interaction_pattern = pattern,
        )
        
    @classmethod
    def external(cls) -> Mode:
        return Mode(
            stop_after_interaction = True,
        )

stark/voice_assistant/mode.py
MarkParker5/STARK
from typing import cast
from datetime import datetime

from ..core import (
    CommandsContext,
    CommandsContextLayer,
    CommandsContextDelegate,
    Response,
    ResponseStatus,
    Pattern
)
from ..interfaces.protocols import SpeechRecognizer, SpeechRecognizerDelegate, SpeechSynthesizer
from .mode import Mode


class ResponseCache(Response):
    # needs to save timeout_before_repeat that was set by mode at the moment of saving because it can change later and may lead to skipping responses
    timeout_before_repeat: int = 0

class VoiceAssistant(SpeechRecognizerDelegate, CommandsContextDelegate):

    speech_recognizer: SpeechRecognizer
    speech_synthesizer: SpeechSynthesizer
    commands_context: CommandsContext
    
    mode: Mode
    ignore_responses: list[ResponseStatus] # TODO: to Mode
    
    _responses: list[ResponseCache]
    _last_interaction_time: datetime

    def __init__(self, speech_recognizer: SpeechRecognizer, speech_synthesizer: SpeechSynthesizer, commands_context: CommandsContext):
        assert isinstance(speech_recognizer, SpeechRecognizer)
        assert isinstance(speech_synthesizer, SpeechSynthesizer)
        assert isinstance(commands_context, CommandsContext)
        
        self.speech_recognizer = speech_recognizer
        self.speech_synthesizer = speech_synthesizer
        self.commands_context = commands_context
        commands_context.delegate = self
        
        self.mode = Mode.active
        self.ignore_responses = []
        self._responses = []
        self._last_interaction_time = datetime.now()

    @property
    def timeout_reached(self):
        return (datetime.now() - self._last_interaction_time).total_seconds() >= self.mode.timeout_after_interaction

    # SpeechRecognizerDelegate

    async def speech_recognizer_did_receive_final_result(self, result: str):
        print(f'\nYou: {result}')
        # check explicit interaction if needed
        if pattern_str := self.mode.explicit_interaction_pattern:
            if not await Pattern(pattern_str).match(result):
                return
        
        # reset context if timeout reached
        if self.timeout_reached:
            self.commands_context.pop_to_root_context()
        
        self._last_interaction_time = datetime.now()
        
        # switch mode if needed
        if self.mode.mode_on_interaction:
            self.mode = self.mode.mode_on_interaction()
        
        await self.commands_context.process_string(result)

    async def speech_recognizer_did_receive_partial_result(self, result: str):
        print(f'\rYou: \x1B[3m{result}\x1B[0m', end = '')

    async def speech_recognizer_did_receive_empty_result(self):
        pass

    # CommandsContextDelegate

    async def commands_context_did_receive_response(self, response: Response):

        if response.status in self.ignore_responses:
            return
        
        timeout_before_repeat = self.mode.timeout_before_repeat
        
        if self.timeout_reached and self.mode.mode_on_timeout: 
            self.mode = self.mode.mode_on_timeout()
        
        if self.timeout_reached and self.mode.collect_responses:
            self._responses.append(ResponseCache(**response.dict(), timeout_before_repeat = timeout_before_repeat))
        
        # play response if needed
            
        if not self.mode.play_responses:
            return
        
        await self._play_response(response)
        
        if self.timeout_reached:
            return
        
        # repeat responses if interaction was recently
        
        while self._responses:
            response = self._responses.pop(0)
            if (datetime.now() - response.time).total_seconds() <= response.timeout_before_repeat:
                self._responses.insert(0, response)
                continue
            
            await self._play_response(response)
            self.commands_context.add_context(CommandsContextLayer(response.commands, response.parameters))
            
            if response.needs_user_input:
                break
        else:
            if self.mode.stop_after_interaction:
                self.speech_recognizer.stop_listening()
    
    def remove_response(self, response: Response):
        if response in self._responses:
            self._responses.remove(cast(ResponseCache, response))
        
    # Private
    
    async def _play_response(self, response: Response):
        self.commands_context.last_response = response
        
        if response.text:
            print(f'S.T.A.R.K.: {response.text}')
        
        if response.voice:
            was_recognizing = self.speech_recognizer.is_recognizing
            self.speech_recognizer.is_recognizing = False
            speech = await self.speech_synthesizer.synthesize(response.voice)
            await speech.play()
            self.speech_recognizer.is_recognizing = was_recognizing
stark/voice_assistant/voice_assistant.py
MarkParker5/STARK
from typing import AsyncGenerator
import time
import contextlib
import pytest
import asyncer
import anyio
from stark.general.dependencies import DependencyManager
from stark.core import (
    CommandsManager,
    CommandsContext,
    CommandsContextDelegate, 
    Response,
    ResponseHandler,
    AsyncResponseHandler
)
from stark.core.types import Word
from stark.interfaces.protocols import SpeechRecognizerDelegate
from stark.voice_assistant import VoiceAssistant


class CommandsContextDelegateMock(CommandsContextDelegate):
    
    responses: list[Response]
    
    def __init__(self):
        self.responses = []
    
    async def commands_context_did_receive_response(self, response: Response):
        self.responses.append(response)
        
    async def remove_response(self, response: Response):
        self.responses.remove(response)
        
class SpeechRecognizerMock:
    is_recognizing: bool = False
    delegate: SpeechRecognizerDelegate | None = None
    
    async def start_listening(self): pass
    def stop_listening(self): pass
        
class SpeechSynthesizerResultMock:
    async def play(self): pass
    
    def __init__(self, text: str):
        self.text = text
    
class SpeechSynthesizerMock:
    
    def __init__(self):
        self.results = []
    
    async def synthesize(self, text: str) -> SpeechSynthesizerResultMock:
        result = SpeechSynthesizerResultMock(text)
        self.results.append(result)
        return result

@pytest.fixture
async def commands_context_flow():
    @contextlib.asynccontextmanager
    async def _commands_context_flow() -> AsyncGenerator[tuple[CommandsManager, CommandsContext, CommandsContextDelegateMock], None]:
        async with asyncer.create_task_group() as main_task_group:
            dependencies = DependencyManager()
            manager = CommandsManager()
            context = CommandsContext(main_task_group, manager, dependencies)
            context_delegate = CommandsContextDelegateMock()
            context.delegate = context_delegate
            
            assert len(context_delegate.responses) == 0
            assert len(context._context_queue) == 1
            
            main_task_group.soonify(context.handle_responses)()
            yield (manager, context, context_delegate)
            context.stop()
    return _commands_context_flow

@pytest.fixture
async def commands_context_flow_filled(commands_context_flow):
    @contextlib.asynccontextmanager
    async def _commands_context_flow_filled() -> AsyncGenerator[tuple[CommandsContext, CommandsContextDelegateMock], None]:
        async with commands_context_flow() as (manager, context, context_delegate):
    
            @manager.new('test')
            def test(): 
                text = voice = 'test'
                return Response(text = text, voice = voice)
            
            @manager.new('lorem * dolor')
            def lorem(): 
                return Response(text = 'Lorem!', voice = 'Lorem!')
            
            @manager.new('hello', hidden = True)
            def hello_context(**params):
                voice = text = f'Hi, {params["name"]}!' 
                return Response(text = text, voice = voice)
            
            @manager.new('bye', hidden = True)
            def bye_context(name: Word, handler: ResponseHandler):
                handler.pop_context()
                return Response(
                    text = f'Bye, {name}!'
                ) 
            
            @manager.new('hello $name:Word')
            def hello(name: Word):
                text = voice = f'Hello, {name}!' 
                return Response(
                    text = text,
                    voice = voice,
                    commands = [hello_context, bye_context],
                    parameters = {'name': name}
                )
                
            @manager.new('repeat')
            def repeat():
                return Response.repeat_last
            
            # background commands
    
            @manager.new('background min')
            async def background(handler: AsyncResponseHandler):
                text = voice = 'Starting background task'
                await handler.respond(Response(text = text, voice = voice))
                await anyio.sleep(1)
                text = voice = 'Finished background task'
                return Response(text = text, voice = voice)
            
            @manager.new('background needs input')
            async def background_needs_input(handler: AsyncResponseHandler):
                await anyio.sleep(1)
                
                for text in ['First response', 'Second response', 'Third response']:
                    await handler.respond(Response(text = text, voice = text))
                    
                text = 'Needs input'
                await handler.respond(Response(text = text, voice = text, needs_user_input = True))
                
                for text in ['Fourth response', 'Fifth response', 'Sixth response']:
                    await handler.respond(Response(text = text, voice = text))
                
                text = voice = 'Finished long background task'
                return Response(text = text, voice = voice)
            
            @manager.new('background with context')
            async def background_multiple_contexts(handler: AsyncResponseHandler):
                await anyio.sleep(1)
                text = voice = 'Finished long background task'
                return Response(text = text, voice = voice, commands = [hello_context, bye_context], parameters = {'name': 'John'})
            
            @manager.new('background remove response')
            async def background_remove_response(handler: AsyncResponseHandler):
                response = Response(text = 'Deleted response', voice = 'Deleted response')
                await handler.respond(response)
                await anyio.sleep(1)
                await handler.unrespond(response)
                return None
    
            yield (context, context_delegate)
            
    return _commands_context_flow_filled

@pytest.fixture
async def voice_assistant(commands_context_flow_filled):
    @contextlib.asynccontextmanager
    async def _voice_assistant() -> AsyncGenerator[VoiceAssistant, None]:
        async with commands_context_flow_filled() as (context, context_delegate):
            voice_assistant = VoiceAssistant(
                speech_recognizer = SpeechRecognizerMock(),
                speech_synthesizer = SpeechSynthesizerMock(),
                commands_context = context
            )
            yield voice_assistant
    return _voice_assistant

tests/conftest.py
MarkParker5/STARK
from typing import AsyncGenerator
import pytest
from stark.core import (
    CommandsManager,
    Response,
    ResponseHandler,
    AsyncResponseHandler,
    ResponseStatus
)


async def test_create_run_async_command():
    manager = CommandsManager()
    
    @manager.new('foo')
    async def foo() -> Response: 
        return Response(text = 'foo!')
    
    assert (await foo()).text == 'foo!'

async def test_call_async_command_from_async_command():
    manager = CommandsManager()
    
    @manager.new('foo')
    async def foo() -> Response: 
        return Response(text = 'foo!')
    
    @manager.new('bar')
    async def bar() -> Response: 
        return await foo()
    
    assert (await bar()).text == 'foo!'
    
async def test_async_command_with_async_response_handler():
    mamnager = CommandsManager()
    
    @mamnager.new('foo')
    async def foo(handler: AsyncResponseHandler): 
        ...

async def test_async_command_with_sync_response_handler():
    manager = CommandsManager()
    
    with pytest.raises(TypeError, match = '`ResponseHandler` is not compatible with command .* because it is async, use `AsyncResponseHandler` instead'):
        @manager.new('foo')
        async def foo(handler: ResponseHandler): 
            ...

async def test_async_command_generator_yielding_response():
    manager = CommandsManager()
    
    @manager.new('foo')
    async def foo() -> AsyncGenerator[Response, None]:
        yield Response(text = 'foo!')
    
    i = 0
    async for response in await foo():
        assert response.text == 'foo!'
        i += 1
    assert i == 1

async def test_async_command_generator_multiple_yielding_response():
    manager = CommandsManager()
    
    @manager.new('foo')
    async def foo() -> AsyncGenerator[Response, None]: 
        yield Response(text = 'foo!')
        yield Response(text = 'bar!')
        yield Response(text = 'baz!')
    
    expected = ['foo!', 'bar!', 'baz!']
    
    async for response in await foo():
        assert response.text == expected.pop(0)

    assert not expected

async def test_exception_in_command():
    manager = CommandsManager()
    
    @manager.new('foo')
    async def foo() -> Response: 
        raise Exception('foo exception')
    
    # with pytest.raises(Exception, match = 'foo exception'):
    #     await foo()
    
    assert (await foo()).status == ResponseStatus.error

tests/test_commands/test_async_command.py
MarkParker5/STARK
import re
import pytest
from stark.core import CommandsManager
from stark.core.types import Word


def test_new():
    manager = CommandsManager()
    manager.new('test')(lambda: None)
    assert len(manager.commands) == 1
    
    @manager.new('foo bar')
    def foo_bar(): pass
    
    assert len(manager.commands) == 2
    assert manager.commands[1].name == 'CommandsManager.foo_bar'
    assert manager.commands[1].pattern._origin == 'foo bar'
    
def test_new_with_extra_parameters_in_pattern():
    manager = CommandsManager()
    
    with pytest.raises(AssertionError, match = re.escape('Command CommandsManager.test must have all parameters from pattern:')):
        @manager.new('test $name:Word, $secondName:Word')
        def test(name: Word): pass
    
async def test_search():
    manager = CommandsManager()
    
    @manager.new('test')
    def test(): pass
    
    @manager.new('hello $name:Word $surname:Word')
    def hello2(name: Word, surname: Word): pass
    
    @manager.new('hello $name:Word')
    def hello(name: Word): pass
    
    # test
    result = await manager.search('test')
    assert result is not None
    assert len(result) == 1
    assert result[0].command.name == 'CommandsManager.test'
    
    # hello
    result = await manager.search('hello world')
    assert result is not None
    assert len(result) == 1
    assert result[0].command.name == 'CommandsManager.hello'
    assert result[0].match_result.substring == 'hello world'
    assert type(result[0].match_result.parameters['name']) is Word
    assert result[0].match_result.parameters['name'].value == 'world'
    
    # hello2
    result = await manager.search('hello new world')
    assert result is not None
    assert len(result) == 1
    assert result[0].command == hello2
    assert result[0].match_result.substring == 'hello new world'
    assert result[0].match_result.parameters == {'name': Word('new'), 'surname': Word('world')}
            
def test_extend_manager():
    root_manager = CommandsManager()
    child_manager = CommandsManager('Child')
    
    @root_manager.new('test')
    def test(): pass
    
    @child_manager.new('test')
    def test(): pass
    
    assert len(child_manager.commands) == 1
    assert len(root_manager.commands) == 1
    assert child_manager.commands[0].name == 'Child.test'
    assert root_manager.commands[0].name == 'CommandsManager.test'
    
    root_manager.extend(child_manager)
    
    assert len(child_manager.commands) == 1
    assert len(root_manager.commands) == 2
    assert child_manager.commands[0].name == 'Child.test'
    assert root_manager.commands[0].name == 'CommandsManager.test'
    assert root_manager.commands[1].name == 'Child.test'
    
def test_manager_get_command_by_name():
    manager = CommandsManager('TestManager')
    child = CommandsManager('Child')
    
    @manager.new('')
    def test(): ...
    
    @manager.new('')
    def test2(): ...
    
    @manager.new('')
    def test3(): ...
    
    @manager.new('')
    def test4(): ...
    
    @child.new('')
    def test5(): ...
    
    manager.extend(child)
    
    assert manager.get_by_name('test2') == test2
    assert manager.get_by_name('TestManager.test3') == test3
    assert manager.get_by_name('test5') == None
    assert manager.get_by_name('Child.test5') == test5

tests/test_commands/test_commands_manager.py
MarkParker5/STARK
from typing import AsyncGenerator, Type, Any, Callable
import pytest
from stark.core import (
    CommandsManager,
    Response,
    ResponseHandler,
    AsyncResponseHandler,
    ResponseStatus
)
from stark.core.types import Word
from stark.general.json_encoder import StarkJsonEncoder
import json


async def test_command_json():
    manager = CommandsManager('TestManager')
    
    @manager.new('test pattern $word:Word')
    def test(var: str, word: Word, foo: int | None = None) -> Response:
        '''test command'''
        return Response(text=var)
    
    string = json.dumps(test, cls = StarkJsonEncoder)
    parsed = json.loads(string)
    
    assert parsed['name'] == 'TestManager.test'
    assert parsed['pattern']['origin'] == r'test pattern $word:Word'
    assert parsed['declaration'] == 'def test(var: str, word: Word, foo: int | None = None) -> Response'
    assert parsed['docstring'] == 'test command'
    
async def test_async_command_complicate_type_json():
    manager = CommandsManager('TestManager')
    
    @manager.new('async test')
    async def test2(
        some: AsyncGenerator[
            Callable[
                [Any], Type
            ], 
            list[None]
        ]
    ):
        return Response()
    
    string = json.dumps(test2, cls = StarkJsonEncoder)
    parsed = json.loads(string)
    
    assert parsed['name'] == 'TestManager.test2'
    assert parsed['pattern']['origin'] == r'async test'
    assert parsed['declaration'] == 'async def test2(some: AsyncGenerator)' # TODO: improve AsyncGenerator to full type
    # assert parsed['declaration'] == 'async def test2(some: AsyncGenerator[Callable[[Any], Type], list[None], None])'
    assert parsed['docstring'] == ''
    
def test_manager_json():
    
    manager = CommandsManager('TestManager')
    
    @manager.new('')
    def test(): ...
    
    @manager.new('')
    def test2(): ...
    
    @manager.new('')
    def test3(): ...
    
    @manager.new('')
    def test4(): ...
    
    string = json.dumps(manager, cls = StarkJsonEncoder)
    parsed = json.loads(string)

    assert parsed['name'] == 'TestManager'
    assert {c['name'] for c in parsed['commands']} == {'TestManager.test', 'TestManager.test2', 'TestManager.test3', 'TestManager.test4',}
    
tests/test_commands/test_commands_to_json.py
MarkParker5/STARK
from typing import Generator
import warnings
import pytest
from asyncer import syncify
from stark.core import (
    CommandsManager,
    Response,
    ResponseHandler,
    AsyncResponseHandler,
    ResponseStatus
)


async def test_create_await_run_sync_command():
    manager = CommandsManager()

    @manager.new('foo')
    def foo() -> Response: 
        return Response(text = 'foo!')

    # commands are always async, no matter runner function is sync or async
    # so we need to await or syncify them
    assert (await foo()).text == 'foo!'
    
def test_create_syncify_run_sync_command():
    manager = CommandsManager()

    @manager.new('foo')
    def foo() -> Response: 
        return Response(text = 'foo!')

    # commands are always async, no matter runner function is sync or async
    # so we need to await or syncify them
    # raise_sync_error needs for tests only because test function doesn't run in worker thread
    sync_foo = syncify(foo, raise_sync_error = False)
    assert sync_foo().text == 'foo!'

async def test_call_sync_command_from_sync_command_await():
    manager = CommandsManager()
    
    @manager.new('foo')
    def foo() -> Response: 
        return Response(text = 'foo!')
    
    @manager.new('bar')
    def bar() -> Response: 
        # commands are always async, so we need to syncify async foo (or delcare current function as async and await foo, check /choice-runner-type)
        sync_foo = syncify(foo)
        return sync_foo()
    
    assert (await bar()).text == 'foo!'

def test_call_sync_command_from_sync_command_syncify():
    manager = CommandsManager()
    
    @manager.new('foo')
    def foo() -> Response: 
        return Response(text = 'foo!')
    
    @manager.new('bar')
    def bar() -> Response: 
        # commands are always async, so we need to syncify async foo (or delcare current function as async and await foo, check /choice-runner-type)
        sync_foo = syncify(foo)
        return sync_foo()
    
    # raise_sync_error needs for tests only because test function doesn't run in worker thread
    sync_bar = syncify(bar, raise_sync_error = False)
    assert sync_bar().text == 'foo!'

def test_sync_command_with_sync_response_handler():
    manager = CommandsManager()
    
    @manager.new('foo')
    def foo(handler: ResponseHandler): 
        ...
    
def test_sync_command_with_async_response_handler():
    manager = CommandsManager()
    
    with pytest.raises(TypeError, match = '`AsyncResponseHandler` is not compatible with command .* because it is sync, use `ResponseHandler` instead'):
        @manager.new('foo')
        def foo(handler: AsyncResponseHandler): 
            ...
            
async def test_sync_command_generator_yielding_response():
    manager = CommandsManager()
    
    with warnings.catch_warnings(record = True) as warnings_list:
        @manager.new('foo')
        def foo() -> Generator[Response, None, None]: 
            yield Response(text = 'foo!')
        
        assert next(await foo()).text == 'foo!'
        
        assert len(warnings_list) == 1
        assert issubclass(warnings_list[0].category, UserWarning)
        assert 'GeneratorType that is not fully supported and may block the main thread' in str(warnings_list[0].message)
    
async def test_sync_command_generator_multiple_yielding_response():
    manager = CommandsManager()
    
    with warnings.catch_warnings(record = True) as warnings_list:
        @manager.new('foo')
        def foo() -> Generator[Response, None, None]: 
            yield Response(text = 'foo!')
            yield Response(text = 'bar!')
            yield Response(text = 'baz!')
        
        expected = ['foo!', 'bar!', 'baz!']
        for response in await foo(): # TODO: make generator async generator and use async for
            assert response.text == expected.pop(0)
            
        assert len(warnings_list) == 1
        assert issubclass(warnings_list[0].category, UserWarning)
        assert 'GeneratorType that is not fully supported and may block the main thread' in str(warnings_list[0].message)

async def test_exception_in_command():
    manager = CommandsManager()
    
    @manager.new('foo')
    def foo() -> Response: 
        raise Exception('foo exception')
    
    # with pytest.raises(Exception, match = 'foo exception'):
    #     await foo()
    
    assert (await foo()).status == ResponseStatus.error

tests/test_commands/test_sync_command.py
MarkParker5/STARK
import pytest
import anyio


async def test_basic_search(commands_context_flow_filled, autojump_clock):
    async with commands_context_flow_filled() as (context, context_delegate):
        assert len(context_delegate.responses) == 0
        assert len(context._context_queue) == 1
        
        await context.process_string('lorem ipsum dolor')
        await anyio.sleep(5)
        assert len(context_delegate.responses) == 1
        assert context_delegate.responses[0].text == 'Lorem!'
        assert len(context._context_queue) == 1

async def test_second_context_layer(commands_context_flow_filled, autojump_clock):
    async with commands_context_flow_filled() as (context, context_delegate):
    
        await context.process_string('hello world')
        await anyio.sleep(5)
        assert len(context_delegate.responses) == 1
        assert context_delegate.responses[0].text == 'Hello, world!'
        assert len(context._context_queue) == 2
        context_delegate.responses.clear()
        
        await context.process_string('hello')
        await anyio.sleep(5)
        assert len(context_delegate.responses) == 1
        assert context_delegate.responses[0].text == 'Hi, world!'
        assert len(context._context_queue) == 2
        context_delegate.responses.clear()
    
async def test_context_pop_on_not_found(commands_context_flow_filled, autojump_clock):
    async with commands_context_flow_filled() as (context, context_delegate):
    
        await context.process_string('hello world')
        await anyio.sleep(5)
        assert len(context._context_queue) == 2
        assert len(context_delegate.responses) == 1
        context_delegate.responses.clear()
        
        await context.process_string('lorem ipsum dolor')
        await anyio.sleep(5)
        assert len(context._context_queue) == 1
        assert len(context_delegate.responses) == 1

async def test_context_pop_context_response_action(commands_context_flow_filled, autojump_clock):
    async with commands_context_flow_filled() as (context, context_delegate):
    
        await context.process_string('hello world')
        await anyio.sleep(5)
        assert len(context_delegate.responses) == 1
        assert context_delegate.responses[0].text == 'Hello, world!'
        assert len(context._context_queue) == 2
        context_delegate.responses.clear()
        
        await context.process_string('bye')
        await anyio.sleep(5)
        assert len(context_delegate.responses) == 1
        assert context_delegate.responses[0].text == 'Bye, world!'
        assert len(context._context_queue) == 1
        context_delegate.responses.clear()
        
        await context.process_string('hello')
        await anyio.sleep(5)
        assert len(context_delegate.responses) == 0
   
async def test_repeat_last_answer_response_action(commands_context_flow_filled, autojump_clock):
    async with commands_context_flow_filled() as (context, context_delegate):
    
        await context.process_string('hello world')
        await anyio.sleep(5)
        assert len(context_delegate.responses) == 1
        assert context_delegate.responses[0].text == 'Hello, world!'
        context_delegate.responses.clear()
        assert len(context_delegate.responses) == 0
        
        await context.process_string('repeat')
        await anyio.sleep(5)
        assert len(context_delegate.responses) == 1
        assert context_delegate.responses[0].text == 'Hello, world!'

tests/test_commands_flow/test_commands_context.py
MarkParker5/STARK
import warnings
import pytest
import anyio
from stark.core import Response


async def test_commands_context_handle_async_generator(commands_context_flow, autojump_clock):
    async with commands_context_flow() as (manager, context, context_delegate):
        
        @manager.new('foo')
        async def foo(): 
            await anyio.sleep(2)
            yield Response(text = 'foo0')
            await anyio.sleep(2)
            yield Response(text = 'foo1')
            await anyio.sleep(2)
            yield Response(text = 'foo2')
            await anyio.sleep(2)
            yield Response(text = 'foo3')
            await anyio.sleep(2)
            yield Response(text = 'foo4')
            # return is not allowed in generators (functions with yield)
        
        await context.process_string('foo')
        
        last_count = 0
        while last_count < 5:
            await anyio.sleep(1)
            if last_count == len(context_delegate.responses): continue
            assert len(context_delegate.responses) == last_count + 1
            last_count += 1
            assert context_delegate.responses[last_count - 1].text == f'foo{last_count - 1}'
            
        assert len(context_delegate.responses) == 5
        assert [r.text for r in context_delegate.responses] == [f'foo{i}' for i in range(5)]

async def test_commands_context_handle_sync_generator(commands_context_flow, autojump_clock):
    async with commands_context_flow() as (manager, context, context_delegate):
        
        @manager.new('foo')
        def foo():
            yield Response(text = 'foo0')
            yield Response(text = 'foo1')
            yield Response(text = 'foo2')
            yield Response(text = 'foo3')
            yield Response(text = 'foo4')
            # return is not allowed in generators (functions with yield)
        
        with warnings.catch_warnings(record = True) as warnings_list:
            await context.process_string('foo')
            await anyio.sleep(1)
        
        assert len(warnings_list) == 2
        for warning in warnings_list:
            assert issubclass(warning.category, UserWarning)
            assert 'GeneratorType that is not fully supported and may block the main thread' in str(warning.message)
            
        assert len(context_delegate.responses) == 5
        assert [r.text for r in context_delegate.responses] == [f'foo{i}' for i in range(5)]

tests/test_commands_flow/test_commands_context_generators_handling.py
MarkParker5/STARK
import anyio
from stark.core import AsyncResponseHandler, Response


async def test_commands_context_inject_dependencies(commands_context_flow, autojump_clock):
    async with commands_context_flow() as (manager, context, context_delegate):
        @manager.new('foo')
        async def foo(handler: AsyncResponseHandler) -> Response: 
            return Response(text = 'foo!')
            
        @manager.new('bar')
        async def bar(inject_dependencies): 
            return await inject_dependencies(foo)()
            
        await context.process_string('bar')
        await anyio.sleep(1)
        
        assert len(context_delegate.responses) == 1
        assert context_delegate.responses[0].text == 'foo!'

tests/test_commands_flow/test_commands_context_inject_dependencies.py
MarkParker5/STARK
import pytest
import warnings
import anyio
from stark.core import Response, ResponseHandler, AsyncResponseHandler


async def test_command_return_response(commands_context_flow, autojump_clock):
    async with commands_context_flow() as (manager, context, context_delegate):
    
        @manager.new('foo')
        async def foo() -> Response: 
            return Response(text = 'foo!')
        
        await context.process_string('foo')
        await anyio.sleep(5)

        assert len(context_delegate.responses) == 1
        assert context_delegate.responses[0].text == 'foo!'

async def test_sync_command_call_sync_respond(commands_context_flow, autojump_clock):
    async with commands_context_flow() as (manager, context, context_delegate):
    
        @manager.new('foo')
        def foo(handler: ResponseHandler): 
            handler.respond(Response(text = 'foo!'))
        
        await context.process_string('foo')
        await anyio.sleep(5)

        assert len(context_delegate.responses) == 1
        assert context_delegate.responses[0].text == 'foo!'

async def test_async_command_call_sync_respond(commands_context_flow, autojump_clock):
    async with commands_context_flow() as (manager, context, context_delegate):
        
        @manager.new('foo')
        async def foo(handler: AsyncResponseHandler): 
            await handler.respond(Response(text = 'foo!'))
        
        await context.process_string('foo')
        await anyio.sleep(5)

        assert len(context_delegate.responses) == 1
        assert context_delegate.responses[0].text == 'foo!'

@pytest.mark.skip(reason = 'deprecated: added checks for DI on command creation')
async def test_sync_command_call_async_respond(commands_context_flow, autojump_clock):
    async with commands_context_flow() as (manager, context, context_delegate):
        
        @manager.new('foo')
        def foo(handler: AsyncResponseHandler): 
            with warnings.catch_warnings(record = True) as warnings_list:
                assert len(warnings_list) == 0
                handler.respond(Response(text = 'foo!')) # type: ignore
                assert len(warnings_list) == 1
                assert issubclass(warnings_list[0].category, RuntimeWarning)
                assert 'was never awaited' in str(warnings_list[0].message)
        
        await context.process_string('foo')
        await anyio.sleep(5)

        assert len(context_delegate.responses) == 0

@pytest.mark.skip(reason = 'deprecated: added checks for DI on command creation')
async def test_async_command_call_async_respond(commands_context_flow, autojump_clock):
    async with commands_context_flow() as (manager, context, context_delegate):
        
        @manager.new('foo')
        async def foo(handler: ResponseHandler):
            with pytest.raises(RuntimeError, match = 'can only be run from an AnyIO worker thread'):
                handler.respond(Response(text = 'foo!'))
        
        await context.process_string('foo')
        await anyio.sleep(5)

        assert len(context_delegate.responses) == 0

async def test_command_multiple_respond(commands_context_flow, autojump_clock):
    async with commands_context_flow() as (manager, context, context_delegate):
        
        @manager.new('foo')
        async def foo(handler: AsyncResponseHandler): 
            await anyio.sleep(2)
            await handler.respond(Response(text = 'foo0'))
            await anyio.sleep(2)
            await handler.respond(Response(text = 'foo1'))
            await anyio.sleep(2)
            await handler.respond(Response(text = 'foo2'))
            await anyio.sleep(2)
            await handler.respond(Response(text = 'foo3'))
            await anyio.sleep(2)
            return Response(text = 'foo4')
        
        await context.process_string('foo')
        
        last_count = 0
        while last_count < 5:
            await anyio.sleep(1)
            if last_count == len(context_delegate.responses): continue
            assert len(context_delegate.responses) == last_count + 1
            last_count += 1
            assert context_delegate.responses[last_count - 1].text == f'foo{last_count - 1}'
            
        assert len(context_delegate.responses) == 5
        assert [r.text for r in context_delegate.responses] == [f'foo{i}' for i in range(5)]

tests/test_commands_flow/test_commands_context_respond.py
MarkParker5/STARK
import pytest
import anyio
from stark.core import Pattern, Response, CommandsManager
from stark.core.types import Object
from stark.general.classproperty import classproperty
import random


async def test_multiple_commands(commands_context_flow, autojump_clock):
    async with commands_context_flow() as (manager, context, context_delegate):
    
        @manager.new('foo bar')
        def foobar(): 
            return Response(text = 'foo!')
        
        @manager.new('lorem * dolor')
        def lorem(): 
            return Response(text = 'lorem!')
        
        await context.process_string('foo bar lorem ipsum dolor')
        await anyio.sleep(5)

        assert len(context_delegate.responses) == 2
        assert {context_delegate.responses[0].text, context_delegate.responses[1].text} == {'foo!', 'lorem!'}

async def test_repeating_command(commands_context_flow, autojump_clock):
    async with commands_context_flow() as (manager, context, context_delegate):
    
        @manager.new('lorem * dolor')
        def lorem(): 
            return Response(text = 'lorem!')
        
        await context.process_string('lorem pisum dolor lorem ipsutest_repeating_commanduum dolor sit amet')
        await anyio.sleep(5)
        
        assert len(context_delegate.responses) == 2
        assert context_delegate.responses[0].text == 'lorem!'
        assert context_delegate.responses[1].text == 'lorem!'

async def test_overlapping_commands_less_priority_cut(commands_context_flow, autojump_clock):
    manager = CommandsManager()

    @manager.new('foo bar *')
    def foobar(): 
        return Response(text = 'foo!')
    
    @manager.new('* baz')
    def baz(): 
        return Response(text = 'baz!')
    
    result = await manager.search('foo bar test baz')
    assert len(result) == 2
    assert result[0].match_result.substring == 'foo bar test'
    assert result[1].match_result.substring == 'baz'

async def test_overlapping_commands_priority_cut(commands_context_flow, autojump_clock):
    manager = CommandsManager()
    
    @manager.new('foo bar *')
    def foobar(): 
        return Response(text = 'foo!')
    
    @manager.new('*t baz')
    def baz(): 
        return Response(text = 'baz!')
    
    result = await manager.search('foo bar test baz')
    
    assert len(result) == 2
    assert result[0].match_result.substring == 'foo bar'
    assert result[1].match_result.substring == 'test baz'

async def test_overlapping_commands_remove(commands_context_flow, autojump_clock):
    manager = CommandsManager()
    
    @manager.new('foo bar')
    def foobar(): 
        return Response(text = 'foo!')
    
    @manager.new('bar baz')
    def barbaz(): 
        return Response(text = 'baz!')
    
    result = await manager.search('foo bar baz')
    assert len(result) == 1
    assert result[0].command == foobar
    
async def test_overlapping_commands_remove_inverse(commands_context_flow, autojump_clock):
    manager = CommandsManager()
     
    @manager.new('bar baz')
    def barbaz(): 
        return Response(text = 'baz!')
    
    @manager.new('foo bar')
    def foobar(): 
        return Response(text = 'foo!')
    
    result = await manager.search('foo bar baz')
    assert len(result) == 1
    assert result[0].command == barbaz
    
@pytest.mark.skip(reason = 'Cache is deprecated and not working properly anymore because of new concurrent algorithm; need new async lru cache implementation')
async def test_objects_parse_caching(commands_context_flow, autojump_clock):
    class Mock(Object):
    
        parsing_counter = 0
        
        @classproperty
        def pattern(cls):
            return Pattern('*')
        
        async def did_parse(self, from_string: str) -> str:
            Mock.parsing_counter += 1
            return from_string
    
    mock_name = f'Mock{random.randint(0, 10**10)}'
    Mock.__name__ = mock_name # prevent name collision on paralell tests
    
    manager = CommandsManager()
    Pattern.add_parameter_type(Mock)
    
    @manager.new(f'hello $mock:{mock_name}')
    def hello(mock: Mock): pass
    
    @manager.new(f'hello $mock:{mock_name} 2')
    def hello2(mock: Mock): pass
    
    @manager.new(f'hello $mock:{mock_name} 22')
    def hello22(mock: Mock): pass
    
    @manager.new(f'test $mock:{mock_name}')
    async def test(mock: Mock): pass
    
    assert Mock.parsing_counter == 0
    await manager.search('hello foobar 22')
    assert Mock.parsing_counter == 1
    await manager.search('hello foobar 22')
    assert Mock.parsing_counter == 2

tests/test_commands_flow/test_multiple_commands.py
MarkParker5/STARK
import re
import pytest
from stark.core import Pattern
from stark.core.types import Object, Word, String
from stark.core.patterns import expressions
from stark.general.classproperty import classproperty


word = fr'[{expressions.alphanumerics}]*'
words = fr'[{expressions.alphanumerics}\s]*'
    
class ExtraParameterInPattern(Object):
    word1: Word
    word2: Word
    
    @classproperty
    def pattern(cls) -> Pattern:
        return Pattern('$word1:Word $word2:Word $word3:Word')

async def test_typed_parameters():
    p = Pattern('lorem $name:Word dolor')
    assert p.parameters == {'name': Word}
    assert p.compiled == fr'lorem (?P<name>{word}) dolor'
    
    m = await p.match('lorem ipsum dolor')
    assert m
    assert m[0].substring == 'lorem ipsum dolor'
    assert m[0].parameters == {'name': Word('ipsum')}
    assert not await p.match('lorem ipsum foo dolor')
    
    p = Pattern('lorem $name:String dolor')
    assert p.parameters == {'name': String}
    m = await p.match('lorem ipsum foo bar dolor')
    assert m
    assert m[0].substring == 'lorem ipsum foo bar dolor'
    assert m[0].parameters == {'name': String('ipsum foo bar')}
    
def test_undefined_typed_parameters():
    pattern = 'lorem $name:Lorem dolor'
    with pytest.raises(NameError, match=re.escape(f'Unknown type: "Lorem" for parameter: "name" in pattern: "{pattern}"')):
        Pattern(pattern)
        
def test_extra_parameter_in_pattern():
    with pytest.raises(AssertionError, match='Can`t add parameter type "ExtraParameterInPattern": pattern parameters do not match properties annotated in class'):
        Pattern.add_parameter_type(ExtraParameterInPattern)
tests/test_patterns/test_pattern_parameters.py
MarkParker5/STARK
from stark.core import Pattern
from stark.core.patterns import expressions


word = fr'[{expressions.alphanumerics}]*'
words = fr'[{expressions.alphanumerics}\s]*'

async def test_leading_star():
    p = Pattern('*text')
    assert p.compiled == fr'{word}text'
    assert await p.match('text')
    assert await p.match('aaatext')
    assert (await p.match('bbb aaaatext cccc'))[0].substring == 'aaaatext'
    assert not await p.match('aaaaext')
    
    p = Pattern('Some *text here')
    assert p.compiled == fr'Some {word}text here'
    assert await p.match('Some text here')
    assert await p.match('Some aaatext here')
    assert (await p.match('bbb Some aaatext here cccc'))[0].substring == 'Some aaatext here'
    assert not await p.match('aaatext here')
    
async def test_trailing_star():
    p = Pattern('text*')
    assert p.compiled == fr'text{word}'
    assert await p.match('text')
    assert await p.match('textaaa')
    assert (await p.match('bbb textaaa cccc'))[0].substring == 'textaaa'
    
    p = Pattern('Some text* here')
    assert p.compiled == fr'Some text{word} here'
    assert await p.match('Some text here')
    assert await p.match('Some textaaa here')
    assert (await p.match('bbb Some textaaa here cccc'))[0].substring == 'Some textaaa here'
    assert not await p.match('Some textaaa ')

async def test_middle_star():
    p = Pattern('te*xt')
    assert p.compiled == fr'te{word}xt'
    assert await p.match('text')
    assert await p.match('teaaaaaxt')
    assert (await p.match('bbb teaaaaaxt cccc'))[0].substring == 'teaaaaaxt'
    
    p = Pattern('Some te*xt here')
    assert p.compiled == fr'Some te{word}xt here'
    assert await p.match('Some text here')
    assert await p.match('Some teaaaaaxt here')
    assert (await p.match('bbb Some teaaeaaaxt here cccc'))[0].substring == 'Some teaaeaaaxt here'
    assert not await p.match('Some teaaaaaxt')
    
async def test_double_star():
    p = Pattern('**')
    assert p.compiled == fr'{words}'
    assert (await p.match('bbb teaaaaaxt cccc'))[0].substring == 'bbb teaaaaaxt cccc'
    
    p = Pattern('Some ** here')
    assert p.compiled == fr'Some {words} here'
    assert await p.match('Some text here')
    assert await p.match('Some lorem ipsum dolor here')
    assert (await p.match('bbb Some lorem ipsum dolor here cccc'))[0].substring == 'Some lorem ipsum dolor here'
    
async def test_one_of():
    p = Pattern('(foo|bar)')
    assert p.compiled == r'(?:foo|bar)'
    assert await p.match('foo')
    assert await p.match('bar')
    assert (await p.match('bbb foo cccc'))[0].substring == 'foo'
    assert (await p.match('bbb bar cccc'))[0].substring == 'bar'
    
    p = Pattern('Some (foo|bar) here')
    assert p.compiled == r'Some (?:foo|bar) here'
    assert await p.match('Some foo here')
    assert await p.match('Some bar here')
    assert (await p.match('bbb Some foo here cccc'))[0].substring == 'Some foo here'
    assert (await p.match('bbb Some bar here cccc'))[0].substring == 'Some bar here'
    assert not await p.match('Some foo')
    
async def test_optional_one_of():
    p = Pattern('(foo|bar)?')
    assert p.compiled == r'(?:foo|bar)?'
    assert await p.match('foo')
    assert await p.match('bar')
    assert not await p.match('')
    assert not await p.match('bbb cccc')
    assert (await p.match('bbb foo cccc'))[0].substring == 'foo'
    assert (await p.match('bbb bar cccc'))[0].substring == 'bar'
    
    p = Pattern('Some (foo|bar)? here')
    assert p.compiled == r'Some (?:foo|bar)? here'
    assert await p.match('Some foo here')
    assert await p.match('Some bar here')
    assert await p.match('Some  here')
    assert (await p.match('bbb Some foo here cccc'))[0].substring == 'Some foo here'
    assert (await p.match('bbb Some bar here cccc'))[0].substring == 'Some bar here'
    assert (await p.match('bbb Some  here cccc'))[0].substring == 'Some  here'
    
    # assert Pattern('[foo|bar]').compiled == Pattern('(foo|bar)?').compiled
    
async def test_one_or_more_of():
    p = Pattern('{foo|bar}')
    assert p.compiled == r'(?:(?:foo|bar)\s?)+'
    assert await p.match('foo')
    assert await p.match('bar')
    assert not await p.match('')
    assert (await p.match('bbb foo cccc'))[0].substring == 'foo'
    assert (await p.match('bbb bar cccc'))[0].substring == 'bar'
    assert (await p.match('bbb foo bar cccc'))[0].substring == 'foo bar'
    assert not await p.match('bbb cccc')
    
    p = Pattern('Some {foo|bar} here')
    assert p.compiled == r'Some (?:(?:foo|bar)\s?)+ here'
    assert await p.match('Some foo here')
    assert await p.match('Some bar here')
    assert not await p.match('Some  here')
    assert (await p.match('bbb Some foo here cccc'))[0].substring == 'Some foo here'
    assert (await p.match('bbb Some bar here cccc'))[0].substring == 'Some bar here'
    assert (await p.match('bbb Some foo bar here cccc'))[0].substring == 'Some foo bar here'
    assert not await p.match('Some foo')

tests/test_patterns/test_patterns.py
MarkParker5/STARK
import pytest
import anyio
from datetime import timedelta
from stark.voice_assistant import Mode


async def test_background_command_with_waiting_mode(voice_assistant, autojump_clock):
    async with voice_assistant() as voice_assistant:
        await voice_assistant.speech_recognizer_did_receive_final_result('background min')
        
        await anyio.sleep(0.2) # allow to capture first command response
        
        # start background task
        assert len(voice_assistant.speech_synthesizer.results) == 1
        assert voice_assistant.speech_synthesizer.results.pop(0).text == 'Starting background task'
        
        # force a timeout
        voice_assistant._last_interaction_time -= timedelta(seconds = voice_assistant.mode.timeout_after_interaction + 1)
        
        # wait for command to finish
        await anyio.sleep(1)
        
        # check finished background command
        assert len(voice_assistant.speech_synthesizer.results) == 1
        assert voice_assistant.speech_synthesizer.results.pop(0).text == 'Finished background task'
        
        # check saved response
        assert len(voice_assistant._responses) == 1
        
        # emulate delay after last response before repeating
        voice_assistant._responses[0].time -= timedelta(seconds = voice_assistant.mode.timeout_before_repeat + 1)
        
        # interact to reset timeout mode and repeat saved response
        await voice_assistant.speech_recognizer_did_receive_final_result('test')
        await anyio.sleep(0.2) # allow to capture command response
        assert len(voice_assistant.speech_synthesizer.results) == 2
        assert [r.text for r in voice_assistant.speech_synthesizer.results] == ['test', 'Finished background task']

async def test_background_command_with_inactive_mode(voice_assistant, autojump_clock):
    async with voice_assistant() as voice_assistant:
        await voice_assistant.speech_recognizer_did_receive_final_result('background min')
        
        await anyio.sleep(0.2) # allow to capture first command response
        
        # start background task
        assert len(voice_assistant.speech_synthesizer.results) == 1
        assert voice_assistant.speech_synthesizer.results.pop(0).text == 'Starting background task'
        
        # set inactive mode
        voice_assistant.mode = Mode.inactive
        
        # wait for command to finish
        await anyio.sleep(1)
        
        # check finished background task
        assert len(voice_assistant.speech_synthesizer.results) == 0
        
        # check saved response
        assert len(voice_assistant._responses) == 1
        
        # interact to reset timeout mode and repeat saved response
        await voice_assistant.speech_recognizer_did_receive_final_result('test')
        await anyio.sleep(0.2) # allow to capture command response
        assert len(voice_assistant.speech_synthesizer.results) == 2
        assert voice_assistant.speech_synthesizer.results.pop(0).text == 'test'   
        assert voice_assistant.speech_synthesizer.results.pop(0).text == 'Finished background task'
    
async def test_background_waiting_needs_input(voice_assistant, autojump_clock):
    async with voice_assistant() as voice_assistant:
        await voice_assistant.speech_recognizer_did_receive_final_result('background needs input')
        
        await anyio.sleep(0.2) # allow to capture first command response
        
        # force a timeout
        voice_assistant._last_interaction_time -= timedelta(seconds = voice_assistant.mode.timeout_after_interaction + 1)
        
        # wait for command to finish
        await anyio.sleep(1)
        
        # voice assistant should save all responses for later
        assert len(voice_assistant._responses) == 8
        assert len(voice_assistant.speech_synthesizer.results) == 8
        voice_assistant.speech_synthesizer.results.clear()
        
        # emulate delay after last response before repeating
        for response in voice_assistant._responses:
            response.time -= timedelta(seconds = voice_assistant.mode.timeout_before_repeat + 1)
        
        # interact to reset timeout mode
        await voice_assistant.speech_recognizer_did_receive_final_result('test')
        
        await anyio.sleep(0.2) # allow to capture command response
        
        # voice assistant should say all responses until needs input
        assert len(voice_assistant.speech_synthesizer.results) == 5
        assert len(voice_assistant._responses) == 4
        
        for response in ['test', 'First response', 'Second response', 'Third response', 'Needs input']:
            assert voice_assistant.speech_synthesizer.results.pop(0).text == response
        
        # interact to emulate user input and continue repeating responses
        await voice_assistant.speech_recognizer_did_receive_final_result('test')
        await anyio.sleep(0.2) # allow to capture command response
        
        # voice assistant should say all left responses
        assert len(voice_assistant.speech_synthesizer.results) == 5
        assert len(voice_assistant._responses) == 0
        for response in ['test', 'Fourth response', 'Fifth response', 'Sixth response', 'Finished long background task']:
            assert voice_assistant.speech_synthesizer.results.pop(0).text == response
    
async def test_background_waiting_with_context(voice_assistant, autojump_clock):
    async with voice_assistant() as voice_assistant:
        await voice_assistant.speech_recognizer_did_receive_final_result('background with context')
        
        # force a timeout
        voice_assistant._last_interaction_time -= timedelta(seconds = voice_assistant.mode.timeout_after_interaction + 1)
        
        # wait for command to finish
        await anyio.sleep(2)
        
        # voice assistant should play all (including adding context) and save all responses for later
        assert len(voice_assistant.speech_synthesizer.results) == 1
        assert len(voice_assistant._responses) == 1 # first response is not saved because it plays immediately
        assert len(voice_assistant.commands_context._context_queue) == 2
        voice_assistant.speech_synthesizer.results.clear()
        
        # emulate delay after last response before repeating
        for response in voice_assistant._responses:
            response.time -= timedelta(seconds = voice_assistant.mode.timeout_before_repeat + 1)
        
        # interact to reset timeout mode, voice assistant should reset context, repeat responses and add response context
        await voice_assistant.speech_recognizer_did_receive_final_result('lorem ipsum dolor')
        await anyio.sleep(0.2) # allow to capture command response
        assert len(voice_assistant.speech_synthesizer.results) == 2
        assert len(voice_assistant.commands_context._context_queue) == 2

async def test_background_waiting_remove_response(voice_assistant, autojump_clock):
    async with voice_assistant() as voice_assistant:
        await voice_assistant.speech_recognizer_did_receive_final_result('background remove response')
        voice_assistant.speech_synthesizer.results.clear()
        
        # force a timeout by settings zero time
        voice_assistant._last_interaction_time -= timedelta(seconds = voice_assistant.mode.timeout_after_interaction + 1)
        
        await anyio.sleep(0.2) # allow to capture command response
        responses_cache = voice_assistant._responses.copy() # cache all responses
        
        # wait for command to finish (and delete first response)
        await anyio.sleep(1)
        
        # check response cached
        assert len(responses_cache) == 1
        assert len(voice_assistant.speech_synthesizer.results) == 1
        assert responses_cache.pop(0).text == 'Deleted response'
        # check voice_assistant doesn't have response
        assert len(voice_assistant._responses) == 0
        voice_assistant.speech_synthesizer.results.clear()
        
        # interact to reset timeout mode, check that removed response is not repeated
        await voice_assistant.speech_recognizer_did_receive_final_result('test')
        await anyio.sleep(1) # allow to capture command response
        assert len(voice_assistant.speech_synthesizer.results) == 1
        assert voice_assistant.speech_synthesizer.results.pop(0).text == 'test'

tests/test_va_modes.py
MarkParker5/STARK

import pytest 
from stark.core import Pattern
from stark.core.types import Object, ParseError
from stark.general.classproperty import classproperty


class Lorem(Object):
    
    @classproperty
    def pattern(cls):
        return Pattern('* ipsum')
    
    async def did_parse(self, from_string: str) -> str:
        if 'lorem' not in from_string:
            raise ParseError('lorem not found')
        self.value = 'lorem'
        return 'lorem'
    
async def test_complex_parsing_failed():
    with pytest.raises(ParseError):
        await Lorem.parse('some lor ipsum')
    
async def test_complex_parsing():
    string = 'some lorem ipsum'
    match = await Lorem.parse(string)
    assert match
    assert match.obj
    assert match.obj.value == 'lorem'
    assert match.substring == 'lorem'
    assert (await Lorem.pattern.match(string))[0].substring == 'lorem ipsum'
    
tests/test_viobjects/test_complex_parsing.py
MarkParker5/STARK
import pytest
from stark.core import Pattern
from stark.core.types import Object, Word
from stark.general.classproperty import classproperty


class FullName(Object):
    first: Word
    second: Word

    @classproperty
    def pattern(cls) -> Pattern:
        return Pattern('$first:Word $second:Word')

class ExtraParameterInAnnotation(Object):
    word1: Word
    word2: Word
    word3: Word
    
    @classproperty
    def pattern(cls) -> Pattern:
        return Pattern('$word1:Word $word2:Word')

async def test_nested_objects():
    Pattern.add_parameter_type(FullName)
    
    p = Pattern('$name:FullName')
    assert p
    assert p.compiled
    
    m = await p.match('John Galt')
    assert m
    assert set(m[0].parameters.keys()) == {'name'}
    assert m[0].parameters['name'].first == Word('John')
    assert m[0].parameters['name'].second == Word('Galt')
    
async def test_extra_parameter_in_annotation():
    Pattern.add_parameter_type(ExtraParameterInAnnotation)
    
    p = Pattern('$name:ExtraParameterInAnnotation')
    assert p
    assert p.compiled
    
    m = await p.match('John Galt')
    assert m
    assert set(m[0].parameters.keys()) == {'name'}
    assert m[0].parameters['name'].word1 == Word('John')
    assert m[0].parameters['name'].word2 == Word('Galt')
    assert not hasattr(m[0].parameters['name'], 'word3')

tests/test_viobjects/test_nested_viobjects.py
MarkParker5/STARK
from stark.core import Pattern
from stark.core.types import String


def test_pattern():
    assert String.pattern == Pattern('**')
    
async def test_parse():
    assert await String.parse('')
    assert (await String.parse('foo bar baz')).obj.value == 'foo bar baz'
    
async def test_match():
    p = Pattern('foo $bar:String baz')
    assert p
    
    m = await p.match('foo qwerty baz')
    assert m
    assert m[0].parameters['bar'] == String('qwerty')
    
    m = await p.match('foo lorem ipsum dolor sit amet baz')
    assert m
    assert m[0].parameters['bar'] == String('lorem ipsum dolor sit amet')
    
async def test_formatted():
    string = (await String.parse('foo bar baz')).obj
    assert str(string) == '<String value: "foo bar baz">'
    assert f'{string}' == 'foo bar baz'

tests/test_viobjects/test_vistring.py
MarkParker5/STARK
from stark.core import Pattern
from stark.core.types import Word


def test_pattern():
    assert Word.pattern == Pattern('*')
    
async def test_parse():
    word = (await Word.parse('foo')).obj
    assert word
    assert word.value == 'foo'
    
async def test_match():
    p = Pattern('foo $bar:Word baz')
    assert p
    
    m = await p.match('foo qwerty baz')
    assert m
    assert m[0].parameters['bar'] == Word('qwerty')
    
    m = await p.match('foo lorem ipsum dolor sit amet baz')
    assert not m
    
async def test_formatted():
    string = (await Word.parse('foo')).obj
    assert str(string) == '<Word value: "foo">'
    assert f'{string}' == 'foo'

tests/test_viobjects/test_viword.py
MarkParker5/STARK